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