"""Tests for muse/plugins/knowtation/events.py — Phase 4.1. Test tiers (per Rule #0 and plan §0.2) -------------------------------------- 1. **Unit** — Every public symbol; positive and negative paths; every validation rule in :meth:`MemoryEventRecord.__post_init__`. 2. **Integration** — Round-trip identity, frozen-instance enforcement, forward-compat extra-key handling, hash determinism across re-imports. 3. **End-to-end** — Simulated commit-metadata embed → JSON dump → re-parse pipeline. 4. **Stress** — 10 000-record construction and 10 000-dict parse loops complete inside a budget. 5. **Data-integrity** — Python ``MemoryEventKind`` value set is byte-equal to the array literal in :file:`knowtation/lib/memory-event.mjs`; the module export ``EVENTS_SCHEMA_HASH`` matches an independently computed reference. 6. **Performance** — 1 000 ``from_dict + to_dict`` round-trips in under 200 ms. 7. **Security** — Sensitive-key rejection across the case-insensitive pattern surface; depth-limit boundary; oversized / malicious inputs are rejected (or accepted) without crashing. The full ``tests/test_knowtation_*.py`` suite must continue to pass alongside this module — see the runner step in the Phase 4.1 commit notes. """ from __future__ import annotations import dataclasses import json import pathlib import re import subprocess import sys import time from typing import Any import pytest from muse.plugins.knowtation.events import ( EVENTS_SCHEMA_HASH, EVENTS_SCHEMA_VERSION, MemoryEventKind, MemoryEventRecord, from_dict, is_valid_event_kind, to_dict, ) # ──────────────────────────────────────────────────────────────────────────── # Constants and helpers # ──────────────────────────────────────────────────────────────────────────── #: The 15 canonical kind values, hard-coded here to act as a static reference #: that tests compare the runtime enum against. Re-used by Tier 1 and 5. _EXPECTED_KIND_VALUES: frozenset[str] = frozenset( { "search", "export", "write", "import", "index", "propose", "agent_interaction", "capture", "error", "session_summary", "user", "consolidation", "consolidation_pass", "maintenance", "insight", } ) #: Path to the JS source-of-truth file, used by the data-integrity test. _JS_SOURCE = pathlib.Path( "/Users/aaronrenecarvajal/knowtation/lib/memory-event.mjs" ) def _minimal_valid_dict(**overrides: Any) -> dict[str, Any]: """Build a minimal valid event dict; allow per-test overrides.""" base: dict[str, Any] = { "event_id": "mem_0123456789ab", "kind": "search", "ts": "2026-05-13T13:57:00+00:00", "vault_id": "default", "data": {"query": "anything"}, } base.update(overrides) return base def _minimal_record(**overrides: Any) -> MemoryEventRecord: """Build a minimal valid :class:`MemoryEventRecord`.""" base = { "event_id": "mem_0123456789ab", "kind": MemoryEventKind.SEARCH, "ts": "2026-05-13T13:57:00+00:00", "vault_id": "default", "data": {"query": "anything"}, } base.update(overrides) return MemoryEventRecord(**base) # type: ignore[arg-type] # ============================================================================ # Tier 1 — Unit # ============================================================================ class TestKindEnumMembership: """Every canonical kind is present and addressable.""" @pytest.mark.parametrize("expected", sorted(_EXPECTED_KIND_VALUES)) def test_each_canonical_value_is_a_member(self, expected: str) -> None: assert MemoryEventKind(expected).value == expected def test_enum_has_exactly_fifteen_members(self) -> None: assert len(list(MemoryEventKind)) == 15 def test_str_enum_string_equality(self) -> None: assert MemoryEventKind.SEARCH == "search" assert str(MemoryEventKind.IMPORT) == "import" class TestIsValidEventKind: """Positive and negative coverage for the predicate.""" @pytest.mark.parametrize("kind", sorted(_EXPECTED_KIND_VALUES)) def test_returns_true_for_canonical_values(self, kind: str) -> None: assert is_valid_event_kind(kind) is True @pytest.mark.parametrize( "kind", [ "recall", "store", "consolidate", "summarize", "note_read", "note_write", "agent_decision", ], ) def test_returns_false_for_non_canonical_strings(self, kind: str) -> None: assert is_valid_event_kind(kind) is False def test_returns_false_for_non_strings(self) -> None: assert is_valid_event_kind(123) is False # type: ignore[arg-type] assert is_valid_event_kind(None) is False # type: ignore[arg-type] assert is_valid_event_kind(["search"]) is False # type: ignore[arg-type] class TestSchemaConstants: def test_schema_version_is_one_zero_zero(self) -> None: assert EVENTS_SCHEMA_VERSION == "1.0.0" def test_schema_hash_is_64_lowercase_hex(self) -> None: assert isinstance(EVENTS_SCHEMA_HASH, str) assert len(EVENTS_SCHEMA_HASH) == 64 assert re.fullmatch(r"[0-9a-f]{64}", EVENTS_SCHEMA_HASH) is not None class TestFromDictMinimal: def test_roundtrips_minimal_dict(self) -> None: d = _minimal_valid_dict() record = from_dict(d) assert record.event_id == d["event_id"] assert record.kind == MemoryEventKind.SEARCH assert record.ts == d["ts"] assert record.vault_id == d["vault_id"] assert record.data == d["data"] assert record.status == "success" assert record.agent_id is None class TestToDictOmitsNone: def test_omits_none_optional_fields(self) -> None: record = _minimal_record() out = to_dict(record) assert "agent_id" not in out assert "model_id" not in out assert "ttl" not in out assert "air_id" not in out assert out["event_id"] == record.event_id assert out["kind"] == "search" assert out["status"] == "success" def test_includes_set_optional_fields(self) -> None: record = _minimal_record( agent_id="agent-x", model_id="claude-sonnet-4-6", ttl="P30D", air_id="air_abc123", ) out = to_dict(record) assert out["agent_id"] == "agent-x" assert out["model_id"] == "claude-sonnet-4-6" assert out["ttl"] == "P30D" assert out["air_id"] == "air_abc123" class TestFromDictRejectsBadInputs: def test_missing_event_id(self) -> None: d = _minimal_valid_dict() del d["event_id"] with pytest.raises(ValueError, match="missing required field"): from_dict(d) def test_missing_kind(self) -> None: d = _minimal_valid_dict() del d["kind"] with pytest.raises(ValueError, match="missing required field"): from_dict(d) def test_missing_ts(self) -> None: d = _minimal_valid_dict() del d["ts"] with pytest.raises(ValueError, match="missing required field"): from_dict(d) def test_missing_vault_id(self) -> None: d = _minimal_valid_dict() del d["vault_id"] with pytest.raises(ValueError, match="missing required field"): from_dict(d) def test_unknown_kind(self) -> None: with pytest.raises(ValueError, match="not one of the 15"): from_dict(_minimal_valid_dict(kind="recall")) def test_malformed_event_id_too_short(self) -> None: with pytest.raises(ValueError, match="event_id must match"): from_dict(_minimal_valid_dict(event_id="mem_abc")) def test_malformed_event_id_wrong_prefix(self) -> None: with pytest.raises(ValueError, match="event_id must match"): from_dict(_minimal_valid_dict(event_id="evt_0123456789ab")) def test_malformed_event_id_uppercase_hex(self) -> None: # The JS regex emits lowercase hex only; we lock the same constraint. with pytest.raises(ValueError, match="event_id must match"): from_dict(_minimal_valid_dict(event_id="mem_ABCDEF012345")) def test_non_parseable_timestamp(self) -> None: with pytest.raises(ValueError, match="ts must be parseable"): from_dict(_minimal_valid_dict(ts="not-a-timestamp")) def test_empty_vault_id(self) -> None: with pytest.raises(ValueError, match="vault_id must be a non-empty string"): from_dict(_minimal_valid_dict(vault_id="")) def test_top_level_not_a_dict(self) -> None: with pytest.raises(ValueError, match="from_dict expects a dict"): from_dict("not-a-dict") # type: ignore[arg-type] def test_data_must_be_dict(self) -> None: with pytest.raises(ValueError, match="data must be a dict"): from_dict(_minimal_valid_dict(data=["not", "a", "dict"])) def test_kind_not_string(self) -> None: with pytest.raises(ValueError, match="kind must be a string"): from_dict(_minimal_valid_dict(kind=42)) class TestPostInitSensitiveData: @pytest.mark.parametrize( "key", ["password", "api_key", "TOKEN", "secret"], ) def test_construction_rejects_sensitive_keys(self, key: str) -> None: with pytest.raises(ValueError, match="sensitive key patterns"): _minimal_record(data={key: "x"}) # ============================================================================ # Tier 2 — Integration # ============================================================================ class TestRoundTripIdentity: @pytest.mark.parametrize( "extras", [ {}, {"agent_id": "agent-x"}, {"agent_id": "a", "model_id": "m"}, {"ttl": "P1D"}, {"air_id": "air_xyz", "status": "failed"}, { "agent_id": "a", "model_id": "m", "ttl": "P30D", "air_id": "air_zzz", "data": {"nested": {"safe_field": [1, 2, {"another": "value"}]}}, }, ], ) def test_from_to_dict_roundtrip(self, extras: dict[str, Any]) -> None: record = _minimal_record(**extras) d = to_dict(record) rebuilt = from_dict(d) assert rebuilt == record class TestForwardCompat: def test_unknown_keys_are_ignored(self) -> None: d = _minimal_valid_dict() d["future_field"] = "x" d["another_future"] = {"nested": True} record = from_dict(d) assert record.kind == MemoryEventKind.SEARCH class TestFrozenness: def test_setattr_raises_frozen_instance_error(self) -> None: record = _minimal_record() with pytest.raises(dataclasses.FrozenInstanceError): record.event_id = "mem_aaaaaaaaaaaa" # type: ignore[misc] def test_data_payload_is_isolated_after_to_dict(self) -> None: record = _minimal_record(data={"k": [1, 2, 3]}) out = to_dict(record) out["data"]["k"].append(4) assert record.data == {"k": [1, 2, 3]} class TestSchemaHashStability: def test_hash_is_deterministic_across_fresh_interpreter(self) -> None: """Spawn a subprocess to prove the hash is not interpreter-state-dependent. ``importlib.reload`` is not a safe substitute here because reloading rebuilds the dataclass and enum classes in place, which breaks ``isinstance`` checks for any record already imported by other tests. A subprocess gives a completely clean state. """ code = ( "from muse.plugins.knowtation.events import EVENTS_SCHEMA_HASH; " "print(EVENTS_SCHEMA_HASH)" ) result = subprocess.check_output( [sys.executable, "-c", code], text=True ).strip() assert result == EVENTS_SCHEMA_HASH # ============================================================================ # Tier 3 — End-to-end # ============================================================================ class TestCommitMetadataPipeline: """Simulate the full commit-metadata round trip.""" def test_record_embedded_in_commit_meta_roundtrips(self) -> None: record = _minimal_record( agent_id="@knowtation-daemon", model_id="claude-sonnet-4-6", data={"path": "vault/notes/example.md", "topic": "alpha"}, ) commit_meta: dict[str, Any] = { "schema_version": EVENTS_SCHEMA_VERSION, "schema_hash": EVENTS_SCHEMA_HASH, "memory_event": to_dict(record), "other_unrelated_field": "preserved", } wire = json.dumps(commit_meta) decoded = json.loads(wire) assert decoded["schema_version"] == "1.0.0" assert decoded["schema_hash"] == EVENTS_SCHEMA_HASH rebuilt = from_dict(decoded["memory_event"]) assert rebuilt == record # ============================================================================ # Tier 4 — Stress # ============================================================================ class TestStress: def test_construct_10k_records_under_2s(self) -> None: kinds = list(MemoryEventKind) start = time.perf_counter() for i in range(10_000): MemoryEventRecord( event_id=f"mem_{i:012x}", kind=kinds[i % len(kinds)], ts="2026-05-13T13:57:00+00:00", vault_id="default", data={"i": i}, ) elapsed = time.perf_counter() - start assert elapsed < 2.0, f"10k constructions took {elapsed:.3f}s (budget 2s)" def test_parse_10k_event_dicts_under_2s(self) -> None: kinds = sorted(_EXPECTED_KIND_VALUES) dicts = [ { "event_id": f"mem_{i:012x}", "kind": kinds[i % len(kinds)], "ts": "2026-05-13T13:57:00+00:00", "vault_id": "default", "data": {"i": i}, } for i in range(10_000) ] start = time.perf_counter() for d in dicts: from_dict(d) elapsed = time.perf_counter() - start assert elapsed < 2.0, f"10k parses took {elapsed:.3f}s (budget 2s)" # ============================================================================ # Tier 5 — Data-integrity # ============================================================================ class TestDataIntegrity: """Cross-language parity with lib/memory-event.mjs.""" def test_python_kinds_match_js_array(self) -> None: assert _JS_SOURCE.exists(), ( f"Cannot find JS source-of-truth at {_JS_SOURCE}; " "this test must run with the knowtation repo checked out." ) text = _JS_SOURCE.read_text(encoding="utf-8") match = re.search( r"MEMORY_EVENT_TYPES\s*=\s*Object\.freeze\(\s*\[(.+?)\]", text, flags=re.DOTALL, ) assert match is not None, ( "Could not locate MEMORY_EVENT_TYPES literal in memory-event.mjs" ) js_kinds = set(re.findall(r"'([^']+)'", match.group(1))) py_kinds = {k.value for k in MemoryEventKind} assert js_kinds == py_kinds, ( f"Drift detected — JS {sorted(js_kinds - py_kinds)!r} not in Python; " f"Python {sorted(py_kinds - js_kinds)!r} not in JS." ) def test_schema_hash_independent_recompute_matches(self) -> None: sorted_values = sorted(k.value for k in MemoryEventKind) canonical = json.dumps(sorted_values, sort_keys=True, separators=(",", ":")) import hashlib expected = hashlib.sha256(canonical.encode("utf-8")).hexdigest() assert expected == EVENTS_SCHEMA_HASH # ============================================================================ # Tier 6 — Performance # ============================================================================ class TestPerformance: def test_1k_roundtrips_under_200ms(self) -> None: record = _minimal_record( agent_id="agent-x", model_id="model-y", data={"a": 1, "b": [1, 2, 3], "c": {"d": "e"}}, ) start = time.perf_counter() for _ in range(1_000): from_dict(to_dict(record)) elapsed_ms = (time.perf_counter() - start) * 1_000.0 assert elapsed_ms < 200.0, ( f"1000 round-trips took {elapsed_ms:.1f}ms (budget 200ms)" ) # ============================================================================ # Tier 7 — Security # ============================================================================ class TestSecurityRejectsSensitiveKeys: @pytest.mark.parametrize( "data", [ {"password": "x"}, {"api_key": "x"}, {"api-key": "x"}, {"PRIVATE_KEY": "x"}, {"Authorization": "Bearer xyz"}, {"BEARER": "x"}, {"my_secret": "x"}, {"some_token": "x"}, {"credential": "x"}, ], ) def test_top_level_sensitive_keys_rejected(self, data: dict[str, Any]) -> None: with pytest.raises(ValueError, match="sensitive key patterns"): from_dict(_minimal_valid_dict(data=data)) def test_recursive_sensitive_key_rejected(self) -> None: data = {"nested": {"api_key": "secret"}} with pytest.raises(ValueError, match="sensitive key patterns"): from_dict(_minimal_valid_dict(data=data)) def test_sensitive_key_inside_list_rejected(self) -> None: data = {"items": [{"safe": 1}, {"private_key": "pk"}]} with pytest.raises(ValueError, match="sensitive key patterns"): from_dict(_minimal_valid_dict(data=data)) class TestSecurityDepthLimit: def test_sensitive_key_beyond_depth_8_is_not_scanned(self) -> None: # Build a structure where the sensitive key sits at nesting depth 9. # Mirrors the JS hasSensitiveKeys depth > 8 short-circuit. deep: dict[str, Any] = {"api_key": "x"} for _ in range(9): deep = {"a": deep} # Sanity: the sensitive key is buried 9 nests below the root. cur: Any = deep steps = 0 while isinstance(cur, dict) and "api_key" not in cur: cur = cur["a"] steps += 1 assert steps == 9, f"expected api_key buried 9 nests deep, got {steps}" record = from_dict(_minimal_valid_dict(data=deep)) assert record.data is deep or record.data == deep class TestSecurityAdversarialInputs: def test_oversized_kind_string_rejected_cleanly(self) -> None: oversized = "x" * 10_000 with pytest.raises(ValueError): from_dict(_minimal_valid_dict(kind=oversized)) def test_10k_safe_keys_accepted_under_1s(self) -> None: big_data: dict[str, Any] = {f"safe_field_{i}": i for i in range(10_000)} d = _minimal_valid_dict(data=big_data) start = time.perf_counter() record = from_dict(d) elapsed = time.perf_counter() - start assert elapsed < 1.0, ( f"parsing 10k safe-key data took {elapsed:.3f}s (budget 1s)" ) assert len(record.data) == 10_000