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