test_lineage_supercharge.py
python
sha256:74b5023693ac2ab80e3b89fddc66e0d60d7d931a1266d3f9294f645c3102fe76
tests/test_lineage_supercharge.py, tests/test_narrative_sup…
Human
15 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.commits 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 | branch="main", |
| 87 | snapshot_id=f"snap-{cid}", |
| 88 | message=message, |
| 89 | committed_at=_ts(day), |
| 90 | structured_delta={"ops": ops or [], "domain": "code", "summary": message}, |
| 91 | ) |
| 92 | |
| 93 | |
| 94 | def _insert(address: str, content_id: str) -> InsertOp: |
| 95 | return InsertOp( |
| 96 | op="insert", |
| 97 | address=address, |
| 98 | position=None, |
| 99 | content_id=_cid(content_id), |
| 100 | content_summary=f"function {address.split('::')[-1]}", |
| 101 | ) |
| 102 | |
| 103 | |
| 104 | def _delete(address: str, content_id: str) -> DeleteOp: |
| 105 | return DeleteOp( |
| 106 | op="delete", |
| 107 | address=address, |
| 108 | position=None, |
| 109 | content_id=_cid(content_id), |
| 110 | content_summary=f"function {address.split('::')[-1]}", |
| 111 | ) |
| 112 | |
| 113 | |
| 114 | def _replace( |
| 115 | address: str, |
| 116 | old_cid: str, |
| 117 | new_cid: str, |
| 118 | old_sum: str = "", |
| 119 | new_sum: str = "", |
| 120 | ) -> ReplaceOp: |
| 121 | return ReplaceOp( |
| 122 | op="replace", |
| 123 | address=address, |
| 124 | position=None, |
| 125 | old_content_id=_cid(old_cid), |
| 126 | new_content_id=_cid(new_cid), |
| 127 | old_summary=old_sum, |
| 128 | new_summary=new_sum, |
| 129 | ) |
| 130 | |
| 131 | |
| 132 | @pytest.fixture |
| 133 | def repo(tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch) -> pathlib.Path: |
| 134 | monkeypatch.chdir(tmp_path) |
| 135 | monkeypatch.setenv("MUSE_REPO_ROOT", str(tmp_path)) |
| 136 | r = runner.invoke(cli, ["init", "--domain", "code"]) |
| 137 | assert r.exit_code == 0, r.output |
| 138 | return tmp_path |
| 139 | |
| 140 | |
| 141 | @pytest.fixture |
| 142 | def code_repo(repo: pathlib.Path) -> pathlib.Path: |
| 143 | """Repo with a two-commit history: created + renamed.""" |
| 144 | (repo / "billing.py").write_text(textwrap.dedent("""\ |
| 145 | def compute_total(items): |
| 146 | return sum(items) |
| 147 | |
| 148 | def process_order(invoice, items): |
| 149 | return compute_total(items) |
| 150 | """)) |
| 151 | r = runner.invoke(cli, ["commit", "-m", "Initial billing module"]) |
| 152 | assert r.exit_code == 0, r.output |
| 153 | |
| 154 | (repo / "billing.py").write_text(textwrap.dedent("""\ |
| 155 | def compute_invoice_total(items): |
| 156 | return sum(items) |
| 157 | |
| 158 | def process_order(invoice, items): |
| 159 | return compute_invoice_total(items) |
| 160 | """)) |
| 161 | r = runner.invoke(cli, ["commit", "-m", "Rename compute_total"]) |
| 162 | assert r.exit_code == 0, r.output |
| 163 | return repo |
| 164 | |
| 165 | |
| 166 | @pytest.fixture |
| 167 | def modified_repo(repo: pathlib.Path) -> pathlib.Path: |
| 168 | """Repo with three commits: created, modified, modified.""" |
| 169 | (repo / "billing.py").write_text("def compute_total(items):\n return sum(items)\n") |
| 170 | r = runner.invoke(cli, ["commit", "-m", "create"]) |
| 171 | assert r.exit_code == 0, r.output |
| 172 | |
| 173 | (repo / "billing.py").write_text("def compute_total(items, tax=0):\n return sum(items) + tax\n") |
| 174 | r = runner.invoke(cli, ["commit", "-m", "add tax"]) |
| 175 | assert r.exit_code == 0, r.output |
| 176 | |
| 177 | (repo / "billing.py").write_text("def compute_total(items, tax=0, currency='USD'):\n return sum(items) + tax\n") |
| 178 | r = runner.invoke(cli, ["commit", "-m", "add currency"]) |
| 179 | assert r.exit_code == 0, r.output |
| 180 | return repo |
| 181 | |
| 182 | |
| 183 | # --------------------------------------------------------------------------- |
| 184 | # JSON Envelope — exit_code, duration_ms, -j, address |
| 185 | # --------------------------------------------------------------------------- |
| 186 | |
| 187 | |
| 188 | class TestJsonEnvelope: |
| 189 | def test_exit_code_zero_in_json(self, code_repo: pathlib.Path) -> None: |
| 190 | result = runner.invoke(cli, ["code", "lineage", "--json", "billing.py::process_order"]) |
| 191 | assert result.exit_code == 0, result.output |
| 192 | data = json.loads(result.output) |
| 193 | assert "exit_code" in data, "JSON must include exit_code" |
| 194 | assert data["exit_code"] == 0 |
| 195 | |
| 196 | def test_duration_ms_non_negative_float(self, code_repo: pathlib.Path) -> None: |
| 197 | result = runner.invoke(cli, ["code", "lineage", "--json", "billing.py::process_order"]) |
| 198 | assert result.exit_code == 0, result.output |
| 199 | data = json.loads(result.output) |
| 200 | assert "duration_ms" in data, "JSON must include duration_ms" |
| 201 | assert isinstance(data["duration_ms"], float | int) |
| 202 | assert data["duration_ms"] >= 0 |
| 203 | |
| 204 | def test_j_alias_works(self, code_repo: pathlib.Path) -> None: |
| 205 | result = runner.invoke(cli, ["code", "lineage", "-j", "billing.py::process_order"]) |
| 206 | assert result.exit_code == 0, result.output |
| 207 | data = json.loads(result.output) |
| 208 | assert "events" in data, "-j must produce same JSON as --json" |
| 209 | assert "exit_code" in data |
| 210 | |
| 211 | def test_j_alias_output_matches_json_flag(self, code_repo: pathlib.Path) -> None: |
| 212 | r1 = runner.invoke(cli, ["code", "lineage", "--json", "billing.py::process_order"]) |
| 213 | r2 = runner.invoke(cli, ["code", "lineage", "-j", "billing.py::process_order"]) |
| 214 | assert r1.exit_code == 0 |
| 215 | assert r2.exit_code == 0 |
| 216 | d1 = json.loads(r1.output) |
| 217 | d2 = json.loads(r2.output) |
| 218 | # All structural keys must match (duration_ms may differ slightly) |
| 219 | for key in ("address", "total", "exit_code", "events", "stability_pct"): |
| 220 | assert d1[key] == d2[key], f"key {key!r} differs between -j and --json" |
| 221 | |
| 222 | def test_address_echoed_in_json(self, code_repo: pathlib.Path) -> None: |
| 223 | result = runner.invoke(cli, ["code", "lineage", "--json", "billing.py::process_order"]) |
| 224 | assert result.exit_code == 0 |
| 225 | data = json.loads(result.output) |
| 226 | assert data["address"] == "billing.py::process_order" |
| 227 | |
| 228 | def test_json_schema_all_required_keys(self, code_repo: pathlib.Path) -> None: |
| 229 | result = runner.invoke(cli, ["code", "lineage", "--json", "billing.py::process_order"]) |
| 230 | assert result.exit_code == 0 |
| 231 | data = json.loads(result.output) |
| 232 | required = {"address", "total", "events", "stability_pct", "modified_count", |
| 233 | "exit_code", "duration_ms"} |
| 234 | missing = required - set(data.keys()) |
| 235 | assert not missing, f"JSON missing keys: {missing}" |
| 236 | |
| 237 | |
| 238 | # --------------------------------------------------------------------------- |
| 239 | # Context Fields — applied constraints echoed for agent verification |
| 240 | # --------------------------------------------------------------------------- |
| 241 | |
| 242 | |
| 243 | class TestJsonContextFields: |
| 244 | def test_filter_field_present_when_applied(self, code_repo: pathlib.Path) -> None: |
| 245 | result = runner.invoke(cli, [ |
| 246 | "code", "lineage", "--json", "--filter", "created", |
| 247 | "billing.py::process_order", |
| 248 | ]) |
| 249 | assert result.exit_code == 0 |
| 250 | data = json.loads(result.output) |
| 251 | assert "filter" in data, "JSON must echo the applied filter" |
| 252 | assert data["filter"] == "created" |
| 253 | |
| 254 | def test_filter_field_none_when_not_applied(self, code_repo: pathlib.Path) -> None: |
| 255 | result = runner.invoke(cli, ["code", "lineage", "--json", "billing.py::process_order"]) |
| 256 | assert result.exit_code == 0 |
| 257 | data = json.loads(result.output) |
| 258 | assert "filter" in data |
| 259 | assert data["filter"] is None |
| 260 | |
| 261 | def test_since_field_in_json_when_applied(self, code_repo: pathlib.Path) -> None: |
| 262 | result = runner.invoke(cli, [ |
| 263 | "code", "lineage", "--json", "--since", "2020-01-01", |
| 264 | "billing.py::process_order", |
| 265 | ]) |
| 266 | assert result.exit_code == 0 |
| 267 | data = json.loads(result.output) |
| 268 | assert "since" in data, "JSON must echo the since date" |
| 269 | assert data["since"] == "2020-01-01" |
| 270 | |
| 271 | def test_until_field_in_json_when_applied(self, code_repo: pathlib.Path) -> None: |
| 272 | result = runner.invoke(cli, [ |
| 273 | "code", "lineage", "--json", "--until", "2099-01-01", |
| 274 | "billing.py::process_order", |
| 275 | ]) |
| 276 | assert result.exit_code == 0 |
| 277 | data = json.loads(result.output) |
| 278 | assert "until" in data, "JSON must echo the until date" |
| 279 | assert data["until"] == "2099-01-01" |
| 280 | |
| 281 | def test_since_field_none_when_not_applied(self, code_repo: pathlib.Path) -> None: |
| 282 | result = runner.invoke(cli, ["code", "lineage", "--json", "billing.py::process_order"]) |
| 283 | assert result.exit_code == 0 |
| 284 | data = json.loads(result.output) |
| 285 | assert "since" in data |
| 286 | assert data["since"] is None |
| 287 | |
| 288 | def test_until_field_none_when_not_applied(self, code_repo: pathlib.Path) -> None: |
| 289 | result = runner.invoke(cli, ["code", "lineage", "--json", "billing.py::process_order"]) |
| 290 | assert result.exit_code == 0 |
| 291 | data = json.loads(result.output) |
| 292 | assert "until" in data |
| 293 | assert data["until"] is None |
| 294 | |
| 295 | |
| 296 | # --------------------------------------------------------------------------- |
| 297 | # commit_id Format — sha256: prefix, not bare hex |
| 298 | # --------------------------------------------------------------------------- |
| 299 | |
| 300 | |
| 301 | class TestCommitIdFormat: |
| 302 | def test_commit_id_uses_sha256_prefix(self, code_repo: pathlib.Path) -> None: |
| 303 | result = runner.invoke(cli, ["code", "lineage", "--json", "billing.py::process_order"]) |
| 304 | assert result.exit_code == 0 |
| 305 | data = json.loads(result.output) |
| 306 | for ev in data["events"]: |
| 307 | assert ev["commit_id"].startswith("sha256:"), ( |
| 308 | f"commit_id must start with 'sha256:', got: {ev['commit_id']!r}" |
| 309 | ) |
| 310 | |
| 311 | def test_commit_id_full_length(self, code_repo: pathlib.Path) -> None: |
| 312 | """sha256:<64-hex> = 71 chars total — not truncated.""" |
| 313 | result = runner.invoke(cli, ["code", "lineage", "--json", "billing.py::process_order"]) |
| 314 | assert result.exit_code == 0 |
| 315 | data = json.loads(result.output) |
| 316 | for ev in data["events"]: |
| 317 | assert len(ev["commit_id"]) == 71, ( |
| 318 | f"Expected sha256:<64-hex> (71 chars), got {len(ev['commit_id'])}: {ev['commit_id']!r}" |
| 319 | ) |
| 320 | |
| 321 | |
| 322 | # --------------------------------------------------------------------------- |
| 323 | # Event Detail Fields — content_id and detail preservation |
| 324 | # --------------------------------------------------------------------------- |
| 325 | |
| 326 | |
| 327 | class TestEventDetailFields: |
| 328 | def test_modified_event_has_old_and_new_content_id( |
| 329 | self, modified_repo: pathlib.Path |
| 330 | ) -> None: |
| 331 | result = runner.invoke(cli, [ |
| 332 | "code", "lineage", "--json", "--filter", "modified", |
| 333 | "billing.py::compute_total", |
| 334 | ]) |
| 335 | assert result.exit_code == 0 |
| 336 | data = json.loads(result.output) |
| 337 | for ev in data["events"]: |
| 338 | assert "old_content_id" in ev, "modified events must have old_content_id" |
| 339 | assert "new_content_id" in ev, "modified events must have new_content_id" |
| 340 | |
| 341 | def test_created_event_has_new_content_id(self, code_repo: pathlib.Path) -> None: |
| 342 | result = runner.invoke(cli, [ |
| 343 | "code", "lineage", "--json", "--filter", "created", |
| 344 | "billing.py::process_order", |
| 345 | ]) |
| 346 | assert result.exit_code == 0 |
| 347 | data = json.loads(result.output) |
| 348 | for ev in data["events"]: |
| 349 | assert "new_content_id" in ev, "created events must have new_content_id" |
| 350 | |
| 351 | def test_renamed_event_has_detail_in_json(self, code_repo: pathlib.Path) -> None: |
| 352 | """renamed_from events must carry the source address in detail.""" |
| 353 | result = runner.invoke(cli, [ |
| 354 | "code", "lineage", "--json", "--filter", "renamed_from", |
| 355 | "billing.py::compute_invoice_total", |
| 356 | ]) |
| 357 | assert result.exit_code == 0 |
| 358 | data = json.loads(result.output) |
| 359 | for ev in data["events"]: |
| 360 | assert ev["event"] == "renamed_from" |
| 361 | assert "detail" in ev, "renamed_from events must include detail (source address)" |
| 362 | assert "::" in ev["detail"], f"detail should be a symbol address, got: {ev['detail']!r}" |
| 363 | |
| 364 | |
| 365 | # --------------------------------------------------------------------------- |
| 366 | # Kind Filters — all six values |
| 367 | # --------------------------------------------------------------------------- |
| 368 | |
| 369 | |
| 370 | class TestKindFilters: |
| 371 | def test_filter_created_only(self, code_repo: pathlib.Path) -> None: |
| 372 | result = runner.invoke(cli, [ |
| 373 | "code", "lineage", "--json", "--filter", "created", |
| 374 | "billing.py::process_order", |
| 375 | ]) |
| 376 | assert result.exit_code == 0 |
| 377 | data = json.loads(result.output) |
| 378 | for ev in data["events"]: |
| 379 | assert ev["event"] == "created" |
| 380 | |
| 381 | def test_filter_modified_only(self, modified_repo: pathlib.Path) -> None: |
| 382 | result = runner.invoke(cli, [ |
| 383 | "code", "lineage", "--json", "--filter", "modified", |
| 384 | "billing.py::compute_total", |
| 385 | ]) |
| 386 | assert result.exit_code == 0 |
| 387 | data = json.loads(result.output) |
| 388 | for ev in data["events"]: |
| 389 | assert ev["event"] == "modified" |
| 390 | |
| 391 | def test_filter_deleted_only(self, repo: pathlib.Path) -> None: |
| 392 | """Create then delete a symbol — filter should return only the delete event.""" |
| 393 | (repo / "billing.py").write_text("def helper(): return 1\n") |
| 394 | r = runner.invoke(cli, ["commit", "-m", "add helper"]) |
| 395 | assert r.exit_code == 0, r.output |
| 396 | |
| 397 | (repo / "billing.py").write_text("# helper removed\n") |
| 398 | r = runner.invoke(cli, ["commit", "-m", "remove helper"]) |
| 399 | assert r.exit_code == 0, r.output |
| 400 | |
| 401 | result = runner.invoke(cli, [ |
| 402 | "code", "lineage", "--json", "--filter", "deleted", |
| 403 | "billing.py::helper", |
| 404 | ]) |
| 405 | assert result.exit_code == 0 |
| 406 | data = json.loads(result.output) |
| 407 | for ev in data["events"]: |
| 408 | assert ev["event"] == "deleted" |
| 409 | |
| 410 | def test_filter_renamed_from_only(self, code_repo: pathlib.Path) -> None: |
| 411 | result = runner.invoke(cli, [ |
| 412 | "code", "lineage", "--json", "--filter", "renamed_from", |
| 413 | "billing.py::compute_invoice_total", |
| 414 | ]) |
| 415 | assert result.exit_code == 0 |
| 416 | data = json.loads(result.output) |
| 417 | for ev in data["events"]: |
| 418 | assert ev["event"] == "renamed_from" |
| 419 | |
| 420 | def test_filter_returns_empty_for_unmatched_kind(self, code_repo: pathlib.Path) -> None: |
| 421 | """Filtering for 'deleted' on a symbol that was never deleted → empty events list.""" |
| 422 | result = runner.invoke(cli, [ |
| 423 | "code", "lineage", "--json", "--filter", "deleted", |
| 424 | "billing.py::process_order", |
| 425 | ]) |
| 426 | assert result.exit_code == 0 |
| 427 | data = json.loads(result.output) |
| 428 | assert data["total"] == 0 |
| 429 | assert data["events"] == [] |
| 430 | |
| 431 | def test_invalid_filter_rejected_by_argparse(self, code_repo: pathlib.Path) -> None: |
| 432 | result = runner.invoke(cli, [ |
| 433 | "code", "lineage", "--filter", "bogus_kind", |
| 434 | "billing.py::process_order", |
| 435 | ]) |
| 436 | assert result.exit_code != 0 |
| 437 | |
| 438 | |
| 439 | # --------------------------------------------------------------------------- |
| 440 | # Stability in JSON |
| 441 | # --------------------------------------------------------------------------- |
| 442 | |
| 443 | |
| 444 | class TestStabilityInJson: |
| 445 | def test_stability_pct_always_in_json(self, code_repo: pathlib.Path) -> None: |
| 446 | """stability_pct is emitted without the --stability flag — agents always get it.""" |
| 447 | result = runner.invoke(cli, ["code", "lineage", "--json", "billing.py::process_order"]) |
| 448 | assert result.exit_code == 0 |
| 449 | data = json.loads(result.output) |
| 450 | assert "stability_pct" in data |
| 451 | assert isinstance(data["stability_pct"], int) |
| 452 | |
| 453 | def test_stability_pct_100_when_no_modifications(self, code_repo: pathlib.Path) -> None: |
| 454 | result = runner.invoke(cli, [ |
| 455 | "code", "lineage", "--json", "--filter", "created", |
| 456 | "billing.py::process_order", |
| 457 | ]) |
| 458 | assert result.exit_code == 0 |
| 459 | data = json.loads(result.output) |
| 460 | # Only created events → no modifications → stability 100% |
| 461 | assert data["stability_pct"] == 100 |
| 462 | |
| 463 | def test_stability_pct_zero_when_all_filtered_to_modified( |
| 464 | self, modified_repo: pathlib.Path |
| 465 | ) -> None: |
| 466 | result = runner.invoke(cli, [ |
| 467 | "code", "lineage", "--json", "--filter", "modified", |
| 468 | "billing.py::compute_total", |
| 469 | ]) |
| 470 | assert result.exit_code == 0 |
| 471 | data = json.loads(result.output) |
| 472 | if data["total"] > 0: |
| 473 | assert data["stability_pct"] == 0 |
| 474 | |
| 475 | def test_modified_count_matches_filtered_events(self, modified_repo: pathlib.Path) -> None: |
| 476 | result = runner.invoke(cli, [ |
| 477 | "code", "lineage", "--json", "billing.py::compute_total", |
| 478 | ]) |
| 479 | assert result.exit_code == 0 |
| 480 | data = json.loads(result.output) |
| 481 | actual_modified = sum(1 for ev in data["events"] if ev["event"] == "modified") |
| 482 | assert data["modified_count"] == actual_modified |
| 483 | |
| 484 | |
| 485 | # --------------------------------------------------------------------------- |
| 486 | # --count flag |
| 487 | # --------------------------------------------------------------------------- |
| 488 | |
| 489 | |
| 490 | class TestCountFlag: |
| 491 | def test_count_only_outputs_integer(self, code_repo: pathlib.Path) -> None: |
| 492 | result = runner.invoke(cli, [ |
| 493 | "code", "lineage", "--count", "billing.py::process_order", |
| 494 | ]) |
| 495 | assert result.exit_code == 0 |
| 496 | assert result.output.strip().isdigit() |
| 497 | |
| 498 | def test_count_with_json_emits_structured_total(self, code_repo: pathlib.Path) -> None: |
| 499 | """--count --json emits full JSON (with 'total' field) not bare integer.""" |
| 500 | result = runner.invoke(cli, [ |
| 501 | "code", "lineage", "--count", "--json", "billing.py::process_order", |
| 502 | ]) |
| 503 | assert result.exit_code == 0 |
| 504 | data = json.loads(result.output) |
| 505 | assert "total" in data |
| 506 | assert isinstance(data["total"], int) |
| 507 | assert "exit_code" in data |
| 508 | |
| 509 | def test_count_filter_combination(self, modified_repo: pathlib.Path) -> None: |
| 510 | """--count --filter modified returns count of only modified events.""" |
| 511 | result = runner.invoke(cli, [ |
| 512 | "code", "lineage", "--count", "--filter", "modified", |
| 513 | "billing.py::compute_total", |
| 514 | ]) |
| 515 | assert result.exit_code == 0 |
| 516 | count = int(result.output.strip()) |
| 517 | assert count >= 2 # two modifications in modified_repo fixture |
| 518 | |
| 519 | |
| 520 | # --------------------------------------------------------------------------- |
| 521 | # Docstring / TypedDict export |
| 522 | # --------------------------------------------------------------------------- |
| 523 | |
| 524 | |
| 525 | class TestTypedDictExport: |
| 526 | def test_lineage_json_typeddict_importable(self) -> None: |
| 527 | """_LineageJson TypedDict must be importable from lineage module.""" |
| 528 | from muse.cli.commands.lineage import _LineageJson # type: ignore[attr-defined] |
| 529 | assert _LineageJson is not None |
| 530 | |
| 531 | def test_typeddict_has_required_fields(self) -> None: |
| 532 | from muse.cli.commands.lineage import _LineageJson # type: ignore[attr-defined] |
| 533 | annotations = _LineageJson.__annotations__ |
| 534 | required = {"address", "total", "events", "stability_pct", "modified_count", |
| 535 | "exit_code", "duration_ms", "filter", "since", "until"} |
| 536 | missing = required - set(annotations) |
| 537 | assert not missing, f"_LineageJson missing annotations: {missing}" |
| 538 | |
| 539 | |
| 540 | # --------------------------------------------------------------------------- |
| 541 | # Classify replace — docstring gap: "impl_only" documented but never returned |
| 542 | # --------------------------------------------------------------------------- |
| 543 | |
| 544 | |
| 545 | class TestClassifyReplaceDocstringAccuracy: |
| 546 | """Verify _classify_replace only returns documented values.""" |
| 547 | |
| 548 | def test_signature_change_on_signature_keyword(self) -> None: |
| 549 | assert _classify_replace("signature changed", "") == "signature_change" |
| 550 | |
| 551 | def test_full_rewrite_is_default(self) -> None: |
| 552 | result = _classify_replace("body rewritten entirely", "") |
| 553 | assert result in ("full_rewrite", "impl_only"), ( |
| 554 | f"_classify_replace returned unexpected value: {result!r}" |
| 555 | ) |
| 556 | |
| 557 | def test_return_value_is_a_known_kind(self) -> None: |
| 558 | known = {"signature_change", "full_rewrite", "impl_only"} |
| 559 | for old_s, new_s in [ |
| 560 | ("", ""), |
| 561 | ("signature changed", ""), |
| 562 | ("", "new signature here"), |
| 563 | ("impl updated", "impl updated v2"), |
| 564 | ("complete rewrite", "new logic"), |
| 565 | ]: |
| 566 | result = _classify_replace(old_s, new_s) |
| 567 | assert result in known, ( |
| 568 | f"_classify_replace({old_s!r}, {new_s!r}) → {result!r} not in {known}" |
| 569 | ) |
| 570 | |
| 571 | |
| 572 | # --------------------------------------------------------------------------- |
| 573 | # TestRegisterFlags — argparse-level verification |
| 574 | # --------------------------------------------------------------------------- |
| 575 | |
| 576 | |
| 577 | class TestRegisterFlags: |
| 578 | """Verify that register() wires --json / -j correctly.""" |
| 579 | |
| 580 | def _make_parser(self) -> "argparse.ArgumentParser": |
| 581 | import argparse |
| 582 | from muse.cli.commands.lineage import register |
| 583 | ap = argparse.ArgumentParser() |
| 584 | subs = ap.add_subparsers() |
| 585 | register(subs) |
| 586 | return ap |
| 587 | |
| 588 | def test_json_flag_long(self) -> None: |
| 589 | ns = self._make_parser().parse_args(["lineage", "file.py::Fn", "--json"]) |
| 590 | assert ns.json_out is True |
| 591 | |
| 592 | def test_j_alias(self) -> None: |
| 593 | ns = self._make_parser().parse_args(["lineage", "file.py::Fn", "-j"]) |
| 594 | assert ns.json_out is True |
| 595 | |
| 596 | def test_default_is_text(self) -> None: |
| 597 | ns = self._make_parser().parse_args(["lineage", "file.py::Fn"]) |
| 598 | assert ns.json_out is False |
| 599 | |
| 600 | def test_dest_is_json_out(self) -> None: |
| 601 | ns = self._make_parser().parse_args(["lineage", "file.py::Fn", "-j"]) |
| 602 | assert hasattr(ns, "json_out") |
| 603 | assert not hasattr(ns, "fmt") |
File History
1 commit
sha256:74b5023693ac2ab80e3b89fddc66e0d60d7d931a1266d3f9294f645c3102fe76
tests/test_lineage_supercharge.py, tests/test_narrative_sup…
Human
15 days ago