"""Comprehensive tests for ``muse coord list``. Coverage -------- Unit — core helpers _format_ttl: all time ranges, zero, negative filter_reservations: run_id, branch, address_glob, include_expired, compound (AND) filters, empty input filter_intents: run_id, branch, address_glob, compound, empty Reservation.ttl_remaining_seconds: positive, negative, near-zero Integration — CLI Empty repo (no .muse/coordination): exits 0, text + JSON Basic list: shows reservations and intents --kind reservations: hides intents --kind intents: hides reservations --run-id filter: exact match --branch filter: exact match --address glob: fnmatch, wildcard, no-match --all flag: expired reservations appear --summary: single line, counts correct --format json / --json: valid JSON, full schema JSON schema fields: all required keys present and typed JSON ttl_remaining_seconds: float, positive for active JSON is_active flag: True for active, False for expired JSON duration_ms: float >= 0 JSON filters object: reflects CLI args Sorting: created_at ascending Multiple reservations: all shown Multiple intents: all shown Expired hidden by default: not shown without --all Compound filter: run_id + branch + address Security ANSI in run_id: stripped in text output Control chars in branch: stripped Control chars in detail: stripped Control chars in address: stripped in text output --address glob no FS access: fnmatch is string-only Null byte in run_id filter: safe (no match, no crash) Stress 500 reservations: listing < 2 s 1 000 intents: listing < 3 s 200 filtered to 1: < 1 s 50 agents, 20 addrs each: JSON schema valid """ from __future__ import annotations import datetime import json import pathlib import time import pytest from tests.cli_test_helper import CliRunner from muse.core.types import fake_id from muse.core.coordination import ( Reservation, Intent, create_intent, create_reservation, filter_intents, filter_reservations, load_all_reservations, load_all_intents, ) from muse.cli.commands.list_coord import _format_ttl from muse.core.paths import coordination_dir cli = None runner = CliRunner() _REQUIRED_JSON_KEYS = { "active_reservations", "expired_reservations", "released_reservations", "total_reservations_shown", "total_intents_shown", "has_conflicts", "filters", "reservations", "intents", "duration_ms", } _REQUIRED_RES_KEYS = { "reservation_id", "run_id", "branch", "addresses", "created_at", "expires_at", "effective_expires_at", "ttl_remaining_seconds", "operation", "is_active", "released", "conflict_count", } _REQUIRED_INT_KEYS = { "intent_id", "reservation_id", "run_id", "branch", "addresses", "operation", "created_at", "detail", } # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- def _now() -> datetime.datetime: return datetime.datetime.now(datetime.timezone.utc) def _future(seconds: int = 3600) -> datetime.datetime: return _now() + datetime.timedelta(seconds=seconds) def _past(seconds: int = 60) -> datetime.datetime: return _now() - datetime.timedelta(seconds=seconds) def _make_reservation( tmp_path: pathlib.Path, *, run_id: str = "agent-1", branch: str = "main", addresses: list[str] | None = None, ttl_seconds: int = 3600, operation: str | None = None, ) -> Reservation: return create_reservation( tmp_path, run_id=run_id, branch=branch, addresses=addresses or ["src/mod.py::foo"], ttl_seconds=ttl_seconds, operation=operation, ) def _make_intent( tmp_path: pathlib.Path, *, run_id: str = "agent-1", branch: str = "main", addresses: list[str] | None = None, operation: str = "modify", detail: str = "", ) -> Intent: return create_intent( tmp_path, reservation_id=fake_id("coord-list-intent-res"), run_id=run_id, branch=branch, addresses=addresses or ["src/mod.py::foo"], operation=operation, detail=detail, ) @pytest.fixture def repo(tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch) -> pathlib.Path: monkeypatch.chdir(tmp_path) monkeypatch.setenv("MUSE_REPO_ROOT", str(tmp_path)) r = runner.invoke(cli, ["init", "--domain", "code"]) assert r.exit_code == 0, r.output return tmp_path # --------------------------------------------------------------------------- # Unit — _format_ttl # --------------------------------------------------------------------------- class TestFormatTtl: def test_hours_minutes_seconds(self) -> None: assert _format_ttl(3661) == "1h 1m 1s" def test_exactly_one_hour(self) -> None: assert _format_ttl(3600) == "1h 0m 0s" def test_minutes_and_seconds(self) -> None: assert _format_ttl(90) == "1m 30s" def test_seconds_only(self) -> None: assert _format_ttl(45) == "45s" def test_one_second(self) -> None: assert _format_ttl(1) == "1s" def test_zero_is_expired(self) -> None: assert _format_ttl(0) == "EXPIRED" def test_negative_is_expired(self) -> None: assert _format_ttl(-100) == "EXPIRED" def test_float_truncates(self) -> None: # 90.9 → 90 seconds → 1m 30s assert _format_ttl(90.9) == "1m 30s" def test_large_hours(self) -> None: result = _format_ttl(86400) # 24 h assert result.startswith("24h") # --------------------------------------------------------------------------- # Unit — Reservation.ttl_remaining_seconds # --------------------------------------------------------------------------- class TestTtlRemainingSeconds: def test_active_reservation_positive(self, tmp_path: pathlib.Path) -> None: res = _make_reservation(tmp_path, ttl_seconds=3600) assert res.ttl_remaining_seconds() > 0 def test_expired_reservation_negative(self, tmp_path: pathlib.Path) -> None: res = _make_reservation(tmp_path, ttl_seconds=1) # Manually expire it. res.expires_at = _past(10) assert res.ttl_remaining_seconds() < 0 def test_near_zero_close_to_expiry(self, tmp_path: pathlib.Path) -> None: res = _make_reservation(tmp_path, ttl_seconds=1) res.expires_at = _now() + datetime.timedelta(seconds=0.5) assert 0 < res.ttl_remaining_seconds() < 2 # --------------------------------------------------------------------------- # Unit — filter_reservations # --------------------------------------------------------------------------- class TestFilterReservations: def test_empty_list(self) -> None: assert filter_reservations([]) == [] def test_active_only_by_default(self, tmp_path: pathlib.Path) -> None: active = _make_reservation(tmp_path, ttl_seconds=3600) expired = _make_reservation(tmp_path, ttl_seconds=1) expired.expires_at = _past(10) result = filter_reservations([active, expired]) assert len(result) == 1 assert result[0].reservation_id == active.reservation_id def test_include_expired(self, tmp_path: pathlib.Path) -> None: active = _make_reservation(tmp_path, ttl_seconds=3600) expired = _make_reservation(tmp_path, ttl_seconds=1) expired.expires_at = _past(10) result = filter_reservations([active, expired], include_expired=True) assert len(result) == 2 def test_run_id_exact_match(self, tmp_path: pathlib.Path) -> None: r1 = _make_reservation(tmp_path, run_id="agent-1") r2 = _make_reservation(tmp_path, run_id="agent-2") result = filter_reservations([r1, r2], run_id="agent-1") assert len(result) == 1 assert result[0].run_id == "agent-1" def test_run_id_no_match(self, tmp_path: pathlib.Path) -> None: r = _make_reservation(tmp_path, run_id="agent-1") assert filter_reservations([r], run_id="agent-99") == [] def test_branch_exact_match(self, tmp_path: pathlib.Path) -> None: r1 = _make_reservation(tmp_path, branch="main") r2 = _make_reservation(tmp_path, branch="feat/x") result = filter_reservations([r1, r2], branch="main") assert len(result) == 1 assert result[0].branch == "main" def test_address_glob_wildcard(self, tmp_path: pathlib.Path) -> None: r1 = _make_reservation(tmp_path, addresses=["billing.py::compute_total"]) r2 = _make_reservation(tmp_path, addresses=["auth.py::login"]) result = filter_reservations([r1, r2], address_glob="billing.py::*") assert len(result) == 1 assert result[0].addresses == ["billing.py::compute_total"] def test_address_glob_no_match(self, tmp_path: pathlib.Path) -> None: r = _make_reservation(tmp_path, addresses=["billing.py::compute_total"]) assert filter_reservations([r], address_glob="auth.py::*") == [] def test_address_glob_any_address_matches(self, tmp_path: pathlib.Path) -> None: """A reservation with multiple addresses: match if ANY address matches.""" r = _make_reservation( tmp_path, addresses=["billing.py::compute_total", "auth.py::login"], ) result = filter_reservations([r], address_glob="auth.py::*") assert len(result) == 1 def test_compound_run_id_and_branch(self, tmp_path: pathlib.Path) -> None: r1 = _make_reservation(tmp_path, run_id="a", branch="main") r2 = _make_reservation(tmp_path, run_id="a", branch="feat") r3 = _make_reservation(tmp_path, run_id="b", branch="main") result = filter_reservations([r1, r2, r3], run_id="a", branch="main") assert len(result) == 1 assert result[0].reservation_id == r1.reservation_id def test_compound_run_id_and_address_glob(self, tmp_path: pathlib.Path) -> None: r1 = _make_reservation(tmp_path, run_id="a", addresses=["billing.py::foo"]) r2 = _make_reservation(tmp_path, run_id="a", addresses=["auth.py::bar"]) r3 = _make_reservation(tmp_path, run_id="b", addresses=["billing.py::foo"]) result = filter_reservations([r1, r2, r3], run_id="a", address_glob="billing.py::*") assert len(result) == 1 assert result[0].reservation_id == r1.reservation_id def test_preserves_order(self, tmp_path: pathlib.Path) -> None: r1 = _make_reservation(tmp_path, run_id="a") r2 = _make_reservation(tmp_path, run_id="b") result = filter_reservations([r1, r2]) assert [r.run_id for r in result] == ["a", "b"] # --------------------------------------------------------------------------- # Unit — filter_intents # --------------------------------------------------------------------------- class TestFilterIntents: def test_empty_list(self) -> None: assert filter_intents([]) == [] def test_all_returned_by_default(self, tmp_path: pathlib.Path) -> None: i1 = _make_intent(tmp_path, run_id="a") i2 = _make_intent(tmp_path, run_id="b") assert len(filter_intents([i1, i2])) == 2 def test_run_id_filter(self, tmp_path: pathlib.Path) -> None: i1 = _make_intent(tmp_path, run_id="a") i2 = _make_intent(tmp_path, run_id="b") result = filter_intents([i1, i2], run_id="a") assert len(result) == 1 def test_branch_filter(self, tmp_path: pathlib.Path) -> None: i1 = _make_intent(tmp_path, branch="main") i2 = _make_intent(tmp_path, branch="feat") result = filter_intents([i1, i2], branch="main") assert len(result) == 1 def test_address_glob(self, tmp_path: pathlib.Path) -> None: i1 = _make_intent(tmp_path, addresses=["billing.py::total"]) i2 = _make_intent(tmp_path, addresses=["auth.py::login"]) result = filter_intents([i1, i2], address_glob="billing.py::*") assert len(result) == 1 def test_no_match_returns_empty(self, tmp_path: pathlib.Path) -> None: i = _make_intent(tmp_path) assert filter_intents([i], run_id="nonexistent") == [] def test_compound_filters(self, tmp_path: pathlib.Path) -> None: i1 = _make_intent(tmp_path, run_id="a", branch="main", addresses=["billing.py::x"]) i2 = _make_intent(tmp_path, run_id="a", branch="feat", addresses=["billing.py::x"]) i3 = _make_intent(tmp_path, run_id="b", branch="main", addresses=["billing.py::x"]) result = filter_intents([i1, i2, i3], run_id="a", branch="main") assert len(result) == 1 assert result[0].intent_id == i1.intent_id # --------------------------------------------------------------------------- # Integration — empty repo # --------------------------------------------------------------------------- class TestCoordListEmpty: def test_empty_exits_zero_text(self, repo: pathlib.Path) -> None: result = runner.invoke(cli, ["coord", "list"]) assert result.exit_code == 0 def test_empty_text_message(self, repo: pathlib.Path) -> None: result = runner.invoke(cli, ["coord", "list"]) assert "no coordination records" in result.output.lower() def test_empty_json_schema(self, repo: pathlib.Path) -> None: result = runner.invoke(cli, ["coord", "list", "--json"]) assert result.exit_code == 0 data = json.loads(result.output) assert _REQUIRED_JSON_KEYS <= set(data) assert data["active_reservations"] == 0 assert data["total_reservations_shown"] == 0 assert data["total_intents_shown"] == 0 assert data["reservations"] == [] assert data["intents"] == [] def test_empty_summary(self, repo: pathlib.Path) -> None: result = runner.invoke(cli, ["coord", "list", "--summary"]) assert result.exit_code == 0 def test_empty_duration_ms_present(self, repo: pathlib.Path) -> None: result = runner.invoke(cli, ["coord", "list", "--json"]) data = json.loads(result.output) assert isinstance(data["duration_ms"], float) assert data["duration_ms"] >= 0 # --------------------------------------------------------------------------- # Integration — basic listing # --------------------------------------------------------------------------- class TestCoordListBasic: @pytest.fixture def populated(self, repo: pathlib.Path) -> pathlib.Path: _make_reservation( repo, run_id="agent-42", branch="feat/billing", addresses=["billing.py::compute_total", "billing.py::apply_discount"], operation="modify", ) _make_reservation( repo, run_id="agent-41", branch="main", addresses=["auth.py::validate_token"], operation="rename", ) _make_intent( repo, run_id="agent-42", branch="feat/billing", addresses=["billing.py::compute_total"], operation="rename", detail="rename to compute_invoice_total", ) return repo def test_shows_reservations(self, populated: pathlib.Path) -> None: result = runner.invoke(cli, ["coord", "list"]) assert result.exit_code == 0 assert "agent-42" in result.output assert "agent-41" in result.output def test_shows_addresses(self, populated: pathlib.Path) -> None: result = runner.invoke(cli, ["coord", "list"]) assert "billing.py::compute_total" in result.output assert "auth.py::validate_token" in result.output def test_shows_intents(self, populated: pathlib.Path) -> None: result = runner.invoke(cli, ["coord", "list"]) assert "rename" in result.output assert "rename to compute_invoice_total" in result.output def test_json_res_count(self, populated: pathlib.Path) -> None: result = runner.invoke(cli, ["coord", "list", "--json"]) data = json.loads(result.output) assert data["total_reservations_shown"] == 2 assert data["total_intents_shown"] == 1 def test_json_res_schema_fields(self, populated: pathlib.Path) -> None: result = runner.invoke(cli, ["coord", "list", "--json"]) data = json.loads(result.output) for entry in data["reservations"]: assert _REQUIRED_RES_KEYS <= set(entry), f"missing keys in {entry}" def test_json_intent_schema_fields(self, populated: pathlib.Path) -> None: result = runner.invoke(cli, ["coord", "list", "--json"]) data = json.loads(result.output) for entry in data["intents"]: assert _REQUIRED_INT_KEYS <= set(entry), f"missing keys in {entry}" def test_json_is_active_true_for_active(self, populated: pathlib.Path) -> None: result = runner.invoke(cli, ["coord", "list", "--json"]) data = json.loads(result.output) for r in data["reservations"]: assert r["is_active"] is True def test_json_ttl_positive_for_active(self, populated: pathlib.Path) -> None: result = runner.invoke(cli, ["coord", "list", "--json"]) data = json.loads(result.output) for r in data["reservations"]: assert r["ttl_remaining_seconds"] > 0 def test_json_duration_ms(self, populated: pathlib.Path) -> None: result = runner.invoke(cli, ["coord", "list", "--json"]) data = json.loads(result.output) assert isinstance(data["duration_ms"], float) assert data["duration_ms"] >= 0 # --------------------------------------------------------------------------- # Integration — --kind flag # --------------------------------------------------------------------------- class TestCoordListKind: @pytest.fixture def both(self, repo: pathlib.Path) -> pathlib.Path: _make_reservation(repo, run_id="agent-1") _make_intent(repo, run_id="agent-1") return repo def test_kind_reservations_hides_intents(self, both: pathlib.Path) -> None: result = runner.invoke(cli, ["coord", "list", "--kind", "reservations"]) assert result.exit_code == 0 data_json = runner.invoke(cli, ["coord", "list", "--kind", "reservations", "--json"]) data = json.loads(data_json.output) assert data["total_intents_shown"] == 0 def test_kind_intents_hides_reservations(self, both: pathlib.Path) -> None: data_json = runner.invoke(cli, ["coord", "list", "--kind", "intents", "--json"]) data = json.loads(data_json.output) assert data["total_reservations_shown"] == 0 assert data["total_intents_shown"] == 1 def test_kind_all_shows_both(self, both: pathlib.Path) -> None: data_json = runner.invoke(cli, ["coord", "list", "--kind", "all", "--json"]) data = json.loads(data_json.output) assert data["total_reservations_shown"] == 1 assert data["total_intents_shown"] == 1 # --------------------------------------------------------------------------- # Integration — filters # --------------------------------------------------------------------------- class TestCoordListFilters: @pytest.fixture def multi(self, repo: pathlib.Path) -> pathlib.Path: _make_reservation( repo, run_id="agent-42", branch="feat/billing", addresses=["billing.py::compute_total"], ) _make_reservation( repo, run_id="agent-41", branch="main", addresses=["auth.py::validate_token"], ) _make_intent(repo, run_id="agent-42", branch="feat/billing", addresses=["billing.py::compute_total"]) _make_intent(repo, run_id="agent-41", branch="main", addresses=["auth.py::validate_token"]) return repo def test_run_id_filter(self, multi: pathlib.Path) -> None: result = runner.invoke(cli, ["coord", "list", "--run-id", "agent-42", "--json"]) data = json.loads(result.output) assert data["total_reservations_shown"] == 1 assert data["reservations"][0]["run_id"] == "agent-42" def test_run_id_no_match(self, multi: pathlib.Path) -> None: result = runner.invoke(cli, ["coord", "list", "--run-id", "nobody", "--json"]) data = json.loads(result.output) assert data["total_reservations_shown"] == 0 assert data["total_intents_shown"] == 0 def test_branch_filter(self, multi: pathlib.Path) -> None: result = runner.invoke(cli, ["coord", "list", "--branch", "main", "--json"]) data = json.loads(result.output) assert data["total_reservations_shown"] == 1 assert data["reservations"][0]["branch"] == "main" def test_address_glob_wildcard(self, multi: pathlib.Path) -> None: result = runner.invoke( cli, ["coord", "list", "--address", "billing.py::*", "--json"] ) data = json.loads(result.output) assert data["total_reservations_shown"] == 1 assert data["total_intents_shown"] == 1 def test_address_glob_no_match(self, multi: pathlib.Path) -> None: result = runner.invoke( cli, ["coord", "list", "--address", "nonexistent.py::*", "--json"] ) data = json.loads(result.output) assert data["total_reservations_shown"] == 0 def test_compound_run_id_and_branch(self, multi: pathlib.Path) -> None: result = runner.invoke( cli, ["coord", "list", "--run-id", "agent-42", "--branch", "feat/billing", "--json"] ) data = json.loads(result.output) assert data["total_reservations_shown"] == 1 assert data["total_intents_shown"] == 1 def test_json_filters_object_reflects_args(self, multi: pathlib.Path) -> None: result = runner.invoke( cli, ["coord", "list", "--run-id", "agent-42", "--branch", "feat/billing", "--address", "billing.py::*", "--json"] ) data = json.loads(result.output) f = data["filters"] assert f["run_id"] == "agent-42" assert f["branch"] == "feat/billing" assert f["address_glob"] == "billing.py::*" assert f["include_expired"] is False assert f["kind"] == "all" # --------------------------------------------------------------------------- # Integration — expired reservations # --------------------------------------------------------------------------- class TestCoordListExpired: @pytest.fixture def with_expired(self, repo: pathlib.Path) -> pathlib.Path: # Active reservation. _make_reservation(repo, run_id="active-agent", ttl_seconds=3600) # Expired: create then manually expire via file. coord_dir = coordination_dir(repo) / "reservations" coord_dir.mkdir(parents=True, exist_ok=True) import json as _json from muse._version import __version__ now = datetime.datetime.now(datetime.timezone.utc) expired_id = fake_id("coord-list-expired-res") record = { "schema_version": __version__, "reservation_id": expired_id, "run_id": "expired-agent", "branch": "main", "addresses": ["old.py::stale_func"], "created_at": (now - datetime.timedelta(hours=2)).isoformat(), "expires_at": (now - datetime.timedelta(hours=1)).isoformat(), "operation": None, } (coord_dir / f"{expired_id}.json").write_text(_json.dumps(record)) return repo def test_expired_hidden_by_default(self, with_expired: pathlib.Path) -> None: result = runner.invoke(cli, ["coord", "list", "--json"]) data = json.loads(result.output) run_ids = [r["run_id"] for r in data["reservations"]] assert "expired-agent" not in run_ids assert "active-agent" in run_ids def test_expired_shown_with_all(self, with_expired: pathlib.Path) -> None: result = runner.invoke(cli, ["coord", "list", "--all", "--json"]) data = json.loads(result.output) run_ids = [r["run_id"] for r in data["reservations"]] assert "expired-agent" in run_ids def test_expired_is_active_false(self, with_expired: pathlib.Path) -> None: result = runner.invoke(cli, ["coord", "list", "--all", "--json"]) data = json.loads(result.output) for r in data["reservations"]: if r["run_id"] == "expired-agent": assert r["is_active"] is False assert r["ttl_remaining_seconds"] < 0 def test_expired_count_reported(self, with_expired: pathlib.Path) -> None: result = runner.invoke(cli, ["coord", "list", "--json"]) data = json.loads(result.output) assert data["expired_reservations"] == 1 assert data["active_reservations"] == 1 # --------------------------------------------------------------------------- # Integration — --summary # --------------------------------------------------------------------------- class TestCoordListSummary: def test_summary_empty(self, repo: pathlib.Path) -> None: result = runner.invoke(cli, ["coord", "list", "--summary"]) assert result.exit_code == 0 def test_summary_shows_count(self, repo: pathlib.Path) -> None: _make_reservation(repo, run_id="a") _make_reservation(repo, run_id="b") _make_intent(repo, run_id="a") result = runner.invoke(cli, ["coord", "list", "--summary"]) assert result.exit_code == 0 assert "2" in result.output # 2 reservations assert "1" in result.output # 1 intent def test_summary_no_detail(self, repo: pathlib.Path) -> None: _make_reservation(repo) result = runner.invoke(cli, ["coord", "list", "--summary"]) # Summary must be a single line — no address lines. lines = [l for l in result.output.splitlines() if l.strip()] assert len(lines) == 1 def test_summary_exits_zero(self, repo: pathlib.Path) -> None: result = runner.invoke(cli, ["coord", "list", "--summary"]) assert result.exit_code == 0 # --------------------------------------------------------------------------- # Integration — --format / --json flag # --------------------------------------------------------------------------- class TestCoordListFormat: def test_format_json_flag(self, repo: pathlib.Path) -> None: result = runner.invoke(cli, ["coord", "list", "--json"]) assert result.exit_code == 0 data = json.loads(result.output) assert "reservations" in data def test_json_shorthand(self, repo: pathlib.Path) -> None: result = runner.invoke(cli, ["coord", "list", "--json"]) assert result.exit_code == 0 data = json.loads(result.output) assert "reservations" in data def test_text_is_default(self, repo: pathlib.Path) -> None: result = runner.invoke(cli, ["coord", "list"]) # Text output contains "Coordination state", not a JSON brace. assert result.output.strip()[0] != "{" # --------------------------------------------------------------------------- # Integration — sorting # --------------------------------------------------------------------------- class TestCoordListSorting: def test_reservations_sorted_by_created_at(self, repo: pathlib.Path) -> None: """Older reservations appear first in JSON output.""" r1 = _make_reservation(repo, run_id="first") # Ensure a distinct created_at by tweaking the file. import json as _json, time as _time _time.sleep(0.01) r2 = _make_reservation(repo, run_id="second") result = runner.invoke(cli, ["coord", "list", "--json"]) data = json.loads(result.output) run_ids = [r["run_id"] for r in data["reservations"]] assert run_ids.index("first") < run_ids.index("second") # --------------------------------------------------------------------------- # Security # --------------------------------------------------------------------------- class TestCoordListSecurity: def test_ansi_in_run_id_stripped_from_text(self, repo: pathlib.Path) -> None: """ANSI escape sequences in run_id must not reach the terminal raw.""" _make_reservation(repo, run_id="\x1b[31mred\x1b[0m") result = runner.invoke(cli, ["coord", "list"]) assert result.exit_code == 0 assert "\x1b" not in result.output def test_control_chars_in_branch_stripped(self, repo: pathlib.Path) -> None: _make_reservation(repo, branch="main\x01;rm -rf /") result = runner.invoke(cli, ["coord", "list"]) assert result.exit_code == 0 assert "\x01" not in result.output def test_control_chars_in_address_stripped(self, repo: pathlib.Path) -> None: _make_reservation(repo, addresses=["billing.py::\x1b[Afoo"]) result = runner.invoke(cli, ["coord", "list"]) assert result.exit_code == 0 assert "\x1b" not in result.output def test_control_chars_in_detail_stripped(self, repo: pathlib.Path) -> None: _make_intent(repo, detail="legit detail\x1b[31m injected color\x1b[0m") result = runner.invoke(cli, ["coord", "list"]) assert result.exit_code == 0 assert "\x1b" not in result.output def test_null_byte_in_run_id_filter_safe(self, repo: pathlib.Path) -> None: """A null byte in --run-id must not crash — it simply matches nothing.""" _make_reservation(repo, run_id="agent-1") result = runner.invoke(cli, ["coord", "list", "--run-id", "agent\x00-1"]) assert result.exit_code == 0 def test_address_glob_no_filesystem_access( self, repo: pathlib.Path, tmp_path: pathlib.Path ) -> None: """Glob pattern must not touch the filesystem — only string matching.""" _make_reservation(repo, addresses=["billing.py::total"]) # A glob that could be interpreted as a filesystem glob. result = runner.invoke(cli, ["coord", "list", "--address", "**/billing*"]) # Should complete without error regardless of match. assert result.exit_code == 0 def test_requires_repo( self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch ) -> None: monkeypatch.chdir(tmp_path) monkeypatch.delenv("MUSE_REPO_ROOT", raising=False) result = runner.invoke(cli, ["coord", "list"]) assert result.exit_code != 0 # --------------------------------------------------------------------------- # Stress # --------------------------------------------------------------------------- class TestCoordListStress: @pytest.fixture def large_swarm( self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch ) -> pathlib.Path: monkeypatch.chdir(tmp_path) monkeypatch.setenv("MUSE_REPO_ROOT", str(tmp_path)) runner.invoke(cli, ["init", "--domain", "code"]) return tmp_path def test_500_reservations_under_2s(self, large_swarm: pathlib.Path) -> None: for i in range(500): _make_reservation( large_swarm, run_id=f"agent-{i}", branch=f"feat/task-{i % 20}", addresses=[f"src/module_{i % 50}.py::func_{i}"], ) start = time.monotonic() result = runner.invoke(cli, ["coord", "list", "--json"]) elapsed = time.monotonic() - start assert result.exit_code == 0 assert elapsed < 2.0, f"500 reservations took {elapsed:.2f}s" data = json.loads(result.output) assert data["total_reservations_shown"] == 500 def test_1000_intents_under_3s(self, large_swarm: pathlib.Path) -> None: for i in range(1000): _make_intent( large_swarm, run_id=f"agent-{i % 50}", branch=f"feat/task-{i % 10}", addresses=[f"src/mod_{i % 30}.py::sym_{i}"], operation="modify", ) start = time.monotonic() result = runner.invoke(cli, ["coord", "list", "--kind", "intents", "--json"]) elapsed = time.monotonic() - start assert result.exit_code == 0 assert elapsed < 3.0, f"1000 intents took {elapsed:.2f}s" data = json.loads(result.output) assert data["total_intents_shown"] == 1000 def test_200_filtered_to_1_under_1s(self, large_swarm: pathlib.Path) -> None: for i in range(199): _make_reservation( large_swarm, run_id=f"agent-{i}", addresses=[f"src/mod_{i}.py::sym"], ) _make_reservation( large_swarm, run_id="needle", addresses=["billing.py::compute_total"], ) start = time.monotonic() result = runner.invoke( cli, ["coord", "list", "--run-id", "needle", "--json"] ) elapsed = time.monotonic() - start assert result.exit_code == 0 assert elapsed < 1.0, f"filter of 200 took {elapsed:.2f}s" data = json.loads(result.output) assert data["total_reservations_shown"] == 1 assert data["reservations"][0]["run_id"] == "needle" def test_50_agents_20_addresses_json_valid(self, large_swarm: pathlib.Path) -> None: """Large JSON payload: every entry has all required fields.""" for agent in range(50): addresses = [f"src/mod_{j}.py::func_{agent}" for j in range(20)] _make_reservation(large_swarm, run_id=f"agent-{agent}", addresses=addresses) result = runner.invoke(cli, ["coord", "list", "--json"]) assert result.exit_code == 0 data = json.loads(result.output) assert data["total_reservations_shown"] == 50 for entry in data["reservations"]: assert _REQUIRED_RES_KEYS <= set(entry) assert len(entry["addresses"]) == 20 # --------------------------------------------------------------------------- # Integration — --op filter # --------------------------------------------------------------------------- class TestCoordListOp: @pytest.fixture def with_ops(self, repo: pathlib.Path) -> pathlib.Path: _make_reservation(repo, run_id="agent-rename", operation="rename", addresses=["billing.py::compute_total"]) _make_reservation(repo, run_id="agent-modify", operation="modify", addresses=["auth.py::validate"]) _make_reservation(repo, run_id="agent-none", operation=None, addresses=["core.py::hash"]) _make_intent(repo, run_id="agent-delete", operation="delete", addresses=["old.py::stale"]) _make_intent(repo, run_id="agent-extract", operation="extract", addresses=["utils.py::helper"]) return repo def test_op_filter_reservations(self, with_ops: pathlib.Path) -> None: result = runner.invoke(cli, ["coord", "list", "--op", "rename", "--json"]) data = json.loads(result.output) assert data["total_reservations_shown"] == 1 assert data["reservations"][0]["run_id"] == "agent-rename" def test_op_filter_intents(self, with_ops: pathlib.Path) -> None: result = runner.invoke(cli, ["coord", "list", "--op", "delete", "--json"]) data = json.loads(result.output) assert data["total_intents_shown"] == 1 assert data["intents"][0]["run_id"] == "agent-delete" def test_op_filter_no_match(self, with_ops: pathlib.Path) -> None: result = runner.invoke(cli, ["coord", "list", "--op", "inline", "--json"]) data = json.loads(result.output) assert data["total_reservations_shown"] == 0 assert data["total_intents_shown"] == 0 def test_op_filter_both_kinds(self, with_ops: pathlib.Path) -> None: """--op filters both reservations and intents simultaneously.""" result = runner.invoke(cli, ["coord", "list", "--op", "extract", "--json"]) data = json.loads(result.output) assert data["total_reservations_shown"] == 0 assert data["total_intents_shown"] == 1 def test_op_reflected_in_filters_object(self, with_ops: pathlib.Path) -> None: result = runner.invoke(cli, ["coord", "list", "--op", "modify", "--json"]) data = json.loads(result.output) assert data["filters"]["operation"] == "modify" def test_op_null_in_filters_when_unset(self, with_ops: pathlib.Path) -> None: result = runner.invoke(cli, ["coord", "list", "--json"]) data = json.loads(result.output) assert data["filters"]["operation"] is None def test_op_compound_with_run_id(self, with_ops: pathlib.Path) -> None: result = runner.invoke( cli, ["coord", "list", "--op", "rename", "--run-id", "agent-rename", "--json"] ) data = json.loads(result.output) assert data["total_reservations_shown"] == 1 def test_op_in_text_output(self, with_ops: pathlib.Path) -> None: result = runner.invoke(cli, ["coord", "list", "--op", "rename"]) assert result.exit_code == 0 assert "agent-rename" in result.output # --------------------------------------------------------------------------- # Integration — --limit flag # --------------------------------------------------------------------------- class TestCoordListLimit: @pytest.fixture def many(self, repo: pathlib.Path) -> pathlib.Path: for i in range(10): _make_reservation(repo, run_id=f"agent-{i}") _make_intent(repo, run_id=f"agent-{i}") return repo def test_limit_caps_reservations(self, many: pathlib.Path) -> None: result = runner.invoke(cli, ["coord", "list", "--limit", "3", "--json"]) data = json.loads(result.output) assert data["total_reservations_shown"] == 3 def test_limit_caps_intents(self, many: pathlib.Path) -> None: result = runner.invoke(cli, ["coord", "list", "--limit", "3", "--json"]) data = json.loads(result.output) assert data["total_intents_shown"] == 3 def test_limit_larger_than_count_returns_all(self, many: pathlib.Path) -> None: result = runner.invoke(cli, ["coord", "list", "--limit", "100", "--json"]) data = json.loads(result.output) assert data["total_reservations_shown"] == 10 def test_limit_one(self, many: pathlib.Path) -> None: result = runner.invoke(cli, ["coord", "list", "--limit", "1", "--json"]) data = json.loads(result.output) assert data["total_reservations_shown"] == 1 assert data["total_intents_shown"] == 1 def test_limit_reflected_in_filters(self, many: pathlib.Path) -> None: result = runner.invoke(cli, ["coord", "list", "--limit", "5", "--json"]) data = json.loads(result.output) assert data["filters"]["limit"] == 5 def test_no_limit_null_in_filters(self, many: pathlib.Path) -> None: result = runner.invoke(cli, ["coord", "list", "--json"]) data = json.loads(result.output) assert data["filters"]["limit"] is None def test_limit_returns_oldest_first(self, many: pathlib.Path) -> None: """--limit preserves sorted order (oldest first).""" result = runner.invoke(cli, ["coord", "list", "--limit", "2", "--json"]) data = json.loads(result.output) created_ats = [r["created_at"] for r in data["reservations"]] assert created_ats == sorted(created_ats) # --------------------------------------------------------------------------- # Integration — conflict detection # --------------------------------------------------------------------------- class TestCoordListConflicts: @pytest.fixture def conflicting(self, repo: pathlib.Path) -> pathlib.Path: """Two agents reserving the same address — classic conflict.""" _make_reservation( repo, run_id="agent-a", addresses=["billing.py::compute_total"] ) _make_reservation( repo, run_id="agent-b", addresses=["billing.py::compute_total"] ) return repo @pytest.fixture def no_conflict(self, repo: pathlib.Path) -> pathlib.Path: """Two agents on distinct addresses — no conflict.""" _make_reservation(repo, run_id="agent-a", addresses=["billing.py::total"]) _make_reservation(repo, run_id="agent-b", addresses=["auth.py::login"]) return repo def test_has_conflicts_true_when_overlap(self, conflicting: pathlib.Path) -> None: result = runner.invoke(cli, ["coord", "list", "--json"]) data = json.loads(result.output) assert data["has_conflicts"] is True def test_has_conflicts_false_when_no_overlap(self, no_conflict: pathlib.Path) -> None: result = runner.invoke(cli, ["coord", "list", "--json"]) data = json.loads(result.output) assert data["has_conflicts"] is False def test_conflict_count_correct(self, conflicting: pathlib.Path) -> None: result = runner.invoke(cli, ["coord", "list", "--json"]) data = json.loads(result.output) for r in data["reservations"]: assert r["conflict_count"] == 1 def test_no_conflict_count_zero(self, no_conflict: pathlib.Path) -> None: result = runner.invoke(cli, ["coord", "list", "--json"]) data = json.loads(result.output) for r in data["reservations"]: assert r["conflict_count"] == 0 def test_conflict_count_multi(self, repo: pathlib.Path) -> None: """Three agents on the same address: each sees conflict_count == 2.""" addr = "billing.py::compute_total" for i in range(3): _make_reservation(repo, run_id=f"agent-{i}", addresses=[addr]) result = runner.invoke(cli, ["coord", "list", "--json"]) data = json.loads(result.output) for r in data["reservations"]: assert r["conflict_count"] == 2 def test_has_conflicts_false_empty(self, repo: pathlib.Path) -> None: result = runner.invoke(cli, ["coord", "list", "--json"]) data = json.loads(result.output) assert data["has_conflicts"] is False def test_conflict_warning_in_text_output(self, conflicting: pathlib.Path) -> None: result = runner.invoke(cli, ["coord", "list"]) assert result.exit_code == 0 assert "conflict" in result.output.lower() def test_reservation_id_in_text_output(self, conflicting: pathlib.Path) -> None: """Short reservation ID must appear in text output so agents can act on it.""" result = runner.invoke(cli, ["coord", "list"]) data_json = runner.invoke(cli, ["coord", "list", "--json"]) data = json.loads(data_json.output) short_id = data["reservations"][0]["reservation_id"][:8] assert short_id in result.output class TestRegisterFlags: def test_default_json_out_is_false(self) -> None: import argparse from muse.cli.commands.list_coord import register p = argparse.ArgumentParser() subs = p.add_subparsers() register(subs) args = p.parse_args(["list"]) assert args.json_out is False def test_json_flag_sets_json_out(self) -> None: import argparse from muse.cli.commands.list_coord import register p = argparse.ArgumentParser() subs = p.add_subparsers() register(subs) args = p.parse_args(["list", "--json"]) assert args.json_out is True def test_j_shorthand_sets_json_out(self) -> None: import argparse from muse.cli.commands.list_coord import register p = argparse.ArgumentParser() subs = p.add_subparsers() register(subs) args = p.parse_args(["list", "-j"]) assert args.json_out is True