gabriel / muse public
test_symbol_log_supercharge.py python
697 lines 31.1 KB
Raw
sha256:81ae324db5ad375fbfe4834c6fcb378312cafad3cc92dec5d3e5c427306621a2 fix: remove commit_exists filter from have anchors — server… Sonnet 4.6 patch 21 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.commits 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.commits import CommitRecord
127 return CommitRecord(
128 commit_id=commit_id,
129 branch="dev",
130 parent_commit_id=None,
131 snapshot_id="snap",
132 message=message,
133 committed_at=committed_at or datetime.datetime(2026, 1, 1, tzinfo=datetime.timezone.utc),
134 structured_delta=structured_delta,
135 )
136
137
138 def _insert_delta(address: str, summary: str = "created") -> Mapping[str, object]:
139 return {"ops": [{"op": "insert", "address": address, "content_summary": summary}]}
140
141
142 def _delete_delta(address: str, summary: str = "deleted") -> Mapping[str, object]:
143 return {"ops": [{"op": "delete", "address": address, "content_summary": summary}]}
144
145
146 def _replace_delta(address: str, new_summary: str = "implementation changed") -> Mapping[str, object]:
147 return {"ops": [{"op": "replace", "address": address, "new_summary": new_summary}]}
148
149
150 def _patch_delta(address: str, new_summary: str = "implementation changed") -> Mapping[str, object]:
151 """replace wrapped in a patch parent — tests _flat_ops flattening."""
152 return {
153 "ops": [
154 {
155 "op": "patch",
156 "address": address,
157 "child_ops": [
158 {"op": "replace", "address": address, "new_summary": new_summary}
159 ],
160 }
161 ]
162 }
163
164
165 # ──────────────────────────────────────────────────────────────────────────────
166 # Unit — TypedDict
167 # ──────────────────────────────────────────────────────────────────────────────
168
169
170 class TestTypedDict:
171 def test_symbol_log_json_exists(self) -> None:
172 from muse.cli.commands.symbol_log import _SymbolLogJson # noqa: F401
173
174 def test_has_schema_version(self) -> None:
175 from muse.cli.commands.symbol_log import _SymbolLogJson
176 assert "schema" in get_type_hints(_SymbolLogJson)
177
178 def test_has_exit_code(self) -> None:
179 from muse.cli.commands.symbol_log import _SymbolLogJson
180 assert "exit_code" in get_type_hints(_SymbolLogJson)
181
182 def test_has_duration_ms(self) -> None:
183 from muse.cli.commands.symbol_log import _SymbolLogJson
184 assert "duration_ms" in get_type_hints(_SymbolLogJson)
185
186 def test_has_core_fields(self) -> None:
187 from muse.cli.commands.symbol_log import _SymbolLogJson
188 hints = get_type_hints(_SymbolLogJson)
189 for field in ("address", "start_ref", "total_commits_scanned", "truncated", "events"):
190 assert field in hints, f"missing field: {field}"
191
192
193 # ──────────────────────────────────────────────────────────────────────────────
194 # Unit — SymbolEvent
195 # ──────────────────────────────────────────────────────────────────────────────
196
197
198 class TestSymbolEvent:
199 def test_constructor_stores_kind(self) -> None:
200 from muse.cli.commands.symbol_log import SymbolEvent
201 ev = SymbolEvent("created", _make_commit(), "f.py::fn", "created")
202 assert ev.kind == "created"
203
204 def test_constructor_stores_address(self) -> None:
205 from muse.cli.commands.symbol_log import SymbolEvent
206 ev = SymbolEvent("modified", _make_commit(), "f.py::fn", "impl changed")
207 assert ev.address == "f.py::fn"
208
209 def test_constructor_stores_detail(self) -> None:
210 from muse.cli.commands.symbol_log import SymbolEvent
211 ev = SymbolEvent("deleted", _make_commit(), "f.py::fn", "removed")
212 assert ev.detail == "removed"
213
214 def test_constructor_default_new_address_is_none(self) -> None:
215 from muse.cli.commands.symbol_log import SymbolEvent
216 ev = SymbolEvent("created", _make_commit(), "f.py::fn", "x")
217 assert ev.new_address is None
218
219 def test_constructor_stores_new_address(self) -> None:
220 from muse.cli.commands.symbol_log import SymbolEvent
221 ev = SymbolEvent("renamed", _make_commit(), "f.py::old", "old → new", "f.py::new")
222 assert ev.new_address == "f.py::new"
223
224 def test_to_dict_has_all_fields(self) -> None:
225 from muse.cli.commands.symbol_log import SymbolEvent
226 commit = _make_commit(commit_id=fake_id("bb"), message="msg")
227 ev = SymbolEvent("modified", commit, "f.py::fn", "detail")
228 d = ev.to_dict()
229 for key in ("event", "commit_id", "message", "committed_at", "address", "detail", "new_address"):
230 assert key in d, f"missing key: {key}"
231
232 def test_to_dict_event_matches_kind(self) -> None:
233 from muse.cli.commands.symbol_log import SymbolEvent
234 ev = SymbolEvent("signature", _make_commit(), "f.py::fn", "sig changed")
235 assert ev.to_dict()["event"] == "signature"
236
237 def test_to_dict_committed_at_is_isoformat(self) -> None:
238 from muse.cli.commands.symbol_log import SymbolEvent
239 dt = datetime.datetime(2026, 3, 14, 12, 0, tzinfo=datetime.timezone.utc)
240 ev = SymbolEvent("created", _make_commit(committed_at=dt), "f.py::fn", "x")
241 iso = ev.to_dict()["committed_at"]
242 assert "2026-03-14" in iso
243 assert "T" in iso
244
245 def test_to_dict_new_address_none_when_not_set(self) -> None:
246 from muse.cli.commands.symbol_log import SymbolEvent
247 ev = SymbolEvent("modified", _make_commit(), "f.py::fn", "x")
248 assert ev.to_dict()["new_address"] is None
249
250
251 # ──────────────────────────────────────────────────────────────────────────────
252 # Unit — _flat_ops
253 # ──────────────────────────────────────────────────────────────────────────────
254
255
256 class TestFlatOps:
257 def test_passthrough_non_patch(self) -> None:
258 from muse.cli.commands.symbol_log import _flat_ops
259 ops = [{"op": "insert", "address": "f.py::fn"}]
260 assert _flat_ops(ops) == ops
261
262 def test_flattens_patch_children(self) -> None:
263 from muse.cli.commands.symbol_log import _flat_ops
264 child = {"op": "replace", "address": "f.py::fn", "new_summary": "x"}
265 ops = [{"op": "patch", "address": "f.py::fn", "child_ops": [child]}]
266 result = _flat_ops(ops)
267 assert result == [child]
268
269 def test_mixed_ops_preserved_in_order(self) -> None:
270 from muse.cli.commands.symbol_log import _flat_ops
271 insert = {"op": "insert", "address": "f.py::a"}
272 child = {"op": "replace", "address": "f.py::b", "new_summary": "y"}
273 patch = {"op": "patch", "address": "f.py::b", "child_ops": [child]}
274 result = _flat_ops([insert, patch])
275 assert result == [insert, child]
276
277 def test_empty_ops_returns_empty(self) -> None:
278 from muse.cli.commands.symbol_log import _flat_ops
279 assert _flat_ops([]) == []
280
281
282 # ──────────────────────────────────────────────────────────────────────────────
283 # Unit — _find_events_in_commit (each EventKind)
284 # ──────────────────────────────────────────────────────────────────────────────
285
286
287 class TestFindEventsInCommit:
288 def test_no_delta_returns_empty(self) -> None:
289 from muse.cli.commands.symbol_log import _find_events_in_commit
290 commit = _make_commit(structured_delta=None)
291 evs, addr = _find_events_in_commit(commit, "f.py::fn")
292 assert evs == []
293 assert addr == "f.py::fn"
294
295 def test_insert_produces_created(self) -> None:
296 from muse.cli.commands.symbol_log import _find_events_in_commit
297 commit = _make_commit(structured_delta=_insert_delta("f.py::fn"))
298 evs, addr = _find_events_in_commit(commit, "f.py::fn")
299 assert len(evs) == 1
300 assert evs[0].kind == "created"
301 assert addr == "f.py::fn"
302
303 def test_delete_produces_deleted(self) -> None:
304 from muse.cli.commands.symbol_log import _find_events_in_commit
305 commit = _make_commit(structured_delta=_delete_delta("f.py::fn"))
306 evs, addr = _find_events_in_commit(commit, "f.py::fn")
307 assert len(evs) == 1
308 assert evs[0].kind == "deleted"
309
310 def test_delete_moved_to_produces_moved(self) -> None:
311 from muse.cli.commands.symbol_log import _find_events_in_commit
312 commit = _make_commit(structured_delta=_delete_delta("f.py::fn", "moved to g.py::fn"))
313 evs, _ = _find_events_in_commit(commit, "f.py::fn")
314 assert evs[0].kind == "moved"
315
316 def test_replace_produces_modified(self) -> None:
317 from muse.cli.commands.symbol_log import _find_events_in_commit
318 commit = _make_commit(structured_delta=_replace_delta("f.py::fn"))
319 evs, addr = _find_events_in_commit(commit, "f.py::fn")
320 assert len(evs) == 1
321 assert evs[0].kind == "modified"
322
323 def test_replace_renamed_to_updates_address(self) -> None:
324 from muse.cli.commands.symbol_log import _find_events_in_commit
325 commit = _make_commit(structured_delta=_replace_delta("f.py::old", "renamed to new"))
326 evs, addr = _find_events_in_commit(commit, "f.py::old")
327 assert evs[0].kind == "renamed"
328 assert addr == "f.py::new"
329 assert evs[0].new_address == "f.py::new"
330
331 def test_replace_moved_to_produces_moved(self) -> None:
332 from muse.cli.commands.symbol_log import _find_events_in_commit
333 commit = _make_commit(structured_delta=_replace_delta("f.py::fn", "moved to g.py::fn"))
334 evs, _ = _find_events_in_commit(commit, "f.py::fn")
335 assert evs[0].kind == "moved"
336
337 def test_replace_signature_produces_signature(self) -> None:
338 from muse.cli.commands.symbol_log import _find_events_in_commit
339 commit = _make_commit(structured_delta=_replace_delta("f.py::fn", "signature changed"))
340 evs, _ = _find_events_in_commit(commit, "f.py::fn")
341 assert evs[0].kind == "signature"
342
343 def test_patch_wrapper_is_flattened(self) -> None:
344 from muse.cli.commands.symbol_log import _find_events_in_commit
345 commit = _make_commit(structured_delta=_patch_delta("f.py::fn"))
346 evs, _ = _find_events_in_commit(commit, "f.py::fn")
347 assert len(evs) == 1
348 assert evs[0].kind == "modified"
349
350 def test_unrelated_address_produces_no_events(self) -> None:
351 from muse.cli.commands.symbol_log import _find_events_in_commit
352 commit = _make_commit(structured_delta=_insert_delta("other.py::other"))
353 evs, addr = _find_events_in_commit(commit, "f.py::fn")
354 assert evs == []
355 assert addr == "f.py::fn"
356
357
358 # ──────────────────────────────────────────────────────────────────────────────
359 # Integration — alias, docstrings, envelope
360 # ──────────────────────────────────────────────────────────────────────────────
361
362
363 class TestAliasRegistration:
364 def test_j_alias_registered(self) -> None:
365 from muse.cli.commands.symbol_log import register
366 import argparse
367 p = argparse.ArgumentParser()
368 sub = p.add_subparsers()
369 register(sub)
370 ns = p.parse_args(["symbol-log", "f.py::fn", "-j"])
371 assert ns.json_out is True
372
373 def test_json_flag_sets_as_json_true(self) -> None:
374 from muse.cli.commands.symbol_log import register
375 import argparse
376 p = argparse.ArgumentParser()
377 sub = p.add_subparsers()
378 register(sub)
379 ns = p.parse_args(["symbol-log", "f.py::fn", "--json"])
380 assert ns.json_out is True
381
382
383 class TestDocstrings:
384 def test_register_mentions_json_alias(self) -> None:
385 from muse.cli.commands.symbol_log import register
386 doc = register.__doc__ or ""
387 assert "--json" in doc or "-j" in doc
388
389 def test_run_mentions_exit_code(self) -> None:
390 from muse.cli.commands.symbol_log import run
391 assert "exit_code" in (run.__doc__ or "")
392
393 def test_run_mentions_duration_ms(self) -> None:
394 from muse.cli.commands.symbol_log import run
395 assert "duration_ms" in (run.__doc__ or "")
396
397 def test_run_mentions_schema_version(self) -> None:
398 from muse.cli.commands.symbol_log import run
399 assert "schema" in (run.__doc__ or "")
400
401
402 class TestJsonEnvelope:
403 def test_schema_version_present(self, sym_repo: pathlib.Path) -> None:
404 r = _symlog(sym_repo, "billing.py::Invoice", "-j")
405 assert r.exit_code == 0
406 assert "schema" in json.loads(r.output)
407
408 def test_exit_code_zero(self, sym_repo: pathlib.Path) -> None:
409 r = _symlog(sym_repo, "billing.py::Invoice", "-j")
410 assert r.exit_code == 0
411 d = json.loads(r.output)
412 assert d["exit_code"] == 0
413
414 def test_duration_ms_present_and_float(self, sym_repo: pathlib.Path) -> None:
415 r = _symlog(sym_repo, "billing.py::Invoice", "-j")
416 assert r.exit_code == 0
417 d = json.loads(r.output)
418 assert "duration_ms" in d
419 assert isinstance(d["duration_ms"], float)
420
421 def test_schema_version_nonempty_string(self, sym_repo: pathlib.Path) -> None:
422 r = _symlog(sym_repo, "billing.py::Invoice", "-j")
423 assert r.exit_code == 0
424 d = json.loads(r.output)
425 assert isinstance(d["schema"], int) and d["schema"] > 0
426
427
428 class TestJsonAlias:
429 def test_j_parity_with_json_flag(self, sym_repo: pathlib.Path) -> None:
430 r1 = _symlog(sym_repo, "billing.py::Invoice", "--json")
431 r2 = _symlog(sym_repo, "billing.py::Invoice", "-j")
432 assert r1.exit_code == 0
433 assert r2.exit_code == 0
434 d1, d2 = json.loads(r1.output), json.loads(r2.output)
435 assert d1["address"] == d2["address"]
436 assert d1["events"] == d2["events"]
437 assert d1["schema"] == d2["schema"]
438 assert d1["exit_code"] == d2["exit_code"]
439
440
441 # ──────────────────────────────────────────────────────────────────────────────
442 # End-to-end
443 # ──────────────────────────────────────────────────────────────────────────────
444
445
446 class TestEndToEnd:
447 def test_valid_symbol_exits_zero(self, sym_repo: pathlib.Path) -> None:
448 r = _symlog(sym_repo, "billing.py::Invoice")
449 assert r.exit_code == 0
450
451 def test_unknown_symbol_exits_zero_with_no_events(self, sym_repo: pathlib.Path) -> None:
452 r = _symlog(sym_repo, "billing.py::DoesNotExistXXX")
453 assert r.exit_code == 0
454 assert "no events found" in r.output
455
456 def test_bad_address_no_double_colon_exits_nonzero(self, sym_repo: pathlib.Path) -> None:
457 r = _symlog(sym_repo, "billing.py")
458 assert r.exit_code != 0
459
460 def test_max_1_sets_truncated_true_in_json(self, sym_repo: pathlib.Path) -> None:
461 r = _symlog(sym_repo, "billing.py::Invoice", "--max", "1", "-j")
462 assert r.exit_code == 0
463 d = json.loads(r.output)
464 assert d["truncated"] is True
465 assert d["total_commits_scanned"] == 1
466
467 def test_max_1_shows_truncation_warning_in_human(self, sym_repo: pathlib.Path) -> None:
468 r = _symlog(sym_repo, "billing.py::Invoice", "--max", "1")
469 assert r.exit_code == 0
470 assert "incomplete" in r.output or "limit" in r.output
471
472 def test_max_zero_exits_nonzero(self, sym_repo: pathlib.Path) -> None:
473 r = _symlog(sym_repo, "billing.py::Invoice", "--max", "0")
474 assert r.exit_code != 0
475
476 def test_invalid_from_ref_exits_nonzero(self, sym_repo: pathlib.Path) -> None:
477 r = _symlog(sym_repo, "billing.py::Invoice", "--from", "deadbeefdeadbeef")
478 assert r.exit_code != 0
479
480 def test_json_address_field_matches_input(self, sym_repo: pathlib.Path) -> None:
481 r = _symlog(sym_repo, "billing.py::Invoice", "-j")
482 assert r.exit_code == 0
483 assert json.loads(r.output)["address"] == "billing.py::Invoice"
484
485 def test_json_events_is_list(self, sym_repo: pathlib.Path) -> None:
486 r = _symlog(sym_repo, "billing.py::Invoice", "-j")
487 assert r.exit_code == 0
488 assert isinstance(json.loads(r.output)["events"], list)
489
490 def test_start_ref_is_head_by_default(self, sym_repo: pathlib.Path) -> None:
491 r = _symlog(sym_repo, "billing.py::Invoice", "-j")
492 assert r.exit_code == 0
493 assert json.loads(r.output)["start_ref"] == "HEAD"
494
495 def test_human_output_shows_symbol_header(self, sym_repo: pathlib.Path) -> None:
496 r = _symlog(sym_repo, "billing.py::Invoice")
497 assert r.exit_code == 0
498 assert "billing.py::Invoice" in r.output
499
500
501 # ──────────────────────────────────────────────────────────────────────────────
502 # Stress
503 # ──────────────────────────────────────────────────────────────────────────────
504
505
506 class TestStress:
507 def test_1000_symbol_event_constructions(self) -> None:
508 from muse.cli.commands.symbol_log import SymbolEvent
509 commit = _make_commit()
510 for i in range(1_000):
511 ev = SymbolEvent("modified", commit, f"f{i}.py::fn", f"detail {i}")
512 assert ev.kind == "modified"
513
514 def test_1000_to_dict_calls(self) -> None:
515 from muse.cli.commands.symbol_log import SymbolEvent
516 commit = _make_commit()
517 ev = SymbolEvent("created", commit, "f.py::fn", "x")
518 for _ in range(1_000):
519 d = ev.to_dict()
520 assert "event" in d
521
522 def test_flat_ops_10000_calls(self) -> None:
523 from muse.cli.commands.symbol_log import _flat_ops
524 ops = [{"op": "insert", "address": f"f{i}.py::fn"} for i in range(10)]
525 for _ in range(10_000):
526 result = _flat_ops(ops)
527 assert len(result) == 10
528
529 def test_concurrent_find_events(self) -> None:
530 from muse.cli.commands.symbol_log import _find_events_in_commit
531 commit = _make_commit(structured_delta=_insert_delta("f.py::fn"))
532 results: list[int] = []
533 lock = threading.Lock()
534
535 def _run() -> None:
536 evs, _ = _find_events_in_commit(commit, "f.py::fn")
537 with lock:
538 results.append(len(evs))
539
540 threads = [threading.Thread(target=_run) for _ in range(50)]
541 for t in threads: t.start()
542 for t in threads: t.join()
543 assert all(n == 1 for n in results)
544 assert len(results) == 50
545
546
547 # ──────────────────────────────────────────────────────────────────────────────
548 # Data integrity
549 # ──────────────────────────────────────────────────────────────────────────────
550
551
552 class TestDataIntegrity:
553 def test_to_dict_preserves_all_fields(self) -> None:
554 from muse.cli.commands.symbol_log import SymbolEvent
555 commit = _make_commit(
556 commit_id=fake_id("cc"),
557 message="fix: something",
558 committed_at=datetime.datetime(2026, 5, 1, tzinfo=datetime.timezone.utc),
559 )
560 ev = SymbolEvent("renamed", commit, "a.py::old", "old → new", "a.py::new")
561 d = ev.to_dict()
562 assert d["event"] == "renamed"
563 assert d["commit_id"] == fake_id("cc")
564 assert d["message"] == "fix: something"
565 assert d["address"] == "a.py::old"
566 assert d["detail"] == "old → new"
567 assert d["new_address"] == "a.py::new"
568 assert "2026-05-01" in d["committed_at"]
569
570 def test_events_in_json_are_chronological(self, sym_repo: pathlib.Path) -> None:
571 r = _symlog(sym_repo, "billing.py::Invoice", "-j")
572 assert r.exit_code == 0
573 events = json.loads(r.output)["events"]
574 if len(events) >= 2:
575 times = [datetime.datetime.fromisoformat(e["committed_at"]) for e in events]
576 assert times == sorted(times), "events not in chronological order"
577
578 def test_total_commits_scanned_is_int(self, sym_repo: pathlib.Path) -> None:
579 r = _symlog(sym_repo, "billing.py::Invoice", "-j")
580 assert r.exit_code == 0
581 assert isinstance(json.loads(r.output)["total_commits_scanned"], int)
582
583 def test_truncated_is_bool(self, sym_repo: pathlib.Path) -> None:
584 r = _symlog(sym_repo, "billing.py::Invoice", "-j")
585 assert r.exit_code == 0
586 assert isinstance(json.loads(r.output)["truncated"], bool)
587
588 def test_rename_tracking_continues_with_new_address(self) -> None:
589 from muse.cli.commands.symbol_log import _find_events_in_commit
590 rename_commit = _make_commit(structured_delta=_replace_delta("f.py::old", "renamed to new"))
591 _, next_addr = _find_events_in_commit(rename_commit, "f.py::old")
592 assert next_addr == "f.py::new"
593 insert_commit = _make_commit(structured_delta=_insert_delta("f.py::new"))
594 evs, _ = _find_events_in_commit(insert_commit, next_addr)
595 assert len(evs) == 1
596 assert evs[0].kind == "created"
597
598 def test_new_address_none_for_modified(self) -> None:
599 from muse.cli.commands.symbol_log import _find_events_in_commit
600 commit = _make_commit(structured_delta=_replace_delta("f.py::fn"))
601 evs, _ = _find_events_in_commit(commit, "f.py::fn")
602 assert evs[0].new_address is None
603
604 def test_json_serialisable(self, sym_repo: pathlib.Path) -> None:
605 r = _symlog(sym_repo, "billing.py::Invoice", "-j")
606 assert r.exit_code == 0
607 json.loads(r.output) # must not raise
608
609
610 # ──────────────────────────────────────────────────────────────────────────────
611 # Security
612 # ──────────────────────────────────────────────────────────────────────────────
613
614
615 class TestSecurity:
616 def test_ansi_in_address_does_not_crash(self, sym_repo: pathlib.Path) -> None:
617 ansi_addr = "\x1b[31mbad\x1b[0m.py::fn"
618 r = _symlog(sym_repo, ansi_addr)
619 assert r.exit_code in (0, 1, 2)
620
621 def test_ansi_in_commit_message_survives_to_dict(self) -> None:
622 from muse.cli.commands.symbol_log import SymbolEvent
623 malicious_msg = "\x1b[31mmalicious\x1b[0m"
624 ev = SymbolEvent("modified", _make_commit(message=malicious_msg), "f.py::fn", "x")
625 d = ev.to_dict()
626 assert d["message"] == malicious_msg
627
628 def test_very_long_address_does_not_crash(self, sym_repo: pathlib.Path) -> None:
629 long_addr = f"f.py::{'x' * 10_000}"
630 r = _symlog(sym_repo, long_addr)
631 assert r.exit_code in (0, 1, 2)
632
633 def test_unicode_in_address_does_not_crash(self, sym_repo: pathlib.Path) -> None:
634 r = _symlog(sym_repo, "音符.py::関数")
635 assert r.exit_code in (0, 1, 2)
636
637 def test_hostile_detail_survives_json_serialisation(self) -> None:
638 from muse.cli.commands.symbol_log import SymbolEvent
639 malicious = '"; DROP TABLE commits; --'
640 ev = SymbolEvent("modified", _make_commit(), "f.py::fn", malicious)
641 d = ev.to_dict()
642 assert json.loads(json.dumps(d))["detail"] == malicious
643
644 def test_very_long_message_in_commit_does_not_crash(self) -> None:
645 from muse.cli.commands.symbol_log import SymbolEvent
646 ev = SymbolEvent("modified", _make_commit(message="x" * 100_000), "f.py::fn", "x")
647 d = ev.to_dict()
648 assert len(d["message"]) == 100_000
649
650
651 # ──────────────────────────────────────────────────────────────────────────────
652 # Performance
653 # ──────────────────────────────────────────────────────────────────────────────
654
655
656 class TestPerformance:
657 def test_1000_to_dict_under_500ms(self) -> None:
658 from muse.cli.commands.symbol_log import SymbolEvent
659 ev = SymbolEvent("modified", _make_commit(), "f.py::fn", "impl changed")
660 start = time.perf_counter()
661 for _ in range(1_000):
662 ev.to_dict()
663 elapsed = time.perf_counter() - start
664 assert elapsed < 0.5, f"1 000 to_dict calls took {elapsed:.2f}s"
665
666 def test_duration_ms_present_and_reasonable(self, sym_repo: pathlib.Path) -> None:
667 r = _symlog(sym_repo, "billing.py::Invoice", "-j")
668 assert r.exit_code == 0
669 d = json.loads(r.output)
670 assert "duration_ms" in d
671 assert 0 <= d["duration_ms"] < 30_000
672
673
674 # Flag registration
675 # ──────────────────────────────────────────────────────────────────────────────
676
677
678 class TestRegisterFlags:
679 def _parse(self, *args: str) -> "argparse.Namespace":
680 import argparse
681 from muse.cli.commands.symbol_log import register
682 p = argparse.ArgumentParser()
683 sub = p.add_subparsers()
684 register(sub)
685 return p.parse_args(["symbol-log", *args])
686
687 def test_default_json_out_is_false(self) -> None:
688 ns = self._parse("src/utils.py::fn")
689 assert ns.json_out is False
690
691 def test_json_flag_sets_json_out(self) -> None:
692 ns = self._parse("--json", "src/utils.py::fn")
693 assert ns.json_out is True
694
695 def test_j_shorthand_sets_json_out(self) -> None:
696 ns = self._parse("-j", "src/utils.py::fn")
697 assert ns.json_out is True
File History 4 commits
sha256:81ae324db5ad375fbfe4834c6fcb378312cafad3cc92dec5d3e5c427306621a2 fix: remove commit_exists filter from have anchors — server… Sonnet 4.6 patch 21 days ago
sha256:36c3cb3e76619d4c30a6d9bf81b5ec4ff148e30dcfed913e3114ca7b43b81c7e fix: rename objects→blobs in push client and all stale test… Sonnet 4.6 patch 22 days ago
sha256:c06a9b9b9fee26c68ea725b44d54b2c0a171301ce9de746d5b656617b4463a9a fix: repair four test failures from post-migration audit Sonnet 4.6 patch 29 days ago
sha256:1900655993c83c4107067375548a7be823e471d2515830842f1a12cba4bd3cdf fix: unified object store migration — idempotent writes, JS… Sonnet 4.6 minor ⚠ 29 days ago