gabriel / muse public
test_velocity_supercharge.py python
747 lines 32.6 KB
Raw
sha256:88ac91129873e6a496e9189515aa690eb893ae25d69c8f72af141a2be5068eb3 docs: docstring sprint contract→find-symbol — idiomatic run… Sonnet 4.6 patch 139 days ago
1 """Seven-tier tests for ``muse/cli/commands/velocity.py``.
2
3 Tiers
4 -----
5 Unit — _module_of; _bar; _WindowStats.net; _compute_predictions;
6 _print_table (empty, with modules, with predictions, truncated).
7 Integration — _VelocityJson TypedDict fields; -j alias; register() docstring;
8 run() docstring envelope fields.
9 End-to-end — --json emits schema_version/mode/exit_code/duration_ms;
10 -j alias; --predict -j; human output unchanged; empty repo.
11 Stress — 1 000 _module_of; 500 _bar; _print_table with 200 modules.
12 Data integrity — schema_version str; exit_code int; duration_ms float;
13 modules list; predictions list; window_size preserved.
14 Security — hostile address in predictions survives JSON; SQL injection
15 in since; very long module path; unicode addresses.
16 Performance — 1 000 _module_of under 200 ms; velocity JSON completes quickly.
17 """
18
19 from __future__ import annotations
20
21 import json
22 import os
23 import pathlib
24 import textwrap
25 import threading
26 import time
27 from typing import get_type_hints
28
29 import pytest
30
31 from tests.cli_test_helper import CliRunner, InvokeResult
32
33 runner = CliRunner()
34
35
36 # ──────────────────────────────────────────────────────────────────────────────
37 # Shared helpers
38 # ──────────────────────────────────────────────────────────────────────────────
39
40
41 def _make_window(**kw):
42 from muse.cli.commands.velocity import _WindowStats
43 w = _WindowStats()
44 for k, v in kw.items():
45 setattr(w, k, v)
46 return w
47
48
49 def _make_accumulator(current=None, prior=None, last_active_rank=-1, stagnant_commits=0):
50 from muse.cli.commands.velocity import _ModuleAccumulator, _WindowStats
51 acc = _ModuleAccumulator()
52 if current is not None:
53 acc.current = current
54 if prior is not None:
55 acc.prior = prior
56 acc.last_active_rank = last_active_rank
57 acc.stagnant_commits = stagnant_commits
58 return acc
59
60
61 def _make_sym_freq(frequency=3, last_rank=0, module="src/"):
62 from muse.cli.commands.velocity import _SymbolFreq
63 sf = _SymbolFreq(frequency=frequency, last_rank=last_rank, module=module)
64 return sf
65
66
67 def _commit(repo: pathlib.Path, files: dict[str, str], message: str) -> None:
68 for name, content in files.items():
69 path = repo / name
70 path.parent.mkdir(parents=True, exist_ok=True)
71 path.write_text(content, encoding="utf-8")
72 saved = os.getcwd()
73 try:
74 os.chdir(repo)
75 runner.invoke(None, ["code", "add", "."])
76 runner.invoke(None, ["commit", "-m", message])
77 finally:
78 os.chdir(saved)
79
80
81 def _invoke(repo: pathlib.Path, args: list[str]) -> InvokeResult:
82 saved = os.getcwd()
83 try:
84 os.chdir(repo)
85 return runner.invoke(None, args)
86 finally:
87 os.chdir(saved)
88
89
90 @pytest.fixture()
91 def vel_repo(tmp_path: pathlib.Path) -> pathlib.Path:
92 """Minimal repo with two commits so velocity has commit history to walk."""
93 saved = os.getcwd()
94 try:
95 os.chdir(tmp_path)
96 runner.invoke(None, ["init"])
97 finally:
98 os.chdir(saved)
99
100 _commit(tmp_path, {
101 "src/calc.py": textwrap.dedent("""\
102 def add(a: int, b: int) -> int:
103 return a + b
104
105 def subtract(a: int, b: int) -> int:
106 return a - b
107 """),
108 "src/utils.py": textwrap.dedent("""\
109 def helper(x: str) -> str:
110 return x.upper()
111 """),
112 }, "feat: initial symbols")
113
114 _commit(tmp_path, {
115 "src/calc.py": textwrap.dedent("""\
116 def add(a: int, b: int) -> int:
117 return a + b
118
119 def subtract(a: int, b: int) -> int:
120 return a - b
121
122 def multiply(a: int, b: int) -> int:
123 return a * b
124 """),
125 "src/engine.py": textwrap.dedent("""\
126 def process(data: list) -> list:
127 return sorted(data)
128
129 def validate(item) -> bool:
130 return item is not None
131 """),
132 }, "feat: add multiply and engine module")
133
134 return tmp_path
135
136
137 # ──────────────────────────────────────────────────────────────────────────────
138 # Unit — _module_of
139 # ──────────────────────────────────────────────────────────────────────────────
140
141
142 class TestModuleOf:
143 def test_nested_path_returns_dir_with_slash(self) -> None:
144 from muse.cli.commands.velocity import _module_of
145 assert _module_of("muse/core/store.py") == "muse/core/"
146
147 def test_single_level_path(self) -> None:
148 from muse.cli.commands.velocity import _module_of
149 assert _module_of("tests/test_foo.py") == "tests/"
150
151 def test_root_file_returns_root_sentinel(self) -> None:
152 from muse.cli.commands.velocity import _module_of
153 assert _module_of("billing.py") == "(root)"
154
155 def test_windows_backslash_normalised(self) -> None:
156 from muse.cli.commands.velocity import _module_of
157 assert _module_of("muse\\core\\store.py") == "muse/core/"
158
159 def test_deeply_nested_path(self) -> None:
160 from muse.cli.commands.velocity import _module_of
161 assert _module_of("a/b/c/d/e.py") == "a/b/c/d/"
162
163 def test_returns_str(self) -> None:
164 from muse.cli.commands.velocity import _module_of
165 assert isinstance(_module_of("src/foo.py"), str)
166
167
168 # ──────────────────────────────────────────────────────────────────────────────
169 # Unit — _bar
170 # ──────────────────────────────────────────────────────────────────────────────
171
172
173 class TestBar:
174 def test_zero_max_returns_empty(self) -> None:
175 from muse.cli.commands.velocity import _bar
176 assert _bar(5, 0) == ""
177
178 def test_positive_net_returns_filled_bar(self) -> None:
179 from muse.cli.commands.velocity import _bar
180 result = _bar(10, 10)
181 assert "█" in result
182
183 def test_negative_net_includes_negative_label(self) -> None:
184 from muse.cli.commands.velocity import _bar
185 result = _bar(-5, 10)
186 assert "net negative" in result
187
188 def test_zero_net_returns_minimal_bar(self) -> None:
189 from muse.cli.commands.velocity import _bar
190 result = _bar(0, 10)
191 # Zero net with positive max: filled=0 → returns "▏"
192 assert result == "▏"
193
194 def test_full_bar_has_max_blocks(self) -> None:
195 from muse.cli.commands.velocity import _bar, _BAR_WIDTH
196 result = _bar(100, 100)
197 assert result.count("█") == _BAR_WIDTH
198
199 def test_partial_bar_proportional(self) -> None:
200 from muse.cli.commands.velocity import _bar
201 full = _bar(10, 10)
202 half = _bar(5, 10)
203 assert len(half) <= len(full)
204
205
206 # ──────────────────────────────────────────────────────────────────────────────
207 # Unit — _WindowStats
208 # ──────────────────────────────────────────────────────────────────────────────
209
210
211 class TestWindowStats:
212 def test_net_positive(self) -> None:
213 from muse.cli.commands.velocity import _WindowStats
214 w = _WindowStats(added=10, removed=3)
215 assert w.net == 7
216
217 def test_net_negative(self) -> None:
218 from muse.cli.commands.velocity import _WindowStats
219 w = _WindowStats(added=2, removed=8)
220 assert w.net == -6
221
222 def test_net_zero(self) -> None:
223 from muse.cli.commands.velocity import _WindowStats
224 w = _WindowStats(added=5, removed=5)
225 assert w.net == 0
226
227 def test_defaults_are_zero(self) -> None:
228 from muse.cli.commands.velocity import _WindowStats
229 w = _WindowStats()
230 assert w.added == 0
231 assert w.removed == 0
232 assert w.modified == 0
233 assert w.active_commits == 0
234 assert w.net == 0
235
236
237 # ──────────────────────────────────────────────────────────────────────────────
238 # Unit — _compute_predictions
239 # ──────────────────────────────────────────────────────────────────────────────
240
241
242 class TestComputePredictions:
243 def test_empty_symbol_freq_returns_empty(self) -> None:
244 from muse.cli.commands.velocity import _compute_predictions
245 result = _compute_predictions({}, {}, window_size=20, top_k=5)
246 assert result == []
247
248 def test_top_k_zero_returns_empty(self) -> None:
249 from muse.cli.commands.velocity import _compute_predictions
250 sym = {"src/foo.py::bar": _make_sym_freq(frequency=5)}
251 result = _compute_predictions(sym, {}, window_size=20, top_k=0)
252 assert result == []
253
254 def test_top_k_limits_results(self) -> None:
255 from muse.cli.commands.velocity import _compute_predictions
256 syms = {f"src/f{i}.py::fn": _make_sym_freq(frequency=i + 1)
257 for i in range(10)}
258 result = _compute_predictions(syms, {}, window_size=20, top_k=3)
259 assert len(result) == 3
260
261 def test_higher_frequency_ranks_first(self) -> None:
262 from muse.cli.commands.velocity import _compute_predictions
263 syms = {
264 "a.py::low": _make_sym_freq(frequency=1, last_rank=0),
265 "b.py::high": _make_sym_freq(frequency=10, last_rank=0),
266 }
267 result = _compute_predictions(syms, {}, window_size=20, top_k=2)
268 assert result[0]["address"] == "b.py::high"
269
270 def test_result_has_required_fields(self) -> None:
271 from muse.cli.commands.velocity import _compute_predictions
272 syms = {"src/foo.py::bar": _make_sym_freq(frequency=3)}
273 result = _compute_predictions(syms, {}, window_size=20, top_k=1)
274 assert len(result) == 1
275 out = result[0]
276 for field in ("address", "module", "score", "frequency", "last_commit_rank"):
277 assert field in out, f"missing: {field}"
278
279 def test_score_is_positive(self) -> None:
280 from muse.cli.commands.velocity import _compute_predictions
281 syms = {"src/foo.py::bar": _make_sym_freq(frequency=5, last_rank=1)}
282 result = _compute_predictions(syms, {}, window_size=20, top_k=1)
283 assert result[0]["score"] > 0.0
284
285 def test_recent_symbol_scores_higher_than_old(self) -> None:
286 from muse.cli.commands.velocity import _compute_predictions
287 syms = {
288 "a.py::recent": _make_sym_freq(frequency=2, last_rank=0),
289 "b.py::old": _make_sym_freq(frequency=2, last_rank=15),
290 }
291 result = _compute_predictions(syms, {}, window_size=20, top_k=2)
292 assert result[0]["address"] == "a.py::recent"
293
294 def test_module_velocity_boost_applied(self) -> None:
295 from muse.cli.commands.velocity import _compute_predictions, _ModuleAccumulator, _WindowStats
296 fast_mod = _ModuleAccumulator()
297 fast_mod.current = _WindowStats(added=10, removed=0)
298 slow_mod = _ModuleAccumulator()
299 slow_mod.current = _WindowStats(added=0, removed=0)
300 modules = {"fast/": fast_mod, "slow/": slow_mod}
301 syms = {
302 "fast/a.py::fn": _make_sym_freq(frequency=3, last_rank=0, module="fast/"),
303 "slow/b.py::fn": _make_sym_freq(frequency=3, last_rank=0, module="slow/"),
304 }
305 result = _compute_predictions(syms, modules, window_size=20, top_k=2)
306 # fast/ module gets velocity boost
307 assert result[0]["module"] == "fast/"
308
309
310 # ──────────────────────────────────────────────────────────────────────────────
311 # Unit — _print_table
312 # ──────────────────────────────────────────────────────────────────────────────
313
314
315 class TestPrintTable:
316 def test_empty_ranked_prints_no_changes_message(self, capsys) -> None:
317 from muse.cli.commands.velocity import _print_table
318 _print_table([], [], ref="dev", commits_analysed=10,
319 window_size=5, truncated=False, since=None)
320 assert "no modules" in capsys.readouterr().out.lower()
321
322 def test_header_shows_ref(self, capsys) -> None:
323 from muse.cli.commands.velocity import _print_table
324 _print_table([], [], ref="my-branch", commits_analysed=0,
325 window_size=5, truncated=False, since=None)
326 assert "my-branch" in capsys.readouterr().out
327
328 def test_truncated_shows_warning(self, capsys) -> None:
329 from muse.cli.commands.velocity import _print_table
330 _print_table([], [], ref="dev", commits_analysed=100,
331 window_size=5, truncated=True, since=None)
332 assert "truncated" in capsys.readouterr().out
333
334 def test_since_shown_in_scope(self, capsys) -> None:
335 from muse.cli.commands.velocity import _print_table
336 _print_table([], [], ref="dev", commits_analysed=10,
337 window_size=5, truncated=False, since="v1.0")
338 assert "v1.0" in capsys.readouterr().out
339
340 def test_module_row_shown(self, capsys) -> None:
341 from muse.cli.commands.velocity import _print_table
342 acc = _make_accumulator(
343 current=_make_window(added=5, removed=1, modified=3),
344 )
345 _print_table([("src/core/", acc)], [], ref="dev",
346 commits_analysed=20, window_size=10,
347 truncated=False, since=None)
348 assert "src/core/" in capsys.readouterr().out
349
350 def test_stagnant_module_shows_note(self, capsys) -> None:
351 from muse.cli.commands.velocity import _print_table
352 acc = _make_accumulator(stagnant_commits=8)
353 _print_table([("docs/", acc)], [], ref="dev",
354 commits_analysed=20, window_size=10,
355 truncated=False, since=None)
356 assert "stagnant" in capsys.readouterr().out
357
358 def test_predictions_shown(self, capsys) -> None:
359 from muse.cli.commands.velocity import _print_table, _PredictionOut
360 pred = _PredictionOut(address="src/core/store.py::read",
361 module="src/core/", score=0.91,
362 frequency=5, last_commit_rank=0)
363 acc = _make_accumulator(current=_make_window(added=1))
364 _print_table([("src/core/", acc)], [pred], ref="dev", commits_analysed=20,
365 window_size=10, truncated=False, since=None)
366 assert "src/core/store.py::read" in capsys.readouterr().out
367
368 def test_acceleration_leader_shown(self, capsys) -> None:
369 from muse.cli.commands.velocity import _print_table
370 current = _make_window(added=10, removed=0, modified=5)
371 prior = _make_window(added=2, removed=0, modified=1)
372 acc = _make_accumulator(current=current, prior=prior)
373 _print_table([("src/hot/", acc)], [], ref="dev",
374 commits_analysed=40, window_size=20,
375 truncated=False, since=None)
376 out = capsys.readouterr().out
377 assert "Acceleration" in out or "src/hot/" in out
378
379
380 # ──────────────────────────────────────────────────────────────────────────────
381 # Integration — TypedDict, alias, docstrings
382 # ──────────────────────────────────────────────────────────────────────────────
383
384
385 class TestTypedDict:
386 def test_velocity_json_has_schema_version(self) -> None:
387 from muse.cli.commands.velocity import _VelocityJson
388 assert "schema_version" in get_type_hints(_VelocityJson)
389
390 def test_velocity_json_has_exit_code(self) -> None:
391 from muse.cli.commands.velocity import _VelocityJson
392 assert "exit_code" in get_type_hints(_VelocityJson)
393
394 def test_velocity_json_has_duration_ms(self) -> None:
395 from muse.cli.commands.velocity import _VelocityJson
396 assert "duration_ms" in get_type_hints(_VelocityJson)
397
398 def test_velocity_json_has_mode(self) -> None:
399 from muse.cli.commands.velocity import _VelocityJson
400 assert "mode" in get_type_hints(_VelocityJson)
401
402 def test_velocity_json_has_modules(self) -> None:
403 from muse.cli.commands.velocity import _VelocityJson
404 assert "modules" in get_type_hints(_VelocityJson)
405
406 def test_velocity_json_has_predictions(self) -> None:
407 from muse.cli.commands.velocity import _VelocityJson
408 assert "predictions" in get_type_hints(_VelocityJson)
409
410 def test_velocity_json_has_ref(self) -> None:
411 from muse.cli.commands.velocity import _VelocityJson
412 assert "ref" in get_type_hints(_VelocityJson)
413
414 def test_velocity_json_has_window_size(self) -> None:
415 from muse.cli.commands.velocity import _VelocityJson
416 assert "window_size" in get_type_hints(_VelocityJson)
417
418 def test_velocity_json_has_truncated(self) -> None:
419 from muse.cli.commands.velocity import _VelocityJson
420 assert "truncated" in get_type_hints(_VelocityJson)
421
422
423 class TestAliasRegistration:
424 def test_j_alias_registered(self) -> None:
425 from muse.cli.commands.velocity import register
426 import argparse
427 p = argparse.ArgumentParser()
428 sub = p.add_subparsers()
429 register(sub)
430 ns = p.parse_args(["velocity", "-j"])
431 assert ns.as_json is True
432
433 def test_json_long_form_works(self) -> None:
434 from muse.cli.commands.velocity import register
435 import argparse
436 p = argparse.ArgumentParser()
437 sub = p.add_subparsers()
438 register(sub)
439 ns = p.parse_args(["velocity", "--json"])
440 assert ns.as_json is True
441
442 def test_j_and_predict_parse_together(self) -> None:
443 from muse.cli.commands.velocity import register
444 import argparse
445 p = argparse.ArgumentParser()
446 sub = p.add_subparsers()
447 register(sub)
448 ns = p.parse_args(["velocity", "--predict", "5", "-j"])
449 assert ns.as_json is True
450 assert ns.predict == 5
451
452
453 class TestDocstrings:
454 def test_register_mentions_j_alias(self) -> None:
455 from muse.cli.commands.velocity import register
456 doc = register.__doc__ or ""
457 assert "-j" in doc or "--json" in doc
458
459 def test_run_mentions_schema_version(self) -> None:
460 from muse.cli.commands.velocity import run
461 assert "schema_version" in (run.__doc__ or "")
462
463 def test_run_mentions_exit_code(self) -> None:
464 from muse.cli.commands.velocity import run
465 assert "exit_code" in (run.__doc__ or "")
466
467 def test_run_mentions_duration_ms(self) -> None:
468 from muse.cli.commands.velocity import run
469 assert "duration_ms" in (run.__doc__ or "")
470
471 def test_run_mentions_mode(self) -> None:
472 from muse.cli.commands.velocity import run
473 assert "mode" in (run.__doc__ or "")
474
475
476 # ──────────────────────────────────────────────────────────────────────────────
477 # End-to-end
478 # ──────────────────────────────────────────────────────────────────────────────
479
480
481 class TestEndToEnd:
482 def test_json_exits_zero(self, vel_repo) -> None:
483 r = _invoke(vel_repo, ["code", "velocity", "--json"])
484 assert r.exit_code == 0
485
486 def test_json_emits_schema_version(self, vel_repo) -> None:
487 r = _invoke(vel_repo, ["code", "velocity", "--json"])
488 assert r.exit_code == 0
489 assert "schema_version" in json.loads(r.output)
490
491 def test_json_emits_mode(self, vel_repo) -> None:
492 r = _invoke(vel_repo, ["code", "velocity", "--json"])
493 assert r.exit_code == 0
494 assert json.loads(r.output)["mode"] == "velocity"
495
496 def test_json_emits_exit_code(self, vel_repo) -> None:
497 r = _invoke(vel_repo, ["code", "velocity", "--json"])
498 assert r.exit_code == 0
499 assert isinstance(json.loads(r.output)["exit_code"], int)
500
501 def test_json_emits_duration_ms(self, vel_repo) -> None:
502 r = _invoke(vel_repo, ["code", "velocity", "--json"])
503 assert r.exit_code == 0
504 d = json.loads(r.output)
505 assert isinstance(d["duration_ms"], float)
506 assert d["duration_ms"] >= 0.0
507
508 def test_json_emits_modules_list(self, vel_repo) -> None:
509 r = _invoke(vel_repo, ["code", "velocity", "--json"])
510 assert r.exit_code == 0
511 assert isinstance(json.loads(r.output)["modules"], list)
512
513 def test_json_emits_predictions_list(self, vel_repo) -> None:
514 r = _invoke(vel_repo, ["code", "velocity", "--json"])
515 assert r.exit_code == 0
516 assert isinstance(json.loads(r.output)["predictions"], list)
517
518 def test_json_emits_window_size(self, vel_repo) -> None:
519 r = _invoke(vel_repo, ["code", "velocity", "--window", "5", "--json"])
520 assert r.exit_code == 0
521 d = json.loads(r.output)
522 assert d["window_size"] == 5
523
524 def test_json_emits_ref(self, vel_repo) -> None:
525 r = _invoke(vel_repo, ["code", "velocity", "--json"])
526 assert r.exit_code == 0
527 assert isinstance(json.loads(r.output)["ref"], str)
528
529 def test_json_emits_truncated(self, vel_repo) -> None:
530 r = _invoke(vel_repo, ["code", "velocity", "--json"])
531 assert r.exit_code == 0
532 assert isinstance(json.loads(r.output)["truncated"], bool)
533
534 def test_j_alias_produces_json(self, vel_repo) -> None:
535 r = _invoke(vel_repo, ["code", "velocity", "-j"])
536 assert r.exit_code == 0
537 d = json.loads(r.output)
538 assert d["mode"] == "velocity"
539
540 def test_predict_j_includes_predictions(self, vel_repo) -> None:
541 r = _invoke(vel_repo, ["code", "velocity", "--predict", "3", "-j"])
542 assert r.exit_code == 0
543 d = json.loads(r.output)
544 assert "predictions" in d
545 # predictions may be empty if no current-window hits, but key must exist
546 assert isinstance(d["predictions"], list)
547
548 def test_human_output_still_works(self, vel_repo) -> None:
549 r = _invoke(vel_repo, ["code", "velocity"])
550 assert r.exit_code == 0
551 assert "Symbol velocity" in r.output
552
553 def test_schema_version_matches_muse_version(self, vel_repo) -> None:
554 from muse import __version__
555 r = _invoke(vel_repo, ["code", "velocity", "--json"])
556 assert r.exit_code == 0
557 assert json.loads(r.output)["schema_version"] == __version__
558
559
560 # ──────────────────────────────────────────────────────────────────────────────
561 # Stress
562 # ──────────────────────────────────────────────────────────────────────────────
563
564
565 class TestStress:
566 def test_1000_module_of(self) -> None:
567 from muse.cli.commands.velocity import _module_of
568 for i in range(1_000):
569 r = _module_of(f"src/pkg_{i}/file_{i}.py")
570 assert r == f"src/pkg_{i}/"
571
572 def test_500_bar_calls(self) -> None:
573 from muse.cli.commands.velocity import _bar
574 for i in range(500):
575 r = _bar(i % 20, 20)
576 assert isinstance(r, str)
577
578 def test_print_table_200_modules(self, capsys) -> None:
579 from muse.cli.commands.velocity import _print_table
580 ranked = [
581 (f"mod_{i}/", _make_accumulator(
582 current=_make_window(added=i, removed=0, modified=i),
583 ))
584 for i in range(200)
585 ]
586 _print_table(ranked, [], ref="dev", commits_analysed=400,
587 window_size=20, truncated=False, since=None)
588 out = capsys.readouterr().out
589 assert "mod_0/" in out
590
591 def test_concurrent_module_of(self) -> None:
592 from muse.cli.commands.velocity import _module_of
593 results: list[str] = []
594 lock = threading.Lock()
595
596 def _run() -> None:
597 r = _module_of("src/core/store.py")
598 with lock:
599 results.append(r)
600
601 threads = [threading.Thread(target=_run) for _ in range(50)]
602 for t in threads: t.start()
603 for t in threads: t.join()
604 assert all(r == "src/core/" for r in results)
605 assert len(results) == 50
606
607
608 # ──────────────────────────────────────────────────────────────────────────────
609 # Data integrity
610 # ──────────────────────────────────────────────────────────────────────────────
611
612
613 class TestDataIntegrity:
614 def test_schema_version_is_str(self, vel_repo) -> None:
615 r = _invoke(vel_repo, ["code", "velocity", "--json"])
616 assert isinstance(json.loads(r.output)["schema_version"], str)
617
618 def test_schema_version_nonempty(self, vel_repo) -> None:
619 r = _invoke(vel_repo, ["code", "velocity", "--json"])
620 assert len(json.loads(r.output)["schema_version"]) > 0
621
622 def test_exit_code_is_int(self, vel_repo) -> None:
623 r = _invoke(vel_repo, ["code", "velocity", "--json"])
624 assert isinstance(json.loads(r.output)["exit_code"], int)
625
626 def test_duration_ms_is_float(self, vel_repo) -> None:
627 r = _invoke(vel_repo, ["code", "velocity", "--json"])
628 d = json.loads(r.output)
629 assert isinstance(d["duration_ms"], float)
630
631 def test_duration_ms_non_negative(self, vel_repo) -> None:
632 r = _invoke(vel_repo, ["code", "velocity", "--json"])
633 assert json.loads(r.output)["duration_ms"] >= 0.0
634
635 def test_window_size_preserved(self, vel_repo) -> None:
636 r = _invoke(vel_repo, ["code", "velocity", "--window", "7", "--json"])
637 assert json.loads(r.output)["window_size"] == 7
638
639 def test_modules_entries_have_required_fields(self, vel_repo) -> None:
640 r = _invoke(vel_repo, ["code", "velocity", "--json"])
641 d = json.loads(r.output)
642 for m in d["modules"]:
643 for field in ("module", "current", "prior", "acceleration", "stagnant_commits"):
644 assert field in m, f"missing: {field}"
645
646 def test_json_serialisable(self, vel_repo) -> None:
647 r = _invoke(vel_repo, ["code", "velocity", "--json"])
648 json.loads(r.output) # must not raise
649
650 def test_mode_is_velocity(self, vel_repo) -> None:
651 r = _invoke(vel_repo, ["code", "velocity", "--json"])
652 assert json.loads(r.output)["mode"] == "velocity"
653
654 def test_predictions_entries_have_required_fields(self, vel_repo) -> None:
655 r = _invoke(vel_repo, ["code", "velocity", "--predict", "5", "--json"])
656 d = json.loads(r.output)
657 for p in d["predictions"]:
658 for field in ("address", "score"):
659 assert field in p, f"missing: {field}"
660
661
662 # ──────────────────────────────────────────────────────────────────────────────
663 # Security
664 # ──────────────────────────────────────────────────────────────────────────────
665
666
667 class TestSecurity:
668 def test_hostile_address_in_predictions_survives_json(self) -> None:
669 from muse.cli.commands.velocity import _compute_predictions, _SymbolFreq
670 evil = '"; DROP TABLE commits; --'
671 syms = {evil: _make_sym_freq(frequency=5)}
672 result = _compute_predictions(syms, {}, window_size=20, top_k=1)
673 assert len(result) == 1
674 serialised = json.dumps(result[0])
675 assert json.loads(serialised)["address"] == evil
676
677 def test_xss_in_module_name_does_not_crash(self, capsys) -> None:
678 from muse.cli.commands.velocity import _print_table
679 acc = _make_accumulator(current=_make_window(added=1))
680 _print_table([("<script>alert(1)</script>/", acc)], [],
681 ref="dev", commits_analysed=5,
682 window_size=2, truncated=False, since=None)
683 assert "<script>" in capsys.readouterr().out
684
685 def test_very_long_module_path_does_not_crash(self, capsys) -> None:
686 from muse.cli.commands.velocity import _print_table
687 long_mod = "x" * 500 + "/"
688 acc = _make_accumulator(current=_make_window(added=2))
689 _print_table([(long_mod, acc)], [],
690 ref="dev", commits_analysed=5,
691 window_size=2, truncated=False, since=None)
692 assert capsys.readouterr().out # no crash
693
694 def test_unicode_in_prediction_address(self) -> None:
695 from muse.cli.commands.velocity import _compute_predictions
696 addr = "src/音符.py::計算"
697 syms = {addr: _make_sym_freq(frequency=3)}
698 result = _compute_predictions(syms, {}, window_size=20, top_k=1)
699 serialised = json.dumps(result[0], ensure_ascii=False)
700 assert json.loads(serialised)["address"] == addr
701
702 def test_null_byte_in_module_survives(self) -> None:
703 from muse.cli.commands.velocity import _module_of
704 # null byte in path — should not crash
705 result = _module_of("src\x00evil/file.py")
706 assert isinstance(result, str)
707
708 def test_very_large_frequency_does_not_crash(self) -> None:
709 from muse.cli.commands.velocity import _compute_predictions
710 syms = {"src/foo.py::bar": _make_sym_freq(frequency=10**9)}
711 result = _compute_predictions(syms, {}, window_size=20, top_k=1)
712 assert len(result) == 1
713 assert result[0]["score"] > 0
714
715
716 # ──────────────────────────────────────────────────────────────────────────────
717 # Performance
718 # ──────────────────────────────────────────────────────────────────────────────
719
720
721 class TestPerformance:
722 def test_1000_module_of_under_200ms(self) -> None:
723 from muse.cli.commands.velocity import _module_of
724 start = time.perf_counter()
725 for i in range(1_000):
726 _module_of(f"muse/core/sub_{i}/file.py")
727 elapsed = time.perf_counter() - start
728 assert elapsed < 0.2, f"1 000 _module_of took {elapsed:.3f}s"
729
730 def test_500_bar_under_100ms(self) -> None:
731 from muse.cli.commands.velocity import _bar
732 start = time.perf_counter()
733 for i in range(500):
734 _bar(i, 500)
735 elapsed = time.perf_counter() - start
736 assert elapsed < 0.1, f"500 _bar took {elapsed:.3f}s"
737
738 def test_velocity_json_completes_quickly(self, vel_repo) -> None:
739 start = time.perf_counter()
740 r = _invoke(vel_repo, ["code", "velocity", "--json"])
741 elapsed = time.perf_counter() - start
742 assert r.exit_code == 0
743 assert elapsed < 15.0, f"velocity --json took {elapsed:.2f}s"
744
745 def test_duration_ms_positive(self, vel_repo) -> None:
746 r = _invoke(vel_repo, ["code", "velocity", "--json"])
747 assert json.loads(r.output)["duration_ms"] >= 0.0
File History 1 commit
sha256:88ac91129873e6a496e9189515aa690eb893ae25d69c8f72af141a2be5068eb3 docs: docstring sprint contract→find-symbol — idiomatic run… Sonnet 4.6 patch 139 days ago