gabriel / muse public
test_symbol_log_supercharge.py python
698 lines 31.1 KB
Raw
sha256:51ce277f663e01a43eaffbe77509b1de7ac2d4251b55d23306304bcdeb92c90d feat(pack): delta-encode snapshots in MPackBundle wire format Sonnet 4.6 minor ⚠ breaking 129 days ago
1 """Seven-tier tests for ``muse/cli/commands/symbol_log.py``.
2
3 Tiers
4 -----
5 Unit — TypedDict fields, SymbolEvent constructor/to_dict, _flat_ops,
6 _find_events_in_commit for each EventKind, _print_human branches.
7 Integration — -j alias parity, JSON envelope fields, rename-tracking,
8 all EventKind paths end-to-end through _find_events_in_commit.
9 End-to-end — CLI invocation: valid symbol, missing symbol, bad address,
10 --from, --max truncation, invalid ref.
11 Stress — 1 000 SymbolEvent constructions; concurrent reads.
12 Data integrity — to_dict round-trip; chronological ordering; counts accurate.
13 Security — ANSI in address/message/detail; hostile strings in address.
14 Performance — 1 000 to_dict calls under 0.5 s; duration_ms < 30 000ms.
15 """
16
17 from __future__ import annotations
18 from collections.abc import Mapping
19
20 import datetime
21 import json
22 import os
23 import pathlib
24 import textwrap
25 import threading
26 import time
27 from typing import TYPE_CHECKING, get_type_hints
28
29 import pytest
30
31 from muse.core.types import fake_id
32 from tests.cli_test_helper import CliRunner, InvokeResult
33
34 if TYPE_CHECKING:
35 from muse.core.store import CommitRecord
36
37 runner = CliRunner()
38
39
40 # ──────────────────────────────────────────────────────────────────────────────
41 # Fixtures
42 # ──────────────────────────────────────────────────────────────────────────────
43
44
45 def _commit(repo: pathlib.Path, files: Mapping[str, str], message: str) -> None:
46 for name, content in files.items():
47 path = repo / name
48 path.parent.mkdir(parents=True, exist_ok=True)
49 path.write_text(content, encoding="utf-8")
50 saved = os.getcwd()
51 try:
52 os.chdir(repo)
53 runner.invoke(None, ["code", "add", "."])
54 runner.invoke(None, ["commit", "-m", message])
55 finally:
56 os.chdir(saved)
57
58
59 def _invoke(repo: pathlib.Path, args: list[str]) -> InvokeResult:
60 saved = os.getcwd()
61 try:
62 os.chdir(repo)
63 return runner.invoke(None, args)
64 finally:
65 os.chdir(saved)
66
67
68 def _symlog(repo: pathlib.Path, *args: str) -> InvokeResult:
69 return _invoke(repo, ["code", "symbol-log", *args])
70
71
72 @pytest.fixture()
73 def sym_repo(tmp_path: pathlib.Path) -> pathlib.Path:
74 """Repo with two commits so symbol-log has real history to walk.
75
76 Commit 1: billing.py with class Invoice + function process_invoice.
77 Commit 2: billing.py with Invoice body modified (new method).
78 """
79 saved = os.getcwd()
80 try:
81 os.chdir(tmp_path)
82 runner.invoke(None, ["init"])
83 finally:
84 os.chdir(saved)
85
86 _commit(tmp_path, {
87 "billing.py": textwrap.dedent("""\
88 class Invoice:
89 def __init__(self, amount):
90 self.amount = amount
91
92 def process_invoice(inv):
93 return inv.amount * 1.1
94 """),
95 }, "feat: add Invoice and process_invoice")
96
97 _commit(tmp_path, {
98 "billing.py": textwrap.dedent("""\
99 class Invoice:
100 def __init__(self, amount):
101 self.amount = amount
102
103 def total(self):
104 return self.amount * 1.1
105
106 def process_invoice(inv):
107 return inv.total()
108 """),
109 }, "feat: add Invoice.total method")
110
111 return tmp_path
112
113
114 # ──────────────────────────────────────────────────────────────────────────────
115 # Shared commit/delta helpers
116 # ──────────────────────────────────────────────────────────────────────────────
117
118
119 def _make_commit(
120 *,
121 commit_id: str = fake_id("aa"),
122 message: str = "feat: hello",
123 committed_at: datetime.datetime | None = None,
124 structured_delta: Mapping[str, object] | None = None,
125 ) -> "CommitRecord":
126 from muse.core.store import CommitRecord
127 return CommitRecord(
128 commit_id=commit_id,
129 repo_id="repo-1",
130 branch="dev",
131 parent_commit_id=None,
132 snapshot_id="snap",
133 message=message,
134 committed_at=committed_at or datetime.datetime(2026, 1, 1, tzinfo=datetime.timezone.utc),
135 structured_delta=structured_delta,
136 )
137
138
139 def _insert_delta(address: str, summary: str = "created") -> Mapping[str, object]:
140 return {"ops": [{"op": "insert", "address": address, "content_summary": summary}]}
141
142
143 def _delete_delta(address: str, summary: str = "deleted") -> Mapping[str, object]:
144 return {"ops": [{"op": "delete", "address": address, "content_summary": summary}]}
145
146
147 def _replace_delta(address: str, new_summary: str = "implementation changed") -> Mapping[str, object]:
148 return {"ops": [{"op": "replace", "address": address, "new_summary": new_summary}]}
149
150
151 def _patch_delta(address: str, new_summary: str = "implementation changed") -> Mapping[str, object]:
152 """replace wrapped in a patch parent — tests _flat_ops flattening."""
153 return {
154 "ops": [
155 {
156 "op": "patch",
157 "address": address,
158 "child_ops": [
159 {"op": "replace", "address": address, "new_summary": new_summary}
160 ],
161 }
162 ]
163 }
164
165
166 # ──────────────────────────────────────────────────────────────────────────────
167 # Unit — TypedDict
168 # ──────────────────────────────────────────────────────────────────────────────
169
170
171 class TestTypedDict:
172 def test_symbol_log_json_exists(self) -> None:
173 from muse.cli.commands.symbol_log import _SymbolLogJson # noqa: F401
174
175 def test_has_schema_version(self) -> None:
176 from muse.cli.commands.symbol_log import _SymbolLogJson
177 assert "schema" in get_type_hints(_SymbolLogJson)
178
179 def test_has_exit_code(self) -> None:
180 from muse.cli.commands.symbol_log import _SymbolLogJson
181 assert "exit_code" in get_type_hints(_SymbolLogJson)
182
183 def test_has_duration_ms(self) -> None:
184 from muse.cli.commands.symbol_log import _SymbolLogJson
185 assert "duration_ms" in get_type_hints(_SymbolLogJson)
186
187 def test_has_core_fields(self) -> None:
188 from muse.cli.commands.symbol_log import _SymbolLogJson
189 hints = get_type_hints(_SymbolLogJson)
190 for field in ("address", "start_ref", "total_commits_scanned", "truncated", "events"):
191 assert field in hints, f"missing field: {field}"
192
193
194 # ──────────────────────────────────────────────────────────────────────────────
195 # Unit — SymbolEvent
196 # ──────────────────────────────────────────────────────────────────────────────
197
198
199 class TestSymbolEvent:
200 def test_constructor_stores_kind(self) -> None:
201 from muse.cli.commands.symbol_log import SymbolEvent
202 ev = SymbolEvent("created", _make_commit(), "f.py::fn", "created")
203 assert ev.kind == "created"
204
205 def test_constructor_stores_address(self) -> None:
206 from muse.cli.commands.symbol_log import SymbolEvent
207 ev = SymbolEvent("modified", _make_commit(), "f.py::fn", "impl changed")
208 assert ev.address == "f.py::fn"
209
210 def test_constructor_stores_detail(self) -> None:
211 from muse.cli.commands.symbol_log import SymbolEvent
212 ev = SymbolEvent("deleted", _make_commit(), "f.py::fn", "removed")
213 assert ev.detail == "removed"
214
215 def test_constructor_default_new_address_is_none(self) -> None:
216 from muse.cli.commands.symbol_log import SymbolEvent
217 ev = SymbolEvent("created", _make_commit(), "f.py::fn", "x")
218 assert ev.new_address is None
219
220 def test_constructor_stores_new_address(self) -> None:
221 from muse.cli.commands.symbol_log import SymbolEvent
222 ev = SymbolEvent("renamed", _make_commit(), "f.py::old", "old → new", "f.py::new")
223 assert ev.new_address == "f.py::new"
224
225 def test_to_dict_has_all_fields(self) -> None:
226 from muse.cli.commands.symbol_log import SymbolEvent
227 commit = _make_commit(commit_id=fake_id("bb"), message="msg")
228 ev = SymbolEvent("modified", commit, "f.py::fn", "detail")
229 d = ev.to_dict()
230 for key in ("event", "commit_id", "message", "committed_at", "address", "detail", "new_address"):
231 assert key in d, f"missing key: {key}"
232
233 def test_to_dict_event_matches_kind(self) -> None:
234 from muse.cli.commands.symbol_log import SymbolEvent
235 ev = SymbolEvent("signature", _make_commit(), "f.py::fn", "sig changed")
236 assert ev.to_dict()["event"] == "signature"
237
238 def test_to_dict_committed_at_is_isoformat(self) -> None:
239 from muse.cli.commands.symbol_log import SymbolEvent
240 dt = datetime.datetime(2026, 3, 14, 12, 0, tzinfo=datetime.timezone.utc)
241 ev = SymbolEvent("created", _make_commit(committed_at=dt), "f.py::fn", "x")
242 iso = ev.to_dict()["committed_at"]
243 assert "2026-03-14" in iso
244 assert "T" in iso
245
246 def test_to_dict_new_address_none_when_not_set(self) -> None:
247 from muse.cli.commands.symbol_log import SymbolEvent
248 ev = SymbolEvent("modified", _make_commit(), "f.py::fn", "x")
249 assert ev.to_dict()["new_address"] is None
250
251
252 # ──────────────────────────────────────────────────────────────────────────────
253 # Unit — _flat_ops
254 # ──────────────────────────────────────────────────────────────────────────────
255
256
257 class TestFlatOps:
258 def test_passthrough_non_patch(self) -> None:
259 from muse.cli.commands.symbol_log import _flat_ops
260 ops = [{"op": "insert", "address": "f.py::fn"}]
261 assert _flat_ops(ops) == ops
262
263 def test_flattens_patch_children(self) -> None:
264 from muse.cli.commands.symbol_log import _flat_ops
265 child = {"op": "replace", "address": "f.py::fn", "new_summary": "x"}
266 ops = [{"op": "patch", "address": "f.py::fn", "child_ops": [child]}]
267 result = _flat_ops(ops)
268 assert result == [child]
269
270 def test_mixed_ops_preserved_in_order(self) -> None:
271 from muse.cli.commands.symbol_log import _flat_ops
272 insert = {"op": "insert", "address": "f.py::a"}
273 child = {"op": "replace", "address": "f.py::b", "new_summary": "y"}
274 patch = {"op": "patch", "address": "f.py::b", "child_ops": [child]}
275 result = _flat_ops([insert, patch])
276 assert result == [insert, child]
277
278 def test_empty_ops_returns_empty(self) -> None:
279 from muse.cli.commands.symbol_log import _flat_ops
280 assert _flat_ops([]) == []
281
282
283 # ──────────────────────────────────────────────────────────────────────────────
284 # Unit — _find_events_in_commit (each EventKind)
285 # ──────────────────────────────────────────────────────────────────────────────
286
287
288 class TestFindEventsInCommit:
289 def test_no_delta_returns_empty(self) -> None:
290 from muse.cli.commands.symbol_log import _find_events_in_commit
291 commit = _make_commit(structured_delta=None)
292 evs, addr = _find_events_in_commit(commit, "f.py::fn")
293 assert evs == []
294 assert addr == "f.py::fn"
295
296 def test_insert_produces_created(self) -> None:
297 from muse.cli.commands.symbol_log import _find_events_in_commit
298 commit = _make_commit(structured_delta=_insert_delta("f.py::fn"))
299 evs, addr = _find_events_in_commit(commit, "f.py::fn")
300 assert len(evs) == 1
301 assert evs[0].kind == "created"
302 assert addr == "f.py::fn"
303
304 def test_delete_produces_deleted(self) -> None:
305 from muse.cli.commands.symbol_log import _find_events_in_commit
306 commit = _make_commit(structured_delta=_delete_delta("f.py::fn"))
307 evs, addr = _find_events_in_commit(commit, "f.py::fn")
308 assert len(evs) == 1
309 assert evs[0].kind == "deleted"
310
311 def test_delete_moved_to_produces_moved(self) -> None:
312 from muse.cli.commands.symbol_log import _find_events_in_commit
313 commit = _make_commit(structured_delta=_delete_delta("f.py::fn", "moved to g.py::fn"))
314 evs, _ = _find_events_in_commit(commit, "f.py::fn")
315 assert evs[0].kind == "moved"
316
317 def test_replace_produces_modified(self) -> None:
318 from muse.cli.commands.symbol_log import _find_events_in_commit
319 commit = _make_commit(structured_delta=_replace_delta("f.py::fn"))
320 evs, addr = _find_events_in_commit(commit, "f.py::fn")
321 assert len(evs) == 1
322 assert evs[0].kind == "modified"
323
324 def test_replace_renamed_to_updates_address(self) -> None:
325 from muse.cli.commands.symbol_log import _find_events_in_commit
326 commit = _make_commit(structured_delta=_replace_delta("f.py::old", "renamed to new"))
327 evs, addr = _find_events_in_commit(commit, "f.py::old")
328 assert evs[0].kind == "renamed"
329 assert addr == "f.py::new"
330 assert evs[0].new_address == "f.py::new"
331
332 def test_replace_moved_to_produces_moved(self) -> None:
333 from muse.cli.commands.symbol_log import _find_events_in_commit
334 commit = _make_commit(structured_delta=_replace_delta("f.py::fn", "moved to g.py::fn"))
335 evs, _ = _find_events_in_commit(commit, "f.py::fn")
336 assert evs[0].kind == "moved"
337
338 def test_replace_signature_produces_signature(self) -> None:
339 from muse.cli.commands.symbol_log import _find_events_in_commit
340 commit = _make_commit(structured_delta=_replace_delta("f.py::fn", "signature changed"))
341 evs, _ = _find_events_in_commit(commit, "f.py::fn")
342 assert evs[0].kind == "signature"
343
344 def test_patch_wrapper_is_flattened(self) -> None:
345 from muse.cli.commands.symbol_log import _find_events_in_commit
346 commit = _make_commit(structured_delta=_patch_delta("f.py::fn"))
347 evs, _ = _find_events_in_commit(commit, "f.py::fn")
348 assert len(evs) == 1
349 assert evs[0].kind == "modified"
350
351 def test_unrelated_address_produces_no_events(self) -> None:
352 from muse.cli.commands.symbol_log import _find_events_in_commit
353 commit = _make_commit(structured_delta=_insert_delta("other.py::other"))
354 evs, addr = _find_events_in_commit(commit, "f.py::fn")
355 assert evs == []
356 assert addr == "f.py::fn"
357
358
359 # ──────────────────────────────────────────────────────────────────────────────
360 # Integration — alias, docstrings, envelope
361 # ──────────────────────────────────────────────────────────────────────────────
362
363
364 class TestAliasRegistration:
365 def test_j_alias_registered(self) -> None:
366 from muse.cli.commands.symbol_log import register
367 import argparse
368 p = argparse.ArgumentParser()
369 sub = p.add_subparsers()
370 register(sub)
371 ns = p.parse_args(["symbol-log", "f.py::fn", "-j"])
372 assert ns.json_out is True
373
374 def test_json_flag_sets_as_json_true(self) -> None:
375 from muse.cli.commands.symbol_log import register
376 import argparse
377 p = argparse.ArgumentParser()
378 sub = p.add_subparsers()
379 register(sub)
380 ns = p.parse_args(["symbol-log", "f.py::fn", "--json"])
381 assert ns.json_out is True
382
383
384 class TestDocstrings:
385 def test_register_mentions_json_alias(self) -> None:
386 from muse.cli.commands.symbol_log import register
387 doc = register.__doc__ or ""
388 assert "--json" in doc or "-j" in doc
389
390 def test_run_mentions_exit_code(self) -> None:
391 from muse.cli.commands.symbol_log import run
392 assert "exit_code" in (run.__doc__ or "")
393
394 def test_run_mentions_duration_ms(self) -> None:
395 from muse.cli.commands.symbol_log import run
396 assert "duration_ms" in (run.__doc__ or "")
397
398 def test_run_mentions_schema_version(self) -> None:
399 from muse.cli.commands.symbol_log import run
400 assert "schema" in (run.__doc__ or "")
401
402
403 class TestJsonEnvelope:
404 def test_schema_version_present(self, sym_repo: pathlib.Path) -> None:
405 r = _symlog(sym_repo, "billing.py::Invoice", "-j")
406 assert r.exit_code == 0
407 assert "schema" in json.loads(r.output)
408
409 def test_exit_code_zero(self, sym_repo: pathlib.Path) -> None:
410 r = _symlog(sym_repo, "billing.py::Invoice", "-j")
411 assert r.exit_code == 0
412 d = json.loads(r.output)
413 assert d["exit_code"] == 0
414
415 def test_duration_ms_present_and_float(self, sym_repo: pathlib.Path) -> None:
416 r = _symlog(sym_repo, "billing.py::Invoice", "-j")
417 assert r.exit_code == 0
418 d = json.loads(r.output)
419 assert "duration_ms" in d
420 assert isinstance(d["duration_ms"], float)
421
422 def test_schema_version_nonempty_string(self, sym_repo: pathlib.Path) -> None:
423 r = _symlog(sym_repo, "billing.py::Invoice", "-j")
424 assert r.exit_code == 0
425 d = json.loads(r.output)
426 assert isinstance(d["schema"], int) and d["schema"] > 0
427
428
429 class TestJsonAlias:
430 def test_j_parity_with_json_flag(self, sym_repo: pathlib.Path) -> None:
431 r1 = _symlog(sym_repo, "billing.py::Invoice", "--json")
432 r2 = _symlog(sym_repo, "billing.py::Invoice", "-j")
433 assert r1.exit_code == 0
434 assert r2.exit_code == 0
435 d1, d2 = json.loads(r1.output), json.loads(r2.output)
436 assert d1["address"] == d2["address"]
437 assert d1["events"] == d2["events"]
438 assert d1["schema"] == d2["schema"]
439 assert d1["exit_code"] == d2["exit_code"]
440
441
442 # ──────────────────────────────────────────────────────────────────────────────
443 # End-to-end
444 # ──────────────────────────────────────────────────────────────────────────────
445
446
447 class TestEndToEnd:
448 def test_valid_symbol_exits_zero(self, sym_repo: pathlib.Path) -> None:
449 r = _symlog(sym_repo, "billing.py::Invoice")
450 assert r.exit_code == 0
451
452 def test_unknown_symbol_exits_zero_with_no_events(self, sym_repo: pathlib.Path) -> None:
453 r = _symlog(sym_repo, "billing.py::DoesNotExistXXX")
454 assert r.exit_code == 0
455 assert "no events found" in r.output
456
457 def test_bad_address_no_double_colon_exits_nonzero(self, sym_repo: pathlib.Path) -> None:
458 r = _symlog(sym_repo, "billing.py")
459 assert r.exit_code != 0
460
461 def test_max_1_sets_truncated_true_in_json(self, sym_repo: pathlib.Path) -> None:
462 r = _symlog(sym_repo, "billing.py::Invoice", "--max", "1", "-j")
463 assert r.exit_code == 0
464 d = json.loads(r.output)
465 assert d["truncated"] is True
466 assert d["total_commits_scanned"] == 1
467
468 def test_max_1_shows_truncation_warning_in_human(self, sym_repo: pathlib.Path) -> None:
469 r = _symlog(sym_repo, "billing.py::Invoice", "--max", "1")
470 assert r.exit_code == 0
471 assert "incomplete" in r.output or "limit" in r.output
472
473 def test_max_zero_exits_nonzero(self, sym_repo: pathlib.Path) -> None:
474 r = _symlog(sym_repo, "billing.py::Invoice", "--max", "0")
475 assert r.exit_code != 0
476
477 def test_invalid_from_ref_exits_nonzero(self, sym_repo: pathlib.Path) -> None:
478 r = _symlog(sym_repo, "billing.py::Invoice", "--from", "deadbeefdeadbeef")
479 assert r.exit_code != 0
480
481 def test_json_address_field_matches_input(self, sym_repo: pathlib.Path) -> None:
482 r = _symlog(sym_repo, "billing.py::Invoice", "-j")
483 assert r.exit_code == 0
484 assert json.loads(r.output)["address"] == "billing.py::Invoice"
485
486 def test_json_events_is_list(self, sym_repo: pathlib.Path) -> None:
487 r = _symlog(sym_repo, "billing.py::Invoice", "-j")
488 assert r.exit_code == 0
489 assert isinstance(json.loads(r.output)["events"], list)
490
491 def test_start_ref_is_head_by_default(self, sym_repo: pathlib.Path) -> None:
492 r = _symlog(sym_repo, "billing.py::Invoice", "-j")
493 assert r.exit_code == 0
494 assert json.loads(r.output)["start_ref"] == "HEAD"
495
496 def test_human_output_shows_symbol_header(self, sym_repo: pathlib.Path) -> None:
497 r = _symlog(sym_repo, "billing.py::Invoice")
498 assert r.exit_code == 0
499 assert "billing.py::Invoice" in r.output
500
501
502 # ──────────────────────────────────────────────────────────────────────────────
503 # Stress
504 # ──────────────────────────────────────────────────────────────────────────────
505
506
507 class TestStress:
508 def test_1000_symbol_event_constructions(self) -> None:
509 from muse.cli.commands.symbol_log import SymbolEvent
510 commit = _make_commit()
511 for i in range(1_000):
512 ev = SymbolEvent("modified", commit, f"f{i}.py::fn", f"detail {i}")
513 assert ev.kind == "modified"
514
515 def test_1000_to_dict_calls(self) -> None:
516 from muse.cli.commands.symbol_log import SymbolEvent
517 commit = _make_commit()
518 ev = SymbolEvent("created", commit, "f.py::fn", "x")
519 for _ in range(1_000):
520 d = ev.to_dict()
521 assert "event" in d
522
523 def test_flat_ops_10000_calls(self) -> None:
524 from muse.cli.commands.symbol_log import _flat_ops
525 ops = [{"op": "insert", "address": f"f{i}.py::fn"} for i in range(10)]
526 for _ in range(10_000):
527 result = _flat_ops(ops)
528 assert len(result) == 10
529
530 def test_concurrent_find_events(self) -> None:
531 from muse.cli.commands.symbol_log import _find_events_in_commit
532 commit = _make_commit(structured_delta=_insert_delta("f.py::fn"))
533 results: list[int] = []
534 lock = threading.Lock()
535
536 def _run() -> None:
537 evs, _ = _find_events_in_commit(commit, "f.py::fn")
538 with lock:
539 results.append(len(evs))
540
541 threads = [threading.Thread(target=_run) for _ in range(50)]
542 for t in threads: t.start()
543 for t in threads: t.join()
544 assert all(n == 1 for n in results)
545 assert len(results) == 50
546
547
548 # ──────────────────────────────────────────────────────────────────────────────
549 # Data integrity
550 # ──────────────────────────────────────────────────────────────────────────────
551
552
553 class TestDataIntegrity:
554 def test_to_dict_preserves_all_fields(self) -> None:
555 from muse.cli.commands.symbol_log import SymbolEvent
556 commit = _make_commit(
557 commit_id=fake_id("cc"),
558 message="fix: something",
559 committed_at=datetime.datetime(2026, 5, 1, tzinfo=datetime.timezone.utc),
560 )
561 ev = SymbolEvent("renamed", commit, "a.py::old", "old → new", "a.py::new")
562 d = ev.to_dict()
563 assert d["event"] == "renamed"
564 assert d["commit_id"] == fake_id("cc")
565 assert d["message"] == "fix: something"
566 assert d["address"] == "a.py::old"
567 assert d["detail"] == "old → new"
568 assert d["new_address"] == "a.py::new"
569 assert "2026-05-01" in d["committed_at"]
570
571 def test_events_in_json_are_chronological(self, sym_repo: pathlib.Path) -> None:
572 r = _symlog(sym_repo, "billing.py::Invoice", "-j")
573 assert r.exit_code == 0
574 events = json.loads(r.output)["events"]
575 if len(events) >= 2:
576 times = [datetime.datetime.fromisoformat(e["committed_at"]) for e in events]
577 assert times == sorted(times), "events not in chronological order"
578
579 def test_total_commits_scanned_is_int(self, sym_repo: pathlib.Path) -> None:
580 r = _symlog(sym_repo, "billing.py::Invoice", "-j")
581 assert r.exit_code == 0
582 assert isinstance(json.loads(r.output)["total_commits_scanned"], int)
583
584 def test_truncated_is_bool(self, sym_repo: pathlib.Path) -> None:
585 r = _symlog(sym_repo, "billing.py::Invoice", "-j")
586 assert r.exit_code == 0
587 assert isinstance(json.loads(r.output)["truncated"], bool)
588
589 def test_rename_tracking_continues_with_new_address(self) -> None:
590 from muse.cli.commands.symbol_log import _find_events_in_commit
591 rename_commit = _make_commit(structured_delta=_replace_delta("f.py::old", "renamed to new"))
592 _, next_addr = _find_events_in_commit(rename_commit, "f.py::old")
593 assert next_addr == "f.py::new"
594 insert_commit = _make_commit(structured_delta=_insert_delta("f.py::new"))
595 evs, _ = _find_events_in_commit(insert_commit, next_addr)
596 assert len(evs) == 1
597 assert evs[0].kind == "created"
598
599 def test_new_address_none_for_modified(self) -> None:
600 from muse.cli.commands.symbol_log import _find_events_in_commit
601 commit = _make_commit(structured_delta=_replace_delta("f.py::fn"))
602 evs, _ = _find_events_in_commit(commit, "f.py::fn")
603 assert evs[0].new_address is None
604
605 def test_json_serialisable(self, sym_repo: pathlib.Path) -> None:
606 r = _symlog(sym_repo, "billing.py::Invoice", "-j")
607 assert r.exit_code == 0
608 json.loads(r.output) # must not raise
609
610
611 # ──────────────────────────────────────────────────────────────────────────────
612 # Security
613 # ──────────────────────────────────────────────────────────────────────────────
614
615
616 class TestSecurity:
617 def test_ansi_in_address_does_not_crash(self, sym_repo: pathlib.Path) -> None:
618 ansi_addr = "\x1b[31mbad\x1b[0m.py::fn"
619 r = _symlog(sym_repo, ansi_addr)
620 assert r.exit_code in (0, 1, 2)
621
622 def test_ansi_in_commit_message_survives_to_dict(self) -> None:
623 from muse.cli.commands.symbol_log import SymbolEvent
624 malicious_msg = "\x1b[31mmalicious\x1b[0m"
625 ev = SymbolEvent("modified", _make_commit(message=malicious_msg), "f.py::fn", "x")
626 d = ev.to_dict()
627 assert d["message"] == malicious_msg
628
629 def test_very_long_address_does_not_crash(self, sym_repo: pathlib.Path) -> None:
630 long_addr = f"f.py::{'x' * 10_000}"
631 r = _symlog(sym_repo, long_addr)
632 assert r.exit_code in (0, 1, 2)
633
634 def test_unicode_in_address_does_not_crash(self, sym_repo: pathlib.Path) -> None:
635 r = _symlog(sym_repo, "音符.py::関数")
636 assert r.exit_code in (0, 1, 2)
637
638 def test_hostile_detail_survives_json_serialisation(self) -> None:
639 from muse.cli.commands.symbol_log import SymbolEvent
640 malicious = '"; DROP TABLE commits; --'
641 ev = SymbolEvent("modified", _make_commit(), "f.py::fn", malicious)
642 d = ev.to_dict()
643 assert json.loads(json.dumps(d))["detail"] == malicious
644
645 def test_very_long_message_in_commit_does_not_crash(self) -> None:
646 from muse.cli.commands.symbol_log import SymbolEvent
647 ev = SymbolEvent("modified", _make_commit(message="x" * 100_000), "f.py::fn", "x")
648 d = ev.to_dict()
649 assert len(d["message"]) == 100_000
650
651
652 # ──────────────────────────────────────────────────────────────────────────────
653 # Performance
654 # ──────────────────────────────────────────────────────────────────────────────
655
656
657 class TestPerformance:
658 def test_1000_to_dict_under_500ms(self) -> None:
659 from muse.cli.commands.symbol_log import SymbolEvent
660 ev = SymbolEvent("modified", _make_commit(), "f.py::fn", "impl changed")
661 start = time.perf_counter()
662 for _ in range(1_000):
663 ev.to_dict()
664 elapsed = time.perf_counter() - start
665 assert elapsed < 0.5, f"1 000 to_dict calls took {elapsed:.2f}s"
666
667 def test_duration_ms_present_and_reasonable(self, sym_repo: pathlib.Path) -> None:
668 r = _symlog(sym_repo, "billing.py::Invoice", "-j")
669 assert r.exit_code == 0
670 d = json.loads(r.output)
671 assert "duration_ms" in d
672 assert 0 <= d["duration_ms"] < 30_000
673
674
675 # Flag registration
676 # ──────────────────────────────────────────────────────────────────────────────
677
678
679 class TestRegisterFlags:
680 def _parse(self, *args: str) -> "argparse.Namespace":
681 import argparse
682 from muse.cli.commands.symbol_log import register
683 p = argparse.ArgumentParser()
684 sub = p.add_subparsers()
685 register(sub)
686 return p.parse_args(["symbol-log", *args])
687
688 def test_default_json_out_is_false(self) -> None:
689 ns = self._parse("src/utils.py::fn")
690 assert ns.json_out is False
691
692 def test_json_flag_sets_json_out(self) -> None:
693 ns = self._parse("--json", "src/utils.py::fn")
694 assert ns.json_out is True
695
696 def test_j_shorthand_sets_json_out(self) -> None:
697 ns = self._parse("-j", "src/utils.py::fn")
698 assert ns.json_out is True
File History 1 commit
sha256:51ce277f663e01a43eaffbe77509b1de7ac2d4251b55d23306304bcdeb92c90d feat(pack): delta-encode snapshots in MPackBundle wire format Sonnet 4.6 minor ⚠ 129 days ago