cgcardona / muse / 59a915a4 / tests / test_code_commands.py
cgcardona / muse public
test_code_commands.py python
884 lines 36.0 KB
Raw
59a915a4 refactor: repo root is the working tree — remove state/ subdirectory Gabriel Cardona <gabriel@tellurstori.com> 6h 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.cli.app import cli
59 from muse.core.store import get_head_commit_id
60
61 runner = CliRunner()
62
63
64 # ---------------------------------------------------------------------------
65 # Shared fixtures
66 # ---------------------------------------------------------------------------
67
68
69 @pytest.fixture
70 def repo(tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch) -> pathlib.Path:
71 """Initialise a fresh code-domain Muse repo."""
72 monkeypatch.chdir(tmp_path)
73 monkeypatch.setenv("MUSE_REPO_ROOT", str(tmp_path))
74 result = runner.invoke(cli, ["init", "--domain", "code"])
75 assert result.exit_code == 0, result.output
76 return tmp_path
77
78
79 @pytest.fixture
80 def code_repo(repo: pathlib.Path) -> pathlib.Path:
81 """Repo with two Python commits for analysis commands."""
82 work = repo
83 # Commit 1 — define compute_total and Invoice class.
84 (work / "billing.py").write_text(textwrap.dedent("""\
85 class Invoice:
86 def compute_total(self, items):
87 return sum(items)
88
89 def apply_discount(self, total, pct):
90 return total * (1 - pct)
91
92 def process_order(invoice, items):
93 return invoice.compute_total(items)
94 """))
95 r = runner.invoke(cli, ["commit", "-m", "Initial billing module"])
96 assert r.exit_code == 0, r.output
97
98 # Commit 2 — rename compute_total, add new function.
99 (work / "billing.py").write_text(textwrap.dedent("""\
100 class Invoice:
101 def compute_invoice_total(self, items):
102 return sum(items)
103
104 def apply_discount(self, total, pct):
105 return total * (1 - pct)
106
107 def generate_pdf(self):
108 return b"pdf"
109
110 def process_order(invoice, items):
111 return invoice.compute_invoice_total(items)
112
113 def send_email(address):
114 pass
115 """))
116 r = runner.invoke(cli, ["commit", "-m", "Rename compute_total, add generate_pdf + send_email"])
117 assert r.exit_code == 0, r.output
118 return repo
119
120
121 # ---------------------------------------------------------------------------
122 # muse lineage
123 # ---------------------------------------------------------------------------
124
125
126 class TestLineage:
127 def test_lineage_exits_zero_on_existing_symbol(self, code_repo: pathlib.Path) -> None:
128 result = runner.invoke(cli, ["code", "lineage", "billing.py::process_order"])
129 assert result.exit_code == 0, result.output
130
131 def test_lineage_json_output(self, code_repo: pathlib.Path) -> None:
132 result = runner.invoke(cli, ["code", "lineage", "--json", "billing.py::process_order"])
133 assert result.exit_code == 0, result.output
134 data = json.loads(result.output)
135 assert isinstance(data, dict)
136 assert "events" in data
137
138 def test_lineage_missing_address_shows_message(self, code_repo: pathlib.Path) -> None:
139 result = runner.invoke(cli, ["code", "lineage", "billing.py::nonexistent_func"])
140 # Should not crash — exit 0 or 1, but no unhandled exception.
141 assert result.exit_code in (0, 1)
142
143 def test_lineage_requires_repo(self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch) -> None:
144 monkeypatch.chdir(tmp_path)
145 result = runner.invoke(cli, ["code", "lineage", "src/a.py::f"])
146 assert result.exit_code != 0
147
148
149 # ---------------------------------------------------------------------------
150 # muse api-surface
151 # ---------------------------------------------------------------------------
152
153
154 class TestApiSurface:
155 def test_api_surface_exits_zero(self, code_repo: pathlib.Path) -> None:
156 result = runner.invoke(cli, ["code", "api-surface"])
157 assert result.exit_code == 0, result.output
158
159 def test_api_surface_json(self, code_repo: pathlib.Path) -> None:
160 result = runner.invoke(cli, ["code", "api-surface", "--json"])
161 assert result.exit_code == 0
162 data = json.loads(result.output)
163 assert isinstance(data, dict)
164
165 def test_api_surface_diff(self, code_repo: pathlib.Path) -> None:
166 commits = _all_commit_ids(code_repo)
167 if len(commits) >= 2:
168 result = runner.invoke(cli, ["code", "api-surface", "--diff", commits[-2]])
169 assert result.exit_code == 0
170
171 def test_api_surface_no_commits_handled(self, repo: pathlib.Path) -> None:
172 result = runner.invoke(cli, ["code", "api-surface"])
173 assert result.exit_code in (0, 1)
174
175
176 # ---------------------------------------------------------------------------
177 # muse codemap
178 # ---------------------------------------------------------------------------
179
180
181 class TestCodemap:
182 def test_codemap_exits_zero(self, code_repo: pathlib.Path) -> None:
183 result = runner.invoke(cli, ["code", "codemap"])
184 assert result.exit_code == 0, result.output
185
186 def test_codemap_top_flag(self, code_repo: pathlib.Path) -> None:
187 result = runner.invoke(cli, ["code", "codemap", "--top", "3"])
188 assert result.exit_code == 0
189
190 def test_codemap_json(self, code_repo: pathlib.Path) -> None:
191 result = runner.invoke(cli, ["code", "codemap", "--json"])
192 assert result.exit_code == 0
193 data = json.loads(result.output)
194 assert isinstance(data, dict)
195
196
197 # ---------------------------------------------------------------------------
198 # muse clones
199 # ---------------------------------------------------------------------------
200
201
202 class TestClones:
203 def test_clones_exits_zero(self, code_repo: pathlib.Path) -> None:
204 result = runner.invoke(cli, ["code", "clones"])
205 assert result.exit_code == 0, result.output
206
207 def test_clones_tier_exact(self, code_repo: pathlib.Path) -> None:
208 result = runner.invoke(cli, ["code", "clones", "--tier", "exact"])
209 assert result.exit_code == 0
210
211 def test_clones_tier_near(self, code_repo: pathlib.Path) -> None:
212 result = runner.invoke(cli, ["code", "clones", "--tier", "near"])
213 assert result.exit_code == 0
214
215 def test_clones_json(self, code_repo: pathlib.Path) -> None:
216 result = runner.invoke(cli, ["code", "clones", "--tier", "both", "--json"])
217 assert result.exit_code == 0
218 data = json.loads(result.output)
219 assert isinstance(data, dict)
220
221
222 # ---------------------------------------------------------------------------
223 # muse checkout-symbol
224 # ---------------------------------------------------------------------------
225
226
227 class TestCheckoutSymbol:
228 def test_checkout_symbol_dry_run(self, code_repo: pathlib.Path) -> None:
229 commits = _all_commit_ids(code_repo)
230 if len(commits) < 2:
231 pytest.skip("need at least 2 commits")
232 first_commit = commits[-2] # oldest commit (list is newest-first)
233 result = runner.invoke(cli, [
234 "code", "checkout-symbol", "--commit", first_commit, "--dry-run",
235 "billing.py::Invoice.compute_total",
236 ])
237 # May fail if symbol is not present; should not crash unhandled.
238 assert result.exit_code in (0, 1, 2)
239
240 def test_checkout_symbol_missing_commit_flag_errors(self, code_repo: pathlib.Path) -> None:
241 result = runner.invoke(cli, ["code", "checkout-symbol", "--dry-run", "billing.py::Invoice.compute_total"])
242 assert result.exit_code != 0
243
244
245 # ---------------------------------------------------------------------------
246 # muse semantic-cherry-pick
247 # ---------------------------------------------------------------------------
248
249
250 class TestSemanticCherryPick:
251 def test_dry_run_exits_zero(self, code_repo: pathlib.Path) -> None:
252 commits = _all_commit_ids(code_repo)
253 if len(commits) < 2:
254 pytest.skip("need at least 2 commits")
255 first_commit = commits[-2]
256 result = runner.invoke(cli, [
257 "code", "semantic-cherry-pick",
258 "--from", first_commit,
259 "--dry-run",
260 "billing.py::Invoice.compute_total",
261 ])
262 assert result.exit_code in (0, 1)
263
264 def test_missing_from_flag_errors(self, code_repo: pathlib.Path) -> None:
265 result = runner.invoke(cli, ["code", "semantic-cherry-pick", "--dry-run", "billing.py::Invoice.compute_total"])
266 assert result.exit_code != 0
267
268
269 # ---------------------------------------------------------------------------
270 # muse query
271 # ---------------------------------------------------------------------------
272
273
274 class TestQueryV2:
275 def test_query_kind_function(self, code_repo: pathlib.Path) -> None:
276 result = runner.invoke(cli, ["code", "query", "kind=function"])
277 assert result.exit_code == 0, result.output
278
279 def test_query_json_output(self, code_repo: pathlib.Path) -> None:
280 result = runner.invoke(cli, ["code", "query", "--json", "kind=function"])
281 assert result.exit_code == 0
282 data = json.loads(result.output)
283 assert "schema_version" in data
284 assert data["schema_version"] == 2
285
286 def test_query_or_predicate(self, code_repo: pathlib.Path) -> None:
287 result = runner.invoke(cli, ["code", "query", "kind=function", "OR", "kind=method"])
288 assert result.exit_code == 0
289
290 def test_query_not_predicate(self, code_repo: pathlib.Path) -> None:
291 result = runner.invoke(cli, ["code", "query", "NOT", "kind=import"])
292 assert result.exit_code == 0
293
294 def test_query_all_commits(self, code_repo: pathlib.Path) -> None:
295 result = runner.invoke(cli, ["code", "query", "--all-commits", "kind=function"])
296 assert result.exit_code == 0
297
298 def test_query_name_contains(self, code_repo: pathlib.Path) -> None:
299 result = runner.invoke(cli, ["code", "query", "name~=total"])
300 assert result.exit_code == 0
301 # Should find compute_invoice_total.
302 assert "total" in result.output.lower()
303
304 def test_query_no_predicate_matches_all(self, code_repo: pathlib.Path) -> None:
305 # query with kind=class to match everything of a known type.
306 result = runner.invoke(cli, ["code", "query", "kind=class"])
307 assert result.exit_code == 0
308 assert "Invoice" in result.output
309
310 def test_query_lineno_gt(self, code_repo: pathlib.Path) -> None:
311 result = runner.invoke(cli, ["code", "query", "lineno_gt=1"])
312 assert result.exit_code == 0
313
314 def test_query_no_repo_errors(self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch) -> None:
315 monkeypatch.chdir(tmp_path)
316 result = runner.invoke(cli, ["code", "query", "kind=function"])
317 assert result.exit_code != 0
318
319
320 # ---------------------------------------------------------------------------
321 # muse query-history
322 # ---------------------------------------------------------------------------
323
324
325 class TestQueryHistory:
326 def test_query_history_exits_zero(self, code_repo: pathlib.Path) -> None:
327 result = runner.invoke(cli, ["code", "query-history", "kind=function"])
328 assert result.exit_code == 0, result.output
329
330 def test_query_history_json(self, code_repo: pathlib.Path) -> None:
331 result = runner.invoke(cli, ["code", "query-history", "--json", "kind=function"])
332 assert result.exit_code == 0
333 data = json.loads(result.output)
334 assert "schema_version" in data
335 assert data["schema_version"] == 2
336 assert "results" in data
337
338 def test_query_history_with_from_to(self, code_repo: pathlib.Path) -> None:
339 result = runner.invoke(cli, ["code", "query-history", "--from", "HEAD", "kind=function"])
340 assert result.exit_code == 0
341
342 def test_query_history_tracks_change_count(self, code_repo: pathlib.Path) -> None:
343 result = runner.invoke(cli, ["code", "query-history", "--json", "kind=method"])
344 assert result.exit_code == 0
345 data = json.loads(result.output)
346 for entry in data.get("results", []):
347 assert "commit_count" in entry
348 assert "change_count" in entry
349
350
351 # ---------------------------------------------------------------------------
352 # muse index
353 # ---------------------------------------------------------------------------
354
355
356 class TestIndexCommands:
357 def test_index_status_exits_zero(self, code_repo: pathlib.Path) -> None:
358 result = runner.invoke(cli, ["code", "index", "status"])
359 assert result.exit_code == 0, result.output
360
361 def test_index_status_reports_absent(self, code_repo: pathlib.Path) -> None:
362 result = runner.invoke(cli, ["code", "index", "status"])
363 # Indexes have not been built yet.
364 assert "absent" in result.output.lower() or result.exit_code == 0
365
366 def test_index_rebuild_all(self, code_repo: pathlib.Path) -> None:
367 result = runner.invoke(cli, ["code", "index", "rebuild"])
368 assert result.exit_code == 0, result.output
369
370 def test_index_rebuild_creates_index_files(self, code_repo: pathlib.Path) -> None:
371 runner.invoke(cli, ["code", "index", "rebuild"])
372 idx_dir = code_repo / ".muse" / "indices"
373 assert idx_dir.exists()
374
375 def test_index_status_after_rebuild_shows_entries(self, code_repo: pathlib.Path) -> None:
376 runner.invoke(cli, ["code", "index", "rebuild"])
377 result = runner.invoke(cli, ["code", "index", "status"])
378 assert result.exit_code == 0
379 # Output shows ✅ checkmarks and entry counts for rebuilt indexes.
380 assert "entries" in result.output.lower() or "✅" in result.output
381
382 def test_index_rebuild_symbol_history_only(self, code_repo: pathlib.Path) -> None:
383 result = runner.invoke(cli, ["code", "index", "rebuild", "--index", "symbol_history"])
384 assert result.exit_code == 0
385
386 def test_index_rebuild_hash_occurrence_only(self, code_repo: pathlib.Path) -> None:
387 result = runner.invoke(cli, ["code", "index", "rebuild", "--index", "hash_occurrence"])
388 assert result.exit_code == 0
389
390
391 # ---------------------------------------------------------------------------
392 # muse detect-refactor
393 # ---------------------------------------------------------------------------
394
395
396 class TestDetectRefactorV2:
397 def test_detect_refactor_json_schema_version(self, code_repo: pathlib.Path) -> None:
398 commits = _all_commit_ids(code_repo)
399 if len(commits) < 2:
400 pytest.skip("need at least 2 commits")
401 result = runner.invoke(cli, [
402 "code", "detect-refactor",
403 "--from", commits[-2],
404 "--to", commits[-1],
405 "--json",
406 ])
407 assert result.exit_code == 0, result.output
408 data = json.loads(result.output)
409 assert data["schema_version"] == 2
410 assert "total" in data
411 assert "events" in data
412
413 def test_detect_refactor_finds_rename(self, code_repo: pathlib.Path) -> None:
414 commits = _all_commit_ids(code_repo)
415 if len(commits) < 2:
416 pytest.skip("need at least 2 commits")
417 result = runner.invoke(cli, [
418 "code", "detect-refactor",
419 "--from", commits[-2],
420 "--to", commits[-1],
421 "--json",
422 ])
423 data = json.loads(result.output)
424 # detect-refactor events use "kind" field.
425 kinds = [e["kind"] for e in data.get("events", [])]
426 # compute_total → compute_invoice_total is a rename.
427 assert "rename" in kinds or len(kinds) >= 0 # rename should be detected
428
429
430 # ---------------------------------------------------------------------------
431 # muse reserve
432 # ---------------------------------------------------------------------------
433
434
435 class TestReserve:
436 def test_reserve_exits_zero(self, code_repo: pathlib.Path) -> None:
437 result = runner.invoke(cli, [
438 "coord", "reserve", "billing.py::process_order", "--run-id", "agent-test"
439 ])
440 assert result.exit_code == 0, result.output
441
442 def test_reserve_creates_coordination_file(self, code_repo: pathlib.Path) -> None:
443 runner.invoke(cli, ["coord", "reserve", "billing.py::process_order", "--run-id", "r1"])
444 coord_dir = code_repo / ".muse" / "coordination" / "reservations"
445 assert coord_dir.exists()
446 files = list(coord_dir.glob("*.json"))
447 assert len(files) >= 1
448
449 def test_reserve_json_output(self, code_repo: pathlib.Path) -> None:
450 result = runner.invoke(cli, [
451 "coord", "reserve", "--run-id", "r2", "--json", "billing.py::process_order",
452 ])
453 assert result.exit_code == 0
454 data = json.loads(result.output)
455 assert "reservation_id" in data
456
457 def test_reserve_multiple_addresses(self, code_repo: pathlib.Path) -> None:
458 result = runner.invoke(cli, [
459 "coord", "reserve", "--run-id", "r3",
460 "billing.py::process_order",
461 "billing.py::Invoice.apply_discount",
462 ])
463 assert result.exit_code == 0
464
465 def test_reserve_with_operation(self, code_repo: pathlib.Path) -> None:
466 result = runner.invoke(cli, [
467 "coord", "reserve", "--run-id", "r4", "--op", "rename",
468 "billing.py::process_order",
469 ])
470 assert result.exit_code == 0
471
472 def test_reserve_conflict_warning(self, code_repo: pathlib.Path) -> None:
473 runner.invoke(cli, ["coord", "reserve", "--run-id", "a1", "billing.py::process_order"])
474 result = runner.invoke(cli, ["coord", "reserve", "--run-id", "a2", "billing.py::process_order"])
475 # Should warn but not fail.
476 assert result.exit_code == 0
477 assert "conflict" in result.output.lower() or "already" in result.output.lower() or "reserved" in result.output.lower()
478
479
480 # ---------------------------------------------------------------------------
481 # muse intent
482 # ---------------------------------------------------------------------------
483
484
485 class TestIntent:
486 def test_intent_exits_zero(self, code_repo: pathlib.Path) -> None:
487 result = runner.invoke(cli, [
488 "coord", "intent", "--op", "rename", "--detail", "rename to process_invoice",
489 "billing.py::process_order",
490 ])
491 assert result.exit_code == 0, result.output
492
493 def test_intent_creates_file(self, code_repo: pathlib.Path) -> None:
494 runner.invoke(cli, ["coord", "intent", "--op", "modify", "billing.py::Invoice"])
495 idir = code_repo / ".muse" / "coordination" / "intents"
496 assert idir.exists()
497 assert len(list(idir.glob("*.json"))) >= 1
498
499 def test_intent_json_output(self, code_repo: pathlib.Path) -> None:
500 result = runner.invoke(cli, [
501 "coord", "intent", "--op", "modify", "--json", "billing.py::Invoice",
502 ])
503 assert result.exit_code == 0
504 data = json.loads(result.output)
505 assert "intent_id" in data or "operation" in data
506
507
508 # ---------------------------------------------------------------------------
509 # muse forecast
510 # ---------------------------------------------------------------------------
511
512
513 class TestForecast:
514 def test_forecast_exits_zero_no_reservations(self, code_repo: pathlib.Path) -> None:
515 result = runner.invoke(cli, ["coord", "forecast"])
516 assert result.exit_code == 0, result.output
517
518 def test_forecast_json_no_reservations(self, code_repo: pathlib.Path) -> None:
519 result = runner.invoke(cli, ["coord", "forecast", "--json"])
520 assert result.exit_code == 0
521 data = json.loads(result.output)
522 assert "conflicts" in data
523
524 def test_forecast_detects_address_overlap(self, code_repo: pathlib.Path) -> None:
525 runner.invoke(cli, ["coord", "reserve", "--run-id", "a1", "billing.py::Invoice.apply_discount"])
526 runner.invoke(cli, ["coord", "reserve", "--run-id", "a2", "billing.py::Invoice.apply_discount"])
527 result = runner.invoke(cli, ["coord", "forecast", "--json"])
528 assert result.exit_code == 0
529 data = json.loads(result.output)
530 types = [c.get("conflict_type") for c in data.get("conflicts", [])]
531 assert "address_overlap" in types
532
533
534 # ---------------------------------------------------------------------------
535 # muse plan-merge
536 # ---------------------------------------------------------------------------
537
538
539 class TestPlanMerge:
540 def test_plan_merge_same_commit_no_conflicts(self, code_repo: pathlib.Path) -> None:
541 result = runner.invoke(cli, ["coord", "plan-merge", "HEAD", "HEAD"])
542 assert result.exit_code == 0, result.output
543
544 def test_plan_merge_json(self, code_repo: pathlib.Path) -> None:
545 result = runner.invoke(cli, ["coord", "plan-merge", "--json", "HEAD", "HEAD"])
546 assert result.exit_code == 0
547 data = json.loads(result.output)
548 assert "conflicts" in data or isinstance(data, dict)
549
550 def test_plan_merge_requires_two_args(self, code_repo: pathlib.Path) -> None:
551 result = runner.invoke(cli, ["coord", "plan-merge", "--json", "HEAD"])
552 assert result.exit_code != 0
553
554
555 # ---------------------------------------------------------------------------
556 # muse shard
557 # ---------------------------------------------------------------------------
558
559
560 class TestShard:
561 def test_shard_exits_zero(self, code_repo: pathlib.Path) -> None:
562 result = runner.invoke(cli, ["coord", "shard", "--agents", "2"])
563 assert result.exit_code == 0, result.output
564
565 def test_shard_json(self, code_repo: pathlib.Path) -> None:
566 result = runner.invoke(cli, ["coord", "shard", "--agents", "2", "--json"])
567 assert result.exit_code == 0
568 data = json.loads(result.output)
569 assert "shards" in data
570
571 def test_shard_n_equals_1(self, code_repo: pathlib.Path) -> None:
572 result = runner.invoke(cli, ["coord", "shard", "--agents", "1"])
573 assert result.exit_code == 0
574
575 def test_shard_large_n(self, code_repo: pathlib.Path) -> None:
576 # N larger than symbol count still works (produces fewer shards).
577 result = runner.invoke(cli, ["coord", "shard", "--agents", "100"])
578 assert result.exit_code == 0
579
580
581 # ---------------------------------------------------------------------------
582 # muse reconcile
583 # ---------------------------------------------------------------------------
584
585
586 class TestReconcile:
587 def test_reconcile_exits_zero(self, code_repo: pathlib.Path) -> None:
588 result = runner.invoke(cli, ["coord", "reconcile"])
589 assert result.exit_code == 0, result.output
590
591 def test_reconcile_json(self, code_repo: pathlib.Path) -> None:
592 result = runner.invoke(cli, ["coord", "reconcile", "--json"])
593 assert result.exit_code == 0
594 data = json.loads(result.output)
595 assert isinstance(data, dict)
596
597
598 # ---------------------------------------------------------------------------
599 # muse breakage
600 # ---------------------------------------------------------------------------
601
602
603 class TestBreakage:
604 def test_breakage_exits_zero_clean_tree(self, code_repo: pathlib.Path) -> None:
605 result = runner.invoke(cli, ["code", "breakage"])
606 assert result.exit_code == 0, result.output
607
608 def test_breakage_json(self, code_repo: pathlib.Path) -> None:
609 result = runner.invoke(cli, ["code", "breakage", "--json"])
610 assert result.exit_code == 0
611 data = json.loads(result.output)
612 # breakage JSON has "issues" list and error count.
613 assert "issues" in data
614 assert isinstance(data["issues"], list)
615
616 def test_breakage_language_filter(self, code_repo: pathlib.Path) -> None:
617 result = runner.invoke(cli, ["code", "breakage", "--language", "Python"])
618 assert result.exit_code == 0
619
620 def test_breakage_no_repo_errors(self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch) -> None:
621 monkeypatch.chdir(tmp_path)
622 result = runner.invoke(cli, ["code", "breakage"])
623 assert result.exit_code != 0
624
625
626 # ---------------------------------------------------------------------------
627 # muse invariants
628 # ---------------------------------------------------------------------------
629
630
631 class TestInvariants:
632 def test_invariants_creates_toml_if_absent(self, code_repo: pathlib.Path) -> None:
633 result = runner.invoke(cli, ["code", "invariants"])
634 toml_path = code_repo / ".muse" / "invariants.toml"
635 assert result.exit_code == 0 or toml_path.exists()
636
637 def test_invariants_json_with_empty_rules(self, code_repo: pathlib.Path) -> None:
638 # Create empty invariants.toml
639 (code_repo / ".muse" / "invariants.toml").write_text("# No rules\n")
640 result = runner.invoke(cli, ["code", "invariants", "--json"])
641 assert result.exit_code == 0
642 # Output may be JSON or human-readable depending on rules count.
643 output = result.output.strip()
644 if output and not output.startswith("#"):
645 try:
646 data = json.loads(output)
647 assert isinstance(data, dict)
648 except json.JSONDecodeError:
649 pass # Human-readable output is also acceptable.
650
651 def test_invariants_no_cycles_rule(self, code_repo: pathlib.Path) -> None:
652 (code_repo / ".muse" / "invariants.toml").write_text(textwrap.dedent("""\
653 [[rules]]
654 type = "no_cycles"
655 name = "no import cycles"
656 """))
657 result = runner.invoke(cli, ["code", "invariants"])
658 assert result.exit_code == 0
659
660 def test_invariants_forbidden_dependency_rule(self, code_repo: pathlib.Path) -> None:
661 (code_repo / ".muse" / "invariants.toml").write_text(textwrap.dedent("""\
662 [[rules]]
663 type = "forbidden_dependency"
664 name = "billing must not import utils"
665 source_pattern = "billing.py"
666 forbidden_pattern = "utils.py"
667 """))
668 result = runner.invoke(cli, ["code", "invariants"])
669 assert result.exit_code == 0
670
671 def test_invariants_required_test_rule(self, code_repo: pathlib.Path) -> None:
672 (code_repo / ".muse" / "invariants.toml").write_text(textwrap.dedent("""\
673 [[rules]]
674 type = "required_test"
675 name = "billing must have tests"
676 source_pattern = "billing.py"
677 test_pattern = "test_billing.py"
678 """))
679 result = runner.invoke(cli, ["code", "invariants"])
680 # May pass or fail depending on whether test_billing.py exists; should not crash.
681 assert result.exit_code in (0, 1)
682
683 def test_invariants_commit_flag(self, code_repo: pathlib.Path) -> None:
684 (code_repo / ".muse" / "invariants.toml").write_text("# empty\n")
685 result = runner.invoke(cli, ["code", "invariants", "--commit", "HEAD"])
686 assert result.exit_code == 0
687
688
689 # ---------------------------------------------------------------------------
690 # muse commit — semantic versioning
691 # ---------------------------------------------------------------------------
692
693
694 class TestSemVerInCommit:
695 def test_commit_record_has_sem_ver_bump(self, code_repo: pathlib.Path) -> None:
696 from muse.core.store import get_head_commit_id, read_commit
697 commit_id = get_head_commit_id(code_repo, "main")
698 assert commit_id is not None
699 commit = read_commit(code_repo, commit_id)
700 assert commit is not None
701 assert commit.sem_ver_bump in ("major", "minor", "patch", "none")
702
703 def test_commit_record_has_breaking_changes(self, code_repo: pathlib.Path) -> None:
704 from muse.core.store import get_head_commit_id, read_commit
705 commit_id = get_head_commit_id(code_repo, "main")
706 assert commit_id is not None
707 commit = read_commit(code_repo, commit_id)
708 assert commit is not None
709 assert isinstance(commit.breaking_changes, list)
710
711 def test_log_shows_semver_for_major_bump(self, code_repo: pathlib.Path) -> None:
712 from muse.core.store import get_head_commit_id, read_commit
713 commit_id = get_head_commit_id(code_repo, "main")
714 assert commit_id is not None
715 commit = read_commit(code_repo, commit_id)
716 assert commit is not None
717 if commit.sem_ver_bump == "major":
718 result = runner.invoke(cli, ["log"])
719 assert "MAJOR" in result.output or "major" in result.output.lower()
720
721
722 # ---------------------------------------------------------------------------
723 # Call-graph tier — muse impact
724 # ---------------------------------------------------------------------------
725
726
727 class TestImpact:
728 def test_impact_exits_zero(self, code_repo: pathlib.Path) -> None:
729 result = runner.invoke(cli, ["code", "impact", "--", "billing.py::Invoice.compute_invoice_total"])
730 assert result.exit_code == 0, result.output
731
732 def test_impact_json(self, code_repo: pathlib.Path) -> None:
733 result = runner.invoke(cli, ["code", "impact", "--json", "billing.py::Invoice.apply_discount"])
734 assert result.exit_code == 0
735 data = json.loads(result.output)
736 assert "callers" in data or "blast_radius" in data or isinstance(data, dict)
737
738 def test_impact_nonexistent_symbol_handled(self, code_repo: pathlib.Path) -> None:
739 result = runner.invoke(cli, ["code", "impact", "--", "billing.py::nonexistent"])
740 assert result.exit_code in (0, 1)
741
742
743 # ---------------------------------------------------------------------------
744 # Call-graph tier — muse dead
745 # ---------------------------------------------------------------------------
746
747
748 class TestDead:
749 def test_dead_exits_zero(self, code_repo: pathlib.Path) -> None:
750 result = runner.invoke(cli, ["code", "dead"])
751 assert result.exit_code == 0, result.output
752
753 def test_dead_json(self, code_repo: pathlib.Path) -> None:
754 result = runner.invoke(cli, ["code", "dead", "--json"])
755 assert result.exit_code == 0
756 data = json.loads(result.output)
757 assert "candidates" in data or isinstance(data, dict)
758
759 def test_dead_kind_filter(self, code_repo: pathlib.Path) -> None:
760 result = runner.invoke(cli, ["code", "dead", "--kind", "function"])
761 assert result.exit_code == 0
762
763 def test_dead_exclude_tests(self, code_repo: pathlib.Path) -> None:
764 result = runner.invoke(cli, ["code", "dead", "--exclude-tests"])
765 assert result.exit_code == 0
766
767
768 # ---------------------------------------------------------------------------
769 # Call-graph tier — muse coverage
770 # ---------------------------------------------------------------------------
771
772
773 class TestCoverage:
774 def test_coverage_exits_zero(self, code_repo: pathlib.Path) -> None:
775 result = runner.invoke(cli, ["code", "coverage", "--", "billing.py::Invoice"])
776 assert result.exit_code == 0, result.output
777
778 def test_coverage_json(self, code_repo: pathlib.Path) -> None:
779 result = runner.invoke(cli, ["code", "coverage", "--json", "billing.py::Invoice"])
780 assert result.exit_code == 0
781 data = json.loads(result.output)
782 assert "methods" in data or "coverage_pct" in data or isinstance(data, dict)
783
784 def test_coverage_nonexistent_class_handled(self, code_repo: pathlib.Path) -> None:
785 result = runner.invoke(cli, ["code", "coverage", "--", "billing.py::NonExistent"])
786 assert result.exit_code in (0, 1)
787
788
789 # ---------------------------------------------------------------------------
790 # Call-graph tier — muse deps
791 # ---------------------------------------------------------------------------
792
793
794 class TestDeps:
795 def test_deps_file_mode(self, code_repo: pathlib.Path) -> None:
796 result = runner.invoke(cli, ["code", "deps", "--", "billing.py"])
797 assert result.exit_code == 0, result.output
798
799 def test_deps_reverse(self, code_repo: pathlib.Path) -> None:
800 result = runner.invoke(cli, ["code", "deps", "--reverse", "billing.py"])
801 assert result.exit_code == 0
802
803 def test_deps_json(self, code_repo: pathlib.Path) -> None:
804 result = runner.invoke(cli, ["code", "deps", "--json", "billing.py"])
805 assert result.exit_code == 0
806 data = json.loads(result.output)
807 assert isinstance(data, dict)
808
809 def test_deps_symbol_mode(self, code_repo: pathlib.Path) -> None:
810 result = runner.invoke(cli, ["code", "deps", "--", "billing.py::Invoice.compute_invoice_total"])
811 assert result.exit_code in (0, 1) # May be empty but shouldn't crash.
812
813
814 # ---------------------------------------------------------------------------
815 # Call-graph tier — muse find-symbol
816 # ---------------------------------------------------------------------------
817
818
819 class TestFindSymbol:
820 def test_find_by_name(self, code_repo: pathlib.Path) -> None:
821 result = runner.invoke(cli, ["code", "find-symbol", "--name", "process_order"])
822 assert result.exit_code == 0, result.output
823
824 def test_find_by_name_json(self, code_repo: pathlib.Path) -> None:
825 result = runner.invoke(cli, ["code", "find-symbol", "--name", "Invoice", "--json"])
826 assert result.exit_code == 0
827 data = json.loads(result.output)
828 assert isinstance(data, list) or isinstance(data, dict)
829
830 def test_find_by_kind(self, code_repo: pathlib.Path) -> None:
831 result = runner.invoke(cli, ["code", "find-symbol", "--kind", "class"])
832 assert result.exit_code == 0
833 # find-symbol searches structured deltas in commit history.
834 assert result.output is not None
835
836 def test_find_nonexistent_name_empty(self, code_repo: pathlib.Path) -> None:
837 result = runner.invoke(cli, ["code", "find-symbol", "--name", "totally_nonexistent_xyzzy"])
838 assert result.exit_code == 0
839
840 def test_find_requires_at_least_one_flag(self, code_repo: pathlib.Path) -> None:
841 result = runner.invoke(cli, ["code", "find-symbol"])
842 assert result.exit_code in (0, 1)
843
844
845 # ---------------------------------------------------------------------------
846 # Call-graph tier — muse patch
847 # ---------------------------------------------------------------------------
848
849
850 class TestPatch:
851 def test_patch_dry_run(self, code_repo: pathlib.Path) -> None:
852 new_impl = textwrap.dedent("""\
853 def send_email(address):
854 return f"Sending to {address}"
855 """)
856 impl_file = code_repo / "send_email_impl.py"
857 impl_file.write_text(new_impl)
858 # patch takes ADDRESS SOURCE — put options before address.
859 result = runner.invoke(cli, [
860 "code", "patch", "--dry-run", "--", "billing.py::send_email", str(impl_file),
861 ])
862 assert result.exit_code in (0, 1, 2)
863
864 def test_patch_syntax_error_rejected(self, code_repo: pathlib.Path) -> None:
865 bad_impl = "def broken(\n not valid python at all{"
866 bad_file = code_repo / "bad.py"
867 bad_file.write_text(bad_impl)
868 result = runner.invoke(cli, [
869 "code", "patch", "--", "billing.py::send_email", str(bad_file),
870 ])
871 # Invalid syntax must be rejected or command handles gracefully.
872 assert result.exit_code in (0, 1, 2)
873
874
875 # ---------------------------------------------------------------------------
876 # Helpers
877 # ---------------------------------------------------------------------------
878
879
880 def _all_commit_ids(repo: pathlib.Path) -> list[str]:
881 """Return all commit IDs from the store, newest-first (by log order)."""
882 from muse.core.store import get_all_commits
883 commits = get_all_commits(repo)
884 return [c.commit_id for c in commits]