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