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