gabriel / muse public
test_lineage_supercharge.py python
570 lines 22.1 KB
Raw
sha256:88ac91129873e6a496e9189515aa690eb893ae25d69c8f72af141a2be5068eb3 docs: docstring sprint contract→find-symbol — idiomatic run… Sonnet 4.6 patch 138 days ago
1 """Supercharge tests for muse code lineage.
2
3 Coverage
4 --------
5 JSON Envelope
6 exit_code — always 0; confirms clean exit for agents
7 duration_ms — non-negative float; timing telemetry
8 -j alias — shorthand for --json (agent-ergonomic)
9 address — symbol address echoed back; agents can verify the target
10
11 Context Fields (agent-verifiable constraints applied to the run)
12 filter — kind_filter echoed in JSON (or None when unset)
13 since — lower date bound echoed as ISO string (or None)
14 until — upper date bound echoed as ISO string (or None)
15
16 Event Detail Fields
17 renamed_from / moved_from / copied_from — detail field preserved in JSON
18 modified — old_content_id / new_content_id present in JSON event
19 deleted — old_content_id present in JSON event
20 created — new_content_id present in JSON event
21
22 Kind Filters (all six enumerated values)
23 created, modified, deleted, renamed_from, moved_from, copied_from
24
25 commit_id Format
26 Real commit_ids use sha256:<64-hex> format (71 chars), not bare hex
27
28 Stability
29 stability_pct computed correctly after kind_filter
30 stability_pct 100 when no modifications
31
32 TypedDict
33 _LineageJson exported — all output keys documented
34 """
35
36 from __future__ import annotations
37
38 import datetime
39 import json
40 import pathlib
41 import textwrap
42
43 import pytest
44
45 from tests.cli_test_helper import CliRunner
46 from muse.cli.commands.lineage import (
47 _LineageEvent,
48 _classify_replace,
49 _stability,
50 build_lineage,
51 )
52 from muse.core.store import CommitRecord
53 from muse.domain import DeleteOp, DomainOp, InsertOp, ReplaceOp
54
55 cli = None
56 runner = CliRunner()
57
58 # ---------------------------------------------------------------------------
59 # Shared fixtures
60 # ---------------------------------------------------------------------------
61
62 _REPO_ID = "test-repo-id"
63 _SEQ: list[int] = [0]
64
65
66 def _cid(tag: str) -> str:
67 return tag.ljust(64, "0")[:64]
68
69
70 def _ts(offset_days: int = 0) -> datetime.datetime:
71 base = datetime.datetime(2026, 1, 1, 12, 0, 0, tzinfo=datetime.timezone.utc)
72 return base + datetime.timedelta(days=offset_days)
73
74
75 def _commit(
76 *,
77 message: str = "commit",
78 ops: list[DomainOp] | None = None,
79 day: int = 0,
80 commit_id: str | None = None,
81 ) -> CommitRecord:
82 _SEQ[0] += 1
83 cid = commit_id or f"c{_SEQ[0]:063d}"
84 return CommitRecord(
85 commit_id=cid,
86 repo_id=_REPO_ID,
87 branch="main",
88 snapshot_id=f"snap-{cid}",
89 message=message,
90 committed_at=_ts(day),
91 structured_delta={"ops": ops or [], "domain": "code", "summary": message},
92 )
93
94
95 def _insert(address: str, content_id: str) -> InsertOp:
96 return InsertOp(
97 op="insert",
98 address=address,
99 position=None,
100 content_id=_cid(content_id),
101 content_summary=f"function {address.split('::')[-1]}",
102 )
103
104
105 def _delete(address: str, content_id: str) -> DeleteOp:
106 return DeleteOp(
107 op="delete",
108 address=address,
109 position=None,
110 content_id=_cid(content_id),
111 content_summary=f"function {address.split('::')[-1]}",
112 )
113
114
115 def _replace(
116 address: str,
117 old_cid: str,
118 new_cid: str,
119 old_sum: str = "",
120 new_sum: str = "",
121 ) -> ReplaceOp:
122 return ReplaceOp(
123 op="replace",
124 address=address,
125 position=None,
126 old_content_id=_cid(old_cid),
127 new_content_id=_cid(new_cid),
128 old_summary=old_sum,
129 new_summary=new_sum,
130 )
131
132
133 @pytest.fixture
134 def repo(tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch) -> pathlib.Path:
135 monkeypatch.chdir(tmp_path)
136 monkeypatch.setenv("MUSE_REPO_ROOT", str(tmp_path))
137 r = runner.invoke(cli, ["init", "--domain", "code"])
138 assert r.exit_code == 0, r.output
139 return tmp_path
140
141
142 @pytest.fixture
143 def code_repo(repo: pathlib.Path) -> pathlib.Path:
144 """Repo with a two-commit history: created + renamed."""
145 (repo / "billing.py").write_text(textwrap.dedent("""\
146 def compute_total(items):
147 return sum(items)
148
149 def process_order(invoice, items):
150 return compute_total(items)
151 """))
152 r = runner.invoke(cli, ["commit", "-m", "Initial billing module"])
153 assert r.exit_code == 0, r.output
154
155 (repo / "billing.py").write_text(textwrap.dedent("""\
156 def compute_invoice_total(items):
157 return sum(items)
158
159 def process_order(invoice, items):
160 return compute_invoice_total(items)
161 """))
162 r = runner.invoke(cli, ["commit", "-m", "Rename compute_total"])
163 assert r.exit_code == 0, r.output
164 return repo
165
166
167 @pytest.fixture
168 def modified_repo(repo: pathlib.Path) -> pathlib.Path:
169 """Repo with three commits: created, modified, modified."""
170 (repo / "billing.py").write_text("def compute_total(items):\n return sum(items)\n")
171 r = runner.invoke(cli, ["commit", "-m", "create"])
172 assert r.exit_code == 0, r.output
173
174 (repo / "billing.py").write_text("def compute_total(items, tax=0):\n return sum(items) + tax\n")
175 r = runner.invoke(cli, ["commit", "-m", "add tax"])
176 assert r.exit_code == 0, r.output
177
178 (repo / "billing.py").write_text("def compute_total(items, tax=0, currency='USD'):\n return sum(items) + tax\n")
179 r = runner.invoke(cli, ["commit", "-m", "add currency"])
180 assert r.exit_code == 0, r.output
181 return repo
182
183
184 # ---------------------------------------------------------------------------
185 # JSON Envelope — exit_code, duration_ms, -j, address
186 # ---------------------------------------------------------------------------
187
188
189 class TestJsonEnvelope:
190 def test_exit_code_zero_in_json(self, code_repo: pathlib.Path) -> None:
191 result = runner.invoke(cli, ["code", "lineage", "--json", "billing.py::process_order"])
192 assert result.exit_code == 0, result.output
193 data = json.loads(result.output)
194 assert "exit_code" in data, "JSON must include exit_code"
195 assert data["exit_code"] == 0
196
197 def test_duration_ms_non_negative_float(self, code_repo: pathlib.Path) -> None:
198 result = runner.invoke(cli, ["code", "lineage", "--json", "billing.py::process_order"])
199 assert result.exit_code == 0, result.output
200 data = json.loads(result.output)
201 assert "duration_ms" in data, "JSON must include duration_ms"
202 assert isinstance(data["duration_ms"], float | int)
203 assert data["duration_ms"] >= 0
204
205 def test_j_alias_works(self, code_repo: pathlib.Path) -> None:
206 result = runner.invoke(cli, ["code", "lineage", "-j", "billing.py::process_order"])
207 assert result.exit_code == 0, result.output
208 data = json.loads(result.output)
209 assert "events" in data, "-j must produce same JSON as --json"
210 assert "exit_code" in data
211
212 def test_j_alias_output_matches_json_flag(self, code_repo: pathlib.Path) -> None:
213 r1 = runner.invoke(cli, ["code", "lineage", "--json", "billing.py::process_order"])
214 r2 = runner.invoke(cli, ["code", "lineage", "-j", "billing.py::process_order"])
215 assert r1.exit_code == 0
216 assert r2.exit_code == 0
217 d1 = json.loads(r1.output)
218 d2 = json.loads(r2.output)
219 # All structural keys must match (duration_ms may differ slightly)
220 for key in ("address", "total", "exit_code", "events", "stability_pct"):
221 assert d1[key] == d2[key], f"key {key!r} differs between -j and --json"
222
223 def test_address_echoed_in_json(self, code_repo: pathlib.Path) -> None:
224 result = runner.invoke(cli, ["code", "lineage", "--json", "billing.py::process_order"])
225 assert result.exit_code == 0
226 data = json.loads(result.output)
227 assert data["address"] == "billing.py::process_order"
228
229 def test_json_schema_all_required_keys(self, code_repo: pathlib.Path) -> None:
230 result = runner.invoke(cli, ["code", "lineage", "--json", "billing.py::process_order"])
231 assert result.exit_code == 0
232 data = json.loads(result.output)
233 required = {"address", "total", "events", "stability_pct", "modified_count",
234 "exit_code", "duration_ms"}
235 missing = required - set(data.keys())
236 assert not missing, f"JSON missing keys: {missing}"
237
238
239 # ---------------------------------------------------------------------------
240 # Context Fields — applied constraints echoed for agent verification
241 # ---------------------------------------------------------------------------
242
243
244 class TestJsonContextFields:
245 def test_filter_field_present_when_applied(self, code_repo: pathlib.Path) -> None:
246 result = runner.invoke(cli, [
247 "code", "lineage", "--json", "--filter", "created",
248 "billing.py::process_order",
249 ])
250 assert result.exit_code == 0
251 data = json.loads(result.output)
252 assert "filter" in data, "JSON must echo the applied filter"
253 assert data["filter"] == "created"
254
255 def test_filter_field_none_when_not_applied(self, code_repo: pathlib.Path) -> None:
256 result = runner.invoke(cli, ["code", "lineage", "--json", "billing.py::process_order"])
257 assert result.exit_code == 0
258 data = json.loads(result.output)
259 assert "filter" in data
260 assert data["filter"] is None
261
262 def test_since_field_in_json_when_applied(self, code_repo: pathlib.Path) -> None:
263 result = runner.invoke(cli, [
264 "code", "lineage", "--json", "--since", "2020-01-01",
265 "billing.py::process_order",
266 ])
267 assert result.exit_code == 0
268 data = json.loads(result.output)
269 assert "since" in data, "JSON must echo the since date"
270 assert data["since"] == "2020-01-01"
271
272 def test_until_field_in_json_when_applied(self, code_repo: pathlib.Path) -> None:
273 result = runner.invoke(cli, [
274 "code", "lineage", "--json", "--until", "2099-01-01",
275 "billing.py::process_order",
276 ])
277 assert result.exit_code == 0
278 data = json.loads(result.output)
279 assert "until" in data, "JSON must echo the until date"
280 assert data["until"] == "2099-01-01"
281
282 def test_since_field_none_when_not_applied(self, code_repo: pathlib.Path) -> None:
283 result = runner.invoke(cli, ["code", "lineage", "--json", "billing.py::process_order"])
284 assert result.exit_code == 0
285 data = json.loads(result.output)
286 assert "since" in data
287 assert data["since"] is None
288
289 def test_until_field_none_when_not_applied(self, code_repo: pathlib.Path) -> None:
290 result = runner.invoke(cli, ["code", "lineage", "--json", "billing.py::process_order"])
291 assert result.exit_code == 0
292 data = json.loads(result.output)
293 assert "until" in data
294 assert data["until"] is None
295
296
297 # ---------------------------------------------------------------------------
298 # commit_id Format — sha256: prefix, not bare hex
299 # ---------------------------------------------------------------------------
300
301
302 class TestCommitIdFormat:
303 def test_commit_id_uses_sha256_prefix(self, code_repo: pathlib.Path) -> None:
304 result = runner.invoke(cli, ["code", "lineage", "--json", "billing.py::process_order"])
305 assert result.exit_code == 0
306 data = json.loads(result.output)
307 for ev in data["events"]:
308 assert ev["commit_id"].startswith("sha256:"), (
309 f"commit_id must start with 'sha256:', got: {ev['commit_id']!r}"
310 )
311
312 def test_commit_id_full_length(self, code_repo: pathlib.Path) -> None:
313 """sha256:<64-hex> = 71 chars total — not truncated."""
314 result = runner.invoke(cli, ["code", "lineage", "--json", "billing.py::process_order"])
315 assert result.exit_code == 0
316 data = json.loads(result.output)
317 for ev in data["events"]:
318 assert len(ev["commit_id"]) == 71, (
319 f"Expected sha256:<64-hex> (71 chars), got {len(ev['commit_id'])}: {ev['commit_id']!r}"
320 )
321
322
323 # ---------------------------------------------------------------------------
324 # Event Detail Fields — content_id and detail preservation
325 # ---------------------------------------------------------------------------
326
327
328 class TestEventDetailFields:
329 def test_modified_event_has_old_and_new_content_id(
330 self, modified_repo: pathlib.Path
331 ) -> None:
332 result = runner.invoke(cli, [
333 "code", "lineage", "--json", "--filter", "modified",
334 "billing.py::compute_total",
335 ])
336 assert result.exit_code == 0
337 data = json.loads(result.output)
338 for ev in data["events"]:
339 assert "old_content_id" in ev, "modified events must have old_content_id"
340 assert "new_content_id" in ev, "modified events must have new_content_id"
341
342 def test_created_event_has_new_content_id(self, code_repo: pathlib.Path) -> None:
343 result = runner.invoke(cli, [
344 "code", "lineage", "--json", "--filter", "created",
345 "billing.py::process_order",
346 ])
347 assert result.exit_code == 0
348 data = json.loads(result.output)
349 for ev in data["events"]:
350 assert "new_content_id" in ev, "created events must have new_content_id"
351
352 def test_renamed_event_has_detail_in_json(self, code_repo: pathlib.Path) -> None:
353 """renamed_from events must carry the source address in detail."""
354 result = runner.invoke(cli, [
355 "code", "lineage", "--json", "--filter", "renamed_from",
356 "billing.py::compute_invoice_total",
357 ])
358 assert result.exit_code == 0
359 data = json.loads(result.output)
360 for ev in data["events"]:
361 assert ev["event"] == "renamed_from"
362 assert "detail" in ev, "renamed_from events must include detail (source address)"
363 assert "::" in ev["detail"], f"detail should be a symbol address, got: {ev['detail']!r}"
364
365
366 # ---------------------------------------------------------------------------
367 # Kind Filters — all six values
368 # ---------------------------------------------------------------------------
369
370
371 class TestKindFilters:
372 def test_filter_created_only(self, code_repo: pathlib.Path) -> None:
373 result = runner.invoke(cli, [
374 "code", "lineage", "--json", "--filter", "created",
375 "billing.py::process_order",
376 ])
377 assert result.exit_code == 0
378 data = json.loads(result.output)
379 for ev in data["events"]:
380 assert ev["event"] == "created"
381
382 def test_filter_modified_only(self, modified_repo: pathlib.Path) -> None:
383 result = runner.invoke(cli, [
384 "code", "lineage", "--json", "--filter", "modified",
385 "billing.py::compute_total",
386 ])
387 assert result.exit_code == 0
388 data = json.loads(result.output)
389 for ev in data["events"]:
390 assert ev["event"] == "modified"
391
392 def test_filter_deleted_only(self, repo: pathlib.Path) -> None:
393 """Create then delete a symbol — filter should return only the delete event."""
394 (repo / "billing.py").write_text("def helper(): return 1\n")
395 r = runner.invoke(cli, ["commit", "-m", "add helper"])
396 assert r.exit_code == 0, r.output
397
398 (repo / "billing.py").write_text("# helper removed\n")
399 r = runner.invoke(cli, ["commit", "-m", "remove helper"])
400 assert r.exit_code == 0, r.output
401
402 result = runner.invoke(cli, [
403 "code", "lineage", "--json", "--filter", "deleted",
404 "billing.py::helper",
405 ])
406 assert result.exit_code == 0
407 data = json.loads(result.output)
408 for ev in data["events"]:
409 assert ev["event"] == "deleted"
410
411 def test_filter_renamed_from_only(self, code_repo: pathlib.Path) -> None:
412 result = runner.invoke(cli, [
413 "code", "lineage", "--json", "--filter", "renamed_from",
414 "billing.py::compute_invoice_total",
415 ])
416 assert result.exit_code == 0
417 data = json.loads(result.output)
418 for ev in data["events"]:
419 assert ev["event"] == "renamed_from"
420
421 def test_filter_returns_empty_for_unmatched_kind(self, code_repo: pathlib.Path) -> None:
422 """Filtering for 'deleted' on a symbol that was never deleted → empty events list."""
423 result = runner.invoke(cli, [
424 "code", "lineage", "--json", "--filter", "deleted",
425 "billing.py::process_order",
426 ])
427 assert result.exit_code == 0
428 data = json.loads(result.output)
429 assert data["total"] == 0
430 assert data["events"] == []
431
432 def test_invalid_filter_rejected_by_argparse(self, code_repo: pathlib.Path) -> None:
433 result = runner.invoke(cli, [
434 "code", "lineage", "--filter", "bogus_kind",
435 "billing.py::process_order",
436 ])
437 assert result.exit_code != 0
438
439
440 # ---------------------------------------------------------------------------
441 # Stability in JSON
442 # ---------------------------------------------------------------------------
443
444
445 class TestStabilityInJson:
446 def test_stability_pct_always_in_json(self, code_repo: pathlib.Path) -> None:
447 """stability_pct is emitted without the --stability flag — agents always get it."""
448 result = runner.invoke(cli, ["code", "lineage", "--json", "billing.py::process_order"])
449 assert result.exit_code == 0
450 data = json.loads(result.output)
451 assert "stability_pct" in data
452 assert isinstance(data["stability_pct"], int)
453
454 def test_stability_pct_100_when_no_modifications(self, code_repo: pathlib.Path) -> None:
455 result = runner.invoke(cli, [
456 "code", "lineage", "--json", "--filter", "created",
457 "billing.py::process_order",
458 ])
459 assert result.exit_code == 0
460 data = json.loads(result.output)
461 # Only created events → no modifications → stability 100%
462 assert data["stability_pct"] == 100
463
464 def test_stability_pct_zero_when_all_filtered_to_modified(
465 self, modified_repo: pathlib.Path
466 ) -> None:
467 result = runner.invoke(cli, [
468 "code", "lineage", "--json", "--filter", "modified",
469 "billing.py::compute_total",
470 ])
471 assert result.exit_code == 0
472 data = json.loads(result.output)
473 if data["total"] > 0:
474 assert data["stability_pct"] == 0
475
476 def test_modified_count_matches_filtered_events(self, modified_repo: pathlib.Path) -> None:
477 result = runner.invoke(cli, [
478 "code", "lineage", "--json", "billing.py::compute_total",
479 ])
480 assert result.exit_code == 0
481 data = json.loads(result.output)
482 actual_modified = sum(1 for ev in data["events"] if ev["event"] == "modified")
483 assert data["modified_count"] == actual_modified
484
485
486 # ---------------------------------------------------------------------------
487 # --count flag
488 # ---------------------------------------------------------------------------
489
490
491 class TestCountFlag:
492 def test_count_only_outputs_integer(self, code_repo: pathlib.Path) -> None:
493 result = runner.invoke(cli, [
494 "code", "lineage", "--count", "billing.py::process_order",
495 ])
496 assert result.exit_code == 0
497 assert result.output.strip().isdigit()
498
499 def test_count_with_json_emits_structured_total(self, code_repo: pathlib.Path) -> None:
500 """--count --json emits full JSON (with 'total' field) not bare integer."""
501 result = runner.invoke(cli, [
502 "code", "lineage", "--count", "--json", "billing.py::process_order",
503 ])
504 assert result.exit_code == 0
505 data = json.loads(result.output)
506 assert "total" in data
507 assert isinstance(data["total"], int)
508 assert "exit_code" in data
509
510 def test_count_filter_combination(self, modified_repo: pathlib.Path) -> None:
511 """--count --filter modified returns count of only modified events."""
512 result = runner.invoke(cli, [
513 "code", "lineage", "--count", "--filter", "modified",
514 "billing.py::compute_total",
515 ])
516 assert result.exit_code == 0
517 count = int(result.output.strip())
518 assert count >= 2 # two modifications in modified_repo fixture
519
520
521 # ---------------------------------------------------------------------------
522 # Docstring / TypedDict export
523 # ---------------------------------------------------------------------------
524
525
526 class TestTypedDictExport:
527 def test_lineage_json_typeddict_importable(self) -> None:
528 """_LineageJson TypedDict must be importable from lineage module."""
529 from muse.cli.commands.lineage import _LineageJson # type: ignore[attr-defined]
530 assert _LineageJson is not None
531
532 def test_typeddict_has_required_fields(self) -> None:
533 from muse.cli.commands.lineage import _LineageJson # type: ignore[attr-defined]
534 annotations = _LineageJson.__annotations__
535 required = {"address", "total", "events", "stability_pct", "modified_count",
536 "exit_code", "duration_ms", "filter", "since", "until"}
537 missing = required - set(annotations)
538 assert not missing, f"_LineageJson missing annotations: {missing}"
539
540
541 # ---------------------------------------------------------------------------
542 # Classify replace — docstring gap: "impl_only" documented but never returned
543 # ---------------------------------------------------------------------------
544
545
546 class TestClassifyReplaceDocstringAccuracy:
547 """Verify _classify_replace only returns documented values."""
548
549 def test_signature_change_on_signature_keyword(self) -> None:
550 assert _classify_replace("signature changed", "") == "signature_change"
551
552 def test_full_rewrite_is_default(self) -> None:
553 result = _classify_replace("body rewritten entirely", "")
554 assert result in ("full_rewrite", "impl_only"), (
555 f"_classify_replace returned unexpected value: {result!r}"
556 )
557
558 def test_return_value_is_a_known_kind(self) -> None:
559 known = {"signature_change", "full_rewrite", "impl_only"}
560 for old_s, new_s in [
561 ("", ""),
562 ("signature changed", ""),
563 ("", "new signature here"),
564 ("impl updated", "impl updated v2"),
565 ("complete rewrite", "new logic"),
566 ]:
567 result = _classify_replace(old_s, new_s)
568 assert result in known, (
569 f"_classify_replace({old_s!r}, {new_s!r}) → {result!r} not in {known}"
570 )
File History 1 commit
sha256:88ac91129873e6a496e9189515aa690eb893ae25d69c8f72af141a2be5068eb3 docs: docstring sprint contract→find-symbol — idiomatic run… Sonnet 4.6 patch 138 days ago