gabriel / muse public
test_harmony_cli.py python
1,170 lines 42.5 KB
Raw
sha256:2eaa5d95f9d9383498e76947410a26e5a3ba23d182f339910c424cf88fad412b fix: try fetch/presign before fetch/mpack to avoid Cloudfla… Sonnet 4.6 patch 6 days ago
1 """Tests for ``muse harmony`` CLI — Phase 2.
2
3 Coverage tiers
4 --------------
5 I Unit — TypedDict field presence; subcommand registration
6 II Integration — every subcommand success path (text + JSON)
7 III Integration — every subcommand error path (bad args, not-found)
8 IV End-to-end — full lifecycle through the CLI layer
9 V Data integrity— JSON round-trips; all fields always present
10 VI Security — path-traversal IDs rejected; invalid hex rejected
11 VII Performance — each subcommand completes within 300 ms
12 """
13 from __future__ import annotations
14 from collections.abc import Mapping
15
16 from muse.core.types import fake_id, short_id
17 from muse.core.paths import muse_dir
18 import json
19 import pathlib
20 import time
21 import typing
22
23 import pytest
24
25 from tests.cli_test_helper import CliRunner
26
27 runner = CliRunner()
28
29 # ---------------------------------------------------------------------------
30 # Helpers
31 # ---------------------------------------------------------------------------
32
33
34
35 # ---------------------------------------------------------------------------
36 # Fixtures
37 # ---------------------------------------------------------------------------
38
39
40 @pytest.fixture()
41 def repo(tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch) -> pathlib.Path:
42 """Minimal Muse repo — .muse/config.toml present so require_repo() succeeds."""
43 dot_muse = muse_dir(tmp_path)
44 dot_muse.mkdir()
45 (dot_muse / "config.toml").write_text('[repo]\nname = "test"\nid = "abc123"\n')
46 monkeypatch.chdir(tmp_path)
47 return tmp_path
48
49
50 def _record(
51 repo: pathlib.Path,
52 *,
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 """Invoke ``muse harmony record`` and return the pattern_id."""
60 r = runner.invoke(None, [
61 "harmony", "record",
62 "--path", path,
63 "--domain", domain,
64 "--conflict-type", conflict_type,
65 "--ours-id", fake_id(ours),
66 "--theirs-id", fake_id(theirs),
67 "--json",
68 ])
69 assert r.exit_code == 0, f"record failed: {r.output}"
70 return json.loads(r.output)["pattern_id"]
71
72
73 def _resolve(
74 repo: pathlib.Path,
75 pattern_id: str,
76 *,
77 strategy: str = "manual",
78 confidence: str = "0.9",
79 outcome: str = "outcome",
80 rationale: str = "test",
81 agent_id: str | None = None,
82 ) -> str:
83 """Invoke ``muse harmony resolve`` and return the resolution_id."""
84 args = [
85 "harmony", "resolve",
86 "--pattern-id", pattern_id,
87 "--strategy", strategy,
88 "--outcome-blob", fake_id(outcome),
89 "--confidence", confidence,
90 "--rationale", rationale,
91 "--json",
92 ]
93 if agent_id:
94 args += ["--agent-id", agent_id]
95 r = runner.invoke(None, args)
96 assert r.exit_code == 0, f"resolve failed: {r.output}"
97 return json.loads(r.output)["resolution_id"]
98
99
100 # ===========================================================================
101 # Tier I — Unit: TypedDict schemas and subcommand registration
102 # ===========================================================================
103
104
105 class TestTypedDictSchemas:
106 """I: All TypedDict output schemas declare expected keys."""
107
108 def _hints(self, name: str) -> Mapping[str, object]:
109 import muse.cli.commands.harmony as h
110 td = getattr(h, name)
111 return typing.get_type_hints(td)
112
113 def test_record_json_has_pattern_id(self) -> None:
114 assert "pattern_id" in self._hints("_HarmonyRecordJson")
115
116 def test_record_json_has_already_existed(self) -> None:
117 assert "already_existed" in self._hints("_HarmonyRecordJson")
118
119 def test_list_json_has_total(self) -> None:
120 assert "total" in self._hints("_HarmonyListJson")
121
122 def test_list_json_has_patterns(self) -> None:
123 assert "patterns" in self._hints("_HarmonyListJson")
124
125 def test_show_json_has_pattern(self) -> None:
126 assert "pattern" in self._hints("_HarmonyShowJson")
127
128 def test_show_json_has_resolutions(self) -> None:
129 assert "resolutions" in self._hints("_HarmonyShowJson")
130
131 def test_resolve_json_has_resolution_id(self) -> None:
132 assert "resolution_id" in self._hints("_HarmonyResolveJson")
133
134 def test_resolve_json_has_pattern_id(self) -> None:
135 assert "pattern_id" in self._hints("_HarmonyResolveJson")
136
137 def test_resolve_json_has_already_existed(self) -> None:
138 assert "already_existed" in self._hints("_HarmonyResolveJson")
139
140 def test_best_json_has_pattern_id(self) -> None:
141 assert "pattern_id" in self._hints("_HarmonyBestJson")
142
143 def test_best_json_has_resolution(self) -> None:
144 assert "resolution" in self._hints("_HarmonyBestJson")
145
146 def test_forget_json_has_pattern_id(self) -> None:
147 assert "pattern_id" in self._hints("_HarmonyForgetJson")
148
149 def test_forget_json_has_removed(self) -> None:
150 assert "removed" in self._hints("_HarmonyForgetJson")
151
152 def test_scalar_json_has_removed(self) -> None:
153 assert "removed" in self._hints("_HarmonyScalarJson")
154
155 def test_policy_add_json_has_policy_id(self) -> None:
156 assert "policy_id" in self._hints("_HarmonyPolicyAddJson")
157
158 def test_policy_list_json_has_total(self) -> None:
159 assert "total" in self._hints("_HarmonyPolicyListJson")
160
161 def test_policy_list_json_has_policies(self) -> None:
162 assert "policies" in self._hints("_HarmonyPolicyListJson")
163
164 def test_policy_remove_json_has_policy_id(self) -> None:
165 assert "policy_id" in self._hints("_HarmonyPolicyRemoveJson")
166
167 def test_policy_remove_json_has_removed(self) -> None:
168 assert "removed" in self._hints("_HarmonyPolicyRemoveJson")
169
170 def test_audit_json_has_total(self) -> None:
171 assert "total" in self._hints("_HarmonyAuditJson")
172
173 def test_audit_json_has_entries(self) -> None:
174 assert "entries" in self._hints("_HarmonyAuditJson")
175
176
177 class TestRegistration:
178 """I: harmony is registered in the CLI and subcommands are reachable."""
179
180 def test_harmony_help_exits_0(self, repo: pathlib.Path) -> None:
181 r = runner.invoke(None, ["harmony", "--help"])
182 assert r.exit_code == 0
183
184 def test_harmony_record_help(self, repo: pathlib.Path) -> None:
185 r = runner.invoke(None, ["harmony", "record", "--help"])
186 assert r.exit_code == 0
187
188 def test_harmony_list_help(self, repo: pathlib.Path) -> None:
189 r = runner.invoke(None, ["harmony", "list", "--help"])
190 assert r.exit_code == 0
191
192 def test_harmony_show_help(self, repo: pathlib.Path) -> None:
193 r = runner.invoke(None, ["harmony", "show", "--help"])
194 assert r.exit_code == 0
195
196 def test_harmony_resolve_help(self, repo: pathlib.Path) -> None:
197 r = runner.invoke(None, ["harmony", "resolve", "--help"])
198 assert r.exit_code == 0
199
200 def test_harmony_best_help(self, repo: pathlib.Path) -> None:
201 r = runner.invoke(None, ["harmony", "best", "--help"])
202 assert r.exit_code == 0
203
204 def test_harmony_forget_help(self, repo: pathlib.Path) -> None:
205 r = runner.invoke(None, ["harmony", "forget", "--help"])
206 assert r.exit_code == 0
207
208 def test_harmony_clear_help(self, repo: pathlib.Path) -> None:
209 r = runner.invoke(None, ["harmony", "clear", "--help"])
210 assert r.exit_code == 0
211
212 def test_harmony_gc_help(self, repo: pathlib.Path) -> None:
213 r = runner.invoke(None, ["harmony", "gc", "--help"])
214 assert r.exit_code == 0
215
216 def test_harmony_policy_add_help(self, repo: pathlib.Path) -> None:
217 r = runner.invoke(None, ["harmony", "policy-add", "--help"])
218 assert r.exit_code == 0
219
220 def test_harmony_policy_list_help(self, repo: pathlib.Path) -> None:
221 r = runner.invoke(None, ["harmony", "policy-list", "--help"])
222 assert r.exit_code == 0
223
224 def test_harmony_policy_remove_help(self, repo: pathlib.Path) -> None:
225 r = runner.invoke(None, ["harmony", "policy-remove", "--help"])
226 assert r.exit_code == 0
227
228 def test_harmony_audit_help(self, repo: pathlib.Path) -> None:
229 r = runner.invoke(None, ["harmony", "audit", "--help"])
230 assert r.exit_code == 0
231
232
233 # ===========================================================================
234 # Tier II — Integration: success paths
235 # ===========================================================================
236
237
238 class TestRecordSuccess:
239 """II: muse harmony record — success paths."""
240
241 def test_record_json_returns_pattern_id(self, repo: pathlib.Path) -> None:
242 r = runner.invoke(None, [
243 "harmony", "record",
244 "--path", "track.mid",
245 "--domain", "midi",
246 "--conflict-type", "content",
247 "--ours-id", fake_id("ours"),
248 "--theirs-id", fake_id("theirs"),
249 "--json",
250 ])
251 assert r.exit_code == 0
252 data = json.loads(r.output)
253 assert len(data["pattern_id"]) == 71
254 assert data["already_existed"] is False
255
256 def test_record_idempotent_sets_already_existed(self, repo: pathlib.Path) -> None:
257 args = [
258 "harmony", "record",
259 "--path", "track.mid",
260 "--domain", "midi",
261 "--conflict-type", "content",
262 "--ours-id", fake_id("ours"),
263 "--theirs-id", fake_id("theirs"),
264 "--json",
265 ]
266 r1 = runner.invoke(None, args)
267 r2 = runner.invoke(None, args)
268 assert r1.exit_code == 0
269 assert r2.exit_code == 0
270 d1 = json.loads(r1.output)
271 d2 = json.loads(r2.output)
272 assert d1["pattern_id"] == d2["pattern_id"]
273 assert d2["already_existed"] is True
274
275 def test_record_text_output(self, repo: pathlib.Path) -> None:
276 r = runner.invoke(None, [
277 "harmony", "record",
278 "--path", "bass.mid",
279 "--domain", "midi",
280 "--conflict-type", "structural",
281 "--ours-id", fake_id("o2"),
282 "--theirs-id", fake_id("t2"),
283 ])
284 assert r.exit_code == 0
285 assert "bass.mid" in r.output
286
287 def test_record_with_semantic_fingerprint(self, repo: pathlib.Path) -> None:
288 r = runner.invoke(None, [
289 "harmony", "record",
290 "--path", "piano.mid",
291 "--domain", "midi",
292 "--conflict-type", "content",
293 "--ours-id", fake_id("ours"),
294 "--theirs-id", fake_id("theirs"),
295 "--semantic-fingerprint", fake_id("custom-semantic"),
296 "--json",
297 ])
298 assert r.exit_code == 0
299 data = json.loads(r.output)
300 assert len(data["pattern_id"]) == 71
301
302 def test_record_with_description(self, repo: pathlib.Path) -> None:
303 r = runner.invoke(None, [
304 "harmony", "record",
305 "--path", "lead.mid",
306 "--domain", "midi",
307 "--conflict-type", "content",
308 "--ours-id", fake_id("ours"),
309 "--theirs-id", fake_id("theirs"),
310 "--description", '{"bar": 4, "key": "Gmaj"}',
311 "--json",
312 ])
313 assert r.exit_code == 0
314
315
316 class TestListSuccess:
317 """II: muse harmony list — success paths."""
318
319 def test_list_empty_json(self, repo: pathlib.Path) -> None:
320 r = runner.invoke(None, ["harmony", "list", "--json"])
321 assert r.exit_code == 0
322 data = json.loads(r.output)
323 assert data["total"] == 0
324 assert data["patterns"] == []
325
326 def test_list_shows_recorded_pattern(self, repo: pathlib.Path) -> None:
327 pid = _record(repo)
328 r = runner.invoke(None, ["harmony", "list", "--json"])
329 assert r.exit_code == 0
330 data = json.loads(r.output)
331 assert data["total"] == 1
332 assert data["patterns"][0]["pattern_id"] == pid
333
334 def test_list_pattern_entry_has_required_fields(self, repo: pathlib.Path) -> None:
335 _record(repo)
336 r = runner.invoke(None, ["harmony", "list", "--json"])
337 entry = json.loads(r.output)["patterns"][0]
338 for field in ("pattern_id", "path", "domain", "conflict_type",
339 "resolution_count", "recorded_at", "recorded_by"):
340 assert field in entry, f"missing field: {field}"
341
342 def test_list_resolution_count_increments(self, repo: pathlib.Path) -> None:
343 pid = _record(repo)
344 _resolve(repo, pid)
345 r = runner.invoke(None, ["harmony", "list", "--json"])
346 entry = json.loads(r.output)["patterns"][0]
347 assert entry["resolution_count"] == 1
348
349 def test_list_text_output(self, repo: pathlib.Path) -> None:
350 _record(repo)
351 r = runner.invoke(None, ["harmony", "list"])
352 assert r.exit_code == 0
353 assert "track.mid" in r.output
354
355 def test_list_filter_by_domain(self, repo: pathlib.Path) -> None:
356 _record(repo, path="a.mid", domain="midi")
357 _record(repo, path="b.py", domain="code", ours="oa", theirs="tb")
358 r = runner.invoke(None, ["harmony", "list", "--domain", "midi", "--json"])
359 data = json.loads(r.output)
360 assert data["total"] == 1
361 assert data["patterns"][0]["domain"] == "midi"
362
363 def test_list_filter_by_conflict_type(self, repo: pathlib.Path) -> None:
364 _record(repo, path="a.mid", conflict_type="content")
365 _record(repo, path="b.mid", conflict_type="structural", ours="o2", theirs="t2")
366 r = runner.invoke(None, ["harmony", "list", "--conflict-type", "structural", "--json"])
367 data = json.loads(r.output)
368 assert data["total"] == 1
369 assert data["patterns"][0]["conflict_type"] == "structural"
370
371
372 class TestShowSuccess:
373 """II: muse harmony show — success paths."""
374
375 def test_show_pattern_json(self, repo: pathlib.Path) -> None:
376 pid = _record(repo)
377 r = runner.invoke(None, ["harmony", "show", pid, "--json"])
378 assert r.exit_code == 0
379 data = json.loads(r.output)
380 assert data["pattern"]["pattern_id"] == pid
381 assert data["resolutions"] == []
382
383 def test_show_includes_resolutions(self, repo: pathlib.Path) -> None:
384 pid = _record(repo)
385 rid = _resolve(repo, pid)
386 r = runner.invoke(None, ["harmony", "show", pid, "--json"])
387 data = json.loads(r.output)
388 assert len(data["resolutions"]) == 1
389 assert data["resolutions"][0]["resolution_id"] == rid
390
391 def test_show_resolution_has_required_fields(self, repo: pathlib.Path) -> None:
392 pid = _record(repo)
393 _resolve(repo, pid)
394 r = runner.invoke(None, ["harmony", "show", pid, "--json"])
395 res = json.loads(r.output)["resolutions"][0]
396 for field in ("resolution_id", "strategy", "confidence", "human_verified",
397 "applied_count", "resolved_by", "resolved_at", "rationale"):
398 assert field in res, f"missing field: {field}"
399
400 def test_show_text_output(self, repo: pathlib.Path) -> None:
401 pid = _record(repo)
402 r = runner.invoke(None, ["harmony", "show", pid])
403 assert r.exit_code == 0
404 assert short_id(pid) in r.output
405
406
407 class TestResolveSuccess:
408 """II: muse harmony resolve — success paths."""
409
410 def test_resolve_json(self, repo: pathlib.Path) -> None:
411 pid = _record(repo)
412 r = runner.invoke(None, [
413 "harmony", "resolve",
414 "--pattern-id", pid,
415 "--strategy", "manual",
416 "--outcome-blob", fake_id("outcome"),
417 "--confidence", "0.85",
418 "--rationale", "looks good",
419 "--json",
420 ])
421 assert r.exit_code == 0
422 data = json.loads(r.output)
423 assert len(data["resolution_id"]) == 71
424 assert data["pattern_id"] == pid
425 assert data["already_existed"] is False
426
427 def test_resolve_idempotent(self, repo: pathlib.Path) -> None:
428 pid = _record(repo)
429 args = [
430 "harmony", "resolve",
431 "--pattern-id", pid,
432 "--strategy", "manual",
433 "--outcome-blob", fake_id("outcome"),
434 "--confidence", "0.9",
435 "--json",
436 ]
437 r1 = runner.invoke(None, args)
438 r2 = runner.invoke(None, args)
439 assert r1.exit_code == 0
440 assert r2.exit_code == 0
441 d1, d2 = json.loads(r1.output), json.loads(r2.output)
442 assert d1["resolution_id"] == d2["resolution_id"]
443 assert d2["already_existed"] is True
444
445 def test_resolve_with_agent_provenance(self, repo: pathlib.Path) -> None:
446 pid = _record(repo)
447 r = runner.invoke(None, [
448 "harmony", "resolve",
449 "--pattern-id", pid,
450 "--strategy", "exact-replay",
451 "--outcome-blob", fake_id("out"),
452 "--confidence", "1.0",
453 "--agent-id", "claude-code",
454 "--model-id", "claude-sonnet-4-6",
455 "--json",
456 ])
457 assert r.exit_code == 0
458 data = json.loads(r.output)
459 assert len(data["resolution_id"]) == 71
460
461 def test_resolve_human_verified(self, repo: pathlib.Path) -> None:
462 pid = _record(repo)
463 r = runner.invoke(None, [
464 "harmony", "resolve",
465 "--pattern-id", pid,
466 "--strategy", "manual",
467 "--outcome-blob", fake_id("out"),
468 "--confidence", "0.95",
469 "--human-verified",
470 "--json",
471 ])
472 assert r.exit_code == 0
473
474 def test_resolve_text_output(self, repo: pathlib.Path) -> None:
475 pid = _record(repo)
476 r = runner.invoke(None, [
477 "harmony", "resolve",
478 "--pattern-id", pid,
479 "--strategy", "manual",
480 "--outcome-blob", fake_id("out"),
481 "--confidence", "0.8",
482 ])
483 assert r.exit_code == 0
484 assert short_id(pid) in r.output
485
486
487 class TestBestSuccess:
488 """II: muse harmony best — success paths."""
489
490 def test_best_returns_null_when_no_resolution(self, repo: pathlib.Path) -> None:
491 pid = _record(repo)
492 r = runner.invoke(None, ["harmony", "best", pid, "--json"])
493 assert r.exit_code == 0
494 data = json.loads(r.output)
495 assert data["pattern_id"] == pid
496 assert data["resolution"] is None
497
498 def test_best_returns_highest_quality(self, repo: pathlib.Path) -> None:
499 pid = _record(repo)
500 _resolve(repo, pid, confidence="0.5", outcome="low")
501 _resolve(repo, pid, confidence="0.9", outcome="high")
502 r = runner.invoke(None, ["harmony", "best", pid, "--json"])
503 data = json.loads(r.output)
504 assert data["resolution"] is not None
505 assert data["resolution"]["confidence"] == pytest.approx(0.9)
506
507 def test_best_text_output(self, repo: pathlib.Path) -> None:
508 pid = _record(repo)
509 _resolve(repo, pid)
510 r = runner.invoke(None, ["harmony", "best", pid])
511 assert r.exit_code == 0
512 assert short_id(pid) in r.output
513
514
515 class TestForgetSuccess:
516 """II: muse harmony forget — success paths."""
517
518 def test_forget_existing_pattern(self, repo: pathlib.Path) -> None:
519 pid = _record(repo)
520 r = runner.invoke(None, ["harmony", "forget", pid, "--json"])
521 assert r.exit_code == 0
522 data = json.loads(r.output)
523 assert data["pattern_id"] == pid
524 assert data["removed"] is True
525
526 def test_forget_nonexistent_returns_false(self, repo: pathlib.Path) -> None:
527 r = runner.invoke(None, ["harmony", "forget", fake_id("a"), "--json"])
528 assert r.exit_code == 0
529 data = json.loads(r.output)
530 assert data["removed"] is False
531
532 def test_forget_text_output(self, repo: pathlib.Path) -> None:
533 pid = _record(repo)
534 r = runner.invoke(None, ["harmony", "forget", pid])
535 assert r.exit_code == 0
536 assert short_id(pid) in r.output
537
538
539 class TestClearSuccess:
540 """II: muse harmony clear — success paths."""
541
542 def test_clear_empty(self, repo: pathlib.Path) -> None:
543 r = runner.invoke(None, ["harmony", "clear", "--yes", "--json"])
544 assert r.exit_code == 0
545 assert json.loads(r.output)["removed"] == 0
546
547 def test_clear_removes_all(self, repo: pathlib.Path) -> None:
548 for i in range(3):
549 _record(repo, path=f"f{i}.mid", ours=f"o{i}", theirs=f"t{i}")
550 r = runner.invoke(None, ["harmony", "clear", "--yes", "--json"])
551 assert r.exit_code == 0
552 assert json.loads(r.output)["removed"] == 3
553
554 def test_clear_text_output(self, repo: pathlib.Path) -> None:
555 _record(repo)
556 r = runner.invoke(None, ["harmony", "clear", "--yes"])
557 assert r.exit_code == 0
558 assert "1" in r.output
559
560
561 class TestGcSuccess:
562 """II: muse harmony gc — success paths."""
563
564 def test_gc_empty_store(self, repo: pathlib.Path) -> None:
565 r = runner.invoke(None, ["harmony", "gc", "--json"])
566 assert r.exit_code == 0
567 data = json.loads(r.output)
568 assert data["removed"] == 0
569 assert "age_days" in data
570
571 def test_gc_removes_stale_unresolved(self, repo: pathlib.Path) -> None:
572 # Record a pattern and manually backdate it
573 pid = _record(repo)
574 import muse.core.harmony as hm
575 meta_p = hm.pattern_dir(pathlib.Path("."), pid) / "pattern.json"
576 pattern_data = json.loads(meta_p.read_text())
577 pattern_data["recorded_at"] = "2020-01-01T00:00:00+00:00"
578 meta_p.write_text(json.dumps(pattern_data))
579
580 r = runner.invoke(None, ["harmony", "gc", "--age", "1", "--json"])
581 assert r.exit_code == 0
582 assert json.loads(r.output)["removed"] == 1
583
584 def test_gc_text_output(self, repo: pathlib.Path) -> None:
585 r = runner.invoke(None, ["harmony", "gc"])
586 assert r.exit_code == 0
587
588
589 class TestPolicyAddSuccess:
590 """II: muse harmony policy-add — success paths."""
591
592 def test_policy_add_json(self, repo: pathlib.Path) -> None:
593 r = runner.invoke(None, [
594 "harmony", "policy-add",
595 "--policy-id", "prefer-ours",
596 "--description", "Always prefer ours for midi",
597 "--scope", "repo",
598 "--action", "prefer-ours",
599 "--json",
600 ])
601 assert r.exit_code == 0
602 data = json.loads(r.output)
603 assert data["policy_id"] == "prefer-ours"
604 assert data["action"] == "prefer-ours"
605 assert data["scope"] == "repo"
606
607 def test_policy_add_with_condition(self, repo: pathlib.Path) -> None:
608 r = runner.invoke(None, [
609 "harmony", "policy-add",
610 "--policy-id", "midi-content",
611 "--description", "Midi content policy",
612 "--scope", "domain",
613 "--action", "prefer-ours",
614 "--conflict-type", "content",
615 "--domain", "midi",
616 "--path-pattern", "*.mid",
617 "--confidence", "0.95",
618 "--json",
619 ])
620 assert r.exit_code == 0
621 data = json.loads(r.output)
622 assert data["policy_id"] == "midi-content"
623
624 def test_policy_add_text_output(self, repo: pathlib.Path) -> None:
625 r = runner.invoke(None, [
626 "harmony", "policy-add",
627 "--policy-id", "my-policy",
628 "--description", "Test",
629 "--scope", "workspace",
630 "--action", "escalate",
631 ])
632 assert r.exit_code == 0
633 assert "my-policy" in r.output
634
635
636 class TestPolicyListSuccess:
637 """II: muse harmony policy-list — success paths."""
638
639 def test_policy_list_empty(self, repo: pathlib.Path) -> None:
640 r = runner.invoke(None, ["harmony", "policy-list", "--json"])
641 assert r.exit_code == 0
642 data = json.loads(r.output)
643 assert data["total"] == 0
644 assert data["policies"] == []
645
646 def test_policy_list_shows_added(self, repo: pathlib.Path) -> None:
647 runner.invoke(None, [
648 "harmony", "policy-add",
649 "--policy-id", "p1",
650 "--description", "d",
651 "--scope", "repo",
652 "--action", "prefer-ours",
653 ])
654 r = runner.invoke(None, ["harmony", "policy-list", "--json"])
655 data = json.loads(r.output)
656 assert data["total"] == 1
657 assert data["policies"][0]["policy_id"] == "p1"
658
659 def test_policy_list_entry_has_required_fields(self, repo: pathlib.Path) -> None:
660 runner.invoke(None, [
661 "harmony", "policy-add",
662 "--policy-id", "p2",
663 "--description", "desc",
664 "--scope", "repo",
665 "--action", "prefer-ours",
666 ])
667 r = runner.invoke(None, ["harmony", "policy-list", "--json"])
668 entry = json.loads(r.output)["policies"][0]
669 for field in ("policy_id", "description", "scope", "action", "confidence",
670 "conflict_type", "domain", "path_pattern", "created_at", "created_by"):
671 assert field in entry, f"missing field: {field}"
672
673 def test_policy_list_scope_sorted(self, repo: pathlib.Path) -> None:
674 for pid, scope in [("f", "file"), ("w", "workspace"), ("d", "domain"), ("r", "repo")]:
675 runner.invoke(None, [
676 "harmony", "policy-add",
677 "--policy-id", pid,
678 "--description", "x",
679 "--scope", scope,
680 "--action", "prefer-ours",
681 ])
682 r = runner.invoke(None, ["harmony", "policy-list", "--json"])
683 scopes = [p["scope"] for p in json.loads(r.output)["policies"]]
684 assert scopes.index("workspace") < scopes.index("repo")
685 assert scopes.index("repo") < scopes.index("domain")
686 assert scopes.index("domain") < scopes.index("file")
687
688
689 class TestPolicyRemoveSuccess:
690 """II: muse harmony policy-remove — success paths."""
691
692 def test_policy_remove_existing(self, repo: pathlib.Path) -> None:
693 runner.invoke(None, [
694 "harmony", "policy-add",
695 "--policy-id", "to-remove",
696 "--description", "x",
697 "--scope", "repo",
698 "--action", "prefer-ours",
699 ])
700 r = runner.invoke(None, ["harmony", "policy-remove", "to-remove", "--json"])
701 assert r.exit_code == 0
702 data = json.loads(r.output)
703 assert data["policy_id"] == "to-remove"
704 assert data["removed"] is True
705
706 def test_policy_remove_nonexistent(self, repo: pathlib.Path) -> None:
707 r = runner.invoke(None, ["harmony", "policy-remove", "no-such-policy", "--json"])
708 assert r.exit_code == 0
709 data = json.loads(r.output)
710 assert data["removed"] is False
711
712
713 class TestAuditSuccess:
714 """II: muse harmony audit — success paths."""
715
716 def test_audit_empty(self, repo: pathlib.Path) -> None:
717 r = runner.invoke(None, ["harmony", "audit", "--json"])
718 assert r.exit_code == 0
719 data = json.loads(r.output)
720 assert data["total"] == 0
721 assert data["entries"] == []
722
723 def test_audit_shows_entries_after_record(self, repo: pathlib.Path) -> None:
724 _record(repo)
725 r = runner.invoke(None, ["harmony", "audit", "--json"])
726 assert r.exit_code == 0
727 data = json.loads(r.output)
728 assert data["total"] >= 1
729
730 def test_audit_entry_has_required_fields(self, repo: pathlib.Path) -> None:
731 _record(repo)
732 r = runner.invoke(None, ["harmony", "audit", "--json"])
733 entry = json.loads(r.output)["entries"][0]
734 for field in ("audit_id", "event_type", "pattern_id", "resolution_id",
735 "policy_id", "acted_by", "occurred_at", "metadata"):
736 assert field in entry, f"missing field: {field}"
737
738 def test_audit_limit(self, repo: pathlib.Path) -> None:
739 for i in range(5):
740 _record(repo, path=f"f{i}.mid", ours=f"o{i}", theirs=f"t{i}")
741 r = runner.invoke(None, ["harmony", "audit", "--limit", "2", "--json"])
742 data = json.loads(r.output)
743 assert len(data["entries"]) <= 2
744
745
746 # ===========================================================================
747 # Tier III — Integration: error paths
748 # ===========================================================================
749
750
751 class TestRecordErrors:
752 """III: muse harmony record — error paths."""
753
754 def test_record_missing_path_exits_nonzero(self, repo: pathlib.Path) -> None:
755 r = runner.invoke(None, [
756 "harmony", "record",
757 "--domain", "midi",
758 "--conflict-type", "content",
759 "--ours-id", fake_id("o"),
760 "--theirs-id", fake_id("t"),
761 ])
762 assert r.exit_code != 0
763
764 def test_record_missing_ours_id_exits_nonzero(self, repo: pathlib.Path) -> None:
765 r = runner.invoke(None, [
766 "harmony", "record",
767 "--path", "track.mid",
768 "--domain", "midi",
769 "--conflict-type", "content",
770 "--theirs-id", fake_id("t"),
771 ])
772 assert r.exit_code != 0
773
774 def test_record_invalid_ours_id_exits_1(self, repo: pathlib.Path) -> None:
775 r = runner.invoke(None, [
776 "harmony", "record",
777 "--path", "track.mid",
778 "--domain", "midi",
779 "--conflict-type", "content",
780 "--ours-id", "not-hex",
781 "--theirs-id", fake_id("t"),
782 "--json",
783 ])
784 assert r.exit_code == 1
785
786 def test_record_bad_description_json_exits_1(self, repo: pathlib.Path) -> None:
787 r = runner.invoke(None, [
788 "harmony", "record",
789 "--path", "track.mid",
790 "--domain", "midi",
791 "--conflict-type", "content",
792 "--ours-id", fake_id("o"),
793 "--theirs-id", fake_id("t"),
794 "--description", "{bad json",
795 "--json",
796 ])
797 assert r.exit_code == 1
798
799
800 class TestResolveErrors:
801 """III: muse harmony resolve — error paths."""
802
803 def test_resolve_missing_pattern_exits_1(self, repo: pathlib.Path) -> None:
804 r = runner.invoke(None, [
805 "harmony", "resolve",
806 "--pattern-id", "a" * 64,
807 "--strategy", "manual",
808 "--outcome-blob", fake_id("out"),
809 "--confidence", "0.9",
810 "--json",
811 ])
812 assert r.exit_code == 1
813
814 def test_resolve_invalid_pattern_id_exits_1(self, repo: pathlib.Path) -> None:
815 r = runner.invoke(None, [
816 "harmony", "resolve",
817 "--pattern-id", "not-hex",
818 "--strategy", "manual",
819 "--outcome-blob", fake_id("out"),
820 "--confidence", "0.9",
821 "--json",
822 ])
823 assert r.exit_code == 1
824
825 def test_resolve_confidence_out_of_range_exits_1(self, repo: pathlib.Path) -> None:
826 pid = _record(repo)
827 r = runner.invoke(None, [
828 "harmony", "resolve",
829 "--pattern-id", pid,
830 "--strategy", "manual",
831 "--outcome-blob", fake_id("out"),
832 "--confidence", "1.5",
833 "--json",
834 ])
835 assert r.exit_code == 1
836
837 def test_resolve_negative_confidence_exits_1(self, repo: pathlib.Path) -> None:
838 pid = _record(repo)
839 r = runner.invoke(None, [
840 "harmony", "resolve",
841 "--pattern-id", pid,
842 "--strategy", "manual",
843 "--outcome-blob", fake_id("out"),
844 "--confidence", "-0.1",
845 "--json",
846 ])
847 assert r.exit_code == 1
848
849
850 class TestShowErrors:
851 """III: muse harmony show — error paths."""
852
853 def test_show_nonexistent_exits_1(self, repo: pathlib.Path) -> None:
854 r = runner.invoke(None, ["harmony", "show", "a" * 64, "--json"])
855 assert r.exit_code == 1
856
857 def test_show_invalid_id_exits_1(self, repo: pathlib.Path) -> None:
858 r = runner.invoke(None, ["harmony", "show", "bad-id", "--json"])
859 assert r.exit_code == 1
860
861
862 class TestBestErrors:
863 """III: muse harmony best — error paths."""
864
865 def test_best_invalid_id_exits_1(self, repo: pathlib.Path) -> None:
866 r = runner.invoke(None, ["harmony", "best", "bad-id", "--json"])
867 assert r.exit_code == 1
868
869
870 class TestForgetErrors:
871 """III: muse harmony forget — error paths."""
872
873 def test_forget_invalid_id_exits_1(self, repo: pathlib.Path) -> None:
874 r = runner.invoke(None, ["harmony", "forget", "bad-id", "--json"])
875 assert r.exit_code == 1
876
877
878 class TestGcErrors:
879 """III: muse harmony gc — error paths."""
880
881 def test_gc_invalid_age_exits_1(self, repo: pathlib.Path) -> None:
882 r = runner.invoke(None, ["harmony", "gc", "--age", "0", "--json"])
883 assert r.exit_code == 1
884
885 def test_gc_negative_age_exits_1(self, repo: pathlib.Path) -> None:
886 r = runner.invoke(None, ["harmony", "gc", "--age", "-5", "--json"])
887 assert r.exit_code == 1
888
889
890 class TestPolicyErrors:
891 """III: muse harmony policy-add — error paths."""
892
893 def test_policy_add_invalid_id_exits_1(self, repo: pathlib.Path) -> None:
894 r = runner.invoke(None, [
895 "harmony", "policy-add",
896 "--policy-id", "bad/id",
897 "--description", "x",
898 "--scope", "repo",
899 "--action", "prefer-ours",
900 "--json",
901 ])
902 assert r.exit_code == 1
903
904 def test_policy_remove_invalid_id_exits_1(self, repo: pathlib.Path) -> None:
905 r = runner.invoke(None, ["harmony", "policy-remove", "bad/id", "--json"])
906 assert r.exit_code == 1
907
908
909 # ===========================================================================
910 # Tier IV — End-to-end lifecycle
911 # ===========================================================================
912
913
914 class TestEndToEnd:
915 """IV: Full lifecycle via CLI."""
916
917 def test_record_resolve_best_lifecycle(self, repo: pathlib.Path) -> None:
918 # Record
919 pid = _record(repo, path="lifecycle.mid", domain="midi")
920
921 # Resolve
922 rid = _resolve(repo, pid, strategy="manual", confidence="0.88")
923
924 # Best
925 r = runner.invoke(None, ["harmony", "best", pid, "--json"])
926 data = json.loads(r.output)
927 assert data["resolution"]["resolution_id"] == rid
928 assert data["resolution"]["confidence"] == pytest.approx(0.88)
929
930 # Audit has entries
931 ra = runner.invoke(None, ["harmony", "audit", "--json"])
932 assert json.loads(ra.output)["total"] >= 2
933
934 def test_policy_controls_match_then_remove(self, repo: pathlib.Path) -> None:
935 # Add policy
936 runner.invoke(None, [
937 "harmony", "policy-add",
938 "--policy-id", "midi-all",
939 "--description", "prefer ours for all midi",
940 "--scope", "domain",
941 "--action", "prefer-ours",
942 "--domain", "midi",
943 "--json",
944 ])
945
946 # List confirms it's there
947 rl = runner.invoke(None, ["harmony", "policy-list", "--json"])
948 assert json.loads(rl.output)["total"] == 1
949
950 # Remove
951 runner.invoke(None, ["harmony", "policy-remove", "midi-all"])
952
953 # List now empty
954 rl2 = runner.invoke(None, ["harmony", "policy-list", "--json"])
955 assert json.loads(rl2.output)["total"] == 0
956
957 def test_forget_removes_from_list(self, repo: pathlib.Path) -> None:
958 pid = _record(repo)
959 runner.invoke(None, ["harmony", "forget", pid])
960 r = runner.invoke(None, ["harmony", "list", "--json"])
961 assert json.loads(r.output)["total"] == 0
962
963 def test_clear_empties_store(self, repo: pathlib.Path) -> None:
964 for i in range(3):
965 _record(repo, path=f"g{i}.mid", ours=f"go{i}", theirs=f"gt{i}")
966 runner.invoke(None, ["harmony", "clear", "--yes"])
967 r = runner.invoke(None, ["harmony", "list", "--json"])
968 assert json.loads(r.output)["total"] == 0
969
970 def test_gc_does_not_remove_resolved_pattern(self, repo: pathlib.Path) -> None:
971 pid = _record(repo)
972 _resolve(repo, pid)
973
974 # Backdate recorded_at to trigger age threshold
975 import muse.core.harmony as hm
976 meta_p = hm.pattern_dir(pathlib.Path("."), pid) / "pattern.json"
977 d = json.loads(meta_p.read_text())
978 d["recorded_at"] = "2020-01-01T00:00:00+00:00"
979 meta_p.write_text(json.dumps(d))
980
981 r = runner.invoke(None, ["harmony", "gc", "--age", "1", "--json"])
982 assert json.loads(r.output)["removed"] == 0
983
984 rl = runner.invoke(None, ["harmony", "list", "--json"])
985 assert json.loads(rl.output)["total"] == 1
986
987
988 # ===========================================================================
989 # Tier V — Data integrity
990 # ===========================================================================
991
992
993 class TestDataIntegrity:
994 """V: JSON schemas always fully populated; round-trips correct."""
995
996 def test_list_entry_fields_always_present(self, repo: pathlib.Path) -> None:
997 _record(repo)
998 r = runner.invoke(None, ["harmony", "list", "--json"])
999 entry = json.loads(r.output)["patterns"][0]
1000 # resolution_count must be 0, not absent
1001 assert entry["resolution_count"] == 0
1002
1003 def test_show_resolutions_field_present_when_empty(self, repo: pathlib.Path) -> None:
1004 pid = _record(repo)
1005 r = runner.invoke(None, ["harmony", "show", pid, "--json"])
1006 data = json.loads(r.output)
1007 assert "resolutions" in data
1008 assert isinstance(data["resolutions"], list)
1009
1010 def test_best_resolution_is_null_not_missing(self, repo: pathlib.Path) -> None:
1011 pid = _record(repo)
1012 r = runner.invoke(None, ["harmony", "best", pid, "--json"])
1013 data = json.loads(r.output)
1014 assert "resolution" in data
1015 assert data["resolution"] is None
1016
1017 def test_gc_json_always_has_age_days(self, repo: pathlib.Path) -> None:
1018 r = runner.invoke(None, ["harmony", "gc", "--json"])
1019 data = json.loads(r.output)
1020 assert "age_days" in data
1021 assert isinstance(data["age_days"], int)
1022
1023 def test_policy_list_null_conditions_present(self, repo: pathlib.Path) -> None:
1024 runner.invoke(None, [
1025 "harmony", "policy-add",
1026 "--policy-id", "no-conds",
1027 "--description", "x",
1028 "--scope", "repo",
1029 "--action", "prefer-ours",
1030 ])
1031 r = runner.invoke(None, ["harmony", "policy-list", "--json"])
1032 entry = json.loads(r.output)["policies"][0]
1033 assert entry["conflict_type"] is None
1034 assert entry["domain"] is None
1035 assert entry["path_pattern"] is None
1036
1037 def test_resolve_confidence_round_trip(self, repo: pathlib.Path) -> None:
1038 pid = _record(repo)
1039 _resolve(repo, pid, confidence="0.73")
1040 r = runner.invoke(None, ["harmony", "best", pid, "--json"])
1041 conf = json.loads(r.output)["resolution"]["confidence"]
1042 assert abs(conf - 0.73) < 0.01
1043
1044
1045 # ===========================================================================
1046 # Tier VI — Security
1047 # ===========================================================================
1048
1049
1050 class TestSecurity:
1051 """VI: path-traversal IDs and crafted inputs are rejected."""
1052
1053 def test_record_path_traversal_ours_id_rejected(self, repo: pathlib.Path) -> None:
1054 r = runner.invoke(None, [
1055 "harmony", "record",
1056 "--path", "track.mid",
1057 "--domain", "midi",
1058 "--conflict-type", "content",
1059 "--ours-id", "../../../etc/passwd",
1060 "--theirs-id", fake_id("t"),
1061 "--json",
1062 ])
1063 assert r.exit_code == 1
1064
1065 def test_show_path_traversal_rejected(self, repo: pathlib.Path) -> None:
1066 r = runner.invoke(None, ["harmony", "show", "../../malicious", "--json"])
1067 assert r.exit_code == 1
1068
1069 def test_forget_path_traversal_rejected(self, repo: pathlib.Path) -> None:
1070 r = runner.invoke(None, ["harmony", "forget", "../../malicious", "--json"])
1071 assert r.exit_code == 1
1072
1073 def test_best_path_traversal_rejected(self, repo: pathlib.Path) -> None:
1074 r = runner.invoke(None, ["harmony", "best", "../../malicious", "--json"])
1075 assert r.exit_code == 1
1076
1077 def test_policy_add_slash_in_id_rejected(self, repo: pathlib.Path) -> None:
1078 r = runner.invoke(None, [
1079 "harmony", "policy-add",
1080 "--policy-id", "malicious/policy",
1081 "--description", "x",
1082 "--scope", "repo",
1083 "--action", "prefer-ours",
1084 "--json",
1085 ])
1086 assert r.exit_code == 1
1087
1088 def test_policy_remove_slash_in_id_rejected(self, repo: pathlib.Path) -> None:
1089 r = runner.invoke(None, ["harmony", "policy-remove", "../etc/passwd", "--json"])
1090 assert r.exit_code == 1
1091
1092 def test_resolve_path_traversal_pattern_id_rejected(self, repo: pathlib.Path) -> None:
1093 r = runner.invoke(None, [
1094 "harmony", "resolve",
1095 "--pattern-id", "../../malicious",
1096 "--strategy", "manual",
1097 "--outcome-blob", fake_id("out"),
1098 "--confidence", "0.9",
1099 "--json",
1100 ])
1101 assert r.exit_code == 1
1102
1103
1104 # ===========================================================================
1105 # Tier VII — Performance
1106 # ===========================================================================
1107
1108
1109 class TestPerformance:
1110 """VII: each subcommand completes within 300 ms (after warm-up)."""
1111
1112 def test_record_under_300ms(self, repo: pathlib.Path) -> None:
1113 start = time.monotonic()
1114 runner.invoke(None, [
1115 "harmony", "record",
1116 "--path", "perf.mid",
1117 "--domain", "midi",
1118 "--conflict-type", "content",
1119 "--ours-id", fake_id("po"),
1120 "--theirs-id", fake_id("pt"),
1121 ])
1122 elapsed = (time.monotonic() - start) * 1000
1123 assert elapsed < 300, f"record took {elapsed:.0f}ms"
1124
1125 def test_list_under_300ms(self, repo: pathlib.Path) -> None:
1126 _record(repo)
1127 start = time.monotonic()
1128 runner.invoke(None, ["harmony", "list", "--json"])
1129 elapsed = (time.monotonic() - start) * 1000
1130 assert elapsed < 300, f"list took {elapsed:.0f}ms"
1131
1132 def test_show_under_300ms(self, repo: pathlib.Path) -> None:
1133 pid = _record(repo)
1134 start = time.monotonic()
1135 runner.invoke(None, ["harmony", "show", pid, "--json"])
1136 elapsed = (time.monotonic() - start) * 1000
1137 assert elapsed < 300, f"show took {elapsed:.0f}ms"
1138
1139 def test_resolve_under_300ms(self, repo: pathlib.Path) -> None:
1140 pid = _record(repo)
1141 start = time.monotonic()
1142 runner.invoke(None, [
1143 "harmony", "resolve",
1144 "--pattern-id", pid,
1145 "--strategy", "manual",
1146 "--outcome-blob", fake_id("po"),
1147 "--confidence", "0.9",
1148 ])
1149 elapsed = (time.monotonic() - start) * 1000
1150 assert elapsed < 300, f"resolve took {elapsed:.0f}ms"
1151
1152 def test_policy_operations_under_300ms(self, repo: pathlib.Path) -> None:
1153 start = time.monotonic()
1154 runner.invoke(None, [
1155 "harmony", "policy-add",
1156 "--policy-id", "perf-policy",
1157 "--description", "x",
1158 "--scope", "repo",
1159 "--action", "prefer-ours",
1160 ])
1161 runner.invoke(None, ["harmony", "policy-list", "--json"])
1162 runner.invoke(None, ["harmony", "policy-remove", "perf-policy"])
1163 elapsed = (time.monotonic() - start) * 1000
1164 assert elapsed < 600, f"policy ops took {elapsed:.0f}ms"
1165
1166 def test_gc_under_300ms(self, repo: pathlib.Path) -> None:
1167 start = time.monotonic()
1168 runner.invoke(None, ["harmony", "gc", "--json"])
1169 elapsed = (time.monotonic() - start) * 1000
1170 assert elapsed < 300, f"gc took {elapsed:.0f}ms"
File History 1 commit
sha256:2eaa5d95f9d9383498e76947410a26e5a3ba23d182f339910c424cf88fad412b fix: try fetch/presign before fetch/mpack to avoid Cloudfla… Sonnet 4.6 patch 6 days ago