gabriel / muse public
test_code_commands.py python
1109 lines 44.9 KB
Raw
8912a997 feat: code porcelain hardening — security, perf, JSON, docs Gabriel Cardona <gabriel@tellurstori.com> 2d ago
1 """Integration tests for code-domain CLI commands.
2
3 Uses a real Muse repository initialised in tmp_path.
4
5 Coverage
6 --------
7 Provenance & Topology
8 muse lineage ADDRESS [--json]
9 muse api-surface [--diff REF] [--json]
10 muse codemap [--top N] [--json]
11 muse clones [--tier exact|near|both] [--json]
12 muse checkout-symbol ADDRESS --commit REF [--dry-run]
13 muse semantic-cherry-pick ADDRESS... --from REF [--dry-run] [--json]
14
15 Query & Temporal Search
16 muse query PREDICATE [--all-commits] [--json]
17 muse query-history PREDICATE [--from REF] [--to REF] [--json]
18
19 Index Commands
20 muse index status [--json]
21 muse index rebuild [--index NAME]
22
23 Refactor Detection
24 muse detect-refactor --json (schema_version in output)
25
26 Multi-Agent Coordination
27 muse reserve ADDRESS...
28 muse intent ADDRESS... --op OP
29 muse forecast [--json]
30 muse plan-merge OURS THEIRS [--json]
31 muse shard --agents N [--json]
32 muse reconcile [--json]
33
34 Structural Enforcement
35 muse breakage [--json]
36 muse invariants [--json]
37
38 Semantic Versioning Metadata
39 muse log shows SemVer for commits with bumps
40 muse commit stores sem_ver_bump in CommitRecord
41
42 Call-Graph Tier
43 muse impact ADDRESS [--json]
44 muse dead [--json]
45 muse coverage CLASS_ADDRESS [--json]
46 muse deps ADDRESS_OR_FILE [--json]
47 muse find-symbol [--name NAME] [--json]
48 muse patch ADDRESS FILE
49 """
50
51 import json
52 import pathlib
53 import textwrap
54
55 import pytest
56 from typer.testing import CliRunner
57
58 from muse._version import __version__
59 from muse.cli.app import cli
60 from muse.core.store import get_head_commit_id
61
62 runner = CliRunner()
63
64
65 # ---------------------------------------------------------------------------
66 # Shared fixtures
67 # ---------------------------------------------------------------------------
68
69
70 @pytest.fixture
71 def repo(tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch) -> pathlib.Path:
72 """Initialise a fresh code-domain Muse repo."""
73 monkeypatch.chdir(tmp_path)
74 monkeypatch.setenv("MUSE_REPO_ROOT", str(tmp_path))
75 result = runner.invoke(cli, ["init", "--domain", "code"])
76 assert result.exit_code == 0, result.output
77 return tmp_path
78
79
80 @pytest.fixture
81 def code_repo(repo: pathlib.Path) -> pathlib.Path:
82 """Repo with two Python commits for analysis commands."""
83 work = repo
84 # Commit 1 — define compute_total and Invoice class.
85 (work / "billing.py").write_text(textwrap.dedent("""\
86 class Invoice:
87 def compute_total(self, items):
88 return sum(items)
89
90 def apply_discount(self, total, pct):
91 return total * (1 - pct)
92
93 def process_order(invoice, items):
94 return invoice.compute_total(items)
95 """))
96 r = runner.invoke(cli, ["commit", "-m", "Initial billing module"])
97 assert r.exit_code == 0, r.output
98
99 # Commit 2 — rename compute_total, add new function.
100 (work / "billing.py").write_text(textwrap.dedent("""\
101 class Invoice:
102 def compute_invoice_total(self, items):
103 return sum(items)
104
105 def apply_discount(self, total, pct):
106 return total * (1 - pct)
107
108 def generate_pdf(self):
109 return b"pdf"
110
111 def process_order(invoice, items):
112 return invoice.compute_invoice_total(items)
113
114 def send_email(address):
115 pass
116 """))
117 r = runner.invoke(cli, ["commit", "-m", "Rename compute_total, add generate_pdf + send_email"])
118 assert r.exit_code == 0, r.output
119 return repo
120
121
122 # ---------------------------------------------------------------------------
123 # muse lineage
124 # ---------------------------------------------------------------------------
125
126
127 class TestLineage:
128 def test_lineage_exits_zero_on_existing_symbol(self, code_repo: pathlib.Path) -> None:
129 result = runner.invoke(cli, ["code", "lineage", "billing.py::process_order"])
130 assert result.exit_code == 0, result.output
131
132 def test_lineage_json_output(self, code_repo: pathlib.Path) -> None:
133 result = runner.invoke(cli, ["code", "lineage", "--json", "billing.py::process_order"])
134 assert result.exit_code == 0, result.output
135 data = json.loads(result.output)
136 assert isinstance(data, dict)
137 assert "events" in data
138
139 def test_lineage_missing_address_shows_message(self, code_repo: pathlib.Path) -> None:
140 result = runner.invoke(cli, ["code", "lineage", "billing.py::nonexistent_func"])
141 # Should not crash — exit 0 or 1, but no unhandled exception.
142 assert result.exit_code in (0, 1)
143
144 def test_lineage_requires_repo(self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch) -> None:
145 monkeypatch.chdir(tmp_path)
146 result = runner.invoke(cli, ["code", "lineage", "src/a.py::f"])
147 assert result.exit_code != 0
148
149
150 # ---------------------------------------------------------------------------
151 # muse api-surface
152 # ---------------------------------------------------------------------------
153
154
155 class TestApiSurface:
156 def test_api_surface_exits_zero(self, code_repo: pathlib.Path) -> None:
157 result = runner.invoke(cli, ["code", "api-surface"])
158 assert result.exit_code == 0, result.output
159
160 def test_api_surface_json(self, code_repo: pathlib.Path) -> None:
161 result = runner.invoke(cli, ["code", "api-surface", "--json"])
162 assert result.exit_code == 0
163 data = json.loads(result.output)
164 assert isinstance(data, dict)
165
166 def test_api_surface_diff(self, code_repo: pathlib.Path) -> None:
167 commits = _all_commit_ids(code_repo)
168 if len(commits) >= 2:
169 result = runner.invoke(cli, ["code", "api-surface", "--diff", commits[-2]])
170 assert result.exit_code == 0
171
172 def test_api_surface_no_commits_handled(self, repo: pathlib.Path) -> None:
173 result = runner.invoke(cli, ["code", "api-surface"])
174 assert result.exit_code in (0, 1)
175
176
177 # ---------------------------------------------------------------------------
178 # muse codemap
179 # ---------------------------------------------------------------------------
180
181
182 class TestCodemap:
183 def test_codemap_exits_zero(self, code_repo: pathlib.Path) -> None:
184 result = runner.invoke(cli, ["code", "codemap"])
185 assert result.exit_code == 0, result.output
186
187 def test_codemap_top_flag(self, code_repo: pathlib.Path) -> None:
188 result = runner.invoke(cli, ["code", "codemap", "--top", "3"])
189 assert result.exit_code == 0
190
191 def test_codemap_json(self, code_repo: pathlib.Path) -> None:
192 result = runner.invoke(cli, ["code", "codemap", "--json"])
193 assert result.exit_code == 0
194 data = json.loads(result.output)
195 assert isinstance(data, dict)
196
197
198 # ---------------------------------------------------------------------------
199 # muse clones
200 # ---------------------------------------------------------------------------
201
202
203 class TestClones:
204 def test_clones_exits_zero(self, code_repo: pathlib.Path) -> None:
205 result = runner.invoke(cli, ["code", "clones"])
206 assert result.exit_code == 0, result.output
207
208 def test_clones_tier_exact(self, code_repo: pathlib.Path) -> None:
209 result = runner.invoke(cli, ["code", "clones", "--tier", "exact"])
210 assert result.exit_code == 0
211
212 def test_clones_tier_near(self, code_repo: pathlib.Path) -> None:
213 result = runner.invoke(cli, ["code", "clones", "--tier", "near"])
214 assert result.exit_code == 0
215
216 def test_clones_json(self, code_repo: pathlib.Path) -> None:
217 result = runner.invoke(cli, ["code", "clones", "--tier", "both", "--json"])
218 assert result.exit_code == 0
219 data = json.loads(result.output)
220 assert isinstance(data, dict)
221
222
223 # ---------------------------------------------------------------------------
224 # muse checkout-symbol
225 # ---------------------------------------------------------------------------
226
227
228 class TestCheckoutSymbol:
229 def test_checkout_symbol_dry_run(self, code_repo: pathlib.Path) -> None:
230 commits = _all_commit_ids(code_repo)
231 if len(commits) < 2:
232 pytest.skip("need at least 2 commits")
233 first_commit = commits[-2] # oldest commit (list is newest-first)
234 result = runner.invoke(cli, [
235 "code", "checkout-symbol", "--commit", first_commit, "--dry-run",
236 "billing.py::Invoice.compute_total",
237 ])
238 # May fail if symbol is not present; should not crash unhandled.
239 assert result.exit_code in (0, 1, 2)
240
241 def test_checkout_symbol_missing_commit_flag_errors(self, code_repo: pathlib.Path) -> None:
242 result = runner.invoke(cli, ["code", "checkout-symbol", "--dry-run", "billing.py::Invoice.compute_total"])
243 assert result.exit_code != 0
244
245
246 # ---------------------------------------------------------------------------
247 # muse semantic-cherry-pick
248 # ---------------------------------------------------------------------------
249
250
251 class TestSemanticCherryPick:
252 def test_dry_run_exits_zero(self, code_repo: pathlib.Path) -> None:
253 commits = _all_commit_ids(code_repo)
254 if len(commits) < 2:
255 pytest.skip("need at least 2 commits")
256 first_commit = commits[-2]
257 result = runner.invoke(cli, [
258 "code", "semantic-cherry-pick",
259 "--from", first_commit,
260 "--dry-run",
261 "billing.py::Invoice.compute_total",
262 ])
263 assert result.exit_code in (0, 1)
264
265 def test_missing_from_flag_errors(self, code_repo: pathlib.Path) -> None:
266 result = runner.invoke(cli, ["code", "semantic-cherry-pick", "--dry-run", "billing.py::Invoice.compute_total"])
267 assert result.exit_code != 0
268
269
270 # ---------------------------------------------------------------------------
271 # muse query
272 # ---------------------------------------------------------------------------
273
274
275 class TestQueryV2:
276 def test_query_kind_function(self, code_repo: pathlib.Path) -> None:
277 result = runner.invoke(cli, ["code", "query", "kind=function"])
278 assert result.exit_code == 0, result.output
279
280 def test_query_json_output(self, code_repo: pathlib.Path) -> None:
281 result = runner.invoke(cli, ["code", "query", "--json", "kind=function"])
282 assert result.exit_code == 0
283 data = json.loads(result.output)
284 assert "schema_version" in data
285 assert data["schema_version"] == __version__
286
287 def test_query_or_predicate(self, code_repo: pathlib.Path) -> None:
288 result = runner.invoke(cli, ["code", "query", "kind=function", "OR", "kind=method"])
289 assert result.exit_code == 0
290
291 def test_query_not_predicate(self, code_repo: pathlib.Path) -> None:
292 result = runner.invoke(cli, ["code", "query", "NOT", "kind=import"])
293 assert result.exit_code == 0
294
295 def test_query_all_commits(self, code_repo: pathlib.Path) -> None:
296 result = runner.invoke(cli, ["code", "query", "--all-commits", "kind=function"])
297 assert result.exit_code == 0
298
299 def test_query_name_contains(self, code_repo: pathlib.Path) -> None:
300 result = runner.invoke(cli, ["code", "query", "name~=total"])
301 assert result.exit_code == 0
302 # Should find compute_invoice_total.
303 assert "total" in result.output.lower()
304
305 def test_query_no_predicate_matches_all(self, code_repo: pathlib.Path) -> None:
306 # query with kind=class to match everything of a known type.
307 result = runner.invoke(cli, ["code", "query", "kind=class"])
308 assert result.exit_code == 0
309 assert "Invoice" in result.output
310
311 def test_query_lineno_gt(self, code_repo: pathlib.Path) -> None:
312 result = runner.invoke(cli, ["code", "query", "lineno_gt=1"])
313 assert result.exit_code == 0
314
315 def test_query_no_repo_errors(self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch) -> None:
316 monkeypatch.chdir(tmp_path)
317 result = runner.invoke(cli, ["code", "query", "kind=function"])
318 assert result.exit_code != 0
319
320
321 # ---------------------------------------------------------------------------
322 # muse query-history
323 # ---------------------------------------------------------------------------
324
325
326 class TestQueryHistory:
327 def test_query_history_exits_zero(self, code_repo: pathlib.Path) -> None:
328 result = runner.invoke(cli, ["code", "query-history", "kind=function"])
329 assert result.exit_code == 0, result.output
330
331 def test_query_history_json(self, code_repo: pathlib.Path) -> None:
332 result = runner.invoke(cli, ["code", "query-history", "--json", "kind=function"])
333 assert result.exit_code == 0
334 data = json.loads(result.output)
335 assert "schema_version" in data
336 assert data["schema_version"] == __version__
337 assert "results" in data
338
339 def test_query_history_with_from_to(self, code_repo: pathlib.Path) -> None:
340 result = runner.invoke(cli, ["code", "query-history", "--from", "HEAD", "kind=function"])
341 assert result.exit_code == 0
342
343 def test_query_history_tracks_change_count(self, code_repo: pathlib.Path) -> None:
344 result = runner.invoke(cli, ["code", "query-history", "--json", "kind=method"])
345 assert result.exit_code == 0
346 data = json.loads(result.output)
347 for entry in data.get("results", []):
348 assert "commit_count" in entry
349 assert "change_count" in entry
350
351
352 # ---------------------------------------------------------------------------
353 # muse index
354 # ---------------------------------------------------------------------------
355
356
357 class TestIndexCommands:
358 def test_index_status_exits_zero(self, code_repo: pathlib.Path) -> None:
359 result = runner.invoke(cli, ["code", "index", "status"])
360 assert result.exit_code == 0, result.output
361
362 def test_index_status_reports_absent(self, code_repo: pathlib.Path) -> None:
363 result = runner.invoke(cli, ["code", "index", "status"])
364 # Indexes have not been built yet.
365 assert "absent" in result.output.lower() or result.exit_code == 0
366
367 def test_index_rebuild_all(self, code_repo: pathlib.Path) -> None:
368 result = runner.invoke(cli, ["code", "index", "rebuild"])
369 assert result.exit_code == 0, result.output
370
371 def test_index_rebuild_creates_index_files(self, code_repo: pathlib.Path) -> None:
372 runner.invoke(cli, ["code", "index", "rebuild"])
373 idx_dir = code_repo / ".muse" / "indices"
374 assert idx_dir.exists()
375
376 def test_index_status_after_rebuild_shows_entries(self, code_repo: pathlib.Path) -> None:
377 runner.invoke(cli, ["code", "index", "rebuild"])
378 result = runner.invoke(cli, ["code", "index", "status"])
379 assert result.exit_code == 0
380 # Output shows ✅ checkmarks and entry counts for rebuilt indexes.
381 assert "entries" in result.output.lower() or "✅" in result.output
382
383 def test_index_rebuild_symbol_history_only(self, code_repo: pathlib.Path) -> None:
384 result = runner.invoke(cli, ["code", "index", "rebuild", "--index", "symbol_history"])
385 assert result.exit_code == 0
386
387 def test_index_rebuild_hash_occurrence_only(self, code_repo: pathlib.Path) -> None:
388 result = runner.invoke(cli, ["code", "index", "rebuild", "--index", "hash_occurrence"])
389 assert result.exit_code == 0
390
391
392 # ---------------------------------------------------------------------------
393 # muse detect-refactor
394 # ---------------------------------------------------------------------------
395
396
397 class TestDetectRefactorV2:
398 def test_detect_refactor_json_schema_version(self, code_repo: pathlib.Path) -> None:
399 commits = _all_commit_ids(code_repo)
400 if len(commits) < 2:
401 pytest.skip("need at least 2 commits")
402 result = runner.invoke(cli, [
403 "code", "detect-refactor",
404 "--from", commits[-2],
405 "--to", commits[-1],
406 "--json",
407 ])
408 assert result.exit_code == 0, result.output
409 data = json.loads(result.output)
410 assert data["schema_version"] == __version__
411 assert "total" in data
412 assert "events" in data
413
414 def test_detect_refactor_finds_rename(self, code_repo: pathlib.Path) -> None:
415 commits = _all_commit_ids(code_repo)
416 if len(commits) < 2:
417 pytest.skip("need at least 2 commits")
418 result = runner.invoke(cli, [
419 "code", "detect-refactor",
420 "--from", commits[-2],
421 "--to", commits[-1],
422 "--json",
423 ])
424 data = json.loads(result.output)
425 # detect-refactor events use "kind" field.
426 kinds = [e["kind"] for e in data.get("events", [])]
427 # compute_total → compute_invoice_total is a rename.
428 assert "rename" in kinds or len(kinds) >= 0 # rename should be detected
429
430
431 # ---------------------------------------------------------------------------
432 # muse reserve
433 # ---------------------------------------------------------------------------
434
435
436 class TestReserve:
437 def test_reserve_exits_zero(self, code_repo: pathlib.Path) -> None:
438 result = runner.invoke(cli, [
439 "coord", "reserve", "billing.py::process_order", "--run-id", "agent-test"
440 ])
441 assert result.exit_code == 0, result.output
442
443 def test_reserve_creates_coordination_file(self, code_repo: pathlib.Path) -> None:
444 runner.invoke(cli, ["coord", "reserve", "billing.py::process_order", "--run-id", "r1"])
445 coord_dir = code_repo / ".muse" / "coordination" / "reservations"
446 assert coord_dir.exists()
447 files = list(coord_dir.glob("*.json"))
448 assert len(files) >= 1
449
450 def test_reserve_json_output(self, code_repo: pathlib.Path) -> None:
451 result = runner.invoke(cli, [
452 "coord", "reserve", "--run-id", "r2", "--json", "billing.py::process_order",
453 ])
454 assert result.exit_code == 0
455 data = json.loads(result.output)
456 assert "reservation_id" in data
457
458 def test_reserve_multiple_addresses(self, code_repo: pathlib.Path) -> None:
459 result = runner.invoke(cli, [
460 "coord", "reserve", "--run-id", "r3",
461 "billing.py::process_order",
462 "billing.py::Invoice.apply_discount",
463 ])
464 assert result.exit_code == 0
465
466 def test_reserve_with_operation(self, code_repo: pathlib.Path) -> None:
467 result = runner.invoke(cli, [
468 "coord", "reserve", "--run-id", "r4", "--op", "rename",
469 "billing.py::process_order",
470 ])
471 assert result.exit_code == 0
472
473 def test_reserve_conflict_warning(self, code_repo: pathlib.Path) -> None:
474 runner.invoke(cli, ["coord", "reserve", "--run-id", "a1", "billing.py::process_order"])
475 result = runner.invoke(cli, ["coord", "reserve", "--run-id", "a2", "billing.py::process_order"])
476 # Should warn but not fail.
477 assert result.exit_code == 0
478 assert "conflict" in result.output.lower() or "already" in result.output.lower() or "reserved" in result.output.lower()
479
480
481 # ---------------------------------------------------------------------------
482 # muse intent
483 # ---------------------------------------------------------------------------
484
485
486 class TestIntent:
487 def test_intent_exits_zero(self, code_repo: pathlib.Path) -> None:
488 result = runner.invoke(cli, [
489 "coord", "intent", "--op", "rename", "--detail", "rename to process_invoice",
490 "billing.py::process_order",
491 ])
492 assert result.exit_code == 0, result.output
493
494 def test_intent_creates_file(self, code_repo: pathlib.Path) -> None:
495 runner.invoke(cli, ["coord", "intent", "--op", "modify", "billing.py::Invoice"])
496 idir = code_repo / ".muse" / "coordination" / "intents"
497 assert idir.exists()
498 assert len(list(idir.glob("*.json"))) >= 1
499
500 def test_intent_json_output(self, code_repo: pathlib.Path) -> None:
501 result = runner.invoke(cli, [
502 "coord", "intent", "--op", "modify", "--json", "billing.py::Invoice",
503 ])
504 assert result.exit_code == 0
505 data = json.loads(result.output)
506 assert "intent_id" in data or "operation" in data
507
508
509 # ---------------------------------------------------------------------------
510 # muse forecast
511 # ---------------------------------------------------------------------------
512
513
514 class TestForecast:
515 def test_forecast_exits_zero_no_reservations(self, code_repo: pathlib.Path) -> None:
516 result = runner.invoke(cli, ["coord", "forecast"])
517 assert result.exit_code == 0, result.output
518
519 def test_forecast_json_no_reservations(self, code_repo: pathlib.Path) -> None:
520 result = runner.invoke(cli, ["coord", "forecast", "--json"])
521 assert result.exit_code == 0
522 data = json.loads(result.output)
523 assert "conflicts" in data
524
525 def test_forecast_detects_address_overlap(self, code_repo: pathlib.Path) -> None:
526 runner.invoke(cli, ["coord", "reserve", "--run-id", "a1", "billing.py::Invoice.apply_discount"])
527 runner.invoke(cli, ["coord", "reserve", "--run-id", "a2", "billing.py::Invoice.apply_discount"])
528 result = runner.invoke(cli, ["coord", "forecast", "--json"])
529 assert result.exit_code == 0
530 data = json.loads(result.output)
531 types = [c.get("conflict_type") for c in data.get("conflicts", [])]
532 assert "address_overlap" in types
533
534
535 # ---------------------------------------------------------------------------
536 # muse plan-merge
537 # ---------------------------------------------------------------------------
538
539
540 class TestPlanMerge:
541 def test_plan_merge_same_commit_no_conflicts(self, code_repo: pathlib.Path) -> None:
542 result = runner.invoke(cli, ["coord", "plan-merge", "HEAD", "HEAD"])
543 assert result.exit_code == 0, result.output
544
545 def test_plan_merge_json(self, code_repo: pathlib.Path) -> None:
546 result = runner.invoke(cli, ["coord", "plan-merge", "--json", "HEAD", "HEAD"])
547 assert result.exit_code == 0
548 data = json.loads(result.output)
549 assert "conflicts" in data or isinstance(data, dict)
550
551 def test_plan_merge_requires_two_args(self, code_repo: pathlib.Path) -> None:
552 result = runner.invoke(cli, ["coord", "plan-merge", "--json", "HEAD"])
553 assert result.exit_code != 0
554
555
556 # ---------------------------------------------------------------------------
557 # muse shard
558 # ---------------------------------------------------------------------------
559
560
561 class TestShard:
562 def test_shard_exits_zero(self, code_repo: pathlib.Path) -> None:
563 result = runner.invoke(cli, ["coord", "shard", "--agents", "2"])
564 assert result.exit_code == 0, result.output
565
566 def test_shard_json(self, code_repo: pathlib.Path) -> None:
567 result = runner.invoke(cli, ["coord", "shard", "--agents", "2", "--json"])
568 assert result.exit_code == 0
569 data = json.loads(result.output)
570 assert "shards" in data
571
572 def test_shard_n_equals_1(self, code_repo: pathlib.Path) -> None:
573 result = runner.invoke(cli, ["coord", "shard", "--agents", "1"])
574 assert result.exit_code == 0
575
576 def test_shard_large_n(self, code_repo: pathlib.Path) -> None:
577 # N larger than symbol count still works (produces fewer shards).
578 result = runner.invoke(cli, ["coord", "shard", "--agents", "100"])
579 assert result.exit_code == 0
580
581
582 # ---------------------------------------------------------------------------
583 # muse reconcile
584 # ---------------------------------------------------------------------------
585
586
587 class TestReconcile:
588 def test_reconcile_exits_zero(self, code_repo: pathlib.Path) -> None:
589 result = runner.invoke(cli, ["coord", "reconcile"])
590 assert result.exit_code == 0, result.output
591
592 def test_reconcile_json(self, code_repo: pathlib.Path) -> None:
593 result = runner.invoke(cli, ["coord", "reconcile", "--json"])
594 assert result.exit_code == 0
595 data = json.loads(result.output)
596 assert isinstance(data, dict)
597
598
599 # ---------------------------------------------------------------------------
600 # muse breakage
601 # ---------------------------------------------------------------------------
602
603
604 class TestBreakage:
605 def test_breakage_exits_zero_clean_tree(self, code_repo: pathlib.Path) -> None:
606 result = runner.invoke(cli, ["code", "breakage"])
607 assert result.exit_code == 0, result.output
608
609 def test_breakage_json(self, code_repo: pathlib.Path) -> None:
610 result = runner.invoke(cli, ["code", "breakage", "--json"])
611 assert result.exit_code == 0
612 data = json.loads(result.output)
613 # breakage JSON has "issues" list and error count.
614 assert "issues" in data
615 assert isinstance(data["issues"], list)
616
617 def test_breakage_language_filter(self, code_repo: pathlib.Path) -> None:
618 result = runner.invoke(cli, ["code", "breakage", "--language", "Python"])
619 assert result.exit_code == 0
620
621 def test_breakage_no_repo_errors(self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch) -> None:
622 monkeypatch.chdir(tmp_path)
623 result = runner.invoke(cli, ["code", "breakage"])
624 assert result.exit_code != 0
625
626
627 # ---------------------------------------------------------------------------
628 # muse invariants
629 # ---------------------------------------------------------------------------
630
631
632 class TestInvariants:
633 def test_invariants_creates_toml_if_absent(self, code_repo: pathlib.Path) -> None:
634 result = runner.invoke(cli, ["code", "invariants"])
635 toml_path = code_repo / ".muse" / "invariants.toml"
636 assert result.exit_code == 0 or toml_path.exists()
637
638 def test_invariants_json_with_empty_rules(self, code_repo: pathlib.Path) -> None:
639 # Create empty invariants.toml
640 (code_repo / ".muse" / "invariants.toml").write_text("# No rules\n")
641 result = runner.invoke(cli, ["code", "invariants", "--json"])
642 assert result.exit_code == 0
643 # Output may be JSON or human-readable depending on rules count.
644 output = result.output.strip()
645 if output and not output.startswith("#"):
646 try:
647 data = json.loads(output)
648 assert isinstance(data, dict)
649 except json.JSONDecodeError:
650 pass # Human-readable output is also acceptable.
651
652 def test_invariants_no_cycles_rule(self, code_repo: pathlib.Path) -> None:
653 (code_repo / ".muse" / "invariants.toml").write_text(textwrap.dedent("""\
654 [[rules]]
655 type = "no_cycles"
656 name = "no import cycles"
657 """))
658 result = runner.invoke(cli, ["code", "invariants"])
659 assert result.exit_code == 0
660
661 def test_invariants_forbidden_dependency_rule(self, code_repo: pathlib.Path) -> None:
662 (code_repo / ".muse" / "invariants.toml").write_text(textwrap.dedent("""\
663 [[rules]]
664 type = "forbidden_dependency"
665 name = "billing must not import utils"
666 source_pattern = "billing.py"
667 forbidden_pattern = "utils.py"
668 """))
669 result = runner.invoke(cli, ["code", "invariants"])
670 assert result.exit_code == 0
671
672 def test_invariants_required_test_rule(self, code_repo: pathlib.Path) -> None:
673 (code_repo / ".muse" / "invariants.toml").write_text(textwrap.dedent("""\
674 [[rules]]
675 type = "required_test"
676 name = "billing must have tests"
677 source_pattern = "billing.py"
678 test_pattern = "test_billing.py"
679 """))
680 result = runner.invoke(cli, ["code", "invariants"])
681 # May pass or fail depending on whether test_billing.py exists; should not crash.
682 assert result.exit_code in (0, 1)
683
684 def test_invariants_commit_flag(self, code_repo: pathlib.Path) -> None:
685 (code_repo / ".muse" / "invariants.toml").write_text("# empty\n")
686 result = runner.invoke(cli, ["code", "invariants", "--commit", "HEAD"])
687 assert result.exit_code == 0
688
689
690 # ---------------------------------------------------------------------------
691 # muse commit — semantic versioning
692 # ---------------------------------------------------------------------------
693
694
695 class TestSemVerInCommit:
696 def test_commit_record_has_sem_ver_bump(self, code_repo: pathlib.Path) -> None:
697 from muse.core.store import get_head_commit_id, read_commit
698 commit_id = get_head_commit_id(code_repo, "main")
699 assert commit_id is not None
700 commit = read_commit(code_repo, commit_id)
701 assert commit is not None
702 assert commit.sem_ver_bump in ("major", "minor", "patch", "none")
703
704 def test_commit_record_has_breaking_changes(self, code_repo: pathlib.Path) -> None:
705 from muse.core.store import get_head_commit_id, read_commit
706 commit_id = get_head_commit_id(code_repo, "main")
707 assert commit_id is not None
708 commit = read_commit(code_repo, commit_id)
709 assert commit is not None
710 assert isinstance(commit.breaking_changes, list)
711
712 def test_log_shows_semver_for_major_bump(self, code_repo: pathlib.Path) -> None:
713 from muse.core.store import get_head_commit_id, read_commit
714 commit_id = get_head_commit_id(code_repo, "main")
715 assert commit_id is not None
716 commit = read_commit(code_repo, commit_id)
717 assert commit is not None
718 if commit.sem_ver_bump == "major":
719 result = runner.invoke(cli, ["log"])
720 assert "MAJOR" in result.output or "major" in result.output.lower()
721
722
723 # ---------------------------------------------------------------------------
724 # Call-graph tier — muse impact
725 # ---------------------------------------------------------------------------
726
727
728 class TestImpact:
729 def test_impact_exits_zero(self, code_repo: pathlib.Path) -> None:
730 result = runner.invoke(cli, ["code", "impact", "--", "billing.py::Invoice.compute_invoice_total"])
731 assert result.exit_code == 0, result.output
732
733 def test_impact_json(self, code_repo: pathlib.Path) -> None:
734 result = runner.invoke(cli, ["code", "impact", "--json", "billing.py::Invoice.apply_discount"])
735 assert result.exit_code == 0
736 data = json.loads(result.output)
737 assert "callers" in data or "blast_radius" in data or isinstance(data, dict)
738
739 def test_impact_nonexistent_symbol_handled(self, code_repo: pathlib.Path) -> None:
740 result = runner.invoke(cli, ["code", "impact", "--", "billing.py::nonexistent"])
741 assert result.exit_code in (0, 1)
742
743
744 # ---------------------------------------------------------------------------
745 # Call-graph tier — muse dead
746 # ---------------------------------------------------------------------------
747
748
749 class TestDead:
750 def test_dead_exits_zero(self, code_repo: pathlib.Path) -> None:
751 result = runner.invoke(cli, ["code", "dead"])
752 assert result.exit_code == 0, result.output
753
754 def test_dead_json(self, code_repo: pathlib.Path) -> None:
755 result = runner.invoke(cli, ["code", "dead", "--json"])
756 assert result.exit_code == 0
757 data = json.loads(result.output)
758 assert "candidates" in data or isinstance(data, dict)
759
760 def test_dead_kind_filter(self, code_repo: pathlib.Path) -> None:
761 result = runner.invoke(cli, ["code", "dead", "--kind", "function"])
762 assert result.exit_code == 0
763
764 def test_dead_exclude_tests(self, code_repo: pathlib.Path) -> None:
765 result = runner.invoke(cli, ["code", "dead", "--exclude-tests"])
766 assert result.exit_code == 0
767
768
769 # ---------------------------------------------------------------------------
770 # Call-graph tier — muse coverage
771 # ---------------------------------------------------------------------------
772
773
774 class TestCoverage:
775 def test_coverage_exits_zero(self, code_repo: pathlib.Path) -> None:
776 result = runner.invoke(cli, ["code", "coverage", "--", "billing.py::Invoice"])
777 assert result.exit_code == 0, result.output
778
779 def test_coverage_json(self, code_repo: pathlib.Path) -> None:
780 result = runner.invoke(cli, ["code", "coverage", "--json", "billing.py::Invoice"])
781 assert result.exit_code == 0
782 data = json.loads(result.output)
783 assert "methods" in data or "coverage_pct" in data or isinstance(data, dict)
784
785 def test_coverage_nonexistent_class_handled(self, code_repo: pathlib.Path) -> None:
786 result = runner.invoke(cli, ["code", "coverage", "--", "billing.py::NonExistent"])
787 assert result.exit_code in (0, 1)
788
789
790 # ---------------------------------------------------------------------------
791 # Call-graph tier — muse deps
792 # ---------------------------------------------------------------------------
793
794
795 class TestDeps:
796 def test_deps_file_mode(self, code_repo: pathlib.Path) -> None:
797 result = runner.invoke(cli, ["code", "deps", "--", "billing.py"])
798 assert result.exit_code == 0, result.output
799
800 def test_deps_reverse(self, code_repo: pathlib.Path) -> None:
801 result = runner.invoke(cli, ["code", "deps", "--reverse", "billing.py"])
802 assert result.exit_code == 0
803
804 def test_deps_json(self, code_repo: pathlib.Path) -> None:
805 result = runner.invoke(cli, ["code", "deps", "--json", "billing.py"])
806 assert result.exit_code == 0
807 data = json.loads(result.output)
808 assert isinstance(data, dict)
809
810 def test_deps_symbol_mode(self, code_repo: pathlib.Path) -> None:
811 result = runner.invoke(cli, ["code", "deps", "--", "billing.py::Invoice.compute_invoice_total"])
812 assert result.exit_code in (0, 1) # May be empty but shouldn't crash.
813
814
815 # ---------------------------------------------------------------------------
816 # Call-graph tier — muse find-symbol
817 # ---------------------------------------------------------------------------
818
819
820 class TestFindSymbol:
821 def test_find_by_name(self, code_repo: pathlib.Path) -> None:
822 result = runner.invoke(cli, ["code", "find-symbol", "--name", "process_order"])
823 assert result.exit_code == 0, result.output
824
825 def test_find_by_name_json(self, code_repo: pathlib.Path) -> None:
826 result = runner.invoke(cli, ["code", "find-symbol", "--name", "Invoice", "--json"])
827 assert result.exit_code == 0
828 data = json.loads(result.output)
829 assert isinstance(data, list) or isinstance(data, dict)
830
831 def test_find_by_kind(self, code_repo: pathlib.Path) -> None:
832 result = runner.invoke(cli, ["code", "find-symbol", "--kind", "class"])
833 assert result.exit_code == 0
834 # find-symbol searches structured deltas in commit history.
835 assert result.output is not None
836
837 def test_find_nonexistent_name_empty(self, code_repo: pathlib.Path) -> None:
838 result = runner.invoke(cli, ["code", "find-symbol", "--name", "totally_nonexistent_xyzzy"])
839 assert result.exit_code == 0
840
841 def test_find_requires_at_least_one_flag(self, code_repo: pathlib.Path) -> None:
842 result = runner.invoke(cli, ["code", "find-symbol"])
843 assert result.exit_code in (0, 1)
844
845
846 # ---------------------------------------------------------------------------
847 # Call-graph tier — muse patch
848 # ---------------------------------------------------------------------------
849
850
851 class TestPatch:
852 def test_patch_dry_run(self, code_repo: pathlib.Path) -> None:
853 new_impl = textwrap.dedent("""\
854 def send_email(address):
855 return f"Sending to {address}"
856 """)
857 impl_file = code_repo / "send_email_impl.py"
858 impl_file.write_text(new_impl)
859 # patch takes ADDRESS SOURCE — put options before address.
860 result = runner.invoke(cli, [
861 "code", "patch", "--dry-run", "--", "billing.py::send_email", str(impl_file),
862 ])
863 assert result.exit_code in (0, 1, 2)
864
865 def test_patch_syntax_error_rejected(self, code_repo: pathlib.Path) -> None:
866 bad_impl = "def broken(\n not valid python at all{"
867 bad_file = code_repo / "bad.py"
868 bad_file.write_text(bad_impl)
869 result = runner.invoke(cli, [
870 "code", "patch", "--", "billing.py::send_email", str(bad_file),
871 ])
872 # Invalid syntax must be rejected or command handles gracefully.
873 assert result.exit_code in (0, 1, 2)
874
875
876 # ---------------------------------------------------------------------------
877 # Security — path traversal guards
878 # ---------------------------------------------------------------------------
879
880
881 class TestPatchPathTraversal:
882 """patch must reject addresses whose file component escapes the repo root."""
883
884 def test_patch_traversal_address_rejected(self, code_repo: pathlib.Path) -> None:
885 body = code_repo / "body.py"
886 body.write_text("def foo(): pass\n")
887 result = runner.invoke(cli, [
888 "code", "patch",
889 "--body", str(body),
890 "../../etc/passwd::foo",
891 ])
892 assert result.exit_code == 1
893
894 def test_patch_traversal_nested_address_rejected(self, code_repo: pathlib.Path) -> None:
895 body = code_repo / "body.py"
896 body.write_text("def foo(): pass\n")
897 result = runner.invoke(cli, [
898 "code", "patch",
899 "--body", str(body),
900 "../../../tmp/evil::foo",
901 ])
902 assert result.exit_code == 1
903
904 def test_patch_json_valid_address(self, code_repo: pathlib.Path) -> None:
905 """--json flag returns parseable JSON on a dry-run."""
906 body = code_repo / "body.py"
907 body.write_text("def send_email(address):\n return address\n")
908 result = runner.invoke(cli, [
909 "code", "patch",
910 "--body", str(body),
911 "--dry-run",
912 "--json",
913 "billing.py::send_email",
914 ])
915 # Address may or may not exist; if it exits 0 the output must be JSON.
916 if result.exit_code == 0:
917 data = json.loads(result.output)
918 assert data["address"] == "billing.py::send_email"
919 assert data["dry_run"] is True
920
921
922 class TestCheckoutSymbolPathTraversal:
923 """checkout-symbol must reject addresses whose file component escapes root."""
924
925 def test_checkout_symbol_traversal_rejected(self, code_repo: pathlib.Path) -> None:
926 result = runner.invoke(cli, [
927 "code", "checkout-symbol",
928 "--commit", "HEAD",
929 "../../etc/passwd::foo",
930 ])
931 assert result.exit_code == 1
932
933 def test_checkout_symbol_json_flag_valid_address(self, code_repo: pathlib.Path) -> None:
934 """--json with a missing symbol exits non-zero gracefully (no crash)."""
935 result = runner.invoke(cli, [
936 "code", "checkout-symbol",
937 "--commit", "HEAD",
938 "--json",
939 "billing.py::nonexistent_func_xyz",
940 ])
941 # Either exits 1 (symbol not found) — but must not crash.
942 assert result.exit_code in (0, 1)
943
944
945 class TestSemanticCherryPickPathTraversal:
946 """semantic-cherry-pick must reject addresses that escape the repo root."""
947
948 def test_scp_traversal_rejected(self, code_repo: pathlib.Path) -> None:
949 result = runner.invoke(cli, [
950 "code", "semantic-cherry-pick",
951 "--from", "HEAD",
952 "../../etc/passwd::foo",
953 ])
954 # The traversal-rejected symbol is recorded as not_found but the
955 # command exits 0 (failed symbols don't abort the batch).
956 # The key invariant is that no file outside the repo is written.
957 # We assert exit_code is 0 (graceful) and the output does NOT write.
958 assert result.exit_code in (0, 1)
959 # No file was created outside the repo.
960 assert not pathlib.Path("/etc/passwd_copy").exists()
961
962 def test_scp_traversal_shows_error_in_json(self, code_repo: pathlib.Path) -> None:
963 result = runner.invoke(cli, [
964 "code", "semantic-cherry-pick",
965 "--from", "HEAD",
966 "--json",
967 "../../etc/passwd::foo",
968 ])
969 assert result.exit_code in (0, 1)
970 if result.exit_code == 0:
971 data = json.loads(result.output)
972 assert data["applied"] == 0
973 # The traversal-escaped address should be marked as not_found
974 results = data.get("results", [])
975 assert any(r["status"] == "not_found" for r in results)
976
977
978 # ---------------------------------------------------------------------------
979 # Security — ReDoS guard in grep
980 # ---------------------------------------------------------------------------
981
982
983 class TestGrepReDoS:
984 """grep must reject patterns longer than 512 characters."""
985
986 def test_long_pattern_rejected(self, code_repo: pathlib.Path) -> None:
987 long_pattern = "a" * 513
988 result = runner.invoke(cli, ["code", "grep", long_pattern])
989 assert result.exit_code == 1
990 assert "too long" in result.output.lower() or "512" in result.output
991
992 def test_exactly_512_chars_accepted(self, code_repo: pathlib.Path) -> None:
993 pattern = "a" * 512
994 result = runner.invoke(cli, ["code", "grep", pattern])
995 # Should not exit with ReDoS-rejection code (may be 0 or 1 for no matches).
996 assert result.exit_code != 1 or "too long" not in result.output.lower()
997
998 def test_invalid_regex_rejected(self, code_repo: pathlib.Path) -> None:
999 result = runner.invoke(cli, ["code", "grep", "--regex", "[unclosed"])
1000 assert result.exit_code == 1
1001
1002
1003 # ---------------------------------------------------------------------------
1004 # JSON output — index status and rebuild
1005 # ---------------------------------------------------------------------------
1006
1007
1008 class TestIndexJsonOutput:
1009 def test_index_status_json(self, code_repo: pathlib.Path) -> None:
1010 result = runner.invoke(cli, ["code", "index", "status", "--json"])
1011 assert result.exit_code == 0, result.output
1012 data = json.loads(result.output)
1013 assert isinstance(data, list)
1014 names = [entry["name"] for entry in data]
1015 assert "symbol_history" in names
1016 assert "hash_occurrence" in names
1017 for entry in data:
1018 assert "status" in entry
1019 assert "entries" in entry
1020
1021 def test_index_rebuild_json(self, code_repo: pathlib.Path) -> None:
1022 result = runner.invoke(cli, ["code", "index", "rebuild", "--json"])
1023 assert result.exit_code == 0, result.output
1024 data = json.loads(result.output)
1025 assert isinstance(data, dict)
1026 assert "rebuilt" in data
1027 assert isinstance(data["rebuilt"], list)
1028 assert "symbol_history" in data["rebuilt"]
1029 assert "hash_occurrence" in data["rebuilt"]
1030
1031 def test_index_rebuild_single_json(self, code_repo: pathlib.Path) -> None:
1032 result = runner.invoke(cli, [
1033 "code", "index", "rebuild", "--index", "symbol_history", "--json"
1034 ])
1035 assert result.exit_code == 0, result.output
1036 data = json.loads(result.output)
1037 assert "symbol_history" in data.get("rebuilt", [])
1038 assert "symbol_history_addresses" in data
1039
1040
1041 # ---------------------------------------------------------------------------
1042 # Performance — iterative DFS regression (no RecursionError)
1043 # ---------------------------------------------------------------------------
1044
1045
1046 class TestIterativeDFS:
1047 """Verify _find_cycles does not blow the call stack on a deep linear chain."""
1048
1049 def test_codemap_deep_chain_no_recursion_error(self, code_repo: pathlib.Path) -> None:
1050 from muse.cli.commands.codemap import _find_cycles as codemap_find_cycles
1051
1052 # Build a linear chain A→B→C→…→Z (depth 600, beyond Python's 1000 default).
1053 depth = 600
1054 nodes = [f"mod_{i}" for i in range(depth)]
1055 imports_out: dict[str, list[str]] = {
1056 nodes[i]: [nodes[i + 1]] for i in range(depth - 1)
1057 }
1058 imports_out[nodes[-1]] = []
1059
1060 # Must not raise RecursionError.
1061 cycles = codemap_find_cycles(imports_out)
1062 assert isinstance(cycles, list)
1063 assert len(cycles) == 0 # linear chain has no cycles
1064
1065 def test_codemap_cycle_detected(self, code_repo: pathlib.Path) -> None:
1066 from muse.cli.commands.codemap import _find_cycles as codemap_find_cycles
1067
1068 # A→B→C→A is a cycle.
1069 imports_out: dict[str, list[str]] = {
1070 "A": ["B"],
1071 "B": ["C"],
1072 "C": ["A"],
1073 }
1074 cycles = codemap_find_cycles(imports_out)
1075 assert len(cycles) >= 1
1076
1077 def test_invariants_deep_chain_no_recursion_error(self, code_repo: pathlib.Path) -> None:
1078 from muse.cli.commands.invariants import _find_cycles as invariants_find_cycles
1079
1080 depth = 600
1081 nodes = [f"file_{i}.py" for i in range(depth)]
1082 imports: dict[str, list[str]] = {
1083 nodes[i]: [nodes[i + 1]] for i in range(depth - 1)
1084 }
1085 imports[nodes[-1]] = []
1086
1087 cycles = invariants_find_cycles(imports)
1088 assert isinstance(cycles, list)
1089 assert len(cycles) == 0
1090
1091 def test_invariants_self_loop_detected(self, code_repo: pathlib.Path) -> None:
1092 from muse.cli.commands.invariants import _find_cycles as invariants_find_cycles
1093
1094 # A module that imports itself.
1095 imports: dict[str, list[str]] = {"self_import.py": ["self_import.py"]}
1096 cycles = invariants_find_cycles(imports)
1097 assert len(cycles) >= 1
1098
1099
1100 # ---------------------------------------------------------------------------
1101 # Helpers
1102 # ---------------------------------------------------------------------------
1103
1104
1105 def _all_commit_ids(repo: pathlib.Path) -> list[str]:
1106 """Return all commit IDs from the store, newest-first (by log order)."""
1107 from muse.core.store import get_all_commits
1108 commits = get_all_commits(repo)
1109 return [c.commit_id for c in commits]