test_knowtation_stats.py
python
sha256:660fcac1df3ab28f61862e961890bd2ca8b754fa0242079d93ca1e25037ec8a6
chore(tests): add docstring to tests/__init__.py so rc14 tr…
Human
27 days ago
| 1 | """Tests for muse/plugins/knowtation/stats.py and the domain-info extension. |
| 2 | |
| 3 | Test tiers covered |
| 4 | ------------------ |
| 5 | 1. Unit — VaultStats TypedDict fields, collect_vault_stats against fixture vaults |
| 6 | of varying shapes (empty, notes only, mixed, notes with full frontmatter). |
| 7 | 2. Integration — CLI snapshot: ``muse plumbing domain-info --domain knowtation`` |
| 8 | and ``muse plumbing domain-info`` from a fixture knowtation repo. |
| 9 | 3. Data-integrity — collect_vault_stats against the real vault; verify that |
| 10 | note_count matches the expected content note count and that all stats |
| 11 | fields are non-negative / correctly typed. |
| 12 | """ |
| 13 | |
| 14 | from __future__ import annotations |
| 15 | |
| 16 | import json |
| 17 | import pathlib |
| 18 | |
| 19 | import pytest |
| 20 | |
| 21 | from muse.plugins.knowtation.stats import VaultStats, collect_vault_stats |
| 22 | from tests.cli_test_helper import CliRunner, InvokeResult |
| 23 | |
| 24 | runner = CliRunner() |
| 25 | |
| 26 | _REAL_VAULT = pathlib.Path("/Users/aaronrenecarvajal/knowtation/vault") |
| 27 | _SKIP_DIRS: frozenset[str] = frozenset({"templates", "meta"}) |
| 28 | |
| 29 | |
| 30 | # --------------------------------------------------------------------------- |
| 31 | # Fixtures |
| 32 | # --------------------------------------------------------------------------- |
| 33 | |
| 34 | |
| 35 | def _make_note(frontmatter: str = "", body: str = "Body text.") -> bytes: |
| 36 | if frontmatter: |
| 37 | return f"---\n{frontmatter}\n---\n{body}".encode() |
| 38 | return body.encode() |
| 39 | |
| 40 | |
| 41 | def _make_knowtation_repo(tmp_path: pathlib.Path, domain: str = "knowtation") -> pathlib.Path: |
| 42 | """Create a minimal Muse repo with domain=knowtation.""" |
| 43 | repo = tmp_path / "vault" |
| 44 | muse = repo / ".muse" |
| 45 | for sub in ("objects", "commits", "snapshots", "refs/heads"): |
| 46 | (muse / sub).mkdir(parents=True) |
| 47 | (muse / "HEAD").write_text("ref: refs/heads/main") |
| 48 | (muse / "repo.json").write_text( |
| 49 | json.dumps({"repo_id": "test-vault", "domain": domain}) |
| 50 | ) |
| 51 | return repo |
| 52 | |
| 53 | |
| 54 | def _di(repo: pathlib.Path | None, *args: str) -> InvokeResult: |
| 55 | from muse.cli.app import main as cli |
| 56 | env = {"MUSE_REPO_ROOT": str(repo)} if repo is not None else {} |
| 57 | return runner.invoke(cli, ["domain-info", *args], env=env) |
| 58 | |
| 59 | |
| 60 | # ============================================================================ |
| 61 | # TestVaultStatsUnit |
| 62 | # ============================================================================ |
| 63 | |
| 64 | |
| 65 | class TestVaultStatsUnit: |
| 66 | def test_empty_vault_returns_zero_counts(self, tmp_path: pathlib.Path) -> None: |
| 67 | stats = collect_vault_stats(tmp_path) |
| 68 | assert stats["note_count"] == 0 |
| 69 | assert stats["source_types"] == [] |
| 70 | assert stats["projects"] == [] |
| 71 | assert stats["entity_count"] == 0 |
| 72 | assert stats["tag_count"] == 0 |
| 73 | assert stats["index_freshness"] is None |
| 74 | |
| 75 | def test_counts_markdown_files(self, tmp_path: pathlib.Path) -> None: |
| 76 | for i in range(3): |
| 77 | (tmp_path / f"note{i}.md").write_bytes(_make_note()) |
| 78 | stats = collect_vault_stats(tmp_path) |
| 79 | assert stats["note_count"] == 3 |
| 80 | |
| 81 | def test_ignores_non_markdown_files(self, tmp_path: pathlib.Path) -> None: |
| 82 | (tmp_path / "note.md").write_bytes(_make_note()) |
| 83 | (tmp_path / "image.png").write_bytes(b"\x89PNG") |
| 84 | (tmp_path / "data.json").write_bytes(b"{}") |
| 85 | stats = collect_vault_stats(tmp_path) |
| 86 | assert stats["note_count"] == 1 |
| 87 | |
| 88 | def test_counts_nested_markdown_files(self, tmp_path: pathlib.Path) -> None: |
| 89 | (tmp_path / "inbox").mkdir() |
| 90 | (tmp_path / "inbox" / "a.md").write_bytes(_make_note()) |
| 91 | (tmp_path / "projects").mkdir() |
| 92 | (tmp_path / "projects" / "b.md").write_bytes(_make_note()) |
| 93 | stats = collect_vault_stats(tmp_path) |
| 94 | assert stats["note_count"] == 2 |
| 95 | |
| 96 | def test_skips_hidden_directories(self, tmp_path: pathlib.Path) -> None: |
| 97 | hidden = tmp_path / ".obsidian" |
| 98 | hidden.mkdir() |
| 99 | (hidden / "config.md").write_bytes(b"---\ntitle: hidden\n---\n") |
| 100 | stats = collect_vault_stats(tmp_path) |
| 101 | assert stats["note_count"] == 0 |
| 102 | |
| 103 | def test_skips_templates_directory(self, tmp_path: pathlib.Path) -> None: |
| 104 | templates = tmp_path / "templates" |
| 105 | templates.mkdir() |
| 106 | (templates / "README.md").write_bytes(b"# Template\nNo frontmatter.") |
| 107 | stats = collect_vault_stats(tmp_path) |
| 108 | assert stats["note_count"] == 0 |
| 109 | |
| 110 | def test_skips_meta_directory(self, tmp_path: pathlib.Path) -> None: |
| 111 | meta = tmp_path / "meta" |
| 112 | meta.mkdir() |
| 113 | (meta / "index.md").write_bytes(b"# Meta\n") |
| 114 | stats = collect_vault_stats(tmp_path) |
| 115 | assert stats["note_count"] == 0 |
| 116 | |
| 117 | def test_source_types_collected(self, tmp_path: pathlib.Path) -> None: |
| 118 | (tmp_path / "a.md").write_bytes( |
| 119 | _make_note("source: telegram\ndate: 2025-01-01") |
| 120 | ) |
| 121 | (tmp_path / "b.md").write_bytes( |
| 122 | _make_note("source: slack\ndate: 2025-01-02") |
| 123 | ) |
| 124 | stats = collect_vault_stats(tmp_path) |
| 125 | assert set(stats["source_types"]) == {"telegram", "slack"} |
| 126 | |
| 127 | def test_source_type_alias_collected(self, tmp_path: pathlib.Path) -> None: |
| 128 | (tmp_path / "a.md").write_bytes( |
| 129 | _make_note("source_type: chatgpt-export\ndate: 2025-01-01") |
| 130 | ) |
| 131 | stats = collect_vault_stats(tmp_path) |
| 132 | assert "chatgpt-export" in stats["source_types"] |
| 133 | |
| 134 | def test_source_types_deduplicated(self, tmp_path: pathlib.Path) -> None: |
| 135 | for i in range(5): |
| 136 | (tmp_path / f"note{i}.md").write_bytes( |
| 137 | _make_note("source: telegram\ndate: 2025-01-01") |
| 138 | ) |
| 139 | stats = collect_vault_stats(tmp_path) |
| 140 | assert stats["source_types"].count("telegram") == 1 |
| 141 | |
| 142 | def test_source_types_sorted(self, tmp_path: pathlib.Path) -> None: |
| 143 | (tmp_path / "z.md").write_bytes(_make_note("source: zzz\ndate: 2025-01-01")) |
| 144 | (tmp_path / "a.md").write_bytes(_make_note("source: aaa\ndate: 2025-01-01")) |
| 145 | stats = collect_vault_stats(tmp_path) |
| 146 | assert stats["source_types"] == sorted(stats["source_types"]) |
| 147 | |
| 148 | def test_projects_collected(self, tmp_path: pathlib.Path) -> None: |
| 149 | (tmp_path / "a.md").write_bytes(_make_note("project: born-free\ndate: 2025-01-01")) |
| 150 | (tmp_path / "b.md").write_bytes(_make_note("project: dreambolt\ndate: 2025-01-01")) |
| 151 | stats = collect_vault_stats(tmp_path) |
| 152 | assert set(stats["projects"]) == {"born-free", "dreambolt"} |
| 153 | |
| 154 | def test_projects_deduplicated(self, tmp_path: pathlib.Path) -> None: |
| 155 | for i in range(3): |
| 156 | (tmp_path / f"note{i}.md").write_bytes( |
| 157 | _make_note("project: born-free\ndate: 2025-01-01") |
| 158 | ) |
| 159 | stats = collect_vault_stats(tmp_path) |
| 160 | assert stats["projects"].count("born-free") == 1 |
| 161 | |
| 162 | def test_projects_sorted(self, tmp_path: pathlib.Path) -> None: |
| 163 | (tmp_path / "z.md").write_bytes(_make_note("project: zzz\ndate: 2025-01-01")) |
| 164 | (tmp_path / "a.md").write_bytes(_make_note("project: aaa\ndate: 2025-01-01")) |
| 165 | stats = collect_vault_stats(tmp_path) |
| 166 | assert stats["projects"] == sorted(stats["projects"]) |
| 167 | |
| 168 | def test_entity_count(self, tmp_path: pathlib.Path) -> None: |
| 169 | (tmp_path / "a.md").write_bytes( |
| 170 | _make_note("entity:\n - alice\n - bob\ndate: 2025-01-01") |
| 171 | ) |
| 172 | (tmp_path / "b.md").write_bytes( |
| 173 | _make_note("entity:\n - bob\n - carol\ndate: 2025-01-01") |
| 174 | ) |
| 175 | stats = collect_vault_stats(tmp_path) |
| 176 | # Unique: alice, bob, carol |
| 177 | assert stats["entity_count"] == 3 |
| 178 | |
| 179 | def test_tag_count(self, tmp_path: pathlib.Path) -> None: |
| 180 | (tmp_path / "a.md").write_bytes( |
| 181 | _make_note("tags:\n - ai\n - memory\ndate: 2025-01-01") |
| 182 | ) |
| 183 | (tmp_path / "b.md").write_bytes( |
| 184 | _make_note("tags:\n - ai\n - research\ndate: 2025-01-01") |
| 185 | ) |
| 186 | stats = collect_vault_stats(tmp_path) |
| 187 | # Unique: ai, memory, research |
| 188 | assert stats["tag_count"] == 3 |
| 189 | |
| 190 | def test_index_freshness_none_when_no_data_dir(self, tmp_path: pathlib.Path) -> None: |
| 191 | stats = collect_vault_stats(tmp_path) |
| 192 | assert stats["index_freshness"] is None |
| 193 | |
| 194 | def test_index_freshness_set_when_data_dir_exists(self, tmp_path: pathlib.Path) -> None: |
| 195 | (tmp_path / "data").mkdir() |
| 196 | stats = collect_vault_stats(tmp_path) |
| 197 | assert stats["index_freshness"] is not None |
| 198 | # Must be ISO 8601 |
| 199 | assert "T" in stats["index_freshness"] |
| 200 | |
| 201 | def test_index_freshness_is_utc(self, tmp_path: pathlib.Path) -> None: |
| 202 | (tmp_path / "data").mkdir() |
| 203 | stats = collect_vault_stats(tmp_path) |
| 204 | freshness = stats["index_freshness"] |
| 205 | assert freshness is not None |
| 206 | # ISO 8601 UTC timestamp ends with +00:00 |
| 207 | assert "+00:00" in freshness |
| 208 | |
| 209 | def test_unreadable_note_skipped_gracefully( |
| 210 | self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch |
| 211 | ) -> None: |
| 212 | note = tmp_path / "bad.md" |
| 213 | note.write_bytes(_make_note("source: telegram\ndate: 2025-01-01")) |
| 214 | |
| 215 | original_read = pathlib.Path.read_bytes |
| 216 | |
| 217 | def _fail_once(self: pathlib.Path) -> bytes: |
| 218 | if self.name == "bad.md": |
| 219 | raise OSError("Permission denied") |
| 220 | return original_read(self) |
| 221 | |
| 222 | monkeypatch.setattr(pathlib.Path, "read_bytes", _fail_once) |
| 223 | stats = collect_vault_stats(tmp_path) |
| 224 | # Counts the file (os.walk sees it) but skips frontmatter parse |
| 225 | assert stats["note_count"] == 1 |
| 226 | assert stats["source_types"] == [] |
| 227 | |
| 228 | def test_vault_stats_typeddict_keys(self, tmp_path: pathlib.Path) -> None: |
| 229 | stats = collect_vault_stats(tmp_path) |
| 230 | assert set(stats.keys()) == { |
| 231 | "note_count", "source_types", "projects", |
| 232 | "entity_count", "tag_count", "index_freshness", |
| 233 | } |
| 234 | |
| 235 | def test_note_without_frontmatter_counted_but_not_parsed( |
| 236 | self, tmp_path: pathlib.Path |
| 237 | ) -> None: |
| 238 | (tmp_path / "plain.md").write_bytes(b"# Just a heading\nNo frontmatter.") |
| 239 | stats = collect_vault_stats(tmp_path) |
| 240 | assert stats["note_count"] == 1 |
| 241 | assert stats["source_types"] == [] |
| 242 | assert stats["projects"] == [] |
| 243 | |
| 244 | def test_markdown_extension_variants(self, tmp_path: pathlib.Path) -> None: |
| 245 | (tmp_path / "a.md").write_bytes(_make_note()) |
| 246 | (tmp_path / "b.markdown").write_bytes(_make_note()) |
| 247 | (tmp_path / "c.mdx").write_bytes(_make_note()) |
| 248 | stats = collect_vault_stats(tmp_path) |
| 249 | assert stats["note_count"] == 3 |
| 250 | |
| 251 | |
| 252 | # ============================================================================ |
| 253 | # TestDomainInfoIntegration (CLI snapshot tests) |
| 254 | # ============================================================================ |
| 255 | |
| 256 | |
| 257 | class TestDomainInfoIntegration: |
| 258 | def test_domain_flag_knowtation_no_vault_stats(self, tmp_path: pathlib.Path) -> None: |
| 259 | """--domain knowtation without a repo gives no vault_stats (no root).""" |
| 260 | result = _di(None, "--domain", "knowtation") |
| 261 | assert result.exit_code == 0 |
| 262 | data = json.loads(result.stdout) |
| 263 | assert data["domain"] == "knowtation" |
| 264 | # No vault_stats when no repo root is available |
| 265 | assert "vault_stats" not in data |
| 266 | |
| 267 | def test_domain_flag_knowtation_schema_present(self, tmp_path: pathlib.Path) -> None: |
| 268 | result = _di(None, "--domain", "knowtation") |
| 269 | assert result.exit_code == 0 |
| 270 | data = json.loads(result.stdout) |
| 271 | assert "schema" in data |
| 272 | assert data["schema"]["domain"] == "knowtation" |
| 273 | |
| 274 | def test_repo_knowtation_vault_stats_present(self, tmp_path: pathlib.Path) -> None: |
| 275 | """When invoked inside a knowtation repo, vault_stats is included.""" |
| 276 | repo = _make_knowtation_repo(tmp_path) |
| 277 | (repo / "note.md").write_bytes( |
| 278 | b"---\nsource: telegram\ndate: 2025-01-01\n---\nBody." |
| 279 | ) |
| 280 | result = _di(repo) |
| 281 | assert result.exit_code == 0 |
| 282 | data = json.loads(result.stdout) |
| 283 | assert data["domain"] == "knowtation" |
| 284 | assert "vault_stats" in data |
| 285 | vs = data["vault_stats"] |
| 286 | assert vs["note_count"] == 1 |
| 287 | assert "telegram" in vs["source_types"] |
| 288 | |
| 289 | def test_repo_knowtation_vault_stats_shape(self, tmp_path: pathlib.Path) -> None: |
| 290 | repo = _make_knowtation_repo(tmp_path) |
| 291 | result = _di(repo) |
| 292 | assert result.exit_code == 0 |
| 293 | data = json.loads(result.stdout) |
| 294 | vs = data["vault_stats"] |
| 295 | assert "note_count" in vs |
| 296 | assert "source_types" in vs |
| 297 | assert "projects" in vs |
| 298 | assert "entity_count" in vs |
| 299 | assert "tag_count" in vs |
| 300 | assert "index_freshness" in vs |
| 301 | |
| 302 | def test_repo_non_knowtation_no_vault_stats(self, tmp_path: pathlib.Path) -> None: |
| 303 | """code domain repos must not get vault_stats.""" |
| 304 | repo = tmp_path / "code-repo" |
| 305 | muse = repo / ".muse" |
| 306 | for sub in ("objects", "commits", "snapshots", "refs/heads"): |
| 307 | (muse / sub).mkdir(parents=True) |
| 308 | (muse / "HEAD").write_text("ref: refs/heads/main") |
| 309 | (muse / "repo.json").write_text( |
| 310 | json.dumps({"repo_id": "test", "domain": "code"}) |
| 311 | ) |
| 312 | result = _di(repo) |
| 313 | assert result.exit_code == 0 |
| 314 | data = json.loads(result.stdout) |
| 315 | assert "vault_stats" not in data |
| 316 | |
| 317 | def test_text_format_knowtation_shows_notes(self, tmp_path: pathlib.Path) -> None: |
| 318 | repo = _make_knowtation_repo(tmp_path) |
| 319 | (repo / "a.md").write_bytes(b"---\nproject: p1\ndate: 2025-01-01\n---\n") |
| 320 | (repo / "b.md").write_bytes(b"---\nproject: p2\ndate: 2025-01-01\n---\n") |
| 321 | result = _di(repo, "--format", "text") |
| 322 | assert result.exit_code == 0 |
| 323 | assert "Notes:" in result.stdout |
| 324 | assert "2" in result.stdout |
| 325 | |
| 326 | def test_text_format_knowtation_shows_projects(self, tmp_path: pathlib.Path) -> None: |
| 327 | repo = _make_knowtation_repo(tmp_path) |
| 328 | (repo / "a.md").write_bytes(b"---\nproject: born-free\ndate: 2025-01-01\n---\n") |
| 329 | result = _di(repo, "--format", "text") |
| 330 | assert result.exit_code == 0 |
| 331 | assert "Projects:" in result.stdout |
| 332 | assert "born-free" in result.stdout |
| 333 | |
| 334 | def test_registered_domains_includes_knowtation(self, tmp_path: pathlib.Path) -> None: |
| 335 | result = _di(None, "--domain", "knowtation") |
| 336 | assert result.exit_code == 0 |
| 337 | data = json.loads(result.stdout) |
| 338 | assert "knowtation" in data["registered_domains"] |
| 339 | |
| 340 | def test_json_output_serialisable(self, tmp_path: pathlib.Path) -> None: |
| 341 | repo = _make_knowtation_repo(tmp_path) |
| 342 | (repo / "note.md").write_bytes( |
| 343 | b"---\nsource: telegram\ndate: 2025-01-01\ntags:\n - ai\n---\n" |
| 344 | ) |
| 345 | result = _di(repo) |
| 346 | assert result.exit_code == 0 |
| 347 | # Must be valid JSON — json.loads raises if not |
| 348 | data = json.loads(result.stdout) |
| 349 | assert isinstance(data, dict) |
| 350 | |
| 351 | def test_vault_stats_empty_vault_zero_counts(self, tmp_path: pathlib.Path) -> None: |
| 352 | repo = _make_knowtation_repo(tmp_path) |
| 353 | result = _di(repo) |
| 354 | assert result.exit_code == 0 |
| 355 | data = json.loads(result.stdout) |
| 356 | vs = data["vault_stats"] |
| 357 | assert vs["note_count"] == 0 |
| 358 | assert vs["source_types"] == [] |
| 359 | assert vs["projects"] == [] |
| 360 | |
| 361 | |
| 362 | # ============================================================================ |
| 363 | # TestRealVaultDataIntegrity |
| 364 | # ============================================================================ |
| 365 | |
| 366 | |
| 367 | @pytest.mark.skipif( |
| 368 | not _REAL_VAULT.exists(), |
| 369 | reason=f"Real vault not found at {_REAL_VAULT}", |
| 370 | ) |
| 371 | class TestRealVaultDataIntegrity: |
| 372 | def _expected_content_note_count(self) -> int: |
| 373 | count = 0 |
| 374 | for md_path in _REAL_VAULT.rglob("*.md"): |
| 375 | parts = md_path.relative_to(_REAL_VAULT).parts |
| 376 | if parts and parts[0] in _SKIP_DIRS: |
| 377 | continue |
| 378 | count += 1 |
| 379 | return count |
| 380 | |
| 381 | def test_note_count_matches_content_notes(self) -> None: |
| 382 | stats = collect_vault_stats(_REAL_VAULT) |
| 383 | expected = self._expected_content_note_count() |
| 384 | assert stats["note_count"] == expected, ( |
| 385 | f"note_count={stats['note_count']} but expected {expected}" |
| 386 | ) |
| 387 | |
| 388 | def test_source_types_are_strings(self) -> None: |
| 389 | stats = collect_vault_stats(_REAL_VAULT) |
| 390 | assert all(isinstance(s, str) for s in stats["source_types"]) |
| 391 | |
| 392 | def test_projects_are_strings(self) -> None: |
| 393 | stats = collect_vault_stats(_REAL_VAULT) |
| 394 | assert all(isinstance(p, str) for p in stats["projects"]) |
| 395 | |
| 396 | def test_entity_count_non_negative(self) -> None: |
| 397 | stats = collect_vault_stats(_REAL_VAULT) |
| 398 | assert stats["entity_count"] >= 0 |
| 399 | |
| 400 | def test_tag_count_non_negative(self) -> None: |
| 401 | stats = collect_vault_stats(_REAL_VAULT) |
| 402 | assert stats["tag_count"] >= 0 |
| 403 | |
| 404 | def test_index_freshness_valid_if_present(self) -> None: |
| 405 | stats = collect_vault_stats(_REAL_VAULT) |
| 406 | if stats["index_freshness"] is not None: |
| 407 | assert "T" in stats["index_freshness"] |
| 408 | |
| 409 | def test_stats_does_not_raise(self) -> None: |
| 410 | try: |
| 411 | collect_vault_stats(_REAL_VAULT) |
| 412 | except Exception as exc: |
| 413 | pytest.fail(f"collect_vault_stats raised: {exc}") |
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