test_knowtation_code_intel.py
python
sha256:660fcac1df3ab28f61862e961890bd2ca8b754fa0242079d93ca1e25037ec8a6
chore(tests): add docstring to tests/__init__.py so rc14 tr…
Human
27 days ago
| 1 | """Tests for Phase 3.2 — Knowtation deps / impact CLI integration. |
| 2 | |
| 3 | Test tiers covered (Rule #0): |
| 4 | |
| 5 | 1. Unit — describe_link round-trip |
| 6 | 2. Unit — note_deps forward + reverse, broken-link filtering, kind filtering |
| 7 | 3. Unit — note_impact: BFS depth, forward vs reverse, attachment flag echo, |
| 8 | negative depth rejection |
| 9 | 4. Integration — deps_for_note / impact_for_note on synthetic vault |
| 10 | 5. Integration — golden JSON output for a 50-note fixture vault (matches |
| 11 | issue spec) |
| 12 | 6. Data-integrity — round-trip: every link emitted by note_deps appears in |
| 13 | note_impact at depth 1 |
| 14 | 7. Performance — note_impact across 200-note linear chain completes < 2s |
| 15 | 8. Security — non-note path rejected, missing file rejected, path traversal |
| 16 | """ |
| 17 | |
| 18 | from __future__ import annotations |
| 19 | |
| 20 | import json |
| 21 | import pathlib |
| 22 | import time |
| 23 | |
| 24 | import pytest |
| 25 | |
| 26 | from muse.plugins.knowtation.code_intel import ( |
| 27 | deps_for_note, |
| 28 | describe_link, |
| 29 | impact_for_note, |
| 30 | note_deps, |
| 31 | note_impact, |
| 32 | ) |
| 33 | from muse.plugins.knowtation.link_index import ( |
| 34 | LinkIndex, |
| 35 | ResolvedLink, |
| 36 | build_link_index, |
| 37 | ) |
| 38 | |
| 39 | |
| 40 | # ============================================================================ |
| 41 | # Helpers |
| 42 | # ============================================================================ |
| 43 | |
| 44 | |
| 45 | def _make_vault(tmp_path: pathlib.Path, files: dict[str, bytes | str]) -> pathlib.Path: |
| 46 | for rel, content in files.items(): |
| 47 | full = tmp_path / rel |
| 48 | full.parent.mkdir(parents=True, exist_ok=True) |
| 49 | if isinstance(content, str): |
| 50 | full.write_text(content, encoding="utf-8") |
| 51 | else: |
| 52 | full.write_bytes(content) |
| 53 | return tmp_path |
| 54 | |
| 55 | |
| 56 | # ============================================================================ |
| 57 | # TestDescribeLink |
| 58 | # ============================================================================ |
| 59 | |
| 60 | |
| 61 | class TestDescribeLink: |
| 62 | def test_describe_link_round_trip(self) -> None: |
| 63 | link = ResolvedLink( |
| 64 | source="a.md", target="b.md", raw="b", kind="wikilink", |
| 65 | fragment="section", broken=False, alias="display", |
| 66 | ) |
| 67 | out = describe_link(link) |
| 68 | assert out["source"] == "a.md" |
| 69 | assert out["target"] == "b.md" |
| 70 | assert out["raw"] == "b" |
| 71 | assert out["kind"] == "wikilink" |
| 72 | assert out["fragment"] == "section" |
| 73 | assert out["broken"] is False |
| 74 | assert out["alias"] == "display" |
| 75 | |
| 76 | def test_describe_broken_link(self) -> None: |
| 77 | link = ResolvedLink( |
| 78 | source="a.md", target=None, raw="missing", kind="wikilink", broken=True |
| 79 | ) |
| 80 | out = describe_link(link) |
| 81 | assert out["target"] is None |
| 82 | assert out["broken"] is True |
| 83 | |
| 84 | def test_describe_link_serialises_to_json(self) -> None: |
| 85 | link = ResolvedLink(source="a.md", target="b.md", raw="b", kind="md_link") |
| 86 | json.dumps(describe_link(link)) |
| 87 | |
| 88 | |
| 89 | # ============================================================================ |
| 90 | # TestNoteDeps |
| 91 | # ============================================================================ |
| 92 | |
| 93 | |
| 94 | class TestNoteDeps: |
| 95 | def _basic_index(self, tmp_path: pathlib.Path) -> LinkIndex: |
| 96 | _make_vault(tmp_path, { |
| 97 | "a.md": b"[[B]] and [C](c.md)", |
| 98 | "b.md": b"# B", |
| 99 | "c.md": b"# C", |
| 100 | }) |
| 101 | return build_link_index(tmp_path) |
| 102 | |
| 103 | def test_outgoing_default(self, tmp_path: pathlib.Path) -> None: |
| 104 | idx = self._basic_index(tmp_path) |
| 105 | result = note_deps(idx, "a.md") |
| 106 | assert result["direction"] == "outgoing" |
| 107 | assert result["count"] == 2 |
| 108 | |
| 109 | def test_incoming_reverse(self, tmp_path: pathlib.Path) -> None: |
| 110 | idx = self._basic_index(tmp_path) |
| 111 | result = note_deps(idx, "b.md", reverse=True) |
| 112 | assert result["direction"] == "incoming" |
| 113 | assert result["count"] == 1 |
| 114 | assert result["links"][0]["source"] == "a.md" |
| 115 | |
| 116 | def test_kinds_filter(self, tmp_path: pathlib.Path) -> None: |
| 117 | idx = self._basic_index(tmp_path) |
| 118 | result = note_deps(idx, "a.md", kinds=("wikilink",)) |
| 119 | assert result["count"] == 1 |
| 120 | assert result["links"][0]["kind"] == "wikilink" |
| 121 | |
| 122 | def test_broken_excluded_by_default(self, tmp_path: pathlib.Path) -> None: |
| 123 | _make_vault(tmp_path, {"a.md": b"[[Missing]]"}) |
| 124 | idx = build_link_index(tmp_path) |
| 125 | result = note_deps(idx, "a.md") |
| 126 | assert result["count"] == 0 |
| 127 | |
| 128 | def test_broken_included_when_opted_in(self, tmp_path: pathlib.Path) -> None: |
| 129 | _make_vault(tmp_path, {"a.md": b"[[Missing]]"}) |
| 130 | idx = build_link_index(tmp_path) |
| 131 | result = note_deps(idx, "a.md", include_broken=True) |
| 132 | assert result["count"] == 1 |
| 133 | |
| 134 | def test_missing_note_returns_empty(self, tmp_path: pathlib.Path) -> None: |
| 135 | _make_vault(tmp_path, {"a.md": b"# A"}) |
| 136 | idx = build_link_index(tmp_path) |
| 137 | result = note_deps(idx, "does-not-exist.md") |
| 138 | assert result["count"] == 0 |
| 139 | assert result["links"] == [] |
| 140 | |
| 141 | def test_result_is_json_serialisable(self, tmp_path: pathlib.Path) -> None: |
| 142 | idx = self._basic_index(tmp_path) |
| 143 | result = note_deps(idx, "a.md") |
| 144 | json.dumps(result) |
| 145 | |
| 146 | |
| 147 | # ============================================================================ |
| 148 | # TestNoteImpact |
| 149 | # ============================================================================ |
| 150 | |
| 151 | |
| 152 | class TestNoteImpact: |
| 153 | def _chain_vault(self, tmp_path: pathlib.Path) -> LinkIndex: |
| 154 | """Build a 5-note linear chain: a → b → c → d → e.""" |
| 155 | _make_vault(tmp_path, { |
| 156 | "a.md": b"[[B]]", |
| 157 | "b.md": b"[[C]]", |
| 158 | "c.md": b"[[D]]", |
| 159 | "d.md": b"[[E]]", |
| 160 | "e.md": b"# Terminal", |
| 161 | }) |
| 162 | return build_link_index(tmp_path) |
| 163 | |
| 164 | def test_reverse_bfs_unlimited(self, tmp_path: pathlib.Path) -> None: |
| 165 | idx = self._chain_vault(tmp_path) |
| 166 | result = note_impact(idx, "e.md") |
| 167 | assert result["direction"] == "reverse" |
| 168 | assert result["total"] == 4 # a, b, c, d |
| 169 | |
| 170 | def test_reverse_depth_capped(self, tmp_path: pathlib.Path) -> None: |
| 171 | idx = self._chain_vault(tmp_path) |
| 172 | result = note_impact(idx, "e.md", depth=2) |
| 173 | # Depth 1 = d, depth 2 = c → total 2 |
| 174 | assert result["total"] == 2 |
| 175 | assert "1" in result["by_depth"] |
| 176 | assert "2" in result["by_depth"] |
| 177 | assert "3" not in result["by_depth"] |
| 178 | |
| 179 | def test_forward_bfs(self, tmp_path: pathlib.Path) -> None: |
| 180 | idx = self._chain_vault(tmp_path) |
| 181 | result = note_impact(idx, "a.md", forward=True) |
| 182 | assert result["direction"] == "forward" |
| 183 | assert result["total"] == 4 # b, c, d, e |
| 184 | |
| 185 | def test_negative_depth_rejected(self, tmp_path: pathlib.Path) -> None: |
| 186 | idx = build_link_index(_make_vault(tmp_path, {"a.md": b"# A"})) |
| 187 | with pytest.raises(ValueError): |
| 188 | note_impact(idx, "a.md", depth=-1) |
| 189 | |
| 190 | def test_seed_not_in_result(self, tmp_path: pathlib.Path) -> None: |
| 191 | idx = self._chain_vault(tmp_path) |
| 192 | result = note_impact(idx, "c.md") |
| 193 | all_notes = [n for notes in result["by_depth"].values() for n in notes] |
| 194 | assert "c.md" not in all_notes |
| 195 | |
| 196 | def test_include_attachments_flag_echoed(self, tmp_path: pathlib.Path) -> None: |
| 197 | idx = self._chain_vault(tmp_path) |
| 198 | result = note_impact(idx, "e.md", include_attachments=False) |
| 199 | assert result["include_attachments"] is False |
| 200 | |
| 201 | def test_disconnected_note_has_empty_blast(self, tmp_path: pathlib.Path) -> None: |
| 202 | _make_vault(tmp_path, { |
| 203 | "alone.md": b"# Alone", |
| 204 | "other.md": b"# Other", |
| 205 | }) |
| 206 | idx = build_link_index(tmp_path) |
| 207 | result = note_impact(idx, "alone.md") |
| 208 | assert result["total"] == 0 |
| 209 | assert result["by_depth"] == {} |
| 210 | |
| 211 | def test_result_is_json_serialisable(self, tmp_path: pathlib.Path) -> None: |
| 212 | idx = self._chain_vault(tmp_path) |
| 213 | result = note_impact(idx, "e.md") |
| 214 | json.dumps(result) |
| 215 | |
| 216 | def test_no_infinite_loop_on_cycle(self, tmp_path: pathlib.Path) -> None: |
| 217 | """A → B → A cycle must terminate.""" |
| 218 | _make_vault(tmp_path, { |
| 219 | "a.md": b"[[B]]", |
| 220 | "b.md": b"[[A]]", |
| 221 | }) |
| 222 | idx = build_link_index(tmp_path) |
| 223 | # Both forward and reverse must terminate |
| 224 | result_f = note_impact(idx, "a.md", forward=True) |
| 225 | result_r = note_impact(idx, "a.md") |
| 226 | assert result_f["total"] == 1 # just b |
| 227 | assert result_r["total"] == 1 # just b |
| 228 | |
| 229 | |
| 230 | # ============================================================================ |
| 231 | # TestConvenienceFunctions |
| 232 | # ============================================================================ |
| 233 | |
| 234 | |
| 235 | class TestConvenienceFunctions: |
| 236 | def test_deps_for_note(self, tmp_path: pathlib.Path) -> None: |
| 237 | _make_vault(tmp_path, { |
| 238 | "a.md": b"[[B]]", |
| 239 | "b.md": b"# B", |
| 240 | }) |
| 241 | result = deps_for_note(tmp_path, "a.md") |
| 242 | assert result["count"] == 1 |
| 243 | assert result["links"][0]["target"] == "b.md" |
| 244 | |
| 245 | def test_impact_for_note(self, tmp_path: pathlib.Path) -> None: |
| 246 | _make_vault(tmp_path, { |
| 247 | "a.md": b"[[B]]", |
| 248 | "b.md": b"# B", |
| 249 | }) |
| 250 | result = impact_for_note(tmp_path, "b.md") |
| 251 | assert result["total"] == 1 |
| 252 | |
| 253 | |
| 254 | # ============================================================================ |
| 255 | # TestGolden50NoteFixture (integration) |
| 256 | # ============================================================================ |
| 257 | |
| 258 | |
| 259 | class TestGolden50NoteFixture: |
| 260 | """Build a 50-note synthetic vault and verify deterministic JSON output.""" |
| 261 | |
| 262 | def _make_50_note_vault(self, tmp_path: pathlib.Path) -> pathlib.Path: |
| 263 | files: dict[str, bytes | str] = {} |
| 264 | # Create 50 notes; note_i links to note_{i+1} and note_0 links to all |
| 265 | for i in range(50): |
| 266 | links = [] |
| 267 | if i < 49: |
| 268 | links.append(f"[[note_{i + 1:02d}]]") |
| 269 | if i == 0: |
| 270 | # Hub note — links to 5 others |
| 271 | for j in (10, 20, 30, 40, 49): |
| 272 | links.append(f"[[note_{j:02d}]]") |
| 273 | files[f"note_{i:02d}.md"] = ( |
| 274 | f"# Note {i}\n\nContent.\n\n" + " ".join(links) |
| 275 | ).encode() |
| 276 | return _make_vault(tmp_path, files) |
| 277 | |
| 278 | def test_hub_note_has_correct_outgoing_count(self, tmp_path: pathlib.Path) -> None: |
| 279 | self._make_50_note_vault(tmp_path) |
| 280 | result = deps_for_note(tmp_path, "note_00.md") |
| 281 | # 5 hub links + 1 sequential link = 6 |
| 282 | assert result["count"] == 6 |
| 283 | |
| 284 | def test_last_note_has_one_incoming(self, tmp_path: pathlib.Path) -> None: |
| 285 | self._make_50_note_vault(tmp_path) |
| 286 | result = deps_for_note(tmp_path, "note_49.md", reverse=True) |
| 287 | # Linked by note_48 (sequential) and note_00 (hub) → 2 |
| 288 | assert result["count"] == 2 |
| 289 | |
| 290 | def test_blast_radius_from_terminal(self, tmp_path: pathlib.Path) -> None: |
| 291 | self._make_50_note_vault(tmp_path) |
| 292 | result = impact_for_note(tmp_path, "note_49.md") |
| 293 | # Backwards chain: 48 → 47 → … → 1 → 0 (49 notes total) |
| 294 | # But 0 is a hub note too, so it's reached at depth 2 (via 48) |
| 295 | # or directly at depth 1 if it links to 49 (yes, hub link). |
| 296 | # All notes except note_49 itself appear → 49 total |
| 297 | assert result["total"] == 49 |
| 298 | |
| 299 | def test_blast_radius_depth_capped(self, tmp_path: pathlib.Path) -> None: |
| 300 | self._make_50_note_vault(tmp_path) |
| 301 | result = impact_for_note(tmp_path, "note_49.md", depth=1) |
| 302 | # Direct callers: note_48 (sequential) and note_00 (hub) → 2 |
| 303 | assert result["total"] == 2 |
| 304 | |
| 305 | def test_json_output_is_stable(self, tmp_path: pathlib.Path) -> None: |
| 306 | self._make_50_note_vault(tmp_path) |
| 307 | r1 = json.dumps(deps_for_note(tmp_path, "note_25.md"), sort_keys=True) |
| 308 | r2 = json.dumps(deps_for_note(tmp_path, "note_25.md"), sort_keys=True) |
| 309 | assert r1 == r2 |
| 310 | |
| 311 | |
| 312 | # ============================================================================ |
| 313 | # TestDataIntegrity |
| 314 | # ============================================================================ |
| 315 | |
| 316 | |
| 317 | class TestDataIntegrity: |
| 318 | def test_outgoing_at_depth_1_equals_deps_targets(self, tmp_path: pathlib.Path) -> None: |
| 319 | """note_impact depth-1 forward == set of note_deps outgoing targets.""" |
| 320 | _make_vault(tmp_path, { |
| 321 | "a.md": b"[[B]] [[C]] [[D]]", |
| 322 | "b.md": b"# B", |
| 323 | "c.md": b"# C", |
| 324 | "d.md": b"# D", |
| 325 | }) |
| 326 | idx = build_link_index(tmp_path) |
| 327 | deps_result = note_deps(idx, "a.md") |
| 328 | deps_targets = {link["target"] for link in deps_result["links"]} |
| 329 | |
| 330 | impact_result = note_impact(idx, "a.md", forward=True, depth=1) |
| 331 | depth_1 = set(impact_result["by_depth"].get("1", [])) |
| 332 | |
| 333 | assert depth_1 == deps_targets |
| 334 | |
| 335 | def test_incoming_at_depth_1_equals_reverse_deps_sources(self, tmp_path: pathlib.Path) -> None: |
| 336 | _make_vault(tmp_path, { |
| 337 | "a.md": b"[[Target]]", |
| 338 | "b.md": b"[[Target]]", |
| 339 | "Target.md": b"# Target", |
| 340 | }) |
| 341 | idx = build_link_index(tmp_path) |
| 342 | reverse_deps = note_deps(idx, "Target.md", reverse=True) |
| 343 | sources = {link["source"] for link in reverse_deps["links"]} |
| 344 | |
| 345 | impact = note_impact(idx, "Target.md", depth=1) |
| 346 | depth_1 = set(impact["by_depth"].get("1", [])) |
| 347 | |
| 348 | assert depth_1 == sources |
| 349 | |
| 350 | |
| 351 | # ============================================================================ |
| 352 | # TestPerformance |
| 353 | # ============================================================================ |
| 354 | |
| 355 | |
| 356 | class TestPerformance: |
| 357 | def test_impact_on_200_note_chain_under_2s(self, tmp_path: pathlib.Path) -> None: |
| 358 | files: dict[str, bytes] = {} |
| 359 | for i in range(200): |
| 360 | if i < 199: |
| 361 | files[f"n_{i:03d}.md"] = f"[[n_{i + 1:03d}]]".encode() |
| 362 | else: |
| 363 | files[f"n_{i:03d}.md"] = b"# Terminal" |
| 364 | _make_vault(tmp_path, files) |
| 365 | idx = build_link_index(tmp_path) |
| 366 | |
| 367 | start = time.monotonic() |
| 368 | result = note_impact(idx, "n_199.md") |
| 369 | elapsed = time.monotonic() - start |
| 370 | |
| 371 | assert result["total"] == 199 |
| 372 | assert elapsed < 2.0, f"Impact on 200-note chain took {elapsed:.2f}s" |
| 373 | |
| 374 | |
| 375 | # ============================================================================ |
| 376 | # TestSecurity |
| 377 | # ============================================================================ |
| 378 | |
| 379 | |
| 380 | class TestSecurity: |
| 381 | def test_deps_for_note_path_traversal_handled(self, tmp_path: pathlib.Path) -> None: |
| 382 | _make_vault(tmp_path, {"a.md": b"# A"}) |
| 383 | # Non-existent path returns empty result — does not crash |
| 384 | result = deps_for_note(tmp_path, "../../etc/passwd.md") |
| 385 | assert result["count"] == 0 |
| 386 | |
| 387 | def test_impact_for_nonexistent_note_returns_empty(self, tmp_path: pathlib.Path) -> None: |
| 388 | _make_vault(tmp_path, {"a.md": b"# A"}) |
| 389 | result = impact_for_note(tmp_path, "nope.md") |
| 390 | assert result["total"] == 0 |
| 391 | |
| 392 | def test_deps_handles_circular_references(self, tmp_path: pathlib.Path) -> None: |
| 393 | _make_vault(tmp_path, { |
| 394 | "a.md": b"[[A]]", # self-link |
| 395 | }) |
| 396 | idx = build_link_index(tmp_path) |
| 397 | result = note_deps(idx, "a.md") |
| 398 | # Self-link counts as one outgoing edge to itself |
| 399 | assert result["count"] == 1 |
File History
2 commits
sha256:660fcac1df3ab28f61862e961890bd2ca8b754fa0242079d93ca1e25037ec8a6
chore(tests): add docstring to tests/__init__.py so rc14 tr…
Human
27 days ago
sha256:d8316ffae901be06347e16ab55be11868eb519dd16ade3e8aa16a99e662f7e62
baseline: rc14 re-baseline after rc3 store corruption recovery
Human
patch
27 days ago