"""Tests for ``muse harmony`` CLI — Phase 2. Coverage tiers -------------- I Unit — TypedDict field presence; subcommand registration II Integration — every subcommand success path (text + JSON) III Integration — every subcommand error path (bad args, not-found) IV End-to-end — full lifecycle through the CLI layer V Data integrity— JSON round-trips; all fields always present VI Security — path-traversal IDs rejected; invalid hex rejected VII Performance — each subcommand completes within 300 ms """ from __future__ import annotations from collections.abc import Mapping from muse.core.types import fake_id, short_id from muse.core.paths import muse_dir import json import pathlib import time import typing import pytest from tests.cli_test_helper import CliRunner runner = CliRunner() # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- # --------------------------------------------------------------------------- # Fixtures # --------------------------------------------------------------------------- @pytest.fixture() def repo(tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch) -> pathlib.Path: """Minimal Muse repo — .muse/config.toml present so require_repo() succeeds.""" dot_muse = muse_dir(tmp_path) dot_muse.mkdir() (dot_muse / "config.toml").write_text('[repo]\nname = "test"\nid = "abc123"\n') monkeypatch.chdir(tmp_path) return tmp_path def _record( repo: pathlib.Path, *, path: str = "track.mid", domain: str = "midi", conflict_type: str = "content", ours: str = "ours", theirs: str = "theirs", ) -> str: """Invoke ``muse harmony record`` and return the pattern_id.""" r = runner.invoke(None, [ "harmony", "record", "--path", path, "--domain", domain, "--conflict-type", conflict_type, "--ours-id", fake_id(ours), "--theirs-id", fake_id(theirs), "--json", ]) assert r.exit_code == 0, f"record failed: {r.output}" return json.loads(r.output)["pattern_id"] def _resolve( repo: pathlib.Path, pattern_id: str, *, strategy: str = "manual", confidence: str = "0.9", outcome: str = "outcome", rationale: str = "test", agent_id: str | None = None, ) -> str: """Invoke ``muse harmony resolve`` and return the resolution_id.""" args = [ "harmony", "resolve", "--pattern-id", pattern_id, "--strategy", strategy, "--outcome-blob", fake_id(outcome), "--confidence", confidence, "--rationale", rationale, "--json", ] if agent_id: args += ["--agent-id", agent_id] r = runner.invoke(None, args) assert r.exit_code == 0, f"resolve failed: {r.output}" return json.loads(r.output)["resolution_id"] # =========================================================================== # Tier I — Unit: TypedDict schemas and subcommand registration # =========================================================================== class TestTypedDictSchemas: """I: All TypedDict output schemas declare expected keys.""" def _hints(self, name: str) -> Mapping[str, object]: import muse.cli.commands.harmony as h td = getattr(h, name) return typing.get_type_hints(td) def test_record_json_has_pattern_id(self) -> None: assert "pattern_id" in self._hints("_HarmonyRecordJson") def test_record_json_has_already_existed(self) -> None: assert "already_existed" in self._hints("_HarmonyRecordJson") def test_list_json_has_total(self) -> None: assert "total" in self._hints("_HarmonyListJson") def test_list_json_has_patterns(self) -> None: assert "patterns" in self._hints("_HarmonyListJson") def test_show_json_has_pattern(self) -> None: assert "pattern" in self._hints("_HarmonyShowJson") def test_show_json_has_resolutions(self) -> None: assert "resolutions" in self._hints("_HarmonyShowJson") def test_resolve_json_has_resolution_id(self) -> None: assert "resolution_id" in self._hints("_HarmonyResolveJson") def test_resolve_json_has_pattern_id(self) -> None: assert "pattern_id" in self._hints("_HarmonyResolveJson") def test_resolve_json_has_already_existed(self) -> None: assert "already_existed" in self._hints("_HarmonyResolveJson") def test_best_json_has_pattern_id(self) -> None: assert "pattern_id" in self._hints("_HarmonyBestJson") def test_best_json_has_resolution(self) -> None: assert "resolution" in self._hints("_HarmonyBestJson") def test_forget_json_has_pattern_id(self) -> None: assert "pattern_id" in self._hints("_HarmonyForgetJson") def test_forget_json_has_removed(self) -> None: assert "removed" in self._hints("_HarmonyForgetJson") def test_scalar_json_has_removed(self) -> None: assert "removed" in self._hints("_HarmonyScalarJson") def test_policy_add_json_has_policy_id(self) -> None: assert "policy_id" in self._hints("_HarmonyPolicyAddJson") def test_policy_list_json_has_total(self) -> None: assert "total" in self._hints("_HarmonyPolicyListJson") def test_policy_list_json_has_policies(self) -> None: assert "policies" in self._hints("_HarmonyPolicyListJson") def test_policy_remove_json_has_policy_id(self) -> None: assert "policy_id" in self._hints("_HarmonyPolicyRemoveJson") def test_policy_remove_json_has_removed(self) -> None: assert "removed" in self._hints("_HarmonyPolicyRemoveJson") def test_audit_json_has_total(self) -> None: assert "total" in self._hints("_HarmonyAuditJson") def test_audit_json_has_entries(self) -> None: assert "entries" in self._hints("_HarmonyAuditJson") class TestRegistration: """I: harmony is registered in the CLI and subcommands are reachable.""" def test_harmony_help_exits_0(self, repo: pathlib.Path) -> None: r = runner.invoke(None, ["harmony", "--help"]) assert r.exit_code == 0 def test_harmony_record_help(self, repo: pathlib.Path) -> None: r = runner.invoke(None, ["harmony", "record", "--help"]) assert r.exit_code == 0 def test_harmony_list_help(self, repo: pathlib.Path) -> None: r = runner.invoke(None, ["harmony", "list", "--help"]) assert r.exit_code == 0 def test_harmony_show_help(self, repo: pathlib.Path) -> None: r = runner.invoke(None, ["harmony", "show", "--help"]) assert r.exit_code == 0 def test_harmony_resolve_help(self, repo: pathlib.Path) -> None: r = runner.invoke(None, ["harmony", "resolve", "--help"]) assert r.exit_code == 0 def test_harmony_best_help(self, repo: pathlib.Path) -> None: r = runner.invoke(None, ["harmony", "best", "--help"]) assert r.exit_code == 0 def test_harmony_forget_help(self, repo: pathlib.Path) -> None: r = runner.invoke(None, ["harmony", "forget", "--help"]) assert r.exit_code == 0 def test_harmony_clear_help(self, repo: pathlib.Path) -> None: r = runner.invoke(None, ["harmony", "clear", "--help"]) assert r.exit_code == 0 def test_harmony_gc_help(self, repo: pathlib.Path) -> None: r = runner.invoke(None, ["harmony", "gc", "--help"]) assert r.exit_code == 0 def test_harmony_policy_add_help(self, repo: pathlib.Path) -> None: r = runner.invoke(None, ["harmony", "policy-add", "--help"]) assert r.exit_code == 0 def test_harmony_policy_list_help(self, repo: pathlib.Path) -> None: r = runner.invoke(None, ["harmony", "policy-list", "--help"]) assert r.exit_code == 0 def test_harmony_policy_remove_help(self, repo: pathlib.Path) -> None: r = runner.invoke(None, ["harmony", "policy-remove", "--help"]) assert r.exit_code == 0 def test_harmony_audit_help(self, repo: pathlib.Path) -> None: r = runner.invoke(None, ["harmony", "audit", "--help"]) assert r.exit_code == 0 # =========================================================================== # Tier II — Integration: success paths # =========================================================================== class TestRecordSuccess: """II: muse harmony record — success paths.""" def test_record_json_returns_pattern_id(self, repo: pathlib.Path) -> None: r = runner.invoke(None, [ "harmony", "record", "--path", "track.mid", "--domain", "midi", "--conflict-type", "content", "--ours-id", fake_id("ours"), "--theirs-id", fake_id("theirs"), "--json", ]) assert r.exit_code == 0 data = json.loads(r.output) assert len(data["pattern_id"]) == 71 assert data["already_existed"] is False def test_record_idempotent_sets_already_existed(self, repo: pathlib.Path) -> None: args = [ "harmony", "record", "--path", "track.mid", "--domain", "midi", "--conflict-type", "content", "--ours-id", fake_id("ours"), "--theirs-id", fake_id("theirs"), "--json", ] r1 = runner.invoke(None, args) r2 = runner.invoke(None, args) assert r1.exit_code == 0 assert r2.exit_code == 0 d1 = json.loads(r1.output) d2 = json.loads(r2.output) assert d1["pattern_id"] == d2["pattern_id"] assert d2["already_existed"] is True def test_record_text_output(self, repo: pathlib.Path) -> None: r = runner.invoke(None, [ "harmony", "record", "--path", "bass.mid", "--domain", "midi", "--conflict-type", "structural", "--ours-id", fake_id("o2"), "--theirs-id", fake_id("t2"), ]) assert r.exit_code == 0 assert "bass.mid" in r.output def test_record_with_semantic_fingerprint(self, repo: pathlib.Path) -> None: r = runner.invoke(None, [ "harmony", "record", "--path", "piano.mid", "--domain", "midi", "--conflict-type", "content", "--ours-id", fake_id("ours"), "--theirs-id", fake_id("theirs"), "--semantic-fingerprint", fake_id("custom-semantic"), "--json", ]) assert r.exit_code == 0 data = json.loads(r.output) assert len(data["pattern_id"]) == 71 def test_record_with_description(self, repo: pathlib.Path) -> None: r = runner.invoke(None, [ "harmony", "record", "--path", "lead.mid", "--domain", "midi", "--conflict-type", "content", "--ours-id", fake_id("ours"), "--theirs-id", fake_id("theirs"), "--description", '{"bar": 4, "key": "Gmaj"}', "--json", ]) assert r.exit_code == 0 class TestListSuccess: """II: muse harmony list — success paths.""" def test_list_empty_json(self, repo: pathlib.Path) -> None: r = runner.invoke(None, ["harmony", "list", "--json"]) assert r.exit_code == 0 data = json.loads(r.output) assert data["total"] == 0 assert data["patterns"] == [] def test_list_shows_recorded_pattern(self, repo: pathlib.Path) -> None: pid = _record(repo) r = runner.invoke(None, ["harmony", "list", "--json"]) assert r.exit_code == 0 data = json.loads(r.output) assert data["total"] == 1 assert data["patterns"][0]["pattern_id"] == pid def test_list_pattern_entry_has_required_fields(self, repo: pathlib.Path) -> None: _record(repo) r = runner.invoke(None, ["harmony", "list", "--json"]) entry = json.loads(r.output)["patterns"][0] for field in ("pattern_id", "path", "domain", "conflict_type", "resolution_count", "recorded_at", "recorded_by"): assert field in entry, f"missing field: {field}" def test_list_resolution_count_increments(self, repo: pathlib.Path) -> None: pid = _record(repo) _resolve(repo, pid) r = runner.invoke(None, ["harmony", "list", "--json"]) entry = json.loads(r.output)["patterns"][0] assert entry["resolution_count"] == 1 def test_list_text_output(self, repo: pathlib.Path) -> None: _record(repo) r = runner.invoke(None, ["harmony", "list"]) assert r.exit_code == 0 assert "track.mid" in r.output def test_list_filter_by_domain(self, repo: pathlib.Path) -> None: _record(repo, path="a.mid", domain="midi") _record(repo, path="b.py", domain="code", ours="oa", theirs="tb") r = runner.invoke(None, ["harmony", "list", "--domain", "midi", "--json"]) data = json.loads(r.output) assert data["total"] == 1 assert data["patterns"][0]["domain"] == "midi" def test_list_filter_by_conflict_type(self, repo: pathlib.Path) -> None: _record(repo, path="a.mid", conflict_type="content") _record(repo, path="b.mid", conflict_type="structural", ours="o2", theirs="t2") r = runner.invoke(None, ["harmony", "list", "--conflict-type", "structural", "--json"]) data = json.loads(r.output) assert data["total"] == 1 assert data["patterns"][0]["conflict_type"] == "structural" class TestShowSuccess: """II: muse harmony show — success paths.""" def test_show_pattern_json(self, repo: pathlib.Path) -> None: pid = _record(repo) r = runner.invoke(None, ["harmony", "show", pid, "--json"]) assert r.exit_code == 0 data = json.loads(r.output) assert data["pattern"]["pattern_id"] == pid assert data["resolutions"] == [] def test_show_includes_resolutions(self, repo: pathlib.Path) -> None: pid = _record(repo) rid = _resolve(repo, pid) r = runner.invoke(None, ["harmony", "show", pid, "--json"]) data = json.loads(r.output) assert len(data["resolutions"]) == 1 assert data["resolutions"][0]["resolution_id"] == rid def test_show_resolution_has_required_fields(self, repo: pathlib.Path) -> None: pid = _record(repo) _resolve(repo, pid) r = runner.invoke(None, ["harmony", "show", pid, "--json"]) res = json.loads(r.output)["resolutions"][0] for field in ("resolution_id", "strategy", "confidence", "human_verified", "applied_count", "resolved_by", "resolved_at", "rationale"): assert field in res, f"missing field: {field}" def test_show_text_output(self, repo: pathlib.Path) -> None: pid = _record(repo) r = runner.invoke(None, ["harmony", "show", pid]) assert r.exit_code == 0 assert short_id(pid) in r.output class TestResolveSuccess: """II: muse harmony resolve — success paths.""" def test_resolve_json(self, repo: pathlib.Path) -> None: pid = _record(repo) r = runner.invoke(None, [ "harmony", "resolve", "--pattern-id", pid, "--strategy", "manual", "--outcome-blob", fake_id("outcome"), "--confidence", "0.85", "--rationale", "looks good", "--json", ]) assert r.exit_code == 0 data = json.loads(r.output) assert len(data["resolution_id"]) == 71 assert data["pattern_id"] == pid assert data["already_existed"] is False def test_resolve_idempotent(self, repo: pathlib.Path) -> None: pid = _record(repo) args = [ "harmony", "resolve", "--pattern-id", pid, "--strategy", "manual", "--outcome-blob", fake_id("outcome"), "--confidence", "0.9", "--json", ] r1 = runner.invoke(None, args) r2 = runner.invoke(None, args) assert r1.exit_code == 0 assert r2.exit_code == 0 d1, d2 = json.loads(r1.output), json.loads(r2.output) assert d1["resolution_id"] == d2["resolution_id"] assert d2["already_existed"] is True def test_resolve_with_agent_provenance(self, repo: pathlib.Path) -> None: pid = _record(repo) r = runner.invoke(None, [ "harmony", "resolve", "--pattern-id", pid, "--strategy", "exact-replay", "--outcome-blob", fake_id("out"), "--confidence", "1.0", "--agent-id", "claude-code", "--model-id", "claude-sonnet-4-6", "--json", ]) assert r.exit_code == 0 data = json.loads(r.output) assert len(data["resolution_id"]) == 71 def test_resolve_human_verified(self, repo: pathlib.Path) -> None: pid = _record(repo) r = runner.invoke(None, [ "harmony", "resolve", "--pattern-id", pid, "--strategy", "manual", "--outcome-blob", fake_id("out"), "--confidence", "0.95", "--human-verified", "--json", ]) assert r.exit_code == 0 def test_resolve_text_output(self, repo: pathlib.Path) -> None: pid = _record(repo) r = runner.invoke(None, [ "harmony", "resolve", "--pattern-id", pid, "--strategy", "manual", "--outcome-blob", fake_id("out"), "--confidence", "0.8", ]) assert r.exit_code == 0 assert short_id(pid) in r.output class TestBestSuccess: """II: muse harmony best — success paths.""" def test_best_returns_null_when_no_resolution(self, repo: pathlib.Path) -> None: pid = _record(repo) r = runner.invoke(None, ["harmony", "best", pid, "--json"]) assert r.exit_code == 0 data = json.loads(r.output) assert data["pattern_id"] == pid assert data["resolution"] is None def test_best_returns_highest_quality(self, repo: pathlib.Path) -> None: pid = _record(repo) _resolve(repo, pid, confidence="0.5", outcome="low") _resolve(repo, pid, confidence="0.9", outcome="high") r = runner.invoke(None, ["harmony", "best", pid, "--json"]) data = json.loads(r.output) assert data["resolution"] is not None assert data["resolution"]["confidence"] == pytest.approx(0.9) def test_best_text_output(self, repo: pathlib.Path) -> None: pid = _record(repo) _resolve(repo, pid) r = runner.invoke(None, ["harmony", "best", pid]) assert r.exit_code == 0 assert short_id(pid) in r.output class TestForgetSuccess: """II: muse harmony forget — success paths.""" def test_forget_existing_pattern(self, repo: pathlib.Path) -> None: pid = _record(repo) r = runner.invoke(None, ["harmony", "forget", pid, "--json"]) assert r.exit_code == 0 data = json.loads(r.output) assert data["pattern_id"] == pid assert data["removed"] is True def test_forget_nonexistent_returns_false(self, repo: pathlib.Path) -> None: r = runner.invoke(None, ["harmony", "forget", fake_id("a"), "--json"]) assert r.exit_code == 0 data = json.loads(r.output) assert data["removed"] is False def test_forget_text_output(self, repo: pathlib.Path) -> None: pid = _record(repo) r = runner.invoke(None, ["harmony", "forget", pid]) assert r.exit_code == 0 assert short_id(pid) in r.output class TestClearSuccess: """II: muse harmony clear — success paths.""" def test_clear_empty(self, repo: pathlib.Path) -> None: r = runner.invoke(None, ["harmony", "clear", "--yes", "--json"]) assert r.exit_code == 0 assert json.loads(r.output)["removed"] == 0 def test_clear_removes_all(self, repo: pathlib.Path) -> None: for i in range(3): _record(repo, path=f"f{i}.mid", ours=f"o{i}", theirs=f"t{i}") r = runner.invoke(None, ["harmony", "clear", "--yes", "--json"]) assert r.exit_code == 0 assert json.loads(r.output)["removed"] == 3 def test_clear_text_output(self, repo: pathlib.Path) -> None: _record(repo) r = runner.invoke(None, ["harmony", "clear", "--yes"]) assert r.exit_code == 0 assert "1" in r.output class TestGcSuccess: """II: muse harmony gc — success paths.""" def test_gc_empty_store(self, repo: pathlib.Path) -> None: r = runner.invoke(None, ["harmony", "gc", "--json"]) assert r.exit_code == 0 data = json.loads(r.output) assert data["removed"] == 0 assert "age_days" in data def test_gc_removes_stale_unresolved(self, repo: pathlib.Path) -> None: # Record a pattern and manually backdate it pid = _record(repo) import muse.core.harmony as hm meta_p = hm.pattern_dir(pathlib.Path("."), pid) / "pattern.json" pattern_data = json.loads(meta_p.read_text()) pattern_data["recorded_at"] = "2020-01-01T00:00:00+00:00" meta_p.write_text(json.dumps(pattern_data)) r = runner.invoke(None, ["harmony", "gc", "--age", "1", "--json"]) assert r.exit_code == 0 assert json.loads(r.output)["removed"] == 1 def test_gc_text_output(self, repo: pathlib.Path) -> None: r = runner.invoke(None, ["harmony", "gc"]) assert r.exit_code == 0 class TestPolicyAddSuccess: """II: muse harmony policy-add — success paths.""" def test_policy_add_json(self, repo: pathlib.Path) -> None: r = runner.invoke(None, [ "harmony", "policy-add", "--policy-id", "prefer-ours", "--description", "Always prefer ours for midi", "--scope", "repo", "--action", "prefer-ours", "--json", ]) assert r.exit_code == 0 data = json.loads(r.output) assert data["policy_id"] == "prefer-ours" assert data["action"] == "prefer-ours" assert data["scope"] == "repo" def test_policy_add_with_condition(self, repo: pathlib.Path) -> None: r = runner.invoke(None, [ "harmony", "policy-add", "--policy-id", "midi-content", "--description", "Midi content policy", "--scope", "domain", "--action", "prefer-ours", "--conflict-type", "content", "--domain", "midi", "--path-pattern", "*.mid", "--confidence", "0.95", "--json", ]) assert r.exit_code == 0 data = json.loads(r.output) assert data["policy_id"] == "midi-content" def test_policy_add_text_output(self, repo: pathlib.Path) -> None: r = runner.invoke(None, [ "harmony", "policy-add", "--policy-id", "my-policy", "--description", "Test", "--scope", "workspace", "--action", "escalate", ]) assert r.exit_code == 0 assert "my-policy" in r.output class TestPolicyListSuccess: """II: muse harmony policy-list — success paths.""" def test_policy_list_empty(self, repo: pathlib.Path) -> None: r = runner.invoke(None, ["harmony", "policy-list", "--json"]) assert r.exit_code == 0 data = json.loads(r.output) assert data["total"] == 0 assert data["policies"] == [] def test_policy_list_shows_added(self, repo: pathlib.Path) -> None: runner.invoke(None, [ "harmony", "policy-add", "--policy-id", "p1", "--description", "d", "--scope", "repo", "--action", "prefer-ours", ]) r = runner.invoke(None, ["harmony", "policy-list", "--json"]) data = json.loads(r.output) assert data["total"] == 1 assert data["policies"][0]["policy_id"] == "p1" def test_policy_list_entry_has_required_fields(self, repo: pathlib.Path) -> None: runner.invoke(None, [ "harmony", "policy-add", "--policy-id", "p2", "--description", "desc", "--scope", "repo", "--action", "prefer-ours", ]) r = runner.invoke(None, ["harmony", "policy-list", "--json"]) entry = json.loads(r.output)["policies"][0] for field in ("policy_id", "description", "scope", "action", "confidence", "conflict_type", "domain", "path_pattern", "created_at", "created_by"): assert field in entry, f"missing field: {field}" def test_policy_list_scope_sorted(self, repo: pathlib.Path) -> None: for pid, scope in [("f", "file"), ("w", "workspace"), ("d", "domain"), ("r", "repo")]: runner.invoke(None, [ "harmony", "policy-add", "--policy-id", pid, "--description", "x", "--scope", scope, "--action", "prefer-ours", ]) r = runner.invoke(None, ["harmony", "policy-list", "--json"]) scopes = [p["scope"] for p in json.loads(r.output)["policies"]] assert scopes.index("workspace") < scopes.index("repo") assert scopes.index("repo") < scopes.index("domain") assert scopes.index("domain") < scopes.index("file") class TestPolicyRemoveSuccess: """II: muse harmony policy-remove — success paths.""" def test_policy_remove_existing(self, repo: pathlib.Path) -> None: runner.invoke(None, [ "harmony", "policy-add", "--policy-id", "to-remove", "--description", "x", "--scope", "repo", "--action", "prefer-ours", ]) r = runner.invoke(None, ["harmony", "policy-remove", "to-remove", "--json"]) assert r.exit_code == 0 data = json.loads(r.output) assert data["policy_id"] == "to-remove" assert data["removed"] is True def test_policy_remove_nonexistent(self, repo: pathlib.Path) -> None: r = runner.invoke(None, ["harmony", "policy-remove", "no-such-policy", "--json"]) assert r.exit_code == 0 data = json.loads(r.output) assert data["removed"] is False class TestAuditSuccess: """II: muse harmony audit — success paths.""" def test_audit_empty(self, repo: pathlib.Path) -> None: r = runner.invoke(None, ["harmony", "audit", "--json"]) assert r.exit_code == 0 data = json.loads(r.output) assert data["total"] == 0 assert data["entries"] == [] def test_audit_shows_entries_after_record(self, repo: pathlib.Path) -> None: _record(repo) r = runner.invoke(None, ["harmony", "audit", "--json"]) assert r.exit_code == 0 data = json.loads(r.output) assert data["total"] >= 1 def test_audit_entry_has_required_fields(self, repo: pathlib.Path) -> None: _record(repo) r = runner.invoke(None, ["harmony", "audit", "--json"]) entry = json.loads(r.output)["entries"][0] for field in ("audit_id", "event_type", "pattern_id", "resolution_id", "policy_id", "acted_by", "occurred_at", "metadata"): assert field in entry, f"missing field: {field}" def test_audit_limit(self, repo: pathlib.Path) -> None: for i in range(5): _record(repo, path=f"f{i}.mid", ours=f"o{i}", theirs=f"t{i}") r = runner.invoke(None, ["harmony", "audit", "--limit", "2", "--json"]) data = json.loads(r.output) assert len(data["entries"]) <= 2 # =========================================================================== # Tier III — Integration: error paths # =========================================================================== class TestRecordErrors: """III: muse harmony record — error paths.""" def test_record_missing_path_exits_nonzero(self, repo: pathlib.Path) -> None: r = runner.invoke(None, [ "harmony", "record", "--domain", "midi", "--conflict-type", "content", "--ours-id", fake_id("o"), "--theirs-id", fake_id("t"), ]) assert r.exit_code != 0 def test_record_missing_ours_id_exits_nonzero(self, repo: pathlib.Path) -> None: r = runner.invoke(None, [ "harmony", "record", "--path", "track.mid", "--domain", "midi", "--conflict-type", "content", "--theirs-id", fake_id("t"), ]) assert r.exit_code != 0 def test_record_invalid_ours_id_exits_1(self, repo: pathlib.Path) -> None: r = runner.invoke(None, [ "harmony", "record", "--path", "track.mid", "--domain", "midi", "--conflict-type", "content", "--ours-id", "not-hex", "--theirs-id", fake_id("t"), "--json", ]) assert r.exit_code == 1 def test_record_bad_description_json_exits_1(self, repo: pathlib.Path) -> None: r = runner.invoke(None, [ "harmony", "record", "--path", "track.mid", "--domain", "midi", "--conflict-type", "content", "--ours-id", fake_id("o"), "--theirs-id", fake_id("t"), "--description", "{bad json", "--json", ]) assert r.exit_code == 1 class TestResolveErrors: """III: muse harmony resolve — error paths.""" def test_resolve_missing_pattern_exits_1(self, repo: pathlib.Path) -> None: r = runner.invoke(None, [ "harmony", "resolve", "--pattern-id", "a" * 64, "--strategy", "manual", "--outcome-blob", fake_id("out"), "--confidence", "0.9", "--json", ]) assert r.exit_code == 1 def test_resolve_invalid_pattern_id_exits_1(self, repo: pathlib.Path) -> None: r = runner.invoke(None, [ "harmony", "resolve", "--pattern-id", "not-hex", "--strategy", "manual", "--outcome-blob", fake_id("out"), "--confidence", "0.9", "--json", ]) assert r.exit_code == 1 def test_resolve_confidence_out_of_range_exits_1(self, repo: pathlib.Path) -> None: pid = _record(repo) r = runner.invoke(None, [ "harmony", "resolve", "--pattern-id", pid, "--strategy", "manual", "--outcome-blob", fake_id("out"), "--confidence", "1.5", "--json", ]) assert r.exit_code == 1 def test_resolve_negative_confidence_exits_1(self, repo: pathlib.Path) -> None: pid = _record(repo) r = runner.invoke(None, [ "harmony", "resolve", "--pattern-id", pid, "--strategy", "manual", "--outcome-blob", fake_id("out"), "--confidence", "-0.1", "--json", ]) assert r.exit_code == 1 class TestShowErrors: """III: muse harmony show — error paths.""" def test_show_nonexistent_exits_1(self, repo: pathlib.Path) -> None: r = runner.invoke(None, ["harmony", "show", "a" * 64, "--json"]) assert r.exit_code == 1 def test_show_invalid_id_exits_1(self, repo: pathlib.Path) -> None: r = runner.invoke(None, ["harmony", "show", "bad-id", "--json"]) assert r.exit_code == 1 class TestBestErrors: """III: muse harmony best — error paths.""" def test_best_invalid_id_exits_1(self, repo: pathlib.Path) -> None: r = runner.invoke(None, ["harmony", "best", "bad-id", "--json"]) assert r.exit_code == 1 class TestForgetErrors: """III: muse harmony forget — error paths.""" def test_forget_invalid_id_exits_1(self, repo: pathlib.Path) -> None: r = runner.invoke(None, ["harmony", "forget", "bad-id", "--json"]) assert r.exit_code == 1 class TestGcErrors: """III: muse harmony gc — error paths.""" def test_gc_invalid_age_exits_1(self, repo: pathlib.Path) -> None: r = runner.invoke(None, ["harmony", "gc", "--age", "0", "--json"]) assert r.exit_code == 1 def test_gc_negative_age_exits_1(self, repo: pathlib.Path) -> None: r = runner.invoke(None, ["harmony", "gc", "--age", "-5", "--json"]) assert r.exit_code == 1 class TestPolicyErrors: """III: muse harmony policy-add — error paths.""" def test_policy_add_invalid_id_exits_1(self, repo: pathlib.Path) -> None: r = runner.invoke(None, [ "harmony", "policy-add", "--policy-id", "bad/id", "--description", "x", "--scope", "repo", "--action", "prefer-ours", "--json", ]) assert r.exit_code == 1 def test_policy_remove_invalid_id_exits_1(self, repo: pathlib.Path) -> None: r = runner.invoke(None, ["harmony", "policy-remove", "bad/id", "--json"]) assert r.exit_code == 1 # =========================================================================== # Tier IV — End-to-end lifecycle # =========================================================================== class TestEndToEnd: """IV: Full lifecycle via CLI.""" def test_record_resolve_best_lifecycle(self, repo: pathlib.Path) -> None: # Record pid = _record(repo, path="lifecycle.mid", domain="midi") # Resolve rid = _resolve(repo, pid, strategy="manual", confidence="0.88") # Best r = runner.invoke(None, ["harmony", "best", pid, "--json"]) data = json.loads(r.output) assert data["resolution"]["resolution_id"] == rid assert data["resolution"]["confidence"] == pytest.approx(0.88) # Audit has entries ra = runner.invoke(None, ["harmony", "audit", "--json"]) assert json.loads(ra.output)["total"] >= 2 def test_policy_controls_match_then_remove(self, repo: pathlib.Path) -> None: # Add policy runner.invoke(None, [ "harmony", "policy-add", "--policy-id", "midi-all", "--description", "prefer ours for all midi", "--scope", "domain", "--action", "prefer-ours", "--domain", "midi", "--json", ]) # List confirms it's there rl = runner.invoke(None, ["harmony", "policy-list", "--json"]) assert json.loads(rl.output)["total"] == 1 # Remove runner.invoke(None, ["harmony", "policy-remove", "midi-all"]) # List now empty rl2 = runner.invoke(None, ["harmony", "policy-list", "--json"]) assert json.loads(rl2.output)["total"] == 0 def test_forget_removes_from_list(self, repo: pathlib.Path) -> None: pid = _record(repo) runner.invoke(None, ["harmony", "forget", pid]) r = runner.invoke(None, ["harmony", "list", "--json"]) assert json.loads(r.output)["total"] == 0 def test_clear_empties_store(self, repo: pathlib.Path) -> None: for i in range(3): _record(repo, path=f"g{i}.mid", ours=f"go{i}", theirs=f"gt{i}") runner.invoke(None, ["harmony", "clear", "--yes"]) r = runner.invoke(None, ["harmony", "list", "--json"]) assert json.loads(r.output)["total"] == 0 def test_gc_does_not_remove_resolved_pattern(self, repo: pathlib.Path) -> None: pid = _record(repo) _resolve(repo, pid) # Backdate recorded_at to trigger age threshold import muse.core.harmony as hm meta_p = hm.pattern_dir(pathlib.Path("."), pid) / "pattern.json" d = json.loads(meta_p.read_text()) d["recorded_at"] = "2020-01-01T00:00:00+00:00" meta_p.write_text(json.dumps(d)) r = runner.invoke(None, ["harmony", "gc", "--age", "1", "--json"]) assert json.loads(r.output)["removed"] == 0 rl = runner.invoke(None, ["harmony", "list", "--json"]) assert json.loads(rl.output)["total"] == 1 # =========================================================================== # Tier V — Data integrity # =========================================================================== class TestDataIntegrity: """V: JSON schemas always fully populated; round-trips correct.""" def test_list_entry_fields_always_present(self, repo: pathlib.Path) -> None: _record(repo) r = runner.invoke(None, ["harmony", "list", "--json"]) entry = json.loads(r.output)["patterns"][0] # resolution_count must be 0, not absent assert entry["resolution_count"] == 0 def test_show_resolutions_field_present_when_empty(self, repo: pathlib.Path) -> None: pid = _record(repo) r = runner.invoke(None, ["harmony", "show", pid, "--json"]) data = json.loads(r.output) assert "resolutions" in data assert isinstance(data["resolutions"], list) def test_best_resolution_is_null_not_missing(self, repo: pathlib.Path) -> None: pid = _record(repo) r = runner.invoke(None, ["harmony", "best", pid, "--json"]) data = json.loads(r.output) assert "resolution" in data assert data["resolution"] is None def test_gc_json_always_has_age_days(self, repo: pathlib.Path) -> None: r = runner.invoke(None, ["harmony", "gc", "--json"]) data = json.loads(r.output) assert "age_days" in data assert isinstance(data["age_days"], int) def test_policy_list_null_conditions_present(self, repo: pathlib.Path) -> None: runner.invoke(None, [ "harmony", "policy-add", "--policy-id", "no-conds", "--description", "x", "--scope", "repo", "--action", "prefer-ours", ]) r = runner.invoke(None, ["harmony", "policy-list", "--json"]) entry = json.loads(r.output)["policies"][0] assert entry["conflict_type"] is None assert entry["domain"] is None assert entry["path_pattern"] is None def test_resolve_confidence_round_trip(self, repo: pathlib.Path) -> None: pid = _record(repo) _resolve(repo, pid, confidence="0.73") r = runner.invoke(None, ["harmony", "best", pid, "--json"]) conf = json.loads(r.output)["resolution"]["confidence"] assert abs(conf - 0.73) < 0.01 # =========================================================================== # Tier VI — Security # =========================================================================== class TestSecurity: """VI: path-traversal IDs and crafted inputs are rejected.""" def test_record_path_traversal_ours_id_rejected(self, repo: pathlib.Path) -> None: r = runner.invoke(None, [ "harmony", "record", "--path", "track.mid", "--domain", "midi", "--conflict-type", "content", "--ours-id", "../../../etc/passwd", "--theirs-id", fake_id("t"), "--json", ]) assert r.exit_code == 1 def test_show_path_traversal_rejected(self, repo: pathlib.Path) -> None: r = runner.invoke(None, ["harmony", "show", "../../malicious", "--json"]) assert r.exit_code == 1 def test_forget_path_traversal_rejected(self, repo: pathlib.Path) -> None: r = runner.invoke(None, ["harmony", "forget", "../../malicious", "--json"]) assert r.exit_code == 1 def test_best_path_traversal_rejected(self, repo: pathlib.Path) -> None: r = runner.invoke(None, ["harmony", "best", "../../malicious", "--json"]) assert r.exit_code == 1 def test_policy_add_slash_in_id_rejected(self, repo: pathlib.Path) -> None: r = runner.invoke(None, [ "harmony", "policy-add", "--policy-id", "malicious/policy", "--description", "x", "--scope", "repo", "--action", "prefer-ours", "--json", ]) assert r.exit_code == 1 def test_policy_remove_slash_in_id_rejected(self, repo: pathlib.Path) -> None: r = runner.invoke(None, ["harmony", "policy-remove", "../etc/passwd", "--json"]) assert r.exit_code == 1 def test_resolve_path_traversal_pattern_id_rejected(self, repo: pathlib.Path) -> None: r = runner.invoke(None, [ "harmony", "resolve", "--pattern-id", "../../malicious", "--strategy", "manual", "--outcome-blob", fake_id("out"), "--confidence", "0.9", "--json", ]) assert r.exit_code == 1 # =========================================================================== # Tier VII — Performance # =========================================================================== class TestPerformance: """VII: each subcommand completes within 300 ms (after warm-up).""" def test_record_under_300ms(self, repo: pathlib.Path) -> None: start = time.monotonic() runner.invoke(None, [ "harmony", "record", "--path", "perf.mid", "--domain", "midi", "--conflict-type", "content", "--ours-id", fake_id("po"), "--theirs-id", fake_id("pt"), ]) elapsed = (time.monotonic() - start) * 1000 assert elapsed < 300, f"record took {elapsed:.0f}ms" def test_list_under_300ms(self, repo: pathlib.Path) -> None: _record(repo) start = time.monotonic() runner.invoke(None, ["harmony", "list", "--json"]) elapsed = (time.monotonic() - start) * 1000 assert elapsed < 300, f"list took {elapsed:.0f}ms" def test_show_under_300ms(self, repo: pathlib.Path) -> None: pid = _record(repo) start = time.monotonic() runner.invoke(None, ["harmony", "show", pid, "--json"]) elapsed = (time.monotonic() - start) * 1000 assert elapsed < 300, f"show took {elapsed:.0f}ms" def test_resolve_under_300ms(self, repo: pathlib.Path) -> None: pid = _record(repo) start = time.monotonic() runner.invoke(None, [ "harmony", "resolve", "--pattern-id", pid, "--strategy", "manual", "--outcome-blob", fake_id("po"), "--confidence", "0.9", ]) elapsed = (time.monotonic() - start) * 1000 assert elapsed < 300, f"resolve took {elapsed:.0f}ms" def test_policy_operations_under_300ms(self, repo: pathlib.Path) -> None: start = time.monotonic() runner.invoke(None, [ "harmony", "policy-add", "--policy-id", "perf-policy", "--description", "x", "--scope", "repo", "--action", "prefer-ours", ]) runner.invoke(None, ["harmony", "policy-list", "--json"]) runner.invoke(None, ["harmony", "policy-remove", "perf-policy"]) elapsed = (time.monotonic() - start) * 1000 assert elapsed < 600, f"policy ops took {elapsed:.0f}ms" def test_gc_under_300ms(self, repo: pathlib.Path) -> None: start = time.monotonic() runner.invoke(None, ["harmony", "gc", "--json"]) elapsed = (time.monotonic() - start) * 1000 assert elapsed < 300, f"gc took {elapsed:.0f}ms"