gabriel / muse public
test_harmony_cli_phase4.py python
583 lines 22.1 KB
Raw
sha256:88ac91129873e6a496e9189515aa690eb893ae25d69c8f72af141a2be5068eb3 docs: docstring sprint contract→find-symbol — idiomatic run… Sonnet 4.6 patch 138 days ago
1 """Tests for Phase 4 CLI additions to ``muse harmony``.
2
3 New subcommands:
4 ``muse harmony escalate <pattern_id>`` — record an escalation
5 ``muse harmony escalations`` — list escalations
6 ``muse harmony resolve-escalation <esc_id>`` — close an escalation
7 ``muse harmony engine … --auto-escalate`` — engine auto-records on Tier 4
8
9 Coverage tiers
10 --------------
11 I Unit — TypedDicts for escalate / escalations / resolve-escalation
12 II Success — escalate recorded; escalations list; resolve-escalation closes
13 III Errors — invalid IDs; missing records; resolution-id required
14 IV E2E — engine --auto-escalate → escalations; full lifecycle
15 V Integrity — all JSON fields always present; timestamps present
16 VI Security — path-traversal IDs rejected
17 VII Perf — all subcommands <300 ms
18 """
19 from __future__ import annotations
20
21 import hashlib
22 import json
23 import pathlib
24 import time
25 import typing
26
27 import pytest
28
29 from tests.cli_test_helper import CliRunner
30
31 runner = CliRunner()
32
33
34 # ---------------------------------------------------------------------------
35 # Helpers
36 # ---------------------------------------------------------------------------
37
38
39 def _fake_id(seed: str) -> str:
40 return hashlib.sha256(seed.encode()).hexdigest()
41
42
43 @pytest.fixture()
44 def repo(tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch) -> pathlib.Path:
45 muse_dir = tmp_path / ".muse"
46 muse_dir.mkdir()
47 (muse_dir / "config.toml").write_text('[repo]\nname = "test"\nid = "abc"\n')
48 monkeypatch.chdir(tmp_path)
49 return tmp_path
50
51
52 def _record_pattern(
53 path: str = "track.mid",
54 domain: str = "midi",
55 conflict_type: str = "content",
56 ours: str = "ours",
57 theirs: str = "theirs",
58 ) -> str:
59 r = runner.invoke(None, [
60 "harmony", "record",
61 "--path", path, "--domain", domain,
62 "--conflict-type", conflict_type,
63 "--ours-id", _fake_id(ours),
64 "--theirs-id", _fake_id(theirs),
65 "--json",
66 ])
67 assert r.exit_code == 0, r.output
68 return json.loads(r.output)["pattern_id"]
69
70
71 def _save_resolution(pattern_id: str, confidence: str = "0.9") -> str:
72 r = runner.invoke(None, [
73 "harmony", "resolve",
74 "--pattern-id", pattern_id,
75 "--strategy", "manual",
76 "--outcome-blob", _fake_id("outcome"),
77 "--confidence", confidence,
78 "--json",
79 ])
80 assert r.exit_code == 0, r.output
81 return json.loads(r.output)["resolution_id"]
82
83
84 def _escalate(pattern_id: str, reason: str = "No match found") -> str:
85 r = runner.invoke(None, [
86 "harmony", "escalate", pattern_id,
87 "--reason", reason,
88 "--json",
89 ])
90 assert r.exit_code == 0, r.output
91 return json.loads(r.output)["escalation_id"]
92
93
94 # ===========================================================================
95 # Tier I — Unit: TypedDict schemas
96 # ===========================================================================
97
98
99 class TestTypedDictSchemas:
100 """I: new TypedDicts declare expected keys."""
101
102 def _hints(self, name: str) -> dict[str, typing.Any]:
103 import muse.cli.commands.harmony as h
104 td = getattr(h, name)
105 return typing.get_type_hints(td)
106
107 def test_escalate_json_has_escalation_id(self) -> None:
108 assert "escalation_id" in self._hints("_HarmonyEscalateJson")
109
110 def test_escalate_json_has_pattern_id(self) -> None:
111 assert "pattern_id" in self._hints("_HarmonyEscalateJson")
112
113 def test_escalate_json_has_already_existed(self) -> None:
114 assert "already_existed" in self._hints("_HarmonyEscalateJson")
115
116 def test_escalation_entry_has_status(self) -> None:
117 assert "status" in self._hints("_HarmonyEscalationEntryJson")
118
119 def test_escalation_entry_has_escalation_id(self) -> None:
120 assert "escalation_id" in self._hints("_HarmonyEscalationEntryJson")
121
122 def test_escalation_entry_has_pattern_id(self) -> None:
123 assert "pattern_id" in self._hints("_HarmonyEscalationEntryJson")
124
125 def test_escalations_json_has_total(self) -> None:
126 assert "total" in self._hints("_HarmonyEscalationsJson")
127
128 def test_escalations_json_has_escalations(self) -> None:
129 assert "escalations" in self._hints("_HarmonyEscalationsJson")
130
131 def test_resolve_escalation_json_has_escalation_id(self) -> None:
132 assert "escalation_id" in self._hints("_HarmonyResolveEscalationJson")
133
134 def test_resolve_escalation_json_has_resolved(self) -> None:
135 assert "resolved" in self._hints("_HarmonyResolveEscalationJson")
136
137
138 class TestRegistration:
139 """I: new subcommands are reachable."""
140
141 def test_escalate_help(self, repo: pathlib.Path) -> None:
142 r = runner.invoke(None, ["harmony", "escalate", "--help"])
143 assert r.exit_code == 0
144
145 def test_escalations_help(self, repo: pathlib.Path) -> None:
146 r = runner.invoke(None, ["harmony", "escalations", "--help"])
147 assert r.exit_code == 0
148
149 def test_resolve_escalation_help(self, repo: pathlib.Path) -> None:
150 r = runner.invoke(None, ["harmony", "resolve-escalation", "--help"])
151 assert r.exit_code == 0
152
153
154 # ===========================================================================
155 # Tier II — Integration: success paths
156 # ===========================================================================
157
158
159 class TestEscalateSuccess:
160 """II: muse harmony escalate — success paths."""
161
162 def test_escalate_returns_escalation_id(self, repo: pathlib.Path) -> None:
163 pid = _record_pattern()
164 r = runner.invoke(None, ["harmony", "escalate", pid, "--json"])
165 assert r.exit_code == 0
166 data = json.loads(r.output)
167 assert "escalation_id" in data
168 assert len(data["escalation_id"]) == 64
169
170 def test_escalate_returns_pattern_id(self, repo: pathlib.Path) -> None:
171 pid = _record_pattern()
172 r = runner.invoke(None, ["harmony", "escalate", pid, "--json"])
173 data = json.loads(r.output)
174 assert data["pattern_id"] == pid
175
176 def test_escalate_already_existed_false_on_first(self, repo: pathlib.Path) -> None:
177 pid = _record_pattern()
178 r = runner.invoke(None, ["harmony", "escalate", pid, "--json"])
179 assert json.loads(r.output)["already_existed"] is False
180
181 def test_escalate_idempotent_already_existed_true(self, repo: pathlib.Path) -> None:
182 pid = _record_pattern()
183 runner.invoke(None, ["harmony", "escalate", pid, "--json"])
184 r2 = runner.invoke(None, ["harmony", "escalate", pid, "--json"])
185 assert r2.exit_code == 0
186 assert json.loads(r2.output)["already_existed"] is True
187
188 def test_escalate_same_id_both_calls(self, repo: pathlib.Path) -> None:
189 pid = _record_pattern()
190 r1 = runner.invoke(None, ["harmony", "escalate", pid, "--json"])
191 r2 = runner.invoke(None, ["harmony", "escalate", pid, "--json"])
192 eid1 = json.loads(r1.output)["escalation_id"]
193 eid2 = json.loads(r2.output)["escalation_id"]
194 assert eid1 == eid2
195
196 def test_escalate_custom_reason(self, repo: pathlib.Path) -> None:
197 pid = _record_pattern()
198 r = runner.invoke(None, [
199 "harmony", "escalate", pid,
200 "--reason", "Custom escalation reason",
201 "--json",
202 ])
203 assert r.exit_code == 0
204 assert "escalation_id" in json.loads(r.output)
205
206 def test_escalate_text_output(self, repo: pathlib.Path) -> None:
207 pid = _record_pattern()
208 r = runner.invoke(None, ["harmony", "escalate", pid])
209 assert r.exit_code == 0
210 assert pid[:12] in r.output
211
212 def test_escalate_unknown_pattern_still_works(self, repo: pathlib.Path) -> None:
213 """Escalation can reference a pattern not in the store."""
214 unknown_pid = _fake_id("unknown-pattern")
215 r = runner.invoke(None, ["harmony", "escalate", unknown_pid, "--json"])
216 assert r.exit_code == 0
217 assert json.loads(r.output)["pattern_id"] == unknown_pid
218
219 def test_escalate_with_agent_id(self, repo: pathlib.Path) -> None:
220 pid = _record_pattern()
221 r = runner.invoke(None, [
222 "harmony", "escalate", pid,
223 "--agent-id", "claude-code",
224 "--json",
225 ])
226 assert r.exit_code == 0
227
228
229 class TestEscalationsSuccess:
230 """II: muse harmony escalations — success paths."""
231
232 def test_empty_store_returns_zero(self, repo: pathlib.Path) -> None:
233 r = runner.invoke(None, ["harmony", "escalations", "--json"])
234 assert r.exit_code == 0
235 data = json.loads(r.output)
236 assert data["total"] == 0
237 assert data["escalations"] == []
238
239 def test_lists_all_by_default(self, repo: pathlib.Path) -> None:
240 pid1, pid2 = _record_pattern("a.mid"), _record_pattern("b.mid", ours="b_ours", theirs="b_theirs")
241 _escalate(pid1)
242 _escalate(pid2)
243 r = runner.invoke(None, ["harmony", "escalations", "--json"])
244 assert r.exit_code == 0
245 data = json.loads(r.output)
246 assert data["total"] == 2
247
248 def test_filter_open(self, repo: pathlib.Path) -> None:
249 pid1 = _record_pattern()
250 pid2 = _record_pattern("b.mid", ours="bo", theirs="bt")
251 eid1 = _escalate(pid1)
252 _escalate(pid2)
253 # resolve first
254 res_id = _save_resolution(pid1)
255 runner.invoke(None, [
256 "harmony", "resolve-escalation", eid1,
257 "--resolution-id", res_id,
258 "--json",
259 ])
260 r = runner.invoke(None, ["harmony", "escalations", "--status", "open", "--json"])
261 data = json.loads(r.output)
262 assert data["total"] == 1
263
264 def test_filter_resolved(self, repo: pathlib.Path) -> None:
265 pid = _record_pattern()
266 eid = _escalate(pid)
267 res_id = _save_resolution(pid)
268 runner.invoke(None, [
269 "harmony", "resolve-escalation", eid,
270 "--resolution-id", res_id, "--json",
271 ])
272 r = runner.invoke(None, ["harmony", "escalations", "--status", "resolved", "--json"])
273 assert json.loads(r.output)["total"] == 1
274
275 def test_text_output(self, repo: pathlib.Path) -> None:
276 pid = _record_pattern()
277 _escalate(pid)
278 r = runner.invoke(None, ["harmony", "escalations"])
279 assert r.exit_code == 0
280
281
282 class TestResolveEscalationSuccess:
283 """II: muse harmony resolve-escalation — success paths."""
284
285 def test_resolved_true_when_found(self, repo: pathlib.Path) -> None:
286 pid = _record_pattern()
287 eid = _escalate(pid)
288 res_id = _save_resolution(pid)
289 r = runner.invoke(None, [
290 "harmony", "resolve-escalation", eid,
291 "--resolution-id", res_id, "--json",
292 ])
293 assert r.exit_code == 0
294 data = json.loads(r.output)
295 assert data["resolved"] is True
296
297 def test_resolve_escalation_id_in_response(self, repo: pathlib.Path) -> None:
298 pid = _record_pattern()
299 eid = _escalate(pid)
300 res_id = _save_resolution(pid)
301 r = runner.invoke(None, [
302 "harmony", "resolve-escalation", eid,
303 "--resolution-id", res_id, "--json",
304 ])
305 assert json.loads(r.output)["escalation_id"] == eid
306
307 def test_resolved_false_when_not_found(self, repo: pathlib.Path) -> None:
308 eid = _fake_id("missing-esc")
309 res_id = _fake_id("res")
310 r = runner.invoke(None, [
311 "harmony", "resolve-escalation", eid,
312 "--resolution-id", res_id, "--json",
313 ])
314 assert r.exit_code == 0
315 assert json.loads(r.output)["resolved"] is False
316
317 def test_text_output(self, repo: pathlib.Path) -> None:
318 pid = _record_pattern()
319 eid = _escalate(pid)
320 res_id = _save_resolution(pid)
321 r = runner.invoke(None, [
322 "harmony", "resolve-escalation", eid,
323 "--resolution-id", res_id,
324 ])
325 assert r.exit_code == 0
326
327
328 class TestEngineAutoEscalate:
329 """II: engine --auto-escalate creates escalation record on Tier 4."""
330
331 def test_auto_escalate_creates_record(self, repo: pathlib.Path) -> None:
332 pid = _record_pattern()
333 runner.invoke(None, ["harmony", "engine", pid, "--auto-escalate", "--json"])
334 r = runner.invoke(None, ["harmony", "escalations", "--json"])
335 data = json.loads(r.output)
336 assert data["total"] >= 1
337
338 def test_auto_escalate_only_on_escalated(self, repo: pathlib.Path) -> None:
339 """If the engine resolves (applied), no escalation record is created."""
340 pid = _record_pattern()
341 _save_resolution(pid, confidence="0.95")
342 runner.invoke(None, ["harmony", "engine", pid, "--auto-escalate", "--json"])
343 r = runner.invoke(None, ["harmony", "escalations", "--json"])
344 assert json.loads(r.output)["total"] == 0
345
346 def test_auto_escalate_pattern_id_in_record(self, repo: pathlib.Path) -> None:
347 pid = _record_pattern()
348 runner.invoke(None, ["harmony", "engine", pid, "--auto-escalate"])
349 r = runner.invoke(None, ["harmony", "escalations", "--json"])
350 recs = json.loads(r.output)["escalations"]
351 assert any(e["pattern_id"] == pid for e in recs)
352
353 def test_engine_without_flag_no_escalation_record(self, repo: pathlib.Path) -> None:
354 """Without --auto-escalate, escalate tier writes audit but NOT a record."""
355 pid = _record_pattern()
356 runner.invoke(None, ["harmony", "engine", pid, "--json"])
357 r = runner.invoke(None, ["harmony", "escalations", "--json"])
358 assert json.loads(r.output)["total"] == 0
359
360
361 # ===========================================================================
362 # Tier III — Error paths
363 # ===========================================================================
364
365
366 class TestEscalateErrors:
367 """III: muse harmony escalate — error paths."""
368
369 def test_invalid_id_exits_1(self, repo: pathlib.Path) -> None:
370 r = runner.invoke(None, ["harmony", "escalate", "bad-id", "--json"])
371 assert r.exit_code == 1
372
373 def test_traversal_id_exits_1(self, repo: pathlib.Path) -> None:
374 r = runner.invoke(None, ["harmony", "escalate", "../../evil", "--json"])
375 assert r.exit_code == 1
376
377
378 class TestResolveEscalationErrors:
379 """III: muse harmony resolve-escalation — error paths."""
380
381 def test_invalid_escalation_id_exits_1(self, repo: pathlib.Path) -> None:
382 r = runner.invoke(None, [
383 "harmony", "resolve-escalation", "bad-id",
384 "--resolution-id", _fake_id("r"), "--json",
385 ])
386 assert r.exit_code == 1
387
388 def test_invalid_resolution_id_exits_1(self, repo: pathlib.Path) -> None:
389 eid = _fake_id("esc")
390 r = runner.invoke(None, [
391 "harmony", "resolve-escalation", eid,
392 "--resolution-id", "bad-res", "--json",
393 ])
394 assert r.exit_code == 1
395
396 def test_missing_resolution_id_flag_exits_non_zero(self, repo: pathlib.Path) -> None:
397 eid = _fake_id("esc")
398 r = runner.invoke(None, ["harmony", "resolve-escalation", eid, "--json"])
399 assert r.exit_code != 0
400
401
402 # ===========================================================================
403 # Tier IV — End-to-end
404 # ===========================================================================
405
406
407 class TestEndToEnd:
408 """IV: Full lifecycle via CLI."""
409
410 def test_escalate_resolve_escalation_audit(self, repo: pathlib.Path) -> None:
411 pid = _record_pattern()
412 eid = _escalate(pid)
413
414 # Verify escalation is open
415 r = runner.invoke(None, ["harmony", "escalations", "--status", "open", "--json"])
416 assert json.loads(r.output)["total"] == 1
417
418 # Save a resolution
419 res_id = _save_resolution(pid)
420
421 # Resolve the escalation
422 r = runner.invoke(None, [
423 "harmony", "resolve-escalation", eid,
424 "--resolution-id", res_id, "--json",
425 ])
426 assert json.loads(r.output)["resolved"] is True
427
428 # Now zero open escalations
429 r = runner.invoke(None, ["harmony", "escalations", "--status", "open", "--json"])
430 assert json.loads(r.output)["total"] == 0
431
432 # Audit log has escalation_resolved event
433 r = runner.invoke(None, ["harmony", "audit", "--json"])
434 event_types = [e["event_type"] for e in json.loads(r.output)["entries"]]
435 assert "escalation_resolved" in event_types
436
437 def test_engine_auto_escalate_then_resolve(self, repo: pathlib.Path) -> None:
438 pid = _record_pattern()
439 runner.invoke(None, ["harmony", "engine", pid, "--auto-escalate"])
440
441 r = runner.invoke(None, ["harmony", "escalations", "--status", "open", "--json"])
442 open_recs = json.loads(r.output)["escalations"]
443 assert len(open_recs) >= 1
444 eid = open_recs[0]["escalation_id"]
445
446 res_id = _save_resolution(pid)
447 r = runner.invoke(None, [
448 "harmony", "resolve-escalation", eid,
449 "--resolution-id", res_id, "--json",
450 ])
451 assert json.loads(r.output)["resolved"] is True
452
453 def test_multiple_escalate_filter_independently(self, repo: pathlib.Path) -> None:
454 pids = [_record_pattern(f"{i}.mid", ours=f"o{i}", theirs=f"t{i}") for i in range(4)]
455 eids = [_escalate(p) for p in pids]
456
457 # Resolve first two
458 for i, (pid, eid) in enumerate(zip(pids[:2], eids[:2])):
459 res_id = _save_resolution(pid, confidence=f"0.{80 + i}")
460 runner.invoke(None, [
461 "harmony", "resolve-escalation", eid,
462 "--resolution-id", res_id, "--json",
463 ])
464
465 r_open = runner.invoke(None, ["harmony", "escalations", "--status", "open", "--json"])
466 r_resolved = runner.invoke(None, ["harmony", "escalations", "--status", "resolved", "--json"])
467 assert json.loads(r_open.output)["total"] == 2
468 assert json.loads(r_resolved.output)["total"] == 2
469
470
471 # ===========================================================================
472 # Tier V — Data integrity
473 # ===========================================================================
474
475
476 class TestDataIntegrity:
477 """V: All JSON fields always present; types correct."""
478
479 def test_escalate_all_fields_present(self, repo: pathlib.Path) -> None:
480 pid = _record_pattern()
481 r = runner.invoke(None, ["harmony", "escalate", pid, "--json"])
482 data = json.loads(r.output)
483 for field in ("escalation_id", "pattern_id", "already_existed"):
484 assert field in data, f"missing: {field}"
485
486 def test_escalations_entry_all_fields_present(self, repo: pathlib.Path) -> None:
487 pid = _record_pattern()
488 _escalate(pid)
489 r = runner.invoke(None, ["harmony", "escalations", "--json"])
490 entry = json.loads(r.output)["escalations"][0]
491 for field in ("escalation_id", "pattern_id", "reason", "status",
492 "escalated_at", "escalated_by"):
493 assert field in entry, f"missing: {field}"
494
495 def test_escalation_id_is_hex64(self, repo: pathlib.Path) -> None:
496 pid = _record_pattern()
497 r = runner.invoke(None, ["harmony", "escalate", pid, "--json"])
498 eid = json.loads(r.output)["escalation_id"]
499 assert len(eid) == 64
500 assert all(c in "0123456789abcdef" for c in eid)
501
502 def test_resolve_escalation_all_fields_present(self, repo: pathlib.Path) -> None:
503 pid = _record_pattern()
504 eid = _escalate(pid)
505 res_id = _save_resolution(pid)
506 r = runner.invoke(None, [
507 "harmony", "resolve-escalation", eid,
508 "--resolution-id", res_id, "--json",
509 ])
510 data = json.loads(r.output)
511 for field in ("escalation_id", "resolved"):
512 assert field in data, f"missing: {field}"
513
514 def test_escalations_empty_list_not_null(self, repo: pathlib.Path) -> None:
515 r = runner.invoke(None, ["harmony", "escalations", "--json"])
516 data = json.loads(r.output)
517 assert isinstance(data["escalations"], list)
518
519
520 # ===========================================================================
521 # Tier VI — Security
522 # ===========================================================================
523
524
525 class TestSecurity:
526 """VI: Path-traversal IDs rejected at all Phase 4 entry points."""
527
528 def test_escalate_traversal_rejected(self, repo: pathlib.Path) -> None:
529 r = runner.invoke(None, ["harmony", "escalate", "../../evil", "--json"])
530 assert r.exit_code == 1
531
532 def test_escalate_null_byte_rejected(self, repo: pathlib.Path) -> None:
533 r = runner.invoke(None, ["harmony", "escalate", "a" * 63 + "\x00", "--json"])
534 assert r.exit_code == 1
535
536 def test_resolve_escalation_traversal_eid(self, repo: pathlib.Path) -> None:
537 r = runner.invoke(None, [
538 "harmony", "resolve-escalation", "../../evil",
539 "--resolution-id", _fake_id("r"), "--json",
540 ])
541 assert r.exit_code == 1
542
543 def test_resolve_escalation_traversal_res_id(self, repo: pathlib.Path) -> None:
544 eid = _fake_id("esc")
545 r = runner.invoke(None, [
546 "harmony", "resolve-escalation", eid,
547 "--resolution-id", "../../evil", "--json",
548 ])
549 assert r.exit_code == 1
550
551
552 # ===========================================================================
553 # Tier VII — Performance
554 # ===========================================================================
555
556
557 class TestPerformance:
558 """VII: all Phase 4 subcommands <300 ms."""
559
560 def test_escalate_under_300ms(self, repo: pathlib.Path) -> None:
561 pid = _record_pattern()
562 start = time.monotonic()
563 runner.invoke(None, ["harmony", "escalate", pid, "--json"])
564 elapsed = (time.monotonic() - start) * 1000
565 assert elapsed < 300, f"escalate took {elapsed:.0f}ms"
566
567 def test_escalations_under_300ms(self, repo: pathlib.Path) -> None:
568 start = time.monotonic()
569 runner.invoke(None, ["harmony", "escalations", "--json"])
570 elapsed = (time.monotonic() - start) * 1000
571 assert elapsed < 300, f"escalations took {elapsed:.0f}ms"
572
573 def test_resolve_escalation_under_300ms(self, repo: pathlib.Path) -> None:
574 pid = _record_pattern()
575 eid = _escalate(pid)
576 res_id = _save_resolution(pid)
577 start = time.monotonic()
578 runner.invoke(None, [
579 "harmony", "resolve-escalation", eid,
580 "--resolution-id", res_id, "--json",
581 ])
582 elapsed = (time.monotonic() - start) * 1000
583 assert elapsed < 300, f"resolve-escalation 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