"""Tests for muse/plugins/knowtation/parser.py — Phase 1.3. Test tiers covered ------------------ 1. Unit — every public function and property, all SPEC §2 field groups, coercion edge cases, slug normalisation, validation, migration pipeline. 2. Data-integrity — parse every content note in the real Knowtation vault without error; verify round-trip fidelity (parse → to_dict → parse). """ from __future__ import annotations import pathlib import textwrap import warnings from typing import Any import pytest from muse.plugins.knowtation.parser import ( FRONTMATTER_SCHEMA_VERSION, SCHEMA_VERSIONS, ParsedFrontmatter, _coerce_optional_str, _coerce_str_or_list, _extract_frontmatter_block, migrate_frontmatter, normalize_slug, parse_frontmatter, validate_inbox_note, validate_slug_field, ) # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- _REAL_VAULT = pathlib.Path("/Users/aaronrenecarvajal/knowtation/vault") # Structural directories that hold scaffold docs, not authored vault notes. _SKIP_DIRS: frozenset[str] = frozenset({"templates", "meta"}) def _note(frontmatter: str, body: str = "") -> bytes: """Build raw note bytes from a frontmatter YAML string and optional body.""" return f"---\n{frontmatter}\n---\n{body}".encode() def _load_content_notes() -> dict[str, bytes]: notes: dict[str, bytes] = {} 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 try: notes[str(md_path.relative_to(_REAL_VAULT))] = md_path.read_bytes() except OSError: pass return notes # ============================================================================ # TestExtractFrontmatterBlock # ============================================================================ class TestExtractFrontmatterBlock: def test_basic_extraction(self) -> None: raw = _extract_frontmatter_block(_note("title: Hello")) assert raw is not None assert "title: Hello" in raw def test_no_frontmatter_returns_none(self) -> None: assert _extract_frontmatter_block(b"# Just a heading\n") is None def test_empty_content_returns_none(self) -> None: assert _extract_frontmatter_block(b"") is None def test_unclosed_block_returns_none(self) -> None: assert _extract_frontmatter_block(b"---\ntitle: oops\n") is None def test_yaml_dots_delimiter(self) -> None: content = b"---\ntitle: dot\n...\nbody here" raw = _extract_frontmatter_block(content) assert raw is not None assert "title: dot" in raw def test_crlf_line_endings(self) -> None: content = b"---\r\ntitle: crlf\r\n---\r\nbody" raw = _extract_frontmatter_block(content) assert raw is not None def test_binary_content_returns_none(self) -> None: assert _extract_frontmatter_block(b"\x00\x01\x02\x03") is None # ============================================================================ # TestCoerceHelpers # ============================================================================ class TestCoerceOptionalStr: def test_none_returns_none(self) -> None: assert _coerce_optional_str(None) is None def test_string_passthrough(self) -> None: assert _coerce_optional_str("hello") == "hello" def test_integer_stringified(self) -> None: assert _coerce_optional_str(42) == "42" def test_bool_stringified(self) -> None: assert _coerce_optional_str(True) == "True" class TestCoerceStrOrList: def test_none_returns_empty(self) -> None: assert _coerce_str_or_list(None) == [] def test_yaml_list_preserved(self) -> None: assert _coerce_str_or_list(["foo", "bar"]) == ["foo", "bar"] def test_comma_separated_string_split(self) -> None: assert _coerce_str_or_list("foo, bar, baz") == ["foo", "bar", "baz"] def test_single_string_no_comma(self) -> None: assert _coerce_str_or_list("solo") == ["solo"] def test_empty_string_returns_empty(self) -> None: assert _coerce_str_or_list("") == [] def test_integer_element_stringified(self) -> None: assert _coerce_str_or_list([1, 2]) == ["1", "2"] def test_mixed_list_stringified(self) -> None: assert _coerce_str_or_list(["a", 1, None]) == ["a", "1", "None"] # ============================================================================ # TestNormalizeSlug # ============================================================================ class TestNormalizeSlug: def test_lowercase(self) -> None: assert normalize_slug("BornFree") == "bornfree" def test_spaces_become_hyphens(self) -> None: assert normalize_slug("born free") == "born-free" def test_underscores_become_hyphens(self) -> None: assert normalize_slug("born_free") == "born-free" def test_leading_trailing_hyphens_stripped(self) -> None: assert normalize_slug("-hello-") == "hello" def test_already_valid_slug_unchanged(self) -> None: assert normalize_slug("born-free") == "born-free" def test_numbers_preserved(self) -> None: assert normalize_slug("phase1") == "phase1" def test_special_chars_replaced(self) -> None: result = normalize_slug("hello!world") assert result == "hello-world" # ============================================================================ # TestParseFrontmatterSection21Common # ============================================================================ class TestParseFrontmatterSection21Common: def test_title_parsed(self) -> None: fm = parse_frontmatter(_note("title: My Note")) assert fm is not None assert fm.title == "My Note" def test_project_parsed(self) -> None: fm = parse_frontmatter(_note("project: born-free")) assert fm is not None assert fm.project == "born-free" def test_tags_yaml_list(self) -> None: fm = parse_frontmatter(_note("tags:\n - ai\n - research")) assert fm is not None assert fm.tags == ["ai", "research"] def test_tags_comma_separated_string(self) -> None: fm = parse_frontmatter(_note("tags: ai, research, memory")) assert fm is not None assert fm.tags == ["ai", "research", "memory"] def test_date_parsed(self) -> None: fm = parse_frontmatter(_note("date: 2025-01-15")) assert fm is not None assert fm.date == "2025-01-15" def test_updated_parsed(self) -> None: fm = parse_frontmatter(_note("updated: 2025-06-01")) assert fm is not None assert fm.updated == "2025-06-01" def test_all_common_fields(self) -> None: content = _note( "title: Full Note\nproject: my-proj\ntags:\n - t1\ndate: 2025-01-01\nupdated: 2025-03-01" ) fm = parse_frontmatter(content) assert fm is not None assert fm.title == "Full Note" assert fm.project == "my-proj" assert fm.tags == ["t1"] assert fm.date == "2025-01-01" assert fm.updated == "2025-03-01" def test_schema_version_set(self) -> None: fm = parse_frontmatter(_note("title: x")) assert fm is not None assert fm.schema_version == FRONTMATTER_SCHEMA_VERSION # ============================================================================ # TestParseFrontmatterSection22Inbox # ============================================================================ class TestParseFrontmatterSection22Inbox: def _inbox_note(self) -> bytes: return _note("source: telegram\ndate: 2025-05-01\nsource_id: msg-abc123") def test_source_parsed(self) -> None: fm = parse_frontmatter(self._inbox_note()) assert fm is not None assert fm.source == "telegram" def test_source_id_parsed(self) -> None: fm = parse_frontmatter(self._inbox_note()) assert fm is not None assert fm.source_id == "msg-abc123" def test_source_type_alias_parsed(self) -> None: fm = parse_frontmatter(_note("source_type: chatgpt-export\ndate: 2025-05-01")) assert fm is not None assert fm.source_type == "chatgpt-export" def test_effective_source_prefers_source(self) -> None: fm = parse_frontmatter( _note("source: telegram\nsource_type: slack\ndate: 2025-05-01") ) assert fm is not None assert fm.effective_source == "telegram" def test_effective_source_falls_back_to_source_type(self) -> None: fm = parse_frontmatter(_note("source_type: chatgpt-export\ndate: 2025-05-01")) assert fm is not None assert fm.effective_source == "chatgpt-export" def test_effective_source_none_when_absent(self) -> None: fm = parse_frontmatter(_note("title: plain")) assert fm is not None assert fm.effective_source is None def test_is_inbox_note_true(self) -> None: fm = parse_frontmatter(self._inbox_note()) assert fm is not None assert fm.is_inbox_note is True def test_is_inbox_note_false_no_source(self) -> None: fm = parse_frontmatter(_note("date: 2025-05-01")) assert fm is not None assert fm.is_inbox_note is False def test_is_inbox_note_false_no_date(self) -> None: fm = parse_frontmatter(_note("source: telegram")) assert fm is not None assert fm.is_inbox_note is False # ============================================================================ # TestParseFrontmatterSection23Temporal # ============================================================================ class TestParseFrontmatterSection23Temporal: def test_follows_single_string(self) -> None: fm = parse_frontmatter(_note("follows: inbox/note-a.md")) assert fm is not None assert fm.follows == ["inbox/note-a.md"] def test_follows_yaml_list(self) -> None: fm = parse_frontmatter(_note("follows:\n - inbox/a.md\n - inbox/b.md")) assert fm is not None assert fm.follows == ["inbox/a.md", "inbox/b.md"] def test_causal_chain_id(self) -> None: fm = parse_frontmatter(_note("causal_chain_id: chain-abc")) assert fm is not None assert fm.causal_chain_id == "chain-abc" def test_entity_list(self) -> None: fm = parse_frontmatter(_note("entity:\n - person:alice\n - project:muse")) assert fm is not None assert fm.entity == ["person:alice", "project:muse"] def test_episode_id(self) -> None: fm = parse_frontmatter(_note("episode_id: ep-2025-01")) assert fm is not None assert fm.episode_id == "ep-2025-01" def test_summarizes_list(self) -> None: fm = parse_frontmatter(_note("summarizes:\n - notes/old.md")) assert fm is not None assert fm.summarizes == ["notes/old.md"] def test_summarizes_range(self) -> None: fm = parse_frontmatter(_note("summarizes_range: 2025-01/2025-03")) assert fm is not None assert fm.summarizes_range == "2025-01/2025-03" def test_state_snapshot_true(self) -> None: fm = parse_frontmatter(_note("state_snapshot: true")) assert fm is not None assert fm.state_snapshot is True def test_state_snapshot_false_by_default(self) -> None: fm = parse_frontmatter(_note("title: no snapshot flag")) assert fm is not None assert fm.state_snapshot is False # ============================================================================ # TestParseFrontmatterSection24Reserved # ============================================================================ class TestParseFrontmatterSection24Reserved: """Reserved Phase-12 fields are stored without validation.""" def test_network_parsed(self) -> None: fm = parse_frontmatter(_note("network: icp")) assert fm is not None assert fm.network == "icp" def test_wallet_address_parsed(self) -> None: # Quote the value so PyYAML does not interpret it as a hex integer. fm = parse_frontmatter(_note('wallet_address: "0xDEAD"')) assert fm is not None assert fm.wallet_address == "0xDEAD" def test_tx_hash_parsed(self) -> None: fm = parse_frontmatter(_note("tx_hash: abc123")) assert fm is not None assert fm.tx_hash == "abc123" def test_payment_status_parsed(self) -> None: fm = parse_frontmatter(_note("payment_status: completed")) assert fm is not None assert fm.payment_status == "completed" def test_reserved_fields_all_none_by_default(self) -> None: fm = parse_frontmatter(_note("title: plain")) assert fm is not None assert fm.network is None assert fm.wallet_address is None assert fm.tx_hash is None assert fm.payment_status is None # ============================================================================ # TestParseFrontmatterAttachments # ============================================================================ class TestParseFrontmatterAttachments: """Phase 1.7 attachments field (mist IDs).""" def test_attachments_yaml_list(self) -> None: fm = parse_frontmatter(_note("attachments:\n - abc123def456\n - 123456abcdef")) assert fm is not None assert fm.attachments == ["abc123def456", "123456abcdef"] def test_attachments_empty_by_default(self) -> None: fm = parse_frontmatter(_note("title: plain")) assert fm is not None assert fm.attachments == [] # ============================================================================ # TestParseFrontmatterExtraKeys # ============================================================================ class TestParseFrontmatterExtraKeys: def test_unknown_key_goes_to_extra(self) -> None: fm = parse_frontmatter(_note("custom_field: my_value")) assert fm is not None assert fm.extra == {"custom_field": "my_value"} def test_known_keys_not_in_extra(self) -> None: fm = parse_frontmatter(_note("title: x\ncustom: y")) assert fm is not None assert "title" not in fm.extra assert fm.extra == {"custom": "y"} def test_no_unknown_keys_extra_empty(self) -> None: fm = parse_frontmatter(_note("title: clean\ndate: 2025-01-01")) assert fm is not None assert fm.extra == {} # ============================================================================ # TestParseFrontmatterEdgeCases # ============================================================================ class TestParseFrontmatterEdgeCases: def test_no_frontmatter_returns_none(self) -> None: assert parse_frontmatter(b"# Just a heading\n") is None def test_empty_content_returns_none(self) -> None: assert parse_frontmatter(b"") is None def test_invalid_yaml_returns_none(self) -> None: content = b"---\n: bad: yaml: here\n---\n" result = parse_frontmatter(content) # Either None (parse error) or a valid ParsedFrontmatter — never raises. assert result is None or isinstance(result, ParsedFrontmatter) def test_frontmatter_is_scalar_returns_none(self) -> None: content = b"---\njust a string\n---\n" assert parse_frontmatter(content) is None def test_frontmatter_is_list_returns_none(self) -> None: content = b"---\n- item1\n- item2\n---\n" assert parse_frontmatter(content) is None def test_binary_content_returns_none(self) -> None: assert parse_frontmatter(b"\x00\x01\x02") is None def test_crlf_note_parsed_correctly(self) -> None: content = b"---\r\nsource: slack\r\ndate: 2025-01-01\r\n---\r\nbody" fm = parse_frontmatter(content) assert fm is not None assert fm.source == "slack" def test_date_as_yaml_date_object_coerced(self) -> None: # PyYAML parses `date: 2025-01-01` as a Python date object; coerce to str. fm = parse_frontmatter(_note("date: 2025-01-01")) assert fm is not None assert fm.date is not None assert "2025" in fm.date # ============================================================================ # TestParsedFrontmatterToDict (round-trip) # ============================================================================ class TestParsedFrontmatterToDict: def test_omits_none_fields(self) -> None: fm = ParsedFrontmatter(title="x") d = fm.to_dict() assert "project" not in d assert "tags" not in d assert d["title"] == "x" def test_omits_empty_lists(self) -> None: fm = ParsedFrontmatter() d = fm.to_dict() assert "tags" not in d assert "follows" not in d def test_omits_false_state_snapshot(self) -> None: fm = ParsedFrontmatter(state_snapshot=False) assert "state_snapshot" not in fm.to_dict() def test_includes_true_state_snapshot(self) -> None: fm = ParsedFrontmatter(state_snapshot=True) assert fm.to_dict()["state_snapshot"] is True def test_extra_keys_included(self) -> None: fm = ParsedFrontmatter(extra={"my_custom": "val"}) assert fm.to_dict()["my_custom"] == "val" def test_round_trip_inbox_note(self) -> None: original = _note( "source: telegram\nsource_id: msg-1\ndate: 2025-05-01\nproject: born-free\ntags:\n - ai\n" ) fm1 = parse_frontmatter(original) assert fm1 is not None # Serialise to dict then wrap in YAML + re-parse. import yaml serialised = ("---\n" + yaml.dump(fm1.to_dict()) + "---\n").encode() fm2 = parse_frontmatter(serialised) assert fm2 is not None assert fm2.source == fm1.source assert fm2.source_id == fm1.source_id assert fm2.date == fm1.date assert fm2.project == fm1.project assert fm2.tags == fm1.tags def test_round_trip_temporal_note(self) -> None: import yaml original = _note( "follows: inbox/a.md\ncausal_chain_id: chain-1\nentity:\n - alice\n" "episode_id: ep-1\nsummarizes:\n - notes/old.md\n" "summarizes_range: 2025-01/2025-03\nstate_snapshot: true\n" ) fm1 = parse_frontmatter(original) assert fm1 is not None serialised = ("---\n" + yaml.dump(fm1.to_dict()) + "---\n").encode() fm2 = parse_frontmatter(serialised) assert fm2 is not None assert fm2.follows == fm1.follows assert fm2.causal_chain_id == fm1.causal_chain_id assert fm2.entity == fm1.entity assert fm2.state_snapshot is True # ============================================================================ # TestValidateInboxNote # ============================================================================ class TestValidateInboxNote: def test_valid_inbox_note_no_errors(self) -> None: fm = ParsedFrontmatter(source="telegram", date="2025-05-01") assert validate_inbox_note(fm) == [] def test_valid_with_source_type_alias(self) -> None: fm = ParsedFrontmatter(source_type="chatgpt-export", date="2025-05-01") assert validate_inbox_note(fm) == [] def test_missing_source_produces_error(self) -> None: fm = ParsedFrontmatter(date="2025-05-01") errors = validate_inbox_note(fm) assert any("source" in e for e in errors) def test_missing_date_produces_error(self) -> None: fm = ParsedFrontmatter(source="telegram") errors = validate_inbox_note(fm) assert any("date" in e for e in errors) def test_missing_both_produces_two_errors(self) -> None: fm = ParsedFrontmatter() assert len(validate_inbox_note(fm)) == 2 # ============================================================================ # TestValidateSlugField # ============================================================================ class TestValidateSlugField: def test_valid_slug_no_errors(self) -> None: assert validate_slug_field("born-free", "project") == [] def test_uppercase_produces_error(self) -> None: errors = validate_slug_field("BornFree", "project") assert len(errors) == 1 # normalize_slug("BornFree") → "bornfree" (no hyphen inserted for case change). assert "bornfree" in errors[0] def test_spaces_produce_error(self) -> None: errors = validate_slug_field("born free", "project") assert len(errors) == 1 def test_underscore_produces_error(self) -> None: errors = validate_slug_field("born_free", "project") assert len(errors) == 1 def test_field_name_in_error_message(self) -> None: errors = validate_slug_field("BAD", "my_field") assert "my_field" in errors[0] # ============================================================================ # TestSchemaVersionConstant # ============================================================================ class TestSchemaVersionConstant: def test_schema_version_is_string(self) -> None: assert isinstance(FRONTMATTER_SCHEMA_VERSION, str) def test_schema_version_in_versions_tuple(self) -> None: assert FRONTMATTER_SCHEMA_VERSION in SCHEMA_VERSIONS def test_schema_versions_ordered(self) -> None: assert list(SCHEMA_VERSIONS) == sorted(SCHEMA_VERSIONS, key=int) def test_parsed_note_carries_current_version(self) -> None: fm = parse_frontmatter(_note("title: x")) assert fm is not None assert fm.schema_version == FRONTMATTER_SCHEMA_VERSION # ============================================================================ # TestSchemaMigration # ============================================================================ class TestSchemaMigration: def test_same_version_returns_same_object(self) -> None: data: dict[str, Any] = {"title": "x"} result = migrate_frontmatter(data, FRONTMATTER_SCHEMA_VERSION) assert result is data # unchanged — same object def test_v1_to_v2_stub_preserves_all_keys(self) -> None: data: dict[str, Any] = { "title": "migration test", "source": "telegram", "date": "2025-01-01", "custom": "value", } result = migrate_frontmatter(data, "1") assert result["title"] == "migration test" # not lost assert result["source"] == "telegram" assert result["custom"] == "value" def test_v1_to_v2_stub_no_data_loss(self) -> None: data: dict[str, Any] = {"tags": ["a", "b"], "project": "born-free"} result = migrate_frontmatter(data, "1") assert result["tags"] == ["a", "b"] assert result["project"] == "born-free" def test_unknown_version_raises(self) -> None: with pytest.raises(ValueError, match="Unknown frontmatter schema version"): migrate_frontmatter({}, "99") def test_migration_result_is_parseable(self) -> None: import yaml data = {"source": "slack", "date": "2025-05-01"} migrated = migrate_frontmatter(data, "1") yaml_str = yaml.dump(migrated) content = f"---\n{yaml_str}---\n".encode() fm = parse_frontmatter(content) assert fm is not None assert fm.source == "slack" # ============================================================================ # TestRealVaultDataIntegrity # ============================================================================ _REAL_VAULT_NOTES: dict[str, bytes] = {} def _get_real_vault_notes() -> dict[str, bytes]: global _REAL_VAULT_NOTES if not _REAL_VAULT_NOTES: _REAL_VAULT_NOTES = _load_content_notes() return _REAL_VAULT_NOTES @pytest.mark.skipif( not _REAL_VAULT.exists(), reason=f"Real vault not found at {_REAL_VAULT}", ) class TestRealVaultDataIntegrity: """Parse every content note in the real vault; verify no crashes and round-trip.""" def test_vault_has_content_notes(self) -> None: notes = _get_real_vault_notes() assert len(notes) > 0, "Vault should contain at least one content note." def test_all_notes_parse_without_exception(self) -> None: """parse_frontmatter must never raise — it returns None or ParsedFrontmatter.""" notes = _get_real_vault_notes() for path, content in notes.items(): try: result = parse_frontmatter(content) assert result is None or isinstance(result, ParsedFrontmatter), ( f"Unexpected return type for {path}: {type(result)}" ) except Exception as exc: pytest.fail(f"parse_frontmatter raised on '{path}': {exc}") def test_notes_with_frontmatter_carry_schema_version(self) -> None: notes = _get_real_vault_notes() for path, content in notes.items(): fm = parse_frontmatter(content) if fm is not None: assert fm.schema_version == FRONTMATTER_SCHEMA_VERSION, ( f"schema_version mismatch for '{path}'" ) def test_round_trip_lossless_for_known_fields(self) -> None: """parse → to_dict → parse produces the same known fields.""" import yaml notes = _get_real_vault_notes() failed: list[str] = [] for path, content in notes.items(): fm1 = parse_frontmatter(content) if fm1 is None: continue d1 = fm1.to_dict() try: serialised = ("---\n" + yaml.dump(d1) + "---\n").encode() fm2 = parse_frontmatter(serialised) except Exception as exc: failed.append(f"{path}: serialisation error — {exc}") continue if fm2 is None: failed.append(f"{path}: re-parse returned None after round-trip") continue # Key known fields must survive the round-trip. for attr in ("title", "project", "date", "source", "source_id"): v1 = getattr(fm1, attr) v2 = getattr(fm2, attr) if v1 != v2: failed.append( f"{path}: field '{attr}' changed: {v1!r} → {v2!r}" ) if failed: warnings.warn( f"{len(failed)} round-trip failures:\n" + "\n".join(failed[:10]), UserWarning, stacklevel=2, ) pytest.fail( f"{len(failed)} notes failed round-trip fidelity check. " "See warnings for details." ) def test_inbox_notes_satisfy_spec_contract(self) -> None: """Every parsed note in inbox/ that claims a source must also have a date.""" notes = _get_real_vault_notes() violations: list[str] = [] for path, content in notes.items(): if "inbox" not in path: continue fm = parse_frontmatter(content) if fm is None: continue errors = validate_inbox_note(fm) if errors: violations.append(f"{path}: {errors}") if violations: warnings.warn( f"{len(violations)} inbox note(s) violate SPEC §2.2:\n" + "\n".join(violations[:10]), UserWarning, stacklevel=2, )