"""Tests for muse/plugins/knowtation/stats.py and the domain-info extension. Test tiers covered ------------------ 1. Unit — VaultStats TypedDict fields, collect_vault_stats against fixture vaults of varying shapes (empty, notes only, mixed, notes with full frontmatter). 2. Integration — CLI snapshot: ``muse plumbing domain-info --domain knowtation`` and ``muse plumbing domain-info`` from a fixture knowtation repo. 3. Data-integrity — collect_vault_stats against the real vault; verify that note_count matches the expected content note count and that all stats fields are non-negative / correctly typed. """ from __future__ import annotations import json import pathlib import pytest from muse.plugins.knowtation.stats import VaultStats, collect_vault_stats from tests.cli_test_helper import CliRunner, InvokeResult runner = CliRunner() _REAL_VAULT = pathlib.Path("/Users/aaronrenecarvajal/knowtation/vault") _SKIP_DIRS: frozenset[str] = frozenset({"templates", "meta"}) # --------------------------------------------------------------------------- # Fixtures # --------------------------------------------------------------------------- def _make_note(frontmatter: str = "", body: str = "Body text.") -> bytes: if frontmatter: return f"---\n{frontmatter}\n---\n{body}".encode() return body.encode() def _make_knowtation_repo(tmp_path: pathlib.Path, domain: str = "knowtation") -> pathlib.Path: """Create a minimal Muse repo with domain=knowtation.""" repo = tmp_path / "vault" muse = repo / ".muse" for sub in ("objects", "commits", "snapshots", "refs/heads"): (muse / sub).mkdir(parents=True) (muse / "HEAD").write_text("ref: refs/heads/main") (muse / "repo.json").write_text( json.dumps({"repo_id": "test-vault", "domain": domain}) ) return repo def _di(repo: pathlib.Path | None, *args: str) -> InvokeResult: from muse.cli.app import main as cli env = {"MUSE_REPO_ROOT": str(repo)} if repo is not None else {} return runner.invoke(cli, ["domain-info", *args], env=env) # ============================================================================ # TestVaultStatsUnit # ============================================================================ class TestVaultStatsUnit: def test_empty_vault_returns_zero_counts(self, tmp_path: pathlib.Path) -> None: stats = collect_vault_stats(tmp_path) assert stats["note_count"] == 0 assert stats["source_types"] == [] assert stats["projects"] == [] assert stats["entity_count"] == 0 assert stats["tag_count"] == 0 assert stats["index_freshness"] is None def test_counts_markdown_files(self, tmp_path: pathlib.Path) -> None: for i in range(3): (tmp_path / f"note{i}.md").write_bytes(_make_note()) stats = collect_vault_stats(tmp_path) assert stats["note_count"] == 3 def test_ignores_non_markdown_files(self, tmp_path: pathlib.Path) -> None: (tmp_path / "note.md").write_bytes(_make_note()) (tmp_path / "image.png").write_bytes(b"\x89PNG") (tmp_path / "data.json").write_bytes(b"{}") stats = collect_vault_stats(tmp_path) assert stats["note_count"] == 1 def test_counts_nested_markdown_files(self, tmp_path: pathlib.Path) -> None: (tmp_path / "inbox").mkdir() (tmp_path / "inbox" / "a.md").write_bytes(_make_note()) (tmp_path / "projects").mkdir() (tmp_path / "projects" / "b.md").write_bytes(_make_note()) stats = collect_vault_stats(tmp_path) assert stats["note_count"] == 2 def test_skips_hidden_directories(self, tmp_path: pathlib.Path) -> None: hidden = tmp_path / ".obsidian" hidden.mkdir() (hidden / "config.md").write_bytes(b"---\ntitle: hidden\n---\n") stats = collect_vault_stats(tmp_path) assert stats["note_count"] == 0 def test_skips_templates_directory(self, tmp_path: pathlib.Path) -> None: templates = tmp_path / "templates" templates.mkdir() (templates / "README.md").write_bytes(b"# Template\nNo frontmatter.") stats = collect_vault_stats(tmp_path) assert stats["note_count"] == 0 def test_skips_meta_directory(self, tmp_path: pathlib.Path) -> None: meta = tmp_path / "meta" meta.mkdir() (meta / "index.md").write_bytes(b"# Meta\n") stats = collect_vault_stats(tmp_path) assert stats["note_count"] == 0 def test_source_types_collected(self, tmp_path: pathlib.Path) -> None: (tmp_path / "a.md").write_bytes( _make_note("source: telegram\ndate: 2025-01-01") ) (tmp_path / "b.md").write_bytes( _make_note("source: slack\ndate: 2025-01-02") ) stats = collect_vault_stats(tmp_path) assert set(stats["source_types"]) == {"telegram", "slack"} def test_source_type_alias_collected(self, tmp_path: pathlib.Path) -> None: (tmp_path / "a.md").write_bytes( _make_note("source_type: chatgpt-export\ndate: 2025-01-01") ) stats = collect_vault_stats(tmp_path) assert "chatgpt-export" in stats["source_types"] def test_source_types_deduplicated(self, tmp_path: pathlib.Path) -> None: for i in range(5): (tmp_path / f"note{i}.md").write_bytes( _make_note("source: telegram\ndate: 2025-01-01") ) stats = collect_vault_stats(tmp_path) assert stats["source_types"].count("telegram") == 1 def test_source_types_sorted(self, tmp_path: pathlib.Path) -> None: (tmp_path / "z.md").write_bytes(_make_note("source: zzz\ndate: 2025-01-01")) (tmp_path / "a.md").write_bytes(_make_note("source: aaa\ndate: 2025-01-01")) stats = collect_vault_stats(tmp_path) assert stats["source_types"] == sorted(stats["source_types"]) def test_projects_collected(self, tmp_path: pathlib.Path) -> None: (tmp_path / "a.md").write_bytes(_make_note("project: born-free\ndate: 2025-01-01")) (tmp_path / "b.md").write_bytes(_make_note("project: dreambolt\ndate: 2025-01-01")) stats = collect_vault_stats(tmp_path) assert set(stats["projects"]) == {"born-free", "dreambolt"} def test_projects_deduplicated(self, tmp_path: pathlib.Path) -> None: for i in range(3): (tmp_path / f"note{i}.md").write_bytes( _make_note("project: born-free\ndate: 2025-01-01") ) stats = collect_vault_stats(tmp_path) assert stats["projects"].count("born-free") == 1 def test_projects_sorted(self, tmp_path: pathlib.Path) -> None: (tmp_path / "z.md").write_bytes(_make_note("project: zzz\ndate: 2025-01-01")) (tmp_path / "a.md").write_bytes(_make_note("project: aaa\ndate: 2025-01-01")) stats = collect_vault_stats(tmp_path) assert stats["projects"] == sorted(stats["projects"]) def test_entity_count(self, tmp_path: pathlib.Path) -> None: (tmp_path / "a.md").write_bytes( _make_note("entity:\n - alice\n - bob\ndate: 2025-01-01") ) (tmp_path / "b.md").write_bytes( _make_note("entity:\n - bob\n - carol\ndate: 2025-01-01") ) stats = collect_vault_stats(tmp_path) # Unique: alice, bob, carol assert stats["entity_count"] == 3 def test_tag_count(self, tmp_path: pathlib.Path) -> None: (tmp_path / "a.md").write_bytes( _make_note("tags:\n - ai\n - memory\ndate: 2025-01-01") ) (tmp_path / "b.md").write_bytes( _make_note("tags:\n - ai\n - research\ndate: 2025-01-01") ) stats = collect_vault_stats(tmp_path) # Unique: ai, memory, research assert stats["tag_count"] == 3 def test_index_freshness_none_when_no_data_dir(self, tmp_path: pathlib.Path) -> None: stats = collect_vault_stats(tmp_path) assert stats["index_freshness"] is None def test_index_freshness_set_when_data_dir_exists(self, tmp_path: pathlib.Path) -> None: (tmp_path / "data").mkdir() stats = collect_vault_stats(tmp_path) assert stats["index_freshness"] is not None # Must be ISO 8601 assert "T" in stats["index_freshness"] def test_index_freshness_is_utc(self, tmp_path: pathlib.Path) -> None: (tmp_path / "data").mkdir() stats = collect_vault_stats(tmp_path) freshness = stats["index_freshness"] assert freshness is not None # ISO 8601 UTC timestamp ends with +00:00 assert "+00:00" in freshness def test_unreadable_note_skipped_gracefully( self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch ) -> None: note = tmp_path / "bad.md" note.write_bytes(_make_note("source: telegram\ndate: 2025-01-01")) original_read = pathlib.Path.read_bytes def _fail_once(self: pathlib.Path) -> bytes: if self.name == "bad.md": raise OSError("Permission denied") return original_read(self) monkeypatch.setattr(pathlib.Path, "read_bytes", _fail_once) stats = collect_vault_stats(tmp_path) # Counts the file (os.walk sees it) but skips frontmatter parse assert stats["note_count"] == 1 assert stats["source_types"] == [] def test_vault_stats_typeddict_keys(self, tmp_path: pathlib.Path) -> None: stats = collect_vault_stats(tmp_path) assert set(stats.keys()) == { "note_count", "source_types", "projects", "entity_count", "tag_count", "index_freshness", } def test_note_without_frontmatter_counted_but_not_parsed( self, tmp_path: pathlib.Path ) -> None: (tmp_path / "plain.md").write_bytes(b"# Just a heading\nNo frontmatter.") stats = collect_vault_stats(tmp_path) assert stats["note_count"] == 1 assert stats["source_types"] == [] assert stats["projects"] == [] def test_markdown_extension_variants(self, tmp_path: pathlib.Path) -> None: (tmp_path / "a.md").write_bytes(_make_note()) (tmp_path / "b.markdown").write_bytes(_make_note()) (tmp_path / "c.mdx").write_bytes(_make_note()) stats = collect_vault_stats(tmp_path) assert stats["note_count"] == 3 # ============================================================================ # TestDomainInfoIntegration (CLI snapshot tests) # ============================================================================ class TestDomainInfoIntegration: def test_domain_flag_knowtation_no_vault_stats(self, tmp_path: pathlib.Path) -> None: """--domain knowtation without a repo gives no vault_stats (no root).""" result = _di(None, "--domain", "knowtation") assert result.exit_code == 0 data = json.loads(result.stdout) assert data["domain"] == "knowtation" # No vault_stats when no repo root is available assert "vault_stats" not in data def test_domain_flag_knowtation_schema_present(self, tmp_path: pathlib.Path) -> None: result = _di(None, "--domain", "knowtation") assert result.exit_code == 0 data = json.loads(result.stdout) assert "schema" in data assert data["schema"]["domain"] == "knowtation" def test_repo_knowtation_vault_stats_present(self, tmp_path: pathlib.Path) -> None: """When invoked inside a knowtation repo, vault_stats is included.""" repo = _make_knowtation_repo(tmp_path) (repo / "note.md").write_bytes( b"---\nsource: telegram\ndate: 2025-01-01\n---\nBody." ) result = _di(repo) assert result.exit_code == 0 data = json.loads(result.stdout) assert data["domain"] == "knowtation" assert "vault_stats" in data vs = data["vault_stats"] assert vs["note_count"] == 1 assert "telegram" in vs["source_types"] def test_repo_knowtation_vault_stats_shape(self, tmp_path: pathlib.Path) -> None: repo = _make_knowtation_repo(tmp_path) result = _di(repo) assert result.exit_code == 0 data = json.loads(result.stdout) vs = data["vault_stats"] assert "note_count" in vs assert "source_types" in vs assert "projects" in vs assert "entity_count" in vs assert "tag_count" in vs assert "index_freshness" in vs def test_repo_non_knowtation_no_vault_stats(self, tmp_path: pathlib.Path) -> None: """code domain repos must not get vault_stats.""" repo = tmp_path / "code-repo" muse = repo / ".muse" for sub in ("objects", "commits", "snapshots", "refs/heads"): (muse / sub).mkdir(parents=True) (muse / "HEAD").write_text("ref: refs/heads/main") (muse / "repo.json").write_text( json.dumps({"repo_id": "test", "domain": "code"}) ) result = _di(repo) assert result.exit_code == 0 data = json.loads(result.stdout) assert "vault_stats" not in data def test_text_format_knowtation_shows_notes(self, tmp_path: pathlib.Path) -> None: repo = _make_knowtation_repo(tmp_path) (repo / "a.md").write_bytes(b"---\nproject: p1\ndate: 2025-01-01\n---\n") (repo / "b.md").write_bytes(b"---\nproject: p2\ndate: 2025-01-01\n---\n") result = _di(repo, "--format", "text") assert result.exit_code == 0 assert "Notes:" in result.stdout assert "2" in result.stdout def test_text_format_knowtation_shows_projects(self, tmp_path: pathlib.Path) -> None: repo = _make_knowtation_repo(tmp_path) (repo / "a.md").write_bytes(b"---\nproject: born-free\ndate: 2025-01-01\n---\n") result = _di(repo, "--format", "text") assert result.exit_code == 0 assert "Projects:" in result.stdout assert "born-free" in result.stdout def test_registered_domains_includes_knowtation(self, tmp_path: pathlib.Path) -> None: result = _di(None, "--domain", "knowtation") assert result.exit_code == 0 data = json.loads(result.stdout) assert "knowtation" in data["registered_domains"] def test_json_output_serialisable(self, tmp_path: pathlib.Path) -> None: repo = _make_knowtation_repo(tmp_path) (repo / "note.md").write_bytes( b"---\nsource: telegram\ndate: 2025-01-01\ntags:\n - ai\n---\n" ) result = _di(repo) assert result.exit_code == 0 # Must be valid JSON — json.loads raises if not data = json.loads(result.stdout) assert isinstance(data, dict) def test_vault_stats_empty_vault_zero_counts(self, tmp_path: pathlib.Path) -> None: repo = _make_knowtation_repo(tmp_path) result = _di(repo) assert result.exit_code == 0 data = json.loads(result.stdout) vs = data["vault_stats"] assert vs["note_count"] == 0 assert vs["source_types"] == [] assert vs["projects"] == [] # ============================================================================ # TestRealVaultDataIntegrity # ============================================================================ @pytest.mark.skipif( not _REAL_VAULT.exists(), reason=f"Real vault not found at {_REAL_VAULT}", ) class TestRealVaultDataIntegrity: def _expected_content_note_count(self) -> int: count = 0 for md_path in _REAL_VAULT.rglob("*.md"): parts = md_path.relative_to(_REAL_VAULT).parts if parts and parts[0] in _SKIP_DIRS: continue count += 1 return count def test_note_count_matches_content_notes(self) -> None: stats = collect_vault_stats(_REAL_VAULT) expected = self._expected_content_note_count() assert stats["note_count"] == expected, ( f"note_count={stats['note_count']} but expected {expected}" ) def test_source_types_are_strings(self) -> None: stats = collect_vault_stats(_REAL_VAULT) assert all(isinstance(s, str) for s in stats["source_types"]) def test_projects_are_strings(self) -> None: stats = collect_vault_stats(_REAL_VAULT) assert all(isinstance(p, str) for p in stats["projects"]) def test_entity_count_non_negative(self) -> None: stats = collect_vault_stats(_REAL_VAULT) assert stats["entity_count"] >= 0 def test_tag_count_non_negative(self) -> None: stats = collect_vault_stats(_REAL_VAULT) assert stats["tag_count"] >= 0 def test_index_freshness_valid_if_present(self) -> None: stats = collect_vault_stats(_REAL_VAULT) if stats["index_freshness"] is not None: assert "T" in stats["index_freshness"] def test_stats_does_not_raise(self) -> None: try: collect_vault_stats(_REAL_VAULT) except Exception as exc: pytest.fail(f"collect_vault_stats raised: {exc}")