gabriel / muse public
test_directory_dimension.py python
514 lines 20.1 KB
Raw
1 """TDD — directory-level insight dimension (issue #3).
2
3 Phase 1: helpers in _query.py
4 flat_directory_ops — yields RenameOp / insert/delete directory ops
5 touched_directories — set of directories affected by an op list
6 dir_of — extract parent directory from a file path
7
8 Phase 2: muse diff -- directories key in JSON output
9 rename, added dir, deleted dir surfaced as distinct dimension
10
11 Phase 3: muse code hotspots --granularity directory
12 churn counted at directory level instead of symbol level
13
14 Phase 4: muse code entangle --granularity directory
15 co-change pairs at directory granularity
16
17 Phase 5: muse code impact --roll-up-to directory
18 blast radius rolled up to directory level
19 """
20
21 from __future__ import annotations
22
23 import json
24 import pathlib
25 import textwrap
26 from typing import cast
27
28 import pytest
29
30 from muse.domain import RenameOp, DomainOp, InsertOp, DeleteOp, PatchOp, ReplaceOp
31
32 # ---------------------------------------------------------------------------
33 # Helpers used across phases
34 # ---------------------------------------------------------------------------
35
36
37 def _insert(address: str, content_summary: str = "") -> InsertOp:
38 return InsertOp(op="insert", address=address, position=None, content_id="", content_summary=content_summary)
39
40
41 def _delete(address: str, content_summary: str = "") -> DeleteOp:
42 return DeleteOp(op="delete", address=address, position=None, content_id="", content_summary=content_summary)
43
44
45 def _rename(from_addr: str, to_addr: str, file_count: int = 2) -> RenameOp:
46 return RenameOp(op="rename", address=to_addr, from_address=from_addr)
47
48
49 def _patch(address: str, children: list[DomainOp] | None = None) -> PatchOp:
50 return PatchOp(
51 op="patch",
52 address=address,
53 child_ops=children or [],
54 child_domain="code",
55 child_summary="",
56 )
57
58
59 def _sym_insert(address: str) -> InsertOp:
60 return InsertOp(op="insert", address=address, position=None, content_id="", content_summary="added function")
61
62
63 # ---------------------------------------------------------------------------
64 # Phase 1A: flat_directory_ops
65 # ---------------------------------------------------------------------------
66
67
68 class TestFlatDirectoryOps:
69 """flat_directory_ops yields directory-level ops and ignores symbol/file ops."""
70
71 def test_yields_directory_rename(self) -> None:
72 from muse.plugins.code._query import flat_directory_ops
73 ops: list[DomainOp] = [_rename("src/old", "src/new")]
74 result = list(flat_directory_ops(ops))
75 assert len(result) == 1
76 assert result[0]["op"] == "rename"
77
78 def test_yields_directory_insert(self) -> None:
79 from muse.plugins.code._query import flat_directory_ops
80 ops: list[DomainOp] = [_insert("src/newdir/", "directory: src/newdir/")]
81 result = list(flat_directory_ops(ops))
82 assert len(result) == 1
83
84 def test_yields_directory_delete(self) -> None:
85 from muse.plugins.code._query import flat_directory_ops
86 ops: list[DomainOp] = [_delete("src/olddir/", "directory: src/olddir/")]
87 result = list(flat_directory_ops(ops))
88 assert len(result) == 1
89
90 def test_skips_symbol_level_patch_children(self) -> None:
91 from muse.plugins.code._query import flat_directory_ops
92 ops: list[DomainOp] = [
93 _patch("src/billing.py", [_sym_insert("src/billing.py::compute_total")]),
94 ]
95 result = list(flat_directory_ops(ops))
96 assert result == []
97
98 def test_skips_plain_file_insert(self) -> None:
99 from muse.plugins.code._query import flat_directory_ops
100 ops: list[DomainOp] = [_insert("src/billing.py", "added file")]
101 result = list(flat_directory_ops(ops))
102 assert result == []
103
104 def test_mixed_ops_only_dir(self) -> None:
105 from muse.plugins.code._query import flat_directory_ops
106 ops: list[DomainOp] = [
107 _rename("api/v1", "api/v2"),
108 _patch("src/billing.py", [_sym_insert("src/billing.py::fn")]),
109 _insert("src/utils.py", "added file"),
110 _insert("tests/", "directory: tests/"),
111 ]
112 result = list(flat_directory_ops(ops))
113 assert len(result) == 2 # rename + tests/ insert
114 op_types = {r["op"] for r in result}
115 assert "rename" in op_types
116
117 def test_empty_ops(self) -> None:
118 from muse.plugins.code._query import flat_directory_ops
119 assert list(flat_directory_ops([])) == []
120
121 def test_rename_carries_from_address(self) -> None:
122 from muse.plugins.code._query import flat_directory_ops
123 ops: list[DomainOp] = [_rename("src/auth_old", "src/auth")]
124 result = list(flat_directory_ops(ops))
125 assert result[0].get("from_address") == "src/auth_old"
126 assert result[0]["address"] == "src/auth"
127
128
129 # ---------------------------------------------------------------------------
130 # Phase 1B: touched_directories
131 # ---------------------------------------------------------------------------
132
133
134 class TestTouchedDirectories:
135 """touched_directories returns the set of directories whose files changed."""
136
137 def test_single_file_returns_its_parent(self) -> None:
138 from muse.plugins.code._query import touched_directories
139 ops: list[DomainOp] = [
140 _patch("src/billing.py", [_sym_insert("src/billing.py::fn")]),
141 ]
142 dirs = touched_directories(ops)
143 assert "src" in dirs
144
145 def test_root_level_file_returns_dot_or_empty(self) -> None:
146 from muse.plugins.code._query import touched_directories
147 ops: list[DomainOp] = [
148 _patch("main.py", [_sym_insert("main.py::fn")]),
149 ]
150 dirs = touched_directories(ops)
151 # root-level files belong to "." (POSIX convention)
152 assert "." in dirs
153
154 def test_multiple_files_same_dir_counted_once(self) -> None:
155 from muse.plugins.code._query import touched_directories
156 ops: list[DomainOp] = [
157 _patch("src/billing.py", [_sym_insert("src/billing.py::fn")]),
158 _patch("src/auth.py", [_sym_insert("src/auth.py::validate")]),
159 ]
160 dirs = touched_directories(ops)
161 assert dirs.count("src") == 1 if isinstance(dirs, list) else len([d for d in dirs if d == "src"]) == 1
162
163 def test_returns_frozenset(self) -> None:
164 from muse.plugins.code._query import touched_directories
165 ops: list[DomainOp] = [
166 _patch("src/a.py", [_sym_insert("src/a.py::fn")]),
167 ]
168 result = touched_directories(ops)
169 assert isinstance(result, frozenset)
170
171 def test_nested_path_returns_immediate_parent(self) -> None:
172 from muse.plugins.code._query import touched_directories
173 ops: list[DomainOp] = [
174 _patch("muse/cli/commands/cat.py", [_sym_insert("muse/cli/commands/cat.py::run")]),
175 ]
176 dirs = touched_directories(ops)
177 assert "muse/cli/commands" in dirs
178
179 def test_directory_rename_op_adds_both_dirs(self) -> None:
180 from muse.plugins.code._query import touched_directories
181 ops: list[DomainOp] = [_rename("api/v1", "api/v2")]
182 dirs = touched_directories(ops)
183 assert "api/v1" in dirs
184 assert "api/v2" in dirs
185
186 def test_empty_ops_returns_empty(self) -> None:
187 from muse.plugins.code._query import touched_directories
188 assert touched_directories([]) == frozenset()
189
190 def test_file_without_symbol_children_not_counted(self) -> None:
191 from muse.plugins.code._query import touched_directories
192 ops: list[DomainOp] = [
193 _patch("src/billing.py", []), # PatchOp with no children — non-semantic
194 ]
195 dirs = touched_directories(ops)
196 assert "src" not in dirs
197
198
199 # ---------------------------------------------------------------------------
200 # Phase 1C: dir_of
201 # ---------------------------------------------------------------------------
202
203
204 class TestDirOf:
205 """dir_of extracts the immediate parent directory from a file path."""
206
207 def test_nested_path(self) -> None:
208 from muse.plugins.code._query import dir_of
209 assert dir_of("src/billing.py") == "src"
210
211 def test_deeply_nested(self) -> None:
212 from muse.plugins.code._query import dir_of
213 assert dir_of("muse/cli/commands/cat.py") == "muse/cli/commands"
214
215 def test_root_level_file(self) -> None:
216 from muse.plugins.code._query import dir_of
217 assert dir_of("main.py") == "."
218
219 def test_directory_address_trailing_slash(self) -> None:
220 from muse.plugins.code._query import dir_of
221 assert dir_of("src/") == "src"
222
223 def test_no_extension_file(self) -> None:
224 from muse.plugins.code._query import dir_of
225 assert dir_of("src/Makefile") == "src"
226
227
228 # ---------------------------------------------------------------------------
229 # Phase 2: muse diff -- directory dimension in JSON output
230 # ---------------------------------------------------------------------------
231
232
233 from tests.cli_test_helper import CliRunner
234
235 cli = None
236 runner = CliRunner()
237
238
239 def _make_diff_repo(tmp_path: pathlib.Path) -> pathlib.Path:
240 """Init a repo and make two commits with a directory rename."""
241 from muse.core.object_store import write_object
242 from muse.core.ids import hash_commit as compute_commit_id, hash_snapshot as compute_snapshot_id
243 from muse.core.commits import (
244 CommitRecord,
245 write_commit,
246 )
247 from muse.core.snapshots import (
248 SnapshotRecord,
249 write_snapshot,
250 )
251 from muse.core.types import long_id, blob_id
252 from muse.core.paths import muse_dir, ref_path
253 import datetime
254
255 dot = muse_dir(tmp_path)
256 for d in ("commits", "snapshots", "objects", "refs/heads"):
257 (dot / d).mkdir(parents=True, exist_ok=True)
258 (dot / "HEAD").write_text("ref: refs/heads/main", encoding="utf-8")
259 (dot / "repo.json").write_text(json.dumps({"repo_id": "dir-dim-test", "domain": "code"}), encoding="utf-8")
260 return tmp_path
261
262
263 # ---------------------------------------------------------------------------
264 # Phase 3: muse code hotspots --granularity directory
265 # ---------------------------------------------------------------------------
266
267
268 class TestHotspotsDirectoryGranularity:
269 """muse code hotspots --granularity directory counts churn at dir level."""
270
271 def test_granularity_flag_accepted(self, tmp_path: pathlib.Path) -> None:
272 import argparse
273 from muse.cli.commands.hotspots import register
274 p = argparse.ArgumentParser()
275 subs = p.add_subparsers(dest="cmd")
276 register(subs)
277 args = p.parse_args(["hotspots", "--granularity", "directory"])
278 assert args.granularity == "directory"
279
280 def test_granularity_default_is_symbol(self, tmp_path: pathlib.Path) -> None:
281 import argparse
282 from muse.cli.commands.hotspots import register
283 p = argparse.ArgumentParser()
284 subs = p.add_subparsers(dest="cmd")
285 register(subs)
286 args = p.parse_args(["hotspots"])
287 assert args.granularity == "symbol"
288
289 def test_directory_granularity_json_addresses_have_no_colons(
290 self, tmp_path: pathlib.Path
291 ) -> None:
292 """Directory addresses are plain paths, never contain '::'."""
293 from tests.cli_test_helper import CliRunner as CR
294 r = CR()
295 from muse.core.object_store import write_object
296 from muse.core.ids import hash_commit as compute_commit_id, hash_snapshot as compute_snapshot_id
297 from muse.core.commits import (
298 CommitRecord,
299 write_commit,
300 )
301 from muse.core.snapshots import (
302 SnapshotRecord,
303 write_snapshot,
304 )
305 from muse.core.types import long_id, blob_id
306 from muse.core.paths import muse_dir, ref_path
307 import datetime
308
309 dot = muse_dir(tmp_path)
310 for d in ("commits", "snapshots", "objects", "refs/heads"):
311 (dot / d).mkdir(parents=True, exist_ok=True)
312 (dot / "HEAD").write_text("ref: refs/heads/main", encoding="utf-8")
313 (dot / "repo.json").write_text(json.dumps({"repo_id": "hs-dir-test", "domain": "code"}), encoding="utf-8")
314
315 result = r.invoke(
316 None,
317 ["code", "hotspots", "--granularity", "directory", "--json"],
318 env={"MUSE_REPO_ROOT": str(tmp_path)},
319 )
320 assert result.exit_code == 0
321 data = json.loads(result.output)
322 assert "hotspots" in data
323 for entry in data["hotspots"]:
324 assert "::" not in entry["address"], f"directory address contains '::': {entry['address']}"
325
326 def test_directory_granularity_json_has_granularity_field(
327 self, tmp_path: pathlib.Path
328 ) -> None:
329 from tests.cli_test_helper import CliRunner as CR
330 from muse.core.paths import muse_dir
331 dot = muse_dir(tmp_path)
332 for d in ("commits", "snapshots", "objects", "refs/heads"):
333 (dot / d).mkdir(parents=True, exist_ok=True)
334 (dot / "HEAD").write_text("ref: refs/heads/main", encoding="utf-8")
335 (dot / "repo.json").write_text(json.dumps({"repo_id": "hs-dir-test2", "domain": "code"}), encoding="utf-8")
336
337 r = CR()
338 result = r.invoke(
339 None,
340 ["code", "hotspots", "--granularity", "directory", "--json"],
341 env={"MUSE_REPO_ROOT": str(tmp_path)},
342 )
343 assert result.exit_code == 0
344 data = json.loads(result.output)
345 assert data.get("granularity") == "directory"
346
347 def test_symbol_granularity_json_addresses_contain_colons(
348 self, tmp_path: pathlib.Path
349 ) -> None:
350 """Default (symbol) granularity addresses are 'file.py::symbol'."""
351 from tests.cli_test_helper import CliRunner as CR
352 from muse.core.paths import muse_dir
353 dot = muse_dir(tmp_path)
354 for d in ("commits", "snapshots", "objects", "refs/heads"):
355 (dot / d).mkdir(parents=True, exist_ok=True)
356 (dot / "HEAD").write_text("ref: refs/heads/main", encoding="utf-8")
357 (dot / "repo.json").write_text(json.dumps({"repo_id": "hs-sym-test", "domain": "code"}), encoding="utf-8")
358
359 r = CR()
360 result = r.invoke(
361 None,
362 ["code", "hotspots", "--json"],
363 env={"MUSE_REPO_ROOT": str(tmp_path)},
364 )
365 assert result.exit_code == 0
366 data = json.loads(result.output)
367 assert data.get("granularity") == "symbol"
368
369
370 # ---------------------------------------------------------------------------
371 # Phase 4: muse code entangle --granularity directory
372 # ---------------------------------------------------------------------------
373
374
375 class TestEntangleDirectoryGranularity:
376 """muse code entangle --granularity directory finds directories that
377 always change together."""
378
379 def test_granularity_flag_accepted(self) -> None:
380 import argparse
381 from muse.cli.commands.entangle import register
382 p = argparse.ArgumentParser()
383 subs = p.add_subparsers(dest="cmd")
384 register(subs)
385 args = p.parse_args(["entangle", "--granularity", "directory"])
386 assert args.granularity == "directory"
387
388 def test_granularity_default_is_symbol(self) -> None:
389 import argparse
390 from muse.cli.commands.entangle import register
391 p = argparse.ArgumentParser()
392 subs = p.add_subparsers(dest="cmd")
393 register(subs)
394 args = p.parse_args(["entangle"])
395 assert args.granularity == "symbol"
396
397 def test_directory_granularity_json_pairs_have_no_colons(
398 self, tmp_path: pathlib.Path
399 ) -> None:
400 from tests.cli_test_helper import CliRunner as CR
401 from muse.core.paths import muse_dir
402 dot = muse_dir(tmp_path)
403 for d in ("commits", "snapshots", "objects", "refs/heads"):
404 (dot / d).mkdir(parents=True, exist_ok=True)
405 (dot / "HEAD").write_text("ref: refs/heads/main", encoding="utf-8")
406 (dot / "repo.json").write_text(json.dumps({"repo_id": "ent-dir-test", "domain": "code"}), encoding="utf-8")
407
408 r = CR()
409 result = r.invoke(
410 None,
411 ["code", "entangle", "--granularity", "directory", "--json"],
412 env={"MUSE_REPO_ROOT": str(tmp_path)},
413 )
414 assert result.exit_code == 0
415 data = json.loads(result.output)
416 assert "pairs" in data
417 for pair in data["pairs"]:
418 assert "::" not in pair["dir_a"]
419 assert "::" not in pair["dir_b"]
420
421 def test_directory_granularity_json_has_granularity_field(
422 self, tmp_path: pathlib.Path
423 ) -> None:
424 from tests.cli_test_helper import CliRunner as CR
425 from muse.core.paths import muse_dir
426 dot = muse_dir(tmp_path)
427 for d in ("commits", "snapshots", "objects", "refs/heads"):
428 (dot / d).mkdir(parents=True, exist_ok=True)
429 (dot / "HEAD").write_text("ref: refs/heads/main", encoding="utf-8")
430 (dot / "repo.json").write_text(json.dumps({"repo_id": "ent-dir-test2", "domain": "code"}), encoding="utf-8")
431
432 r = CR()
433 result = r.invoke(
434 None,
435 ["code", "entangle", "--granularity", "directory", "--json"],
436 env={"MUSE_REPO_ROOT": str(tmp_path)},
437 )
438 assert result.exit_code == 0
439 data = json.loads(result.output)
440 assert data.get("granularity") == "directory"
441
442
443 # ---------------------------------------------------------------------------
444 # Phase 5: muse code impact --roll-up-to directory
445 # ---------------------------------------------------------------------------
446
447
448 class TestImpactDirectoryRollup:
449 """muse code impact --roll-up-to directory aggregates blast radius by dir."""
450
451 def test_roll_up_to_flag_accepted(self) -> None:
452 import argparse
453 from muse.cli.commands.impact import register
454 p = argparse.ArgumentParser()
455 subs = p.add_subparsers(dest="cmd")
456 register(subs)
457 args = p.parse_args(["impact", "src/billing.py::compute_total", "--roll-up-to", "directory"])
458 assert args.roll_up_to == "directory"
459
460 def test_roll_up_to_default_is_none(self) -> None:
461 import argparse
462 from muse.cli.commands.impact import register
463 p = argparse.ArgumentParser()
464 subs = p.add_subparsers(dest="cmd")
465 register(subs)
466 args = p.parse_args(["impact", "src/billing.py::compute_total"])
467 assert args.roll_up_to is None
468
469 def test_directory_rollup_json_has_directory_blast_radius(
470 self, tmp_path: pathlib.Path
471 ) -> None:
472 from tests.cli_test_helper import CliRunner as CR
473 from muse.core.paths import muse_dir
474 dot = muse_dir(tmp_path)
475 for d in ("commits", "snapshots", "objects", "refs/heads"):
476 (dot / d).mkdir(parents=True, exist_ok=True)
477 (dot / "HEAD").write_text("ref: refs/heads/main", encoding="utf-8")
478 (dot / "repo.json").write_text(json.dumps({"repo_id": "imp-dir-test", "domain": "code"}), encoding="utf-8")
479
480 r = CR()
481 result = r.invoke(
482 None,
483 ["code", "impact", "src/billing.py::compute_total",
484 "--roll-up-to", "directory", "--json"],
485 env={"MUSE_REPO_ROOT": str(tmp_path)},
486 )
487 # May exit 1 (symbol not found in empty repo) — just verify schema when successful
488 if result.exit_code == 0:
489 data = json.loads(result.output)
490 assert "directory_blast_radius" in data
491
492 def test_directory_rollup_addresses_have_no_colons(
493 self, tmp_path: pathlib.Path
494 ) -> None:
495 """directory_blast_radius keys are plain directory paths, never 'file::sym'."""
496 from tests.cli_test_helper import CliRunner as CR
497 from muse.core.paths import muse_dir
498 dot = muse_dir(tmp_path)
499 for d in ("commits", "snapshots", "objects", "refs/heads"):
500 (dot / d).mkdir(parents=True, exist_ok=True)
501 (dot / "HEAD").write_text("ref: refs/heads/main", encoding="utf-8")
502 (dot / "repo.json").write_text(json.dumps({"repo_id": "imp-dir-test2", "domain": "code"}), encoding="utf-8")
503
504 r = CR()
505 result = r.invoke(
506 None,
507 ["code", "impact", "src/billing.py::compute_total",
508 "--roll-up-to", "directory", "--json"],
509 env={"MUSE_REPO_ROOT": str(tmp_path)},
510 )
511 if result.exit_code == 0:
512 data = json.loads(result.output)
513 for dir_path in data.get("directory_blast_radius", {}).keys():
514 assert "::" not in dir_path
File History 1 commit