gabriel / muse public
test_lineage.py python
573 lines 21.8 KB
Raw
sha256:b89fa4fd9ca0d692fc66f6b9aef4c3a0c13c8e9b439faf42da8e91e09f048d4f tests/test_cmd_revert_hardening.py, tests/test_cmd_semantic… Human 41 days ago
1 """Comprehensive tests for muse code lineage.
2
3 Test layers
4 -----------
5 Unit
6 ``build_lineage`` exercised directly with synthetic ``CommitRecord`` objects
7 carrying hand-crafted ``structured_delta`` data. No repo, no disk I/O.
8
9 Integration
10 CLI invocations via ``CliRunner`` against a real tmp-path repo with two
11 Python commits (the shared ``code_repo`` fixture).
12
13 Edge-case
14 Empty history, deleted-then-re-created, address without ``::`` guard,
15 unknown branch / ref, ``--filter`` narrowing, ``--since``/``--until``
16 date bounds, ``--count`` output, ``--stability`` output.
17
18 Stress
19 Programmatically generate N commits each carrying an InsertOp, ModifyOp,
20 or DeleteOp and verify ``build_lineage`` produces the expected event count
21 and kind sequence without error.
22 """
23
24 from __future__ import annotations
25
26 import datetime
27 import json
28 import pathlib
29 import textwrap
30
31 import pytest
32
33 from tests.cli_test_helper import CliRunner
34 from muse.cli.commands.lineage import _LineageEvent, _classify_replace, _stability, build_lineage
35 from muse.core.store import CommitRecord
36 from muse.domain import DeleteOp, DomainOp, InsertOp, PatchOp, ReplaceOp
37
38 cli = None # argparse migration — CliRunner ignores this arg
39 runner = CliRunner()
40
41 # ---------------------------------------------------------------------------
42 # Helpers
43 # ---------------------------------------------------------------------------
44
45 _REPO_ID = "test-repo-id"
46 _SEQ = [0]
47
48
49 def _cid(tag: str) -> str:
50 """Return a deterministic 64-char hex content_id from a short tag."""
51 return tag.ljust(64, "0")[:64]
52
53
54 def _ts(offset_days: int = 0) -> datetime.datetime:
55 base = datetime.datetime(2026, 1, 1, 12, 0, 0, tzinfo=datetime.timezone.utc)
56 return base + datetime.timedelta(days=offset_days)
57
58
59 def _commit(
60 *,
61 message: str = "commit",
62 ops: list[DomainOp] | None = None,
63 day: int = 0,
64 commit_id: str | None = None,
65 ) -> CommitRecord:
66 """Build a synthetic CommitRecord with the given symbol-level ops."""
67 _SEQ[0] += 1
68 cid = commit_id or f"c{_SEQ[0]:063d}"
69 return CommitRecord(
70 commit_id=cid,
71 repo_id=_REPO_ID,
72 branch="main",
73 snapshot_id=f"snap-{cid}",
74 message=message,
75 committed_at=_ts(day),
76 structured_delta={"ops": ops or [], "domain": "code", "summary": message},
77 )
78
79
80 def _insert(address: str, content_id: str) -> InsertOp:
81 return InsertOp(
82 op="insert",
83 address=address,
84 position=None,
85 content_id=_cid(content_id),
86 content_summary=f"function {address.split('::')[-1]}",
87 )
88
89
90 def _delete(address: str, content_id: str) -> DeleteOp:
91 return DeleteOp(
92 op="delete",
93 address=address,
94 position=None,
95 content_id=_cid(content_id),
96 content_summary=f"function {address.split('::')[-1]}",
97 )
98
99
100 def _replace(address: str, old_cid: str, new_cid: str, old_sum: str = "", new_sum: str = "") -> ReplaceOp:
101 return ReplaceOp(
102 op="replace",
103 address=address,
104 position=None,
105 old_content_id=_cid(old_cid),
106 new_content_id=_cid(new_cid),
107 old_summary=old_sum,
108 new_summary=new_sum,
109 )
110
111
112 def _patch(*child_ops: DomainOp, file: str = "billing.py") -> PatchOp:
113 """Wrap symbol-level ops in a PatchOp (as Muse emits for file changes)."""
114 return PatchOp(
115 op="patch",
116 address=file,
117 child_ops=list(child_ops),
118 child_domain="code",
119 child_summary="",
120 )
121
122
123 ADDR = "billing.py::compute_total"
124 OTHER = "billing.py::compute_total_v2"
125 OTHER_FILE = "utils.py::compute_total"
126
127
128 # ---------------------------------------------------------------------------
129 # Unit: _classify_replace
130 # ---------------------------------------------------------------------------
131
132
133 class TestClassifyReplace:
134 def test_signature_change_detected_in_old(self) -> None:
135 assert _classify_replace("signature changed", "") == "signature_change"
136
137 def test_signature_change_detected_in_new(self) -> None:
138 assert _classify_replace("", "new signature") == "signature_change"
139
140 def test_full_rewrite_when_no_signature(self) -> None:
141 assert _classify_replace("impl updated", "impl updated v2") == "full_rewrite"
142
143 def test_empty_summaries_full_rewrite(self) -> None:
144 assert _classify_replace("", "") == "full_rewrite"
145
146
147 # ---------------------------------------------------------------------------
148 # Unit: _stability
149 # ---------------------------------------------------------------------------
150
151
152 class TestStability:
153 def test_no_events(self) -> None:
154 assert _stability([]) == (0, 0)
155
156 def test_all_created(self) -> None:
157 evs = [_LineageEvent("c1", "2026-01-01", "init", "created")]
158 assert _stability(evs) == (0, 1)
159
160 def test_mixed(self) -> None:
161 evs = [
162 _LineageEvent("c1", "2026-01-01", "init", "created"),
163 _LineageEvent("c2", "2026-01-02", "fix", "modified", detail="impl_only"),
164 _LineageEvent("c3", "2026-01-03", "fix2", "modified", detail="full_rewrite"),
165 ]
166 assert _stability(evs) == (2, 3)
167
168
169 # ---------------------------------------------------------------------------
170 # Unit: build_lineage — core event kinds
171 # ---------------------------------------------------------------------------
172
173
174 class TestBuildLineageCreated:
175 def test_no_commits(self) -> None:
176 assert build_lineage(ADDR, []) == []
177
178 def test_no_structured_delta(self) -> None:
179 c = CommitRecord(
180 commit_id="c" * 64,
181 repo_id=_REPO_ID,
182 branch="main",
183 snapshot_id="snap",
184 message="empty",
185 committed_at=_ts(),
186 structured_delta=None,
187 )
188 assert build_lineage(ADDR, [c]) == []
189
190 def test_single_insert_emits_created(self) -> None:
191 c = _commit(ops=[_insert(ADDR, "aaa")], message="add fn", day=0)
192 events = build_lineage(ADDR, [c])
193 assert len(events) == 1
194 assert events[0].kind == "created"
195 assert events[0].message == "add fn"
196 assert events[0].new_content_id == _cid("aaa")
197
198 def test_insert_inside_patch_op(self) -> None:
199 """flat_symbol_ops must recurse into PatchOp.child_ops."""
200 c = _commit(ops=[_patch(_insert(ADDR, "bbb"))], message="patch add")
201 events = build_lineage(ADDR, [c])
202 assert len(events) == 1
203 assert events[0].kind == "created"
204
205 def test_unrelated_insert_ignored(self) -> None:
206 c = _commit(ops=[_insert("billing.py::other_fn", "ccc")])
207 assert build_lineage(ADDR, [c]) == []
208
209
210 class TestBuildLineageModified:
211 def test_replace_emits_modified(self) -> None:
212 c1 = _commit(ops=[_insert(ADDR, "v1")], day=0)
213 c2 = _commit(ops=[_replace(ADDR, "v1", "v2")], day=1, message="fix")
214 events = build_lineage(ADDR, [c1, c2])
215 kinds = [e.kind for e in events]
216 assert kinds == ["created", "modified"]
217 assert events[1].detail == "full_rewrite"
218 assert events[1].message == "fix"
219
220 def test_replace_with_signature_detail(self) -> None:
221 c1 = _commit(ops=[_insert(ADDR, "v1")], day=0)
222 c2 = _commit(ops=[_replace(ADDR, "v1", "v2", old_sum="signature changed")], day=1)
223 events = build_lineage(ADDR, [c1, c2])
224 assert events[1].kind == "modified"
225 assert events[1].detail == "signature_change"
226
227 def test_multiple_modifications_in_sequence(self) -> None:
228 commits = [
229 _commit(ops=[_insert(ADDR, "v1")], day=0),
230 _commit(ops=[_replace(ADDR, "v1", "v2")], day=1),
231 _commit(ops=[_replace(ADDR, "v2", "v3")], day=2),
232 _commit(ops=[_replace(ADDR, "v3", "v4")], day=3),
233 ]
234 events = build_lineage(ADDR, commits)
235 assert len(events) == 4
236 assert events[0].kind == "created"
237 assert all(e.kind == "modified" for e in events[1:])
238
239
240 class TestBuildLineageDeleted:
241 def test_delete_emits_deleted(self) -> None:
242 c1 = _commit(ops=[_insert(ADDR, "v1")], day=0)
243 c2 = _commit(ops=[_delete(ADDR, "v1")], day=1, message="remove fn")
244 events = build_lineage(ADDR, [c1, c2])
245 assert events[-1].kind == "deleted"
246 assert events[-1].message == "remove fn"
247
248 def test_delete_marks_address_not_live(self) -> None:
249 """After delete, re-inserting the same content should emit 'created', not 'copied_from'."""
250 c1 = _commit(ops=[_insert(ADDR, "v1")], day=0)
251 c2 = _commit(ops=[_delete(ADDR, "v1")], day=1)
252 c3 = _commit(ops=[_insert(ADDR, "v1")], day=2)
253 events = build_lineage(ADDR, [c1, c2, c3])
254 kinds = [e.kind for e in events]
255 assert kinds == ["created", "deleted", "created"]
256
257
258 class TestBuildLineageRenamedMoved:
259 def test_rename_within_same_file(self) -> None:
260 """InsertOp at ADDR + DeleteOp at OTHER (same file, same content_id) → renamed_from."""
261 c1 = _commit(ops=[_insert(OTHER, "v1")], day=0)
262 c2 = _commit(ops=[_insert(ADDR, "v1"), _delete(OTHER, "v1")], day=1, message="rename")
263 events = build_lineage(ADDR, [c1, c2])
264 ev = next(e for e in events if e.kind == "renamed_from")
265 assert ev.detail == OTHER
266 assert ev.message == "rename"
267
268 def test_move_across_files(self) -> None:
269 """InsertOp at ADDR + DeleteOp at OTHER_FILE (different file) → moved_from."""
270 c1 = _commit(ops=[_insert(OTHER_FILE, "v1")], day=0)
271 c2 = _commit(ops=[_insert(ADDR, "v1"), _delete(OTHER_FILE, "v1")], day=1, message="move")
272 events = build_lineage(ADDR, [c1, c2])
273 ev = next(e for e in events if e.kind == "moved_from")
274 assert ev.detail == OTHER_FILE
275 assert ev.message == "move"
276
277 def test_rename_file_correctly_classified(self) -> None:
278 """Same file → renamed_from, not moved_from."""
279 c1 = _commit(ops=[_insert(OTHER, "v1")], day=0)
280 c2 = _commit(ops=[_insert(ADDR, "v1"), _delete(OTHER, "v1")], day=1)
281 events = build_lineage(ADDR, [c1, c2])
282 assert any(e.kind == "renamed_from" for e in events)
283 assert not any(e.kind == "moved_from" for e in events)
284
285
286 class TestBuildLineageCopied:
287 def test_copied_from_living_symbol(self) -> None:
288 """Insert at ADDR with content_id already live at OTHER → copied_from."""
289 c1 = _commit(ops=[_insert(OTHER, "v1")], day=0)
290 c2 = _commit(ops=[_insert(ADDR, "v1")], day=1, message="copy fn")
291 events = build_lineage(ADDR, [c1, c2])
292 assert events[0].kind == "copied_from"
293 assert events[0].detail == OTHER
294 assert events[0].message == "copy fn"
295
296 def test_not_copied_when_no_living_symbol(self) -> None:
297 """Insert with unique content_id → created, not copied_from."""
298 c = _commit(ops=[_insert(ADDR, "unique_content")], day=0)
299 events = build_lineage(ADDR, [c])
300 assert events[0].kind == "created"
301
302
303 # ---------------------------------------------------------------------------
304 # Unit: build_lineage — complex lifecycle
305 # ---------------------------------------------------------------------------
306
307
308 class TestBuildLineageLifecycle:
309 def test_full_lifecycle(self) -> None:
310 """create → modify → rename_away → recreate → delete."""
311 # Phase 1: created at ADDR
312 c1 = _commit(ops=[_insert(ADDR, "v1")], day=0, message="create")
313 # Phase 2: modified
314 c2 = _commit(ops=[_replace(ADDR, "v1", "v2")], day=1, message="modify")
315 # Phase 3: renamed away — ADDR is deleted, NEW_ADDR is inserted
316 new_addr = "billing.py::compute_total_renamed"
317 c3 = _commit(ops=[_insert(new_addr, "v2"), _delete(ADDR, "v2")], day=2, message="rename away")
318 # Phase 4: ADDR re-created with fresh content
319 c4 = _commit(ops=[_insert(ADDR, "v3")], day=3, message="recreate")
320 # Phase 5: deleted
321 c5 = _commit(ops=[_delete(ADDR, "v3")], day=4, message="delete")
322
323 events = build_lineage(ADDR, [c1, c2, c3, c4, c5])
324 kinds = [e.kind for e in events]
325 assert kinds == ["created", "modified", "deleted", "created", "deleted"]
326
327 def test_commit_message_propagated(self) -> None:
328 c1 = _commit(ops=[_insert(ADDR, "v1")], message="Initial commit")
329 events = build_lineage(ADDR, [c1])
330 assert events[0].message == "Initial commit"
331
332 def test_to_dict_has_full_commit_id(self) -> None:
333 c1 = _commit(ops=[_insert(ADDR, "v1")], commit_id="a" * 64)
334 events = build_lineage(ADDR, [c1])
335 d = events[0].to_dict()
336 assert d["commit_id"] == "a" * 64 # not truncated
337
338 def test_to_dict_has_message(self) -> None:
339 c1 = _commit(ops=[_insert(ADDR, "v1")], message="My message")
340 events = build_lineage(ADDR, [c1])
341 assert events[0].to_dict()["message"] == "My message"
342
343 def test_commits_without_symbol_ops_skipped(self) -> None:
344 """File-level ops (no '::') must not generate any events."""
345 file_op = ReplaceOp(
346 op="replace",
347 address="billing.py",
348 position=None,
349 old_content_id=_cid("old"),
350 new_content_id=_cid("new"),
351 old_summary="",
352 new_summary="",
353 )
354 c = _commit(ops=[file_op])
355 assert build_lineage(ADDR, [c]) == []
356
357
358 # ---------------------------------------------------------------------------
359 # Unit: build_lineage — date filtering via _gather_commits (tested indirectly
360 # through the CLI --since/--until flags in integration tests below)
361 # ---------------------------------------------------------------------------
362
363
364 # ---------------------------------------------------------------------------
365 # Stress: large commit sequence
366 # ---------------------------------------------------------------------------
367
368
369 class TestBuildLineageStress:
370 def test_many_modifications(self) -> None:
371 """500 sequential modifications produce 501 events without error."""
372 n = 500
373 commits: list[CommitRecord] = [_commit(ops=[_insert(ADDR, "v0")], day=0)]
374 for i in range(1, n + 1):
375 commits.append(_commit(
376 ops=[_replace(ADDR, f"v{i-1}", f"v{i}")],
377 day=i,
378 ))
379 events = build_lineage(ADDR, commits)
380 assert len(events) == n + 1
381 assert events[0].kind == "created"
382 assert all(e.kind == "modified" for e in events[1:])
383
384 def test_many_unrelated_commits_skipped_efficiently(self) -> None:
385 """1000 commits touching only unrelated symbols → 0 events for ADDR."""
386 commits = [
387 _commit(ops=[_insert(f"billing.py::other_{i}", f"uid_{i}")], day=i)
388 for i in range(1000)
389 ]
390 events = build_lineage(ADDR, commits)
391 assert events == []
392
393 def test_interleaved_symbol_and_unrelated_ops(self) -> None:
394 """Mix of target-symbol ops and noise — only target events emitted."""
395 commits: list[CommitRecord] = []
396 for i in range(200):
397 ops: list[DomainOp] = [_insert(f"billing.py::noise_{i}", f"n{i}")]
398 if i == 50:
399 ops.append(_insert(ADDR, "start"))
400 if i == 100:
401 ops.append(_replace(ADDR, "start", "mid"))
402 if i == 150:
403 ops.append(_delete(ADDR, "mid"))
404 commits.append(_commit(ops=ops, day=i))
405
406 events = build_lineage(ADDR, commits)
407 kinds = [e.kind for e in events]
408 assert kinds == ["created", "modified", "deleted"]
409
410 def test_delete_recreate_cycle(self) -> None:
411 """Symbol deleted and recreated 10 times → 10 deletes + 11 creates."""
412 commits: list[CommitRecord] = [_commit(ops=[_insert(ADDR, "v0")], day=0)]
413 for cycle in range(10):
414 base = cycle * 2 + 1
415 commits.append(_commit(ops=[_delete(ADDR, f"v{cycle}")], day=base))
416 commits.append(_commit(ops=[_insert(ADDR, f"v{cycle+1}")], day=base + 1))
417
418 events = build_lineage(ADDR, commits)
419 kinds = [e.kind for e in events]
420 assert kinds.count("created") == 11
421 assert kinds.count("deleted") == 10
422
423
424 # ---------------------------------------------------------------------------
425 # Integration: CLI
426 # ---------------------------------------------------------------------------
427
428
429 @pytest.fixture
430 def repo(tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch) -> pathlib.Path:
431 monkeypatch.chdir(tmp_path)
432 monkeypatch.setenv("MUSE_REPO_ROOT", str(tmp_path))
433 r = runner.invoke(cli, ["init", "--domain", "code"])
434 assert r.exit_code == 0, r.output
435 return tmp_path
436
437
438 @pytest.fixture
439 def code_repo(repo: pathlib.Path) -> pathlib.Path:
440 """Two-commit repo: billing.py created then function renamed."""
441 (repo / "billing.py").write_text(textwrap.dedent("""\
442 def compute_total(items):
443 return sum(items)
444
445 def process_order(invoice, items):
446 return compute_total(items)
447 """))
448 r = runner.invoke(cli, ["commit", "-m", "Initial billing module"])
449 assert r.exit_code == 0, r.output
450
451 (repo / "billing.py").write_text(textwrap.dedent("""\
452 def compute_invoice_total(items):
453 return sum(items)
454
455 def process_order(invoice, items):
456 return compute_invoice_total(items)
457 """))
458 r = runner.invoke(cli, ["commit", "-m", "Rename compute_total"])
459 assert r.exit_code == 0, r.output
460 return repo
461
462
463 class TestLineageCLI:
464 def test_exits_zero_for_existing_symbol(self, code_repo: pathlib.Path) -> None:
465 result = runner.invoke(cli, ["code", "lineage", "billing.py::process_order"])
466 assert result.exit_code == 0, result.output
467
468 def test_json_schema(self, code_repo: pathlib.Path) -> None:
469 result = runner.invoke(cli, ["code", "lineage", "--json", "billing.py::process_order"])
470 assert result.exit_code == 0, result.output
471 data = json.loads(result.output)
472 assert "events" in data
473 assert "total" in data
474 assert "stability_pct" in data
475 assert "modified_count" in data
476 assert isinstance(data["events"], list)
477 for ev in data["events"]:
478 assert "commit_id" in ev
479 assert "committed_at" in ev
480 assert "event" in ev
481 assert "message" in ev
482 # Full SHA — not truncated
483 assert len(ev["commit_id"]) == 64
484
485 def test_no_address_separator_rejected(self, code_repo: pathlib.Path) -> None:
486 result = runner.invoke(cli, ["code", "lineage", "billing.py"])
487 assert result.exit_code == 1
488
489 def test_missing_symbol_returns_zero_events(self, code_repo: pathlib.Path) -> None:
490 result = runner.invoke(cli, ["code", "lineage", "billing.py::nonexistent_xyz"])
491 assert result.exit_code == 0
492 assert "no events found" in result.output
493
494 def test_count_only_outputs_integer(self, code_repo: pathlib.Path) -> None:
495 result = runner.invoke(cli, ["code", "lineage", "--count", "billing.py::process_order"])
496 assert result.exit_code == 0
497 assert result.output.strip().isdigit()
498
499 def test_filter_created_subset(self, code_repo: pathlib.Path) -> None:
500 result = runner.invoke(cli, [
501 "code", "lineage", "--filter", "created",
502 "--json", "billing.py::process_order",
503 ])
504 assert result.exit_code == 0
505 data = json.loads(result.output)
506 for ev in data["events"]:
507 assert ev["event"] == "created"
508
509 def test_filter_modified_subset(self, code_repo: pathlib.Path) -> None:
510 result = runner.invoke(cli, [
511 "code", "lineage", "--filter", "modified",
512 "--json", "billing.py::process_order",
513 ])
514 assert result.exit_code == 0
515 data = json.loads(result.output)
516 for ev in data["events"]:
517 assert ev["event"] == "modified"
518
519 def test_since_future_returns_empty(self, code_repo: pathlib.Path) -> None:
520 result = runner.invoke(cli, [
521 "code", "lineage", "--since", "2099-01-01",
522 "billing.py::process_order",
523 ])
524 assert result.exit_code == 0
525 assert "no events found" in result.output
526
527 def test_since_invalid_date_rejected(self, code_repo: pathlib.Path) -> None:
528 result = runner.invoke(cli, [
529 "code", "lineage", "--since", "not-a-date",
530 "billing.py::process_order",
531 ])
532 assert result.exit_code == 1
533
534 def test_until_invalid_date_rejected(self, code_repo: pathlib.Path) -> None:
535 result = runner.invoke(cli, [
536 "code", "lineage", "--until", "99/99/99",
537 "billing.py::process_order",
538 ])
539 assert result.exit_code == 1
540
541 def test_commit_flag_accepted(self, code_repo: pathlib.Path) -> None:
542 result = runner.invoke(cli, [
543 "code", "lineage", "--commit", "HEAD",
544 "billing.py::process_order",
545 ])
546 assert result.exit_code == 0
547
548 def test_stability_flag_present_in_output(self, code_repo: pathlib.Path) -> None:
549 result = runner.invoke(cli, [
550 "code", "lineage", "--stability", "billing.py::process_order",
551 ])
552 assert result.exit_code == 0
553 # Stability line only appears when there are events.
554 # Just check no crash.
555
556 def test_unknown_branch_rejected(self, code_repo: pathlib.Path) -> None:
557 result = runner.invoke(cli, [
558 "code", "lineage", "--branch", "nonexistent-branch-xyz",
559 "billing.py::process_order",
560 ])
561 assert result.exit_code == 1
562
563 def test_requires_repo(self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch) -> None:
564 monkeypatch.chdir(tmp_path)
565 result = runner.invoke(cli, ["code", "lineage", "src/a.py::f"])
566 assert result.exit_code != 0
567
568 def test_json_events_have_full_commit_id(self, code_repo: pathlib.Path) -> None:
569 result = runner.invoke(cli, ["code", "lineage", "--json", "billing.py::process_order"])
570 assert result.exit_code == 0
571 data = json.loads(result.output)
572 for ev in data["events"]:
573 assert len(ev["commit_id"]) == 64, "commit_id should be full SHA, not truncated"
File History 1 commit
sha256:b89fa4fd9ca0d692fc66f6b9aef4c3a0c13c8e9b439faf42da8e91e09f048d4f tests/test_cmd_revert_hardening.py, tests/test_cmd_semantic… Human 41 days ago