gabriel / muse public
test_harmony_cli_phase3.py python
492 lines 18.6 KB
Raw
sha256:b636f72dcba9e190afb980bece906fa5b717fbde014b76ef023df8cb96e01eb9 docs: expand cache plan with all seven testing tiers and do… Sonnet 4.6 132 days ago
1 """Tests for Phase 3 CLI additions to ``muse harmony``.
2
3 New subcommands:
4 ``muse harmony engine <pattern_id>`` — run the three-tier resolution engine
5 ``muse harmony similar <pattern_id>`` — find semantically similar patterns
6
7 Coverage tiers
8 --------------
9 I Unit — TypedDict schemas for engine + similar JSON output
10 II Success — engine applied/proposed/escalated; similar with matches
11 III Errors — invalid IDs; pattern not found
12 IV E2E — full policy → engine → audit lifecycle via CLI
13 V Integrity — all JSON fields always present; confidence in range
14 VI Security — path-traversal IDs rejected
15 VII Perf — both subcommands <300 ms
16 """
17 from __future__ import annotations
18 from collections.abc import Mapping
19
20 from muse.core._types import fake_id
21 import json
22 import pathlib
23 import time
24 import typing
25
26 import pytest
27
28 from tests.cli_test_helper import CliRunner
29
30 runner = CliRunner()
31
32
33 # ---------------------------------------------------------------------------
34 # Helpers
35 # ---------------------------------------------------------------------------
36
37
38
39 @pytest.fixture()
40 def repo(tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch) -> pathlib.Path:
41 muse_dir = tmp_path / ".muse"
42 muse_dir.mkdir()
43 (muse_dir / "config.toml").write_text('[repo]\nname = "test"\nid = "abc"\n')
44 monkeypatch.chdir(tmp_path)
45 return tmp_path
46
47
48 def _record(
49 path: str = "track.mid",
50 domain: str = "midi",
51 conflict_type: str = "content",
52 ours: str = "ours",
53 theirs: str = "theirs",
54 semantic_fp: str | None = None,
55 ) -> str:
56 args = [
57 "harmony", "record",
58 "--path", path,
59 "--domain", domain,
60 "--conflict-type", conflict_type,
61 "--ours-id", fake_id(ours),
62 "--theirs-id", fake_id(theirs),
63 "--json",
64 ]
65 if semantic_fp is not None:
66 args += ["--semantic-fingerprint", semantic_fp]
67 r = runner.invoke(None, args)
68 assert r.exit_code == 0, r.output
69 return json.loads(r.output)["pattern_id"]
70
71
72 def _resolve(
73 pattern_id: str,
74 confidence: str = "0.9",
75 strategy: str = "manual",
76 outcome: str = "outcome",
77 ) -> str:
78 r = runner.invoke(None, [
79 "harmony", "resolve",
80 "--pattern-id", pattern_id,
81 "--strategy", strategy,
82 "--outcome-blob", fake_id(outcome),
83 "--confidence", confidence,
84 "--json",
85 ])
86 assert r.exit_code == 0, r.output
87 return json.loads(r.output)["resolution_id"]
88
89
90 def _add_policy(
91 policy_id: str = "auto-policy",
92 scope: str = "repo",
93 action: str = "prefer-ours",
94 confidence: str = "0.95",
95 domain: str | None = None,
96 ) -> None:
97 args = [
98 "harmony", "policy-add",
99 "--policy-id", policy_id,
100 "--description", "Test policy",
101 "--scope", scope,
102 "--action", action,
103 "--confidence", confidence,
104 ]
105 if domain:
106 args += ["--domain", domain]
107 runner.invoke(None, args)
108
109
110 # ===========================================================================
111 # Tier I — Unit: TypedDict schemas
112 # ===========================================================================
113
114
115 class TestTypedDictSchemas:
116 """I: Engine and similar TypedDicts declare expected keys."""
117
118 def _hints(self, name: str) -> Mapping[str, object]:
119 import muse.cli.commands.harmony as h
120 td = getattr(h, name)
121 return typing.get_type_hints(td)
122
123 def test_engine_json_has_status(self) -> None:
124 assert "status" in self._hints("_HarmonyEngineJson")
125
126 def test_engine_json_has_pattern_id(self) -> None:
127 assert "pattern_id" in self._hints("_HarmonyEngineJson")
128
129 def test_engine_json_has_proposal(self) -> None:
130 assert "proposal" in self._hints("_HarmonyEngineJson")
131
132 def test_engine_json_has_applied_resolution_id(self) -> None:
133 assert "applied_resolution_id" in self._hints("_HarmonyEngineJson")
134
135 def test_engine_json_has_escalation_reason(self) -> None:
136 assert "escalation_reason" in self._hints("_HarmonyEngineJson")
137
138 def test_similar_json_has_pattern_id(self) -> None:
139 assert "pattern_id" in self._hints("_HarmonySimilarJson")
140
141 def test_similar_json_has_total(self) -> None:
142 assert "total" in self._hints("_HarmonySimilarJson")
143
144 def test_similar_json_has_proposals(self) -> None:
145 assert "proposals" in self._hints("_HarmonySimilarJson")
146
147
148 class TestRegistration:
149 """I: engine and similar subcommands are reachable."""
150
151 def test_engine_help(self, repo: pathlib.Path) -> None:
152 r = runner.invoke(None, ["harmony", "engine", "--help"])
153 assert r.exit_code == 0
154
155 def test_similar_help(self, repo: pathlib.Path) -> None:
156 r = runner.invoke(None, ["harmony", "similar", "--help"])
157 assert r.exit_code == 0
158
159
160 # ===========================================================================
161 # Tier II — Integration: success paths
162 # ===========================================================================
163
164
165 class TestEngineSuccess:
166 """II: muse harmony engine — success paths for all three statuses."""
167
168 def test_engine_escalates_no_policy_no_resolution(self, repo: pathlib.Path) -> None:
169 pid = _record()
170 r = runner.invoke(None, ["harmony", "engine", pid, "--json"])
171 assert r.exit_code == 0
172 data = json.loads(r.output)
173 assert data["status"] == "escalated"
174 assert data["pattern_id"] == pid
175 assert data["escalation_reason"] is not None
176
177 def test_engine_applied_via_policy(self, repo: pathlib.Path) -> None:
178 _add_policy(confidence="0.95", action="prefer-ours")
179 pid = _record()
180 r = runner.invoke(None, ["harmony", "engine", pid, "--json"])
181 assert r.exit_code == 0
182 data = json.loads(r.output)
183 assert data["status"] == "applied"
184 assert data["proposal"] is not None
185 assert data["proposal"]["strategy"] == "policy"
186
187 def test_engine_applied_via_exact_replay(self, repo: pathlib.Path) -> None:
188 pid = _record()
189 _resolve(pid, confidence="0.90")
190 r = runner.invoke(None, ["harmony", "engine", pid, "--json"])
191 assert r.exit_code == 0
192 data = json.loads(r.output)
193 assert data["status"] == "applied"
194 assert data["applied_resolution_id"] is not None
195
196 def test_engine_proposed_low_confidence(self, repo: pathlib.Path) -> None:
197 pid = _record()
198 _resolve(pid, confidence="0.60")
199 r = runner.invoke(None, ["harmony", "engine", pid, "--json"])
200 assert r.exit_code == 0
201 data = json.loads(r.output)
202 assert data["status"] == "proposed"
203 assert data["proposal"] is not None
204 assert data["proposal"]["requires_confirmation"] is True
205
206 def test_engine_text_output(self, repo: pathlib.Path) -> None:
207 pid = _record()
208 r = runner.invoke(None, ["harmony", "engine", pid])
209 assert r.exit_code == 0
210 assert pid[:12] in r.output
211
212 def test_engine_with_custom_threshold(self, repo: pathlib.Path) -> None:
213 """--auto-apply-threshold overrides default."""
214 pid = _record()
215 _resolve(pid, confidence="0.80")
216 # Below default threshold (0.85) → would be proposed. Above 0.75 → applied.
217 r = runner.invoke(None, ["harmony", "engine", pid, "--auto-apply-threshold", "0.75", "--json"])
218 assert r.exit_code == 0
219 data = json.loads(r.output)
220 assert data["status"] == "applied"
221
222 def test_engine_proposed_via_policy_low_confidence(self, repo: pathlib.Path) -> None:
223 _add_policy(policy_id="low-conf", confidence="0.60", action="prefer-ours")
224 pid = _record()
225 r = runner.invoke(None, ["harmony", "engine", pid, "--json"])
226 assert r.exit_code == 0
227 data = json.loads(r.output)
228 assert data["status"] == "proposed"
229
230 def test_engine_escalated_via_escalate_policy(self, repo: pathlib.Path) -> None:
231 _add_policy(policy_id="esc-policy", confidence="1.0", action="escalate")
232 pid = _record()
233 r = runner.invoke(None, ["harmony", "engine", pid, "--json"])
234 assert r.exit_code == 0
235 assert json.loads(r.output)["status"] == "escalated"
236
237
238 class TestSimilarSuccess:
239 """II: muse harmony similar — success paths."""
240
241 def test_similar_empty_when_no_match(self, repo: pathlib.Path) -> None:
242 pid = _record()
243 r = runner.invoke(None, ["harmony", "similar", pid, "--json"])
244 assert r.exit_code == 0
245 data = json.loads(r.output)
246 assert data["pattern_id"] == pid
247 assert data["total"] == 0
248 assert data["proposals"] == []
249
250 def test_similar_finds_shared_semantic_fingerprint(self, repo: pathlib.Path) -> None:
251 shared_fp = fake_id("shared-semantic-cli")
252 source_pid = _record(path="source.mid", ours="so", theirs="st", semantic_fp=shared_fp)
253 target_pid = _record(path="target.mid", ours="to", theirs="tt", semantic_fp=shared_fp)
254
255 # Give source a resolution
256 _resolve(source_pid, confidence="0.88")
257
258 r = runner.invoke(None, ["harmony", "similar", target_pid, "--json"])
259 assert r.exit_code == 0
260 data = json.loads(r.output)
261 assert data["total"] >= 1
262 assert data["proposals"][0]["similar_pattern_id"] == source_pid
263
264 def test_similar_entry_has_required_fields(self, repo: pathlib.Path) -> None:
265 shared_fp = fake_id("shared-fields")
266 source_pid = _record(path="s.mid", ours="so", theirs="st", semantic_fp=shared_fp)
267 target_pid = _record(path="t.mid", ours="to", theirs="tt", semantic_fp=shared_fp)
268 _resolve(source_pid, confidence="0.85")
269
270 r = runner.invoke(None, ["harmony", "similar", target_pid, "--json"])
271 entry = json.loads(r.output)["proposals"][0]
272 for field in ("similar_pattern_id", "similarity", "confidence", "strategy", "rationale"):
273 assert field in entry, f"missing field: {field}"
274
275 def test_similar_text_output(self, repo: pathlib.Path) -> None:
276 shared_fp = fake_id("shared-text")
277 src = _record(path="text-src.mid", ours="so", theirs="st", semantic_fp=shared_fp)
278 tgt = _record(path="text-tgt.mid", ours="to", theirs="tt", semantic_fp=shared_fp)
279 _resolve(src)
280 r = runner.invoke(None, ["harmony", "similar", tgt])
281 assert r.exit_code == 0
282
283 def test_similar_limit(self, repo: pathlib.Path) -> None:
284 shared_fp = fake_id("limit-shared")
285 target_pid = _record(path="lim-tgt.mid", ours="to", theirs="tt", semantic_fp=shared_fp)
286 for i in range(10):
287 src = _record(
288 path=f"lim{i}.mid", ours=f"o{i}", theirs=f"t{i}", semantic_fp=shared_fp
289 )
290 _resolve(src, confidence=f"0.{70+i}", outcome=f"o{i}")
291
292 r = runner.invoke(None, ["harmony", "similar", target_pid, "--limit", "3", "--json"])
293 data = json.loads(r.output)
294 assert len(data["proposals"]) <= 3
295
296
297 # ===========================================================================
298 # Tier III — Error paths
299 # ===========================================================================
300
301
302 class TestEngineErrors:
303 """III: muse harmony engine — error paths."""
304
305 def test_engine_invalid_id_exits_1(self, repo: pathlib.Path) -> None:
306 r = runner.invoke(None, ["harmony", "engine", "bad-id", "--json"])
307 assert r.exit_code == 1
308
309 def test_engine_nonexistent_id_exits_0_escalated(self, repo: pathlib.Path) -> None:
310 # Unknown pattern → engine escalates rather than errors
311 r = runner.invoke(None, ["harmony", "engine", fake_id("nonexistent"), "--json"])
312 assert r.exit_code == 0
313 assert json.loads(r.output)["status"] == "escalated"
314
315 def test_engine_invalid_threshold_exits_1(self, repo: pathlib.Path) -> None:
316 pid = _record()
317 r = runner.invoke(None, [
318 "harmony", "engine", pid,
319 "--auto-apply-threshold", "1.5",
320 "--json",
321 ])
322 assert r.exit_code == 1
323
324 def test_engine_negative_threshold_exits_1(self, repo: pathlib.Path) -> None:
325 pid = _record()
326 r = runner.invoke(None, [
327 "harmony", "engine", pid,
328 "--auto-apply-threshold", "-0.1",
329 "--json",
330 ])
331 assert r.exit_code == 1
332
333
334 class TestSimilarErrors:
335 """III: muse harmony similar — error paths."""
336
337 def test_similar_invalid_id_exits_1(self, repo: pathlib.Path) -> None:
338 r = runner.invoke(None, ["harmony", "similar", "bad-id", "--json"])
339 assert r.exit_code == 1
340
341 def test_similar_nonexistent_exits_0_empty(self, repo: pathlib.Path) -> None:
342 r = runner.invoke(None, ["harmony", "similar", fake_id("nonexistent"), "--json"])
343 assert r.exit_code == 0
344 data = json.loads(r.output)
345 assert data["total"] == 0
346
347
348 # ===========================================================================
349 # Tier IV — End-to-end
350 # ===========================================================================
351
352
353 class TestEndToEnd:
354 """IV: Full lifecycle via CLI layer."""
355
356 def test_policy_engine_audit_trail(self, repo: pathlib.Path) -> None:
357 _add_policy(confidence="0.95", action="prefer-ours")
358 pid = _record()
359
360 runner.invoke(None, ["harmony", "engine", pid])
361
362 r = runner.invoke(None, ["harmony", "audit", "--json"])
363 event_types = [e["event_type"] for e in json.loads(r.output)["entries"]]
364 assert "resolution_applied" in event_types
365
366 def test_escalation_audit_trail(self, repo: pathlib.Path) -> None:
367 pid = _record()
368 runner.invoke(None, ["harmony", "engine", pid])
369
370 r = runner.invoke(None, ["harmony", "audit", "--json"])
371 event_types = [e["event_type"] for e in json.loads(r.output)["entries"]]
372 assert "escalation_recorded" in event_types
373
374 def test_exact_replay_increments_applied_count_via_cli(self, repo: pathlib.Path) -> None:
375 pid = _record()
376 _resolve(pid, confidence="0.90")
377
378 runner.invoke(None, ["harmony", "engine", pid])
379
380 r = runner.invoke(None, ["harmony", "show", pid, "--json"])
381 res = json.loads(r.output)["resolutions"][0]
382 assert res["applied_count"] == 1
383
384 def test_similar_then_engine_workflow(self, repo: pathlib.Path) -> None:
385 """Agent workflow: find_similar to discover candidates, engine to resolve."""
386 shared_fp = fake_id("workflow-shared")
387 src = _record(path="src.mid", ours="so", theirs="st", semantic_fp=shared_fp)
388 tgt = _record(path="tgt.mid", ours="to", theirs="tt", semantic_fp=shared_fp)
389 _resolve(src, confidence="0.88")
390
391 # Step 1: find similar
392 rs = runner.invoke(None, ["harmony", "similar", tgt, "--json"])
393 assert json.loads(rs.output)["total"] >= 1
394
395 # Step 2: run engine (semantic tier fires, requires confirmation)
396 re = runner.invoke(None, ["harmony", "engine", tgt, "--json"])
397 data = json.loads(re.output)
398 assert data["status"] == "proposed"
399 assert data["proposal"]["strategy"] == "semantic-proposal"
400
401
402 # ===========================================================================
403 # Tier V — Data integrity
404 # ===========================================================================
405
406
407 class TestDataIntegrity:
408 """V: All JSON fields always present; types correct."""
409
410 def test_engine_escalated_fields_all_present(self, repo: pathlib.Path) -> None:
411 pid = _record()
412 r = runner.invoke(None, ["harmony", "engine", pid, "--json"])
413 data = json.loads(r.output)
414 for field in ("status", "pattern_id", "proposal", "applied_resolution_id",
415 "escalation_reason"):
416 assert field in data, f"missing field: {field}"
417
418 def test_engine_applied_applied_resolution_id_is_hex64(self, repo: pathlib.Path) -> None:
419 pid = _record()
420 _resolve(pid, confidence="0.90")
421 r = runner.invoke(None, ["harmony", "engine", pid, "--json"])
422 rid = json.loads(r.output)["applied_resolution_id"]
423 assert rid is not None
424 assert len(rid) == 71
425
426 def test_similar_empty_proposals_is_list_not_null(self, repo: pathlib.Path) -> None:
427 pid = _record()
428 r = runner.invoke(None, ["harmony", "similar", pid, "--json"])
429 data = json.loads(r.output)
430 assert isinstance(data["proposals"], list)
431
432 def test_engine_proposed_proposal_confidence_in_range(self, repo: pathlib.Path) -> None:
433 pid = _record()
434 _resolve(pid, confidence="0.60")
435 r = runner.invoke(None, ["harmony", "engine", pid, "--json"])
436 prop = json.loads(r.output)["proposal"]
437 assert prop is not None
438 assert 0.0 <= prop["confidence"] <= 1.0
439
440 def test_similar_similarity_in_range(self, repo: pathlib.Path) -> None:
441 shared_fp = fake_id("range-check")
442 src = _record(path="rc-src.mid", ours="so", theirs="st", semantic_fp=shared_fp)
443 tgt = _record(path="rc-tgt.mid", ours="to", theirs="tt", semantic_fp=shared_fp)
444 _resolve(src)
445
446 r = runner.invoke(None, ["harmony", "similar", tgt, "--json"])
447 for prop in json.loads(r.output)["proposals"]:
448 assert 0.0 <= prop["similarity"] <= 1.0
449
450
451 # ===========================================================================
452 # Tier VI — Security
453 # ===========================================================================
454
455
456 class TestSecurity:
457 """VI: Path-traversal IDs rejected at engine and similar entry points."""
458
459 def test_engine_traversal_rejected(self, repo: pathlib.Path) -> None:
460 r = runner.invoke(None, ["harmony", "engine", "../../evil", "--json"])
461 assert r.exit_code == 1
462
463 def test_similar_traversal_rejected(self, repo: pathlib.Path) -> None:
464 r = runner.invoke(None, ["harmony", "similar", "../../evil", "--json"])
465 assert r.exit_code == 1
466
467 def test_engine_null_byte_rejected(self, repo: pathlib.Path) -> None:
468 r = runner.invoke(None, ["harmony", "engine", "a" * 63 + "\x00", "--json"])
469 assert r.exit_code == 1
470
471
472 # ===========================================================================
473 # Tier VII — Performance
474 # ===========================================================================
475
476
477 class TestPerformance:
478 """VII: engine and similar complete within 300 ms."""
479
480 def test_engine_under_300ms(self, repo: pathlib.Path) -> None:
481 pid = _record()
482 start = time.monotonic()
483 runner.invoke(None, ["harmony", "engine", pid, "--json"])
484 elapsed = (time.monotonic() - start) * 1000
485 assert elapsed < 300, f"engine took {elapsed:.0f}ms"
486
487 def test_similar_under_300ms(self, repo: pathlib.Path) -> None:
488 pid = _record()
489 start = time.monotonic()
490 runner.invoke(None, ["harmony", "similar", pid, "--json"])
491 elapsed = (time.monotonic() - start) * 1000
492 assert elapsed < 300, f"similar took {elapsed:.0f}ms"
File History 2 commits
sha256:b636f72dcba9e190afb980bece906fa5b717fbde014b76ef023df8cb96e01eb9 docs: expand cache plan with all seven testing tiers and do… Sonnet 4.6 132 days ago
sha256:7f9e2ef5286aedad9c1e6011b4c46ca27f39dbdad6e3409357e36b26e46b3b7c docs: docstring sprint for-each-ref→hotspots — idiomatic ru… Sonnet 4.6 patch 138 days ago