test_cmd_coord_list.py
python
sha256:1c4b3e3a9a1f300774c3ee662b572a698d5fd405bf765a71e6011a2e9c3eaaaa
feat: Muse — version control for the agent era
Human
73 days ago
| 1 | """Comprehensive tests for ``muse coord list``. |
| 2 | |
| 3 | Coverage |
| 4 | -------- |
| 5 | Unit — core helpers |
| 6 | _format_ttl: all time ranges, zero, negative |
| 7 | filter_reservations: run_id, branch, address_glob, include_expired, |
| 8 | compound (AND) filters, empty input |
| 9 | filter_intents: run_id, branch, address_glob, compound, empty |
| 10 | Reservation.ttl_remaining_seconds: positive, negative, near-zero |
| 11 | |
| 12 | Integration — CLI |
| 13 | Empty repo (no .muse/coordination): exits 0, text + JSON |
| 14 | Basic list: shows reservations and intents |
| 15 | --kind reservations: hides intents |
| 16 | --kind intents: hides reservations |
| 17 | --run-id filter: exact match |
| 18 | --branch filter: exact match |
| 19 | --address glob: fnmatch, wildcard, no-match |
| 20 | --all flag: expired reservations appear |
| 21 | --summary: single line, counts correct |
| 22 | --format json / --json: valid JSON, full schema |
| 23 | JSON schema fields: all required keys present and typed |
| 24 | JSON ttl_remaining_seconds: float, positive for active |
| 25 | JSON is_active flag: True for active, False for expired |
| 26 | JSON elapsed_seconds: float >= 0 |
| 27 | JSON filters object: reflects CLI args |
| 28 | Sorting: created_at ascending |
| 29 | Multiple reservations: all shown |
| 30 | Multiple intents: all shown |
| 31 | Expired hidden by default: not shown without --all |
| 32 | Compound filter: run_id + branch + address |
| 33 | |
| 34 | Security |
| 35 | ANSI in run_id: stripped in text output |
| 36 | Control chars in branch: stripped |
| 37 | Control chars in detail: stripped |
| 38 | Control chars in address: stripped in text output |
| 39 | --address glob no FS access: fnmatch is string-only |
| 40 | Null byte in run_id filter: safe (no match, no crash) |
| 41 | |
| 42 | Stress |
| 43 | 500 reservations: listing < 2 s |
| 44 | 1 000 intents: listing < 3 s |
| 45 | 200 filtered to 1: < 1 s |
| 46 | 50 agents, 20 addrs each: JSON schema valid |
| 47 | """ |
| 48 | |
| 49 | from __future__ import annotations |
| 50 | |
| 51 | import datetime |
| 52 | import json |
| 53 | import pathlib |
| 54 | import time |
| 55 | import uuid |
| 56 | |
| 57 | import pytest |
| 58 | |
| 59 | from tests.cli_test_helper import CliRunner |
| 60 | from muse.core.coordination import ( |
| 61 | Reservation, |
| 62 | Intent, |
| 63 | create_intent, |
| 64 | create_reservation, |
| 65 | filter_intents, |
| 66 | filter_reservations, |
| 67 | load_all_reservations, |
| 68 | load_all_intents, |
| 69 | ) |
| 70 | from muse.cli.commands.list_coord import _format_ttl |
| 71 | |
| 72 | cli = None |
| 73 | runner = CliRunner() |
| 74 | |
| 75 | _REQUIRED_JSON_KEYS = { |
| 76 | "active_reservations", |
| 77 | "expired_reservations", |
| 78 | "released_reservations", |
| 79 | "total_reservations_shown", |
| 80 | "total_intents_shown", |
| 81 | "has_conflicts", |
| 82 | "filters", |
| 83 | "reservations", |
| 84 | "intents", |
| 85 | "elapsed_seconds", |
| 86 | } |
| 87 | _REQUIRED_RES_KEYS = { |
| 88 | "reservation_id", |
| 89 | "run_id", |
| 90 | "branch", |
| 91 | "addresses", |
| 92 | "created_at", |
| 93 | "expires_at", |
| 94 | "effective_expires_at", |
| 95 | "ttl_remaining_seconds", |
| 96 | "operation", |
| 97 | "is_active", |
| 98 | "released", |
| 99 | "conflict_count", |
| 100 | } |
| 101 | _REQUIRED_INT_KEYS = { |
| 102 | "intent_id", |
| 103 | "reservation_id", |
| 104 | "run_id", |
| 105 | "branch", |
| 106 | "addresses", |
| 107 | "operation", |
| 108 | "created_at", |
| 109 | "detail", |
| 110 | } |
| 111 | |
| 112 | |
| 113 | # --------------------------------------------------------------------------- |
| 114 | # Helpers |
| 115 | # --------------------------------------------------------------------------- |
| 116 | |
| 117 | def _now() -> datetime.datetime: |
| 118 | return datetime.datetime.now(datetime.timezone.utc) |
| 119 | |
| 120 | |
| 121 | def _future(seconds: int = 3600) -> datetime.datetime: |
| 122 | return _now() + datetime.timedelta(seconds=seconds) |
| 123 | |
| 124 | |
| 125 | def _past(seconds: int = 60) -> datetime.datetime: |
| 126 | return _now() - datetime.timedelta(seconds=seconds) |
| 127 | |
| 128 | |
| 129 | def _make_reservation( |
| 130 | tmp_path: pathlib.Path, |
| 131 | *, |
| 132 | run_id: str = "agent-1", |
| 133 | branch: str = "main", |
| 134 | addresses: list[str] | None = None, |
| 135 | ttl_seconds: int = 3600, |
| 136 | operation: str | None = None, |
| 137 | ) -> Reservation: |
| 138 | return create_reservation( |
| 139 | tmp_path, |
| 140 | run_id=run_id, |
| 141 | branch=branch, |
| 142 | addresses=addresses or ["src/mod.py::foo"], |
| 143 | ttl_seconds=ttl_seconds, |
| 144 | operation=operation, |
| 145 | ) |
| 146 | |
| 147 | |
| 148 | def _make_intent( |
| 149 | tmp_path: pathlib.Path, |
| 150 | *, |
| 151 | run_id: str = "agent-1", |
| 152 | branch: str = "main", |
| 153 | addresses: list[str] | None = None, |
| 154 | operation: str = "modify", |
| 155 | detail: str = "", |
| 156 | ) -> Intent: |
| 157 | return create_intent( |
| 158 | tmp_path, |
| 159 | reservation_id=str(uuid.uuid4()), |
| 160 | run_id=run_id, |
| 161 | branch=branch, |
| 162 | addresses=addresses or ["src/mod.py::foo"], |
| 163 | operation=operation, |
| 164 | detail=detail, |
| 165 | ) |
| 166 | |
| 167 | |
| 168 | @pytest.fixture |
| 169 | def repo(tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch) -> pathlib.Path: |
| 170 | monkeypatch.chdir(tmp_path) |
| 171 | monkeypatch.setenv("MUSE_REPO_ROOT", str(tmp_path)) |
| 172 | r = runner.invoke(cli, ["init", "--domain", "code"]) |
| 173 | assert r.exit_code == 0, r.output |
| 174 | return tmp_path |
| 175 | |
| 176 | |
| 177 | # --------------------------------------------------------------------------- |
| 178 | # Unit — _format_ttl |
| 179 | # --------------------------------------------------------------------------- |
| 180 | |
| 181 | |
| 182 | class TestFormatTtl: |
| 183 | def test_hours_minutes_seconds(self) -> None: |
| 184 | assert _format_ttl(3661) == "1h 1m 1s" |
| 185 | |
| 186 | def test_exactly_one_hour(self) -> None: |
| 187 | assert _format_ttl(3600) == "1h 0m 0s" |
| 188 | |
| 189 | def test_minutes_and_seconds(self) -> None: |
| 190 | assert _format_ttl(90) == "1m 30s" |
| 191 | |
| 192 | def test_seconds_only(self) -> None: |
| 193 | assert _format_ttl(45) == "45s" |
| 194 | |
| 195 | def test_one_second(self) -> None: |
| 196 | assert _format_ttl(1) == "1s" |
| 197 | |
| 198 | def test_zero_is_expired(self) -> None: |
| 199 | assert _format_ttl(0) == "EXPIRED" |
| 200 | |
| 201 | def test_negative_is_expired(self) -> None: |
| 202 | assert _format_ttl(-100) == "EXPIRED" |
| 203 | |
| 204 | def test_float_truncates(self) -> None: |
| 205 | # 90.9 → 90 seconds → 1m 30s |
| 206 | assert _format_ttl(90.9) == "1m 30s" |
| 207 | |
| 208 | def test_large_hours(self) -> None: |
| 209 | result = _format_ttl(86400) # 24 h |
| 210 | assert result.startswith("24h") |
| 211 | |
| 212 | |
| 213 | # --------------------------------------------------------------------------- |
| 214 | # Unit — Reservation.ttl_remaining_seconds |
| 215 | # --------------------------------------------------------------------------- |
| 216 | |
| 217 | |
| 218 | class TestTtlRemainingSeconds: |
| 219 | def test_active_reservation_positive(self, tmp_path: pathlib.Path) -> None: |
| 220 | res = _make_reservation(tmp_path, ttl_seconds=3600) |
| 221 | assert res.ttl_remaining_seconds() > 0 |
| 222 | |
| 223 | def test_expired_reservation_negative(self, tmp_path: pathlib.Path) -> None: |
| 224 | res = _make_reservation(tmp_path, ttl_seconds=1) |
| 225 | # Manually expire it. |
| 226 | res.expires_at = _past(10) |
| 227 | assert res.ttl_remaining_seconds() < 0 |
| 228 | |
| 229 | def test_near_zero_close_to_expiry(self, tmp_path: pathlib.Path) -> None: |
| 230 | res = _make_reservation(tmp_path, ttl_seconds=1) |
| 231 | res.expires_at = _now() + datetime.timedelta(seconds=0.5) |
| 232 | assert 0 < res.ttl_remaining_seconds() < 2 |
| 233 | |
| 234 | |
| 235 | # --------------------------------------------------------------------------- |
| 236 | # Unit — filter_reservations |
| 237 | # --------------------------------------------------------------------------- |
| 238 | |
| 239 | |
| 240 | class TestFilterReservations: |
| 241 | def test_empty_list(self) -> None: |
| 242 | assert filter_reservations([]) == [] |
| 243 | |
| 244 | def test_active_only_by_default(self, tmp_path: pathlib.Path) -> None: |
| 245 | active = _make_reservation(tmp_path, ttl_seconds=3600) |
| 246 | expired = _make_reservation(tmp_path, ttl_seconds=1) |
| 247 | expired.expires_at = _past(10) |
| 248 | result = filter_reservations([active, expired]) |
| 249 | assert len(result) == 1 |
| 250 | assert result[0].reservation_id == active.reservation_id |
| 251 | |
| 252 | def test_include_expired(self, tmp_path: pathlib.Path) -> None: |
| 253 | active = _make_reservation(tmp_path, ttl_seconds=3600) |
| 254 | expired = _make_reservation(tmp_path, ttl_seconds=1) |
| 255 | expired.expires_at = _past(10) |
| 256 | result = filter_reservations([active, expired], include_expired=True) |
| 257 | assert len(result) == 2 |
| 258 | |
| 259 | def test_run_id_exact_match(self, tmp_path: pathlib.Path) -> None: |
| 260 | r1 = _make_reservation(tmp_path, run_id="agent-1") |
| 261 | r2 = _make_reservation(tmp_path, run_id="agent-2") |
| 262 | result = filter_reservations([r1, r2], run_id="agent-1") |
| 263 | assert len(result) == 1 |
| 264 | assert result[0].run_id == "agent-1" |
| 265 | |
| 266 | def test_run_id_no_match(self, tmp_path: pathlib.Path) -> None: |
| 267 | r = _make_reservation(tmp_path, run_id="agent-1") |
| 268 | assert filter_reservations([r], run_id="agent-99") == [] |
| 269 | |
| 270 | def test_branch_exact_match(self, tmp_path: pathlib.Path) -> None: |
| 271 | r1 = _make_reservation(tmp_path, branch="main") |
| 272 | r2 = _make_reservation(tmp_path, branch="feat/x") |
| 273 | result = filter_reservations([r1, r2], branch="main") |
| 274 | assert len(result) == 1 |
| 275 | assert result[0].branch == "main" |
| 276 | |
| 277 | def test_address_glob_wildcard(self, tmp_path: pathlib.Path) -> None: |
| 278 | r1 = _make_reservation(tmp_path, addresses=["billing.py::compute_total"]) |
| 279 | r2 = _make_reservation(tmp_path, addresses=["auth.py::login"]) |
| 280 | result = filter_reservations([r1, r2], address_glob="billing.py::*") |
| 281 | assert len(result) == 1 |
| 282 | assert result[0].addresses == ["billing.py::compute_total"] |
| 283 | |
| 284 | def test_address_glob_no_match(self, tmp_path: pathlib.Path) -> None: |
| 285 | r = _make_reservation(tmp_path, addresses=["billing.py::compute_total"]) |
| 286 | assert filter_reservations([r], address_glob="auth.py::*") == [] |
| 287 | |
| 288 | def test_address_glob_any_address_matches(self, tmp_path: pathlib.Path) -> None: |
| 289 | """A reservation with multiple addresses: match if ANY address matches.""" |
| 290 | r = _make_reservation( |
| 291 | tmp_path, |
| 292 | addresses=["billing.py::compute_total", "auth.py::login"], |
| 293 | ) |
| 294 | result = filter_reservations([r], address_glob="auth.py::*") |
| 295 | assert len(result) == 1 |
| 296 | |
| 297 | def test_compound_run_id_and_branch(self, tmp_path: pathlib.Path) -> None: |
| 298 | r1 = _make_reservation(tmp_path, run_id="a", branch="main") |
| 299 | r2 = _make_reservation(tmp_path, run_id="a", branch="feat") |
| 300 | r3 = _make_reservation(tmp_path, run_id="b", branch="main") |
| 301 | result = filter_reservations([r1, r2, r3], run_id="a", branch="main") |
| 302 | assert len(result) == 1 |
| 303 | assert result[0].reservation_id == r1.reservation_id |
| 304 | |
| 305 | def test_compound_run_id_and_address_glob(self, tmp_path: pathlib.Path) -> None: |
| 306 | r1 = _make_reservation(tmp_path, run_id="a", addresses=["billing.py::foo"]) |
| 307 | r2 = _make_reservation(tmp_path, run_id="a", addresses=["auth.py::bar"]) |
| 308 | r3 = _make_reservation(tmp_path, run_id="b", addresses=["billing.py::foo"]) |
| 309 | result = filter_reservations([r1, r2, r3], run_id="a", address_glob="billing.py::*") |
| 310 | assert len(result) == 1 |
| 311 | assert result[0].reservation_id == r1.reservation_id |
| 312 | |
| 313 | def test_preserves_order(self, tmp_path: pathlib.Path) -> None: |
| 314 | r1 = _make_reservation(tmp_path, run_id="a") |
| 315 | r2 = _make_reservation(tmp_path, run_id="b") |
| 316 | result = filter_reservations([r1, r2]) |
| 317 | assert [r.run_id for r in result] == ["a", "b"] |
| 318 | |
| 319 | |
| 320 | # --------------------------------------------------------------------------- |
| 321 | # Unit — filter_intents |
| 322 | # --------------------------------------------------------------------------- |
| 323 | |
| 324 | |
| 325 | class TestFilterIntents: |
| 326 | def test_empty_list(self) -> None: |
| 327 | assert filter_intents([]) == [] |
| 328 | |
| 329 | def test_all_returned_by_default(self, tmp_path: pathlib.Path) -> None: |
| 330 | i1 = _make_intent(tmp_path, run_id="a") |
| 331 | i2 = _make_intent(tmp_path, run_id="b") |
| 332 | assert len(filter_intents([i1, i2])) == 2 |
| 333 | |
| 334 | def test_run_id_filter(self, tmp_path: pathlib.Path) -> None: |
| 335 | i1 = _make_intent(tmp_path, run_id="a") |
| 336 | i2 = _make_intent(tmp_path, run_id="b") |
| 337 | result = filter_intents([i1, i2], run_id="a") |
| 338 | assert len(result) == 1 |
| 339 | |
| 340 | def test_branch_filter(self, tmp_path: pathlib.Path) -> None: |
| 341 | i1 = _make_intent(tmp_path, branch="main") |
| 342 | i2 = _make_intent(tmp_path, branch="feat") |
| 343 | result = filter_intents([i1, i2], branch="main") |
| 344 | assert len(result) == 1 |
| 345 | |
| 346 | def test_address_glob(self, tmp_path: pathlib.Path) -> None: |
| 347 | i1 = _make_intent(tmp_path, addresses=["billing.py::total"]) |
| 348 | i2 = _make_intent(tmp_path, addresses=["auth.py::login"]) |
| 349 | result = filter_intents([i1, i2], address_glob="billing.py::*") |
| 350 | assert len(result) == 1 |
| 351 | |
| 352 | def test_no_match_returns_empty(self, tmp_path: pathlib.Path) -> None: |
| 353 | i = _make_intent(tmp_path) |
| 354 | assert filter_intents([i], run_id="nonexistent") == [] |
| 355 | |
| 356 | def test_compound_filters(self, tmp_path: pathlib.Path) -> None: |
| 357 | i1 = _make_intent(tmp_path, run_id="a", branch="main", addresses=["billing.py::x"]) |
| 358 | i2 = _make_intent(tmp_path, run_id="a", branch="feat", addresses=["billing.py::x"]) |
| 359 | i3 = _make_intent(tmp_path, run_id="b", branch="main", addresses=["billing.py::x"]) |
| 360 | result = filter_intents([i1, i2, i3], run_id="a", branch="main") |
| 361 | assert len(result) == 1 |
| 362 | assert result[0].intent_id == i1.intent_id |
| 363 | |
| 364 | |
| 365 | # --------------------------------------------------------------------------- |
| 366 | # Integration — empty repo |
| 367 | # --------------------------------------------------------------------------- |
| 368 | |
| 369 | |
| 370 | class TestCoordListEmpty: |
| 371 | def test_empty_exits_zero_text(self, repo: pathlib.Path) -> None: |
| 372 | result = runner.invoke(cli, ["coord", "list"]) |
| 373 | assert result.exit_code == 0 |
| 374 | |
| 375 | def test_empty_text_message(self, repo: pathlib.Path) -> None: |
| 376 | result = runner.invoke(cli, ["coord", "list"]) |
| 377 | assert "no coordination records" in result.output.lower() |
| 378 | |
| 379 | def test_empty_json_schema(self, repo: pathlib.Path) -> None: |
| 380 | result = runner.invoke(cli, ["coord", "list", "--json"]) |
| 381 | assert result.exit_code == 0 |
| 382 | data = json.loads(result.output) |
| 383 | assert _REQUIRED_JSON_KEYS <= set(data) |
| 384 | assert data["active_reservations"] == 0 |
| 385 | assert data["total_reservations_shown"] == 0 |
| 386 | assert data["total_intents_shown"] == 0 |
| 387 | assert data["reservations"] == [] |
| 388 | assert data["intents"] == [] |
| 389 | |
| 390 | def test_empty_summary(self, repo: pathlib.Path) -> None: |
| 391 | result = runner.invoke(cli, ["coord", "list", "--summary"]) |
| 392 | assert result.exit_code == 0 |
| 393 | |
| 394 | def test_empty_elapsed_seconds_present(self, repo: pathlib.Path) -> None: |
| 395 | result = runner.invoke(cli, ["coord", "list", "--json"]) |
| 396 | data = json.loads(result.output) |
| 397 | assert isinstance(data["elapsed_seconds"], float) |
| 398 | assert data["elapsed_seconds"] >= 0 |
| 399 | |
| 400 | |
| 401 | # --------------------------------------------------------------------------- |
| 402 | # Integration — basic listing |
| 403 | # --------------------------------------------------------------------------- |
| 404 | |
| 405 | |
| 406 | class TestCoordListBasic: |
| 407 | @pytest.fixture |
| 408 | def populated(self, repo: pathlib.Path) -> pathlib.Path: |
| 409 | _make_reservation( |
| 410 | repo, |
| 411 | run_id="agent-42", |
| 412 | branch="feat/billing", |
| 413 | addresses=["billing.py::compute_total", "billing.py::apply_discount"], |
| 414 | operation="modify", |
| 415 | ) |
| 416 | _make_reservation( |
| 417 | repo, |
| 418 | run_id="agent-41", |
| 419 | branch="main", |
| 420 | addresses=["auth.py::validate_token"], |
| 421 | operation="rename", |
| 422 | ) |
| 423 | _make_intent( |
| 424 | repo, |
| 425 | run_id="agent-42", |
| 426 | branch="feat/billing", |
| 427 | addresses=["billing.py::compute_total"], |
| 428 | operation="rename", |
| 429 | detail="rename to compute_invoice_total", |
| 430 | ) |
| 431 | return repo |
| 432 | |
| 433 | def test_shows_reservations(self, populated: pathlib.Path) -> None: |
| 434 | result = runner.invoke(cli, ["coord", "list"]) |
| 435 | assert result.exit_code == 0 |
| 436 | assert "agent-42" in result.output |
| 437 | assert "agent-41" in result.output |
| 438 | |
| 439 | def test_shows_addresses(self, populated: pathlib.Path) -> None: |
| 440 | result = runner.invoke(cli, ["coord", "list"]) |
| 441 | assert "billing.py::compute_total" in result.output |
| 442 | assert "auth.py::validate_token" in result.output |
| 443 | |
| 444 | def test_shows_intents(self, populated: pathlib.Path) -> None: |
| 445 | result = runner.invoke(cli, ["coord", "list"]) |
| 446 | assert "rename" in result.output |
| 447 | assert "rename to compute_invoice_total" in result.output |
| 448 | |
| 449 | def test_json_res_count(self, populated: pathlib.Path) -> None: |
| 450 | result = runner.invoke(cli, ["coord", "list", "--json"]) |
| 451 | data = json.loads(result.output) |
| 452 | assert data["total_reservations_shown"] == 2 |
| 453 | assert data["total_intents_shown"] == 1 |
| 454 | |
| 455 | def test_json_res_schema_fields(self, populated: pathlib.Path) -> None: |
| 456 | result = runner.invoke(cli, ["coord", "list", "--json"]) |
| 457 | data = json.loads(result.output) |
| 458 | for entry in data["reservations"]: |
| 459 | assert _REQUIRED_RES_KEYS <= set(entry), f"missing keys in {entry}" |
| 460 | |
| 461 | def test_json_intent_schema_fields(self, populated: pathlib.Path) -> None: |
| 462 | result = runner.invoke(cli, ["coord", "list", "--json"]) |
| 463 | data = json.loads(result.output) |
| 464 | for entry in data["intents"]: |
| 465 | assert _REQUIRED_INT_KEYS <= set(entry), f"missing keys in {entry}" |
| 466 | |
| 467 | def test_json_is_active_true_for_active(self, populated: pathlib.Path) -> None: |
| 468 | result = runner.invoke(cli, ["coord", "list", "--json"]) |
| 469 | data = json.loads(result.output) |
| 470 | for r in data["reservations"]: |
| 471 | assert r["is_active"] is True |
| 472 | |
| 473 | def test_json_ttl_positive_for_active(self, populated: pathlib.Path) -> None: |
| 474 | result = runner.invoke(cli, ["coord", "list", "--json"]) |
| 475 | data = json.loads(result.output) |
| 476 | for r in data["reservations"]: |
| 477 | assert r["ttl_remaining_seconds"] > 0 |
| 478 | |
| 479 | def test_json_elapsed_seconds(self, populated: pathlib.Path) -> None: |
| 480 | result = runner.invoke(cli, ["coord", "list", "--json"]) |
| 481 | data = json.loads(result.output) |
| 482 | assert isinstance(data["elapsed_seconds"], float) |
| 483 | assert data["elapsed_seconds"] >= 0 |
| 484 | |
| 485 | |
| 486 | # --------------------------------------------------------------------------- |
| 487 | # Integration — --kind flag |
| 488 | # --------------------------------------------------------------------------- |
| 489 | |
| 490 | |
| 491 | class TestCoordListKind: |
| 492 | @pytest.fixture |
| 493 | def both(self, repo: pathlib.Path) -> pathlib.Path: |
| 494 | _make_reservation(repo, run_id="agent-1") |
| 495 | _make_intent(repo, run_id="agent-1") |
| 496 | return repo |
| 497 | |
| 498 | def test_kind_reservations_hides_intents(self, both: pathlib.Path) -> None: |
| 499 | result = runner.invoke(cli, ["coord", "list", "--kind", "reservations"]) |
| 500 | assert result.exit_code == 0 |
| 501 | data_json = runner.invoke(cli, ["coord", "list", "--kind", "reservations", "--json"]) |
| 502 | data = json.loads(data_json.output) |
| 503 | assert data["total_intents_shown"] == 0 |
| 504 | |
| 505 | def test_kind_intents_hides_reservations(self, both: pathlib.Path) -> None: |
| 506 | data_json = runner.invoke(cli, ["coord", "list", "--kind", "intents", "--json"]) |
| 507 | data = json.loads(data_json.output) |
| 508 | assert data["total_reservations_shown"] == 0 |
| 509 | assert data["total_intents_shown"] == 1 |
| 510 | |
| 511 | def test_kind_all_shows_both(self, both: pathlib.Path) -> None: |
| 512 | data_json = runner.invoke(cli, ["coord", "list", "--kind", "all", "--json"]) |
| 513 | data = json.loads(data_json.output) |
| 514 | assert data["total_reservations_shown"] == 1 |
| 515 | assert data["total_intents_shown"] == 1 |
| 516 | |
| 517 | |
| 518 | # --------------------------------------------------------------------------- |
| 519 | # Integration — filters |
| 520 | # --------------------------------------------------------------------------- |
| 521 | |
| 522 | |
| 523 | class TestCoordListFilters: |
| 524 | @pytest.fixture |
| 525 | def multi(self, repo: pathlib.Path) -> pathlib.Path: |
| 526 | _make_reservation( |
| 527 | repo, run_id="agent-42", branch="feat/billing", |
| 528 | addresses=["billing.py::compute_total"], |
| 529 | ) |
| 530 | _make_reservation( |
| 531 | repo, run_id="agent-41", branch="main", |
| 532 | addresses=["auth.py::validate_token"], |
| 533 | ) |
| 534 | _make_intent(repo, run_id="agent-42", branch="feat/billing", |
| 535 | addresses=["billing.py::compute_total"]) |
| 536 | _make_intent(repo, run_id="agent-41", branch="main", |
| 537 | addresses=["auth.py::validate_token"]) |
| 538 | return repo |
| 539 | |
| 540 | def test_run_id_filter(self, multi: pathlib.Path) -> None: |
| 541 | result = runner.invoke(cli, ["coord", "list", "--run-id", "agent-42", "--json"]) |
| 542 | data = json.loads(result.output) |
| 543 | assert data["total_reservations_shown"] == 1 |
| 544 | assert data["reservations"][0]["run_id"] == "agent-42" |
| 545 | |
| 546 | def test_run_id_no_match(self, multi: pathlib.Path) -> None: |
| 547 | result = runner.invoke(cli, ["coord", "list", "--run-id", "nobody", "--json"]) |
| 548 | data = json.loads(result.output) |
| 549 | assert data["total_reservations_shown"] == 0 |
| 550 | assert data["total_intents_shown"] == 0 |
| 551 | |
| 552 | def test_branch_filter(self, multi: pathlib.Path) -> None: |
| 553 | result = runner.invoke(cli, ["coord", "list", "--branch", "main", "--json"]) |
| 554 | data = json.loads(result.output) |
| 555 | assert data["total_reservations_shown"] == 1 |
| 556 | assert data["reservations"][0]["branch"] == "main" |
| 557 | |
| 558 | def test_address_glob_wildcard(self, multi: pathlib.Path) -> None: |
| 559 | result = runner.invoke( |
| 560 | cli, ["coord", "list", "--address", "billing.py::*", "--json"] |
| 561 | ) |
| 562 | data = json.loads(result.output) |
| 563 | assert data["total_reservations_shown"] == 1 |
| 564 | assert data["total_intents_shown"] == 1 |
| 565 | |
| 566 | def test_address_glob_no_match(self, multi: pathlib.Path) -> None: |
| 567 | result = runner.invoke( |
| 568 | cli, ["coord", "list", "--address", "nonexistent.py::*", "--json"] |
| 569 | ) |
| 570 | data = json.loads(result.output) |
| 571 | assert data["total_reservations_shown"] == 0 |
| 572 | |
| 573 | def test_compound_run_id_and_branch(self, multi: pathlib.Path) -> None: |
| 574 | result = runner.invoke( |
| 575 | cli, ["coord", "list", "--run-id", "agent-42", "--branch", "feat/billing", "--json"] |
| 576 | ) |
| 577 | data = json.loads(result.output) |
| 578 | assert data["total_reservations_shown"] == 1 |
| 579 | assert data["total_intents_shown"] == 1 |
| 580 | |
| 581 | def test_json_filters_object_reflects_args(self, multi: pathlib.Path) -> None: |
| 582 | result = runner.invoke( |
| 583 | cli, ["coord", "list", "--run-id", "agent-42", "--branch", "feat/billing", |
| 584 | "--address", "billing.py::*", "--json"] |
| 585 | ) |
| 586 | data = json.loads(result.output) |
| 587 | f = data["filters"] |
| 588 | assert f["run_id"] == "agent-42" |
| 589 | assert f["branch"] == "feat/billing" |
| 590 | assert f["address_glob"] == "billing.py::*" |
| 591 | assert f["include_expired"] is False |
| 592 | assert f["kind"] == "all" |
| 593 | |
| 594 | |
| 595 | # --------------------------------------------------------------------------- |
| 596 | # Integration — expired reservations |
| 597 | # --------------------------------------------------------------------------- |
| 598 | |
| 599 | |
| 600 | class TestCoordListExpired: |
| 601 | @pytest.fixture |
| 602 | def with_expired(self, repo: pathlib.Path) -> pathlib.Path: |
| 603 | # Active reservation. |
| 604 | _make_reservation(repo, run_id="active-agent", ttl_seconds=3600) |
| 605 | # Expired: create then manually expire via file. |
| 606 | coord_dir = repo / ".muse" / "coordination" / "reservations" |
| 607 | coord_dir.mkdir(parents=True, exist_ok=True) |
| 608 | import json as _json |
| 609 | from muse._version import __version__ |
| 610 | now = datetime.datetime.now(datetime.timezone.utc) |
| 611 | expired_id = str(uuid.uuid4()) |
| 612 | record = { |
| 613 | "schema_version": __version__, |
| 614 | "reservation_id": expired_id, |
| 615 | "run_id": "expired-agent", |
| 616 | "branch": "main", |
| 617 | "addresses": ["old.py::stale_func"], |
| 618 | "created_at": (now - datetime.timedelta(hours=2)).isoformat(), |
| 619 | "expires_at": (now - datetime.timedelta(hours=1)).isoformat(), |
| 620 | "operation": None, |
| 621 | } |
| 622 | (coord_dir / f"{expired_id}.json").write_text(_json.dumps(record)) |
| 623 | return repo |
| 624 | |
| 625 | def test_expired_hidden_by_default(self, with_expired: pathlib.Path) -> None: |
| 626 | result = runner.invoke(cli, ["coord", "list", "--json"]) |
| 627 | data = json.loads(result.output) |
| 628 | run_ids = [r["run_id"] for r in data["reservations"]] |
| 629 | assert "expired-agent" not in run_ids |
| 630 | assert "active-agent" in run_ids |
| 631 | |
| 632 | def test_expired_shown_with_all(self, with_expired: pathlib.Path) -> None: |
| 633 | result = runner.invoke(cli, ["coord", "list", "--all", "--json"]) |
| 634 | data = json.loads(result.output) |
| 635 | run_ids = [r["run_id"] for r in data["reservations"]] |
| 636 | assert "expired-agent" in run_ids |
| 637 | |
| 638 | def test_expired_is_active_false(self, with_expired: pathlib.Path) -> None: |
| 639 | result = runner.invoke(cli, ["coord", "list", "--all", "--json"]) |
| 640 | data = json.loads(result.output) |
| 641 | for r in data["reservations"]: |
| 642 | if r["run_id"] == "expired-agent": |
| 643 | assert r["is_active"] is False |
| 644 | assert r["ttl_remaining_seconds"] < 0 |
| 645 | |
| 646 | def test_expired_count_reported(self, with_expired: pathlib.Path) -> None: |
| 647 | result = runner.invoke(cli, ["coord", "list", "--json"]) |
| 648 | data = json.loads(result.output) |
| 649 | assert data["expired_reservations"] == 1 |
| 650 | assert data["active_reservations"] == 1 |
| 651 | |
| 652 | |
| 653 | # --------------------------------------------------------------------------- |
| 654 | # Integration — --summary |
| 655 | # --------------------------------------------------------------------------- |
| 656 | |
| 657 | |
| 658 | class TestCoordListSummary: |
| 659 | def test_summary_empty(self, repo: pathlib.Path) -> None: |
| 660 | result = runner.invoke(cli, ["coord", "list", "--summary"]) |
| 661 | assert result.exit_code == 0 |
| 662 | |
| 663 | def test_summary_shows_count(self, repo: pathlib.Path) -> None: |
| 664 | _make_reservation(repo, run_id="a") |
| 665 | _make_reservation(repo, run_id="b") |
| 666 | _make_intent(repo, run_id="a") |
| 667 | result = runner.invoke(cli, ["coord", "list", "--summary"]) |
| 668 | assert result.exit_code == 0 |
| 669 | assert "2" in result.output # 2 reservations |
| 670 | assert "1" in result.output # 1 intent |
| 671 | |
| 672 | def test_summary_no_detail(self, repo: pathlib.Path) -> None: |
| 673 | _make_reservation(repo) |
| 674 | result = runner.invoke(cli, ["coord", "list", "--summary"]) |
| 675 | # Summary must be a single line — no address lines. |
| 676 | lines = [l for l in result.output.splitlines() if l.strip()] |
| 677 | assert len(lines) == 1 |
| 678 | |
| 679 | def test_summary_exits_zero(self, repo: pathlib.Path) -> None: |
| 680 | result = runner.invoke(cli, ["coord", "list", "--summary"]) |
| 681 | assert result.exit_code == 0 |
| 682 | |
| 683 | |
| 684 | # --------------------------------------------------------------------------- |
| 685 | # Integration — --format / --json flag |
| 686 | # --------------------------------------------------------------------------- |
| 687 | |
| 688 | |
| 689 | class TestCoordListFormat: |
| 690 | def test_format_json_flag(self, repo: pathlib.Path) -> None: |
| 691 | result = runner.invoke(cli, ["coord", "list", "--format", "json"]) |
| 692 | assert result.exit_code == 0 |
| 693 | data = json.loads(result.output) |
| 694 | assert "reservations" in data |
| 695 | |
| 696 | def test_json_shorthand(self, repo: pathlib.Path) -> None: |
| 697 | result = runner.invoke(cli, ["coord", "list", "--json"]) |
| 698 | assert result.exit_code == 0 |
| 699 | data = json.loads(result.output) |
| 700 | assert "reservations" in data |
| 701 | |
| 702 | def test_text_is_default(self, repo: pathlib.Path) -> None: |
| 703 | result = runner.invoke(cli, ["coord", "list"]) |
| 704 | # Text output contains "Coordination state", not a JSON brace. |
| 705 | assert result.output.strip()[0] != "{" |
| 706 | |
| 707 | |
| 708 | # --------------------------------------------------------------------------- |
| 709 | # Integration — sorting |
| 710 | # --------------------------------------------------------------------------- |
| 711 | |
| 712 | |
| 713 | class TestCoordListSorting: |
| 714 | def test_reservations_sorted_by_created_at(self, repo: pathlib.Path) -> None: |
| 715 | """Older reservations appear first in JSON output.""" |
| 716 | r1 = _make_reservation(repo, run_id="first") |
| 717 | # Ensure a distinct created_at by tweaking the file. |
| 718 | import json as _json, time as _time |
| 719 | _time.sleep(0.01) |
| 720 | r2 = _make_reservation(repo, run_id="second") |
| 721 | result = runner.invoke(cli, ["coord", "list", "--json"]) |
| 722 | data = json.loads(result.output) |
| 723 | run_ids = [r["run_id"] for r in data["reservations"]] |
| 724 | assert run_ids.index("first") < run_ids.index("second") |
| 725 | |
| 726 | |
| 727 | # --------------------------------------------------------------------------- |
| 728 | # Security |
| 729 | # --------------------------------------------------------------------------- |
| 730 | |
| 731 | |
| 732 | class TestCoordListSecurity: |
| 733 | def test_ansi_in_run_id_stripped_from_text(self, repo: pathlib.Path) -> None: |
| 734 | """ANSI escape sequences in run_id must not reach the terminal raw.""" |
| 735 | _make_reservation(repo, run_id="\x1b[31mred\x1b[0m") |
| 736 | result = runner.invoke(cli, ["coord", "list"]) |
| 737 | assert result.exit_code == 0 |
| 738 | assert "\x1b" not in result.output |
| 739 | |
| 740 | def test_control_chars_in_branch_stripped(self, repo: pathlib.Path) -> None: |
| 741 | _make_reservation(repo, branch="main\x01;rm -rf /") |
| 742 | result = runner.invoke(cli, ["coord", "list"]) |
| 743 | assert result.exit_code == 0 |
| 744 | assert "\x01" not in result.output |
| 745 | |
| 746 | def test_control_chars_in_address_stripped(self, repo: pathlib.Path) -> None: |
| 747 | _make_reservation(repo, addresses=["billing.py::\x1b[Afoo"]) |
| 748 | result = runner.invoke(cli, ["coord", "list"]) |
| 749 | assert result.exit_code == 0 |
| 750 | assert "\x1b" not in result.output |
| 751 | |
| 752 | def test_control_chars_in_detail_stripped(self, repo: pathlib.Path) -> None: |
| 753 | _make_intent(repo, detail="legit detail\x1b[31m injected color\x1b[0m") |
| 754 | result = runner.invoke(cli, ["coord", "list"]) |
| 755 | assert result.exit_code == 0 |
| 756 | assert "\x1b" not in result.output |
| 757 | |
| 758 | def test_null_byte_in_run_id_filter_safe(self, repo: pathlib.Path) -> None: |
| 759 | """A null byte in --run-id must not crash — it simply matches nothing.""" |
| 760 | _make_reservation(repo, run_id="agent-1") |
| 761 | result = runner.invoke(cli, ["coord", "list", "--run-id", "agent\x00-1"]) |
| 762 | assert result.exit_code == 0 |
| 763 | |
| 764 | def test_address_glob_no_filesystem_access( |
| 765 | self, repo: pathlib.Path, tmp_path: pathlib.Path |
| 766 | ) -> None: |
| 767 | """Glob pattern must not touch the filesystem — only string matching.""" |
| 768 | _make_reservation(repo, addresses=["billing.py::total"]) |
| 769 | # A glob that could be interpreted as a filesystem glob. |
| 770 | result = runner.invoke(cli, ["coord", "list", "--address", "**/billing*"]) |
| 771 | # Should complete without error regardless of match. |
| 772 | assert result.exit_code == 0 |
| 773 | |
| 774 | def test_requires_repo( |
| 775 | self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch |
| 776 | ) -> None: |
| 777 | monkeypatch.chdir(tmp_path) |
| 778 | monkeypatch.delenv("MUSE_REPO_ROOT", raising=False) |
| 779 | result = runner.invoke(cli, ["coord", "list"]) |
| 780 | assert result.exit_code != 0 |
| 781 | |
| 782 | |
| 783 | # --------------------------------------------------------------------------- |
| 784 | # Stress |
| 785 | # --------------------------------------------------------------------------- |
| 786 | |
| 787 | |
| 788 | class TestCoordListStress: |
| 789 | @pytest.fixture |
| 790 | def large_swarm( |
| 791 | self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch |
| 792 | ) -> pathlib.Path: |
| 793 | monkeypatch.chdir(tmp_path) |
| 794 | monkeypatch.setenv("MUSE_REPO_ROOT", str(tmp_path)) |
| 795 | runner.invoke(cli, ["init", "--domain", "code"]) |
| 796 | return tmp_path |
| 797 | |
| 798 | def test_500_reservations_under_2s(self, large_swarm: pathlib.Path) -> None: |
| 799 | for i in range(500): |
| 800 | _make_reservation( |
| 801 | large_swarm, |
| 802 | run_id=f"agent-{i}", |
| 803 | branch=f"feat/task-{i % 20}", |
| 804 | addresses=[f"src/module_{i % 50}.py::func_{i}"], |
| 805 | ) |
| 806 | start = time.monotonic() |
| 807 | result = runner.invoke(cli, ["coord", "list", "--json"]) |
| 808 | elapsed = time.monotonic() - start |
| 809 | assert result.exit_code == 0 |
| 810 | assert elapsed < 2.0, f"500 reservations took {elapsed:.2f}s" |
| 811 | data = json.loads(result.output) |
| 812 | assert data["total_reservations_shown"] == 500 |
| 813 | |
| 814 | def test_1000_intents_under_3s(self, large_swarm: pathlib.Path) -> None: |
| 815 | for i in range(1000): |
| 816 | _make_intent( |
| 817 | large_swarm, |
| 818 | run_id=f"agent-{i % 50}", |
| 819 | branch=f"feat/task-{i % 10}", |
| 820 | addresses=[f"src/mod_{i % 30}.py::sym_{i}"], |
| 821 | operation="modify", |
| 822 | ) |
| 823 | start = time.monotonic() |
| 824 | result = runner.invoke(cli, ["coord", "list", "--kind", "intents", "--json"]) |
| 825 | elapsed = time.monotonic() - start |
| 826 | assert result.exit_code == 0 |
| 827 | assert elapsed < 3.0, f"1000 intents took {elapsed:.2f}s" |
| 828 | data = json.loads(result.output) |
| 829 | assert data["total_intents_shown"] == 1000 |
| 830 | |
| 831 | def test_200_filtered_to_1_under_1s(self, large_swarm: pathlib.Path) -> None: |
| 832 | for i in range(199): |
| 833 | _make_reservation( |
| 834 | large_swarm, |
| 835 | run_id=f"agent-{i}", |
| 836 | addresses=[f"src/mod_{i}.py::sym"], |
| 837 | ) |
| 838 | _make_reservation( |
| 839 | large_swarm, |
| 840 | run_id="needle", |
| 841 | addresses=["billing.py::compute_total"], |
| 842 | ) |
| 843 | start = time.monotonic() |
| 844 | result = runner.invoke( |
| 845 | cli, ["coord", "list", "--run-id", "needle", "--json"] |
| 846 | ) |
| 847 | elapsed = time.monotonic() - start |
| 848 | assert result.exit_code == 0 |
| 849 | assert elapsed < 1.0, f"filter of 200 took {elapsed:.2f}s" |
| 850 | data = json.loads(result.output) |
| 851 | assert data["total_reservations_shown"] == 1 |
| 852 | assert data["reservations"][0]["run_id"] == "needle" |
| 853 | |
| 854 | def test_50_agents_20_addresses_json_valid(self, large_swarm: pathlib.Path) -> None: |
| 855 | """Large JSON payload: every entry has all required fields.""" |
| 856 | for agent in range(50): |
| 857 | addresses = [f"src/mod_{j}.py::func_{agent}" for j in range(20)] |
| 858 | _make_reservation(large_swarm, run_id=f"agent-{agent}", addresses=addresses) |
| 859 | result = runner.invoke(cli, ["coord", "list", "--json"]) |
| 860 | assert result.exit_code == 0 |
| 861 | data = json.loads(result.output) |
| 862 | assert data["total_reservations_shown"] == 50 |
| 863 | for entry in data["reservations"]: |
| 864 | assert _REQUIRED_RES_KEYS <= set(entry) |
| 865 | assert len(entry["addresses"]) == 20 |
| 866 | |
| 867 | |
| 868 | # --------------------------------------------------------------------------- |
| 869 | # Integration — --op filter |
| 870 | # --------------------------------------------------------------------------- |
| 871 | |
| 872 | |
| 873 | class TestCoordListOp: |
| 874 | @pytest.fixture |
| 875 | def with_ops(self, repo: pathlib.Path) -> pathlib.Path: |
| 876 | _make_reservation(repo, run_id="agent-rename", operation="rename", |
| 877 | addresses=["billing.py::compute_total"]) |
| 878 | _make_reservation(repo, run_id="agent-modify", operation="modify", |
| 879 | addresses=["auth.py::validate"]) |
| 880 | _make_reservation(repo, run_id="agent-none", operation=None, |
| 881 | addresses=["core.py::hash"]) |
| 882 | _make_intent(repo, run_id="agent-delete", operation="delete", |
| 883 | addresses=["old.py::stale"]) |
| 884 | _make_intent(repo, run_id="agent-extract", operation="extract", |
| 885 | addresses=["utils.py::helper"]) |
| 886 | return repo |
| 887 | |
| 888 | def test_op_filter_reservations(self, with_ops: pathlib.Path) -> None: |
| 889 | result = runner.invoke(cli, ["coord", "list", "--op", "rename", "--json"]) |
| 890 | data = json.loads(result.output) |
| 891 | assert data["total_reservations_shown"] == 1 |
| 892 | assert data["reservations"][0]["run_id"] == "agent-rename" |
| 893 | |
| 894 | def test_op_filter_intents(self, with_ops: pathlib.Path) -> None: |
| 895 | result = runner.invoke(cli, ["coord", "list", "--op", "delete", "--json"]) |
| 896 | data = json.loads(result.output) |
| 897 | assert data["total_intents_shown"] == 1 |
| 898 | assert data["intents"][0]["run_id"] == "agent-delete" |
| 899 | |
| 900 | def test_op_filter_no_match(self, with_ops: pathlib.Path) -> None: |
| 901 | result = runner.invoke(cli, ["coord", "list", "--op", "inline", "--json"]) |
| 902 | data = json.loads(result.output) |
| 903 | assert data["total_reservations_shown"] == 0 |
| 904 | assert data["total_intents_shown"] == 0 |
| 905 | |
| 906 | def test_op_filter_both_kinds(self, with_ops: pathlib.Path) -> None: |
| 907 | """--op filters both reservations and intents simultaneously.""" |
| 908 | result = runner.invoke(cli, ["coord", "list", "--op", "extract", "--json"]) |
| 909 | data = json.loads(result.output) |
| 910 | assert data["total_reservations_shown"] == 0 |
| 911 | assert data["total_intents_shown"] == 1 |
| 912 | |
| 913 | def test_op_reflected_in_filters_object(self, with_ops: pathlib.Path) -> None: |
| 914 | result = runner.invoke(cli, ["coord", "list", "--op", "modify", "--json"]) |
| 915 | data = json.loads(result.output) |
| 916 | assert data["filters"]["operation"] == "modify" |
| 917 | |
| 918 | def test_op_null_in_filters_when_unset(self, with_ops: pathlib.Path) -> None: |
| 919 | result = runner.invoke(cli, ["coord", "list", "--json"]) |
| 920 | data = json.loads(result.output) |
| 921 | assert data["filters"]["operation"] is None |
| 922 | |
| 923 | def test_op_compound_with_run_id(self, with_ops: pathlib.Path) -> None: |
| 924 | result = runner.invoke( |
| 925 | cli, ["coord", "list", "--op", "rename", "--run-id", "agent-rename", "--json"] |
| 926 | ) |
| 927 | data = json.loads(result.output) |
| 928 | assert data["total_reservations_shown"] == 1 |
| 929 | |
| 930 | def test_op_in_text_output(self, with_ops: pathlib.Path) -> None: |
| 931 | result = runner.invoke(cli, ["coord", "list", "--op", "rename"]) |
| 932 | assert result.exit_code == 0 |
| 933 | assert "agent-rename" in result.output |
| 934 | |
| 935 | |
| 936 | # --------------------------------------------------------------------------- |
| 937 | # Integration — --limit flag |
| 938 | # --------------------------------------------------------------------------- |
| 939 | |
| 940 | |
| 941 | class TestCoordListLimit: |
| 942 | @pytest.fixture |
| 943 | def many(self, repo: pathlib.Path) -> pathlib.Path: |
| 944 | for i in range(10): |
| 945 | _make_reservation(repo, run_id=f"agent-{i}") |
| 946 | _make_intent(repo, run_id=f"agent-{i}") |
| 947 | return repo |
| 948 | |
| 949 | def test_limit_caps_reservations(self, many: pathlib.Path) -> None: |
| 950 | result = runner.invoke(cli, ["coord", "list", "--limit", "3", "--json"]) |
| 951 | data = json.loads(result.output) |
| 952 | assert data["total_reservations_shown"] == 3 |
| 953 | |
| 954 | def test_limit_caps_intents(self, many: pathlib.Path) -> None: |
| 955 | result = runner.invoke(cli, ["coord", "list", "--limit", "3", "--json"]) |
| 956 | data = json.loads(result.output) |
| 957 | assert data["total_intents_shown"] == 3 |
| 958 | |
| 959 | def test_limit_larger_than_count_returns_all(self, many: pathlib.Path) -> None: |
| 960 | result = runner.invoke(cli, ["coord", "list", "--limit", "100", "--json"]) |
| 961 | data = json.loads(result.output) |
| 962 | assert data["total_reservations_shown"] == 10 |
| 963 | |
| 964 | def test_limit_one(self, many: pathlib.Path) -> None: |
| 965 | result = runner.invoke(cli, ["coord", "list", "--limit", "1", "--json"]) |
| 966 | data = json.loads(result.output) |
| 967 | assert data["total_reservations_shown"] == 1 |
| 968 | assert data["total_intents_shown"] == 1 |
| 969 | |
| 970 | def test_limit_reflected_in_filters(self, many: pathlib.Path) -> None: |
| 971 | result = runner.invoke(cli, ["coord", "list", "--limit", "5", "--json"]) |
| 972 | data = json.loads(result.output) |
| 973 | assert data["filters"]["limit"] == 5 |
| 974 | |
| 975 | def test_no_limit_null_in_filters(self, many: pathlib.Path) -> None: |
| 976 | result = runner.invoke(cli, ["coord", "list", "--json"]) |
| 977 | data = json.loads(result.output) |
| 978 | assert data["filters"]["limit"] is None |
| 979 | |
| 980 | def test_limit_returns_oldest_first(self, many: pathlib.Path) -> None: |
| 981 | """--limit preserves sorted order (oldest first).""" |
| 982 | result = runner.invoke(cli, ["coord", "list", "--limit", "2", "--json"]) |
| 983 | data = json.loads(result.output) |
| 984 | created_ats = [r["created_at"] for r in data["reservations"]] |
| 985 | assert created_ats == sorted(created_ats) |
| 986 | |
| 987 | |
| 988 | # --------------------------------------------------------------------------- |
| 989 | # Integration — conflict detection |
| 990 | # --------------------------------------------------------------------------- |
| 991 | |
| 992 | |
| 993 | class TestCoordListConflicts: |
| 994 | @pytest.fixture |
| 995 | def conflicting(self, repo: pathlib.Path) -> pathlib.Path: |
| 996 | """Two agents reserving the same address — classic conflict.""" |
| 997 | _make_reservation( |
| 998 | repo, run_id="agent-a", addresses=["billing.py::compute_total"] |
| 999 | ) |
| 1000 | _make_reservation( |
| 1001 | repo, run_id="agent-b", addresses=["billing.py::compute_total"] |
| 1002 | ) |
| 1003 | return repo |
| 1004 | |
| 1005 | @pytest.fixture |
| 1006 | def no_conflict(self, repo: pathlib.Path) -> pathlib.Path: |
| 1007 | """Two agents on distinct addresses — no conflict.""" |
| 1008 | _make_reservation(repo, run_id="agent-a", addresses=["billing.py::total"]) |
| 1009 | _make_reservation(repo, run_id="agent-b", addresses=["auth.py::login"]) |
| 1010 | return repo |
| 1011 | |
| 1012 | def test_has_conflicts_true_when_overlap(self, conflicting: pathlib.Path) -> None: |
| 1013 | result = runner.invoke(cli, ["coord", "list", "--json"]) |
| 1014 | data = json.loads(result.output) |
| 1015 | assert data["has_conflicts"] is True |
| 1016 | |
| 1017 | def test_has_conflicts_false_when_no_overlap(self, no_conflict: pathlib.Path) -> None: |
| 1018 | result = runner.invoke(cli, ["coord", "list", "--json"]) |
| 1019 | data = json.loads(result.output) |
| 1020 | assert data["has_conflicts"] is False |
| 1021 | |
| 1022 | def test_conflict_count_correct(self, conflicting: pathlib.Path) -> None: |
| 1023 | result = runner.invoke(cli, ["coord", "list", "--json"]) |
| 1024 | data = json.loads(result.output) |
| 1025 | for r in data["reservations"]: |
| 1026 | assert r["conflict_count"] == 1 |
| 1027 | |
| 1028 | def test_no_conflict_count_zero(self, no_conflict: pathlib.Path) -> None: |
| 1029 | result = runner.invoke(cli, ["coord", "list", "--json"]) |
| 1030 | data = json.loads(result.output) |
| 1031 | for r in data["reservations"]: |
| 1032 | assert r["conflict_count"] == 0 |
| 1033 | |
| 1034 | def test_conflict_count_multi(self, repo: pathlib.Path) -> None: |
| 1035 | """Three agents on the same address: each sees conflict_count == 2.""" |
| 1036 | addr = "billing.py::compute_total" |
| 1037 | for i in range(3): |
| 1038 | _make_reservation(repo, run_id=f"agent-{i}", addresses=[addr]) |
| 1039 | result = runner.invoke(cli, ["coord", "list", "--json"]) |
| 1040 | data = json.loads(result.output) |
| 1041 | for r in data["reservations"]: |
| 1042 | assert r["conflict_count"] == 2 |
| 1043 | |
| 1044 | def test_has_conflicts_false_empty(self, repo: pathlib.Path) -> None: |
| 1045 | result = runner.invoke(cli, ["coord", "list", "--json"]) |
| 1046 | data = json.loads(result.output) |
| 1047 | assert data["has_conflicts"] is False |
| 1048 | |
| 1049 | def test_conflict_warning_in_text_output(self, conflicting: pathlib.Path) -> None: |
| 1050 | result = runner.invoke(cli, ["coord", "list"]) |
| 1051 | assert result.exit_code == 0 |
| 1052 | assert "conflict" in result.output.lower() |
| 1053 | |
| 1054 | def test_reservation_id_in_text_output(self, conflicting: pathlib.Path) -> None: |
| 1055 | """Short reservation ID must appear in text output so agents can act on it.""" |
| 1056 | result = runner.invoke(cli, ["coord", "list"]) |
| 1057 | data_json = runner.invoke(cli, ["coord", "list", "--json"]) |
| 1058 | data = json.loads(data_json.output) |
| 1059 | short_id = data["reservations"][0]["reservation_id"][:8] |
| 1060 | assert short_id in result.output |
File History
1 commit
sha256:1c4b3e3a9a1f300774c3ee662b572a698d5fd405bf765a71e6011a2e9c3eaaaa
feat: Muse — version control for the agent era
Human
73 days ago