test: expand coverage from 20 to 92 tests, migrate to uv
- Migrate from setuptools to hatchling build backend - Add dependency-groups (PEP 735) for dev tooling (pytest, ruff) - Remove redundant requirements.txt in favor of uv.lock - Fix __version__ mismatch (2.0.0 -> 3.0.0 to match pyproject.toml) New test files: - conftest.py: shared fixtures (isolated palace, KG, ChromaDB collection) - test_knowledge_graph.py: 17 tests (entity CRUD, temporal queries, timeline) - test_mcp_server.py: 25 tests (protocol dispatch, read/write/KG/diary tools) - test_searcher.py: 7 tests (search_memories API, filters, error handling) - test_dialect.py: 13 tests (AAAK compression, entity/emotion detection, zettel encoding) All 92 tests pass on Python 3.13 with chromadb 0.6.3.
This commit is contained in:
@@ -0,0 +1,124 @@
|
||||
"""
|
||||
test_knowledge_graph.py — Tests for the temporal knowledge graph.
|
||||
|
||||
Covers: entity CRUD, triple CRUD, temporal queries, invalidation,
|
||||
timeline, stats, and edge cases (duplicate triples, ID collisions).
|
||||
"""
|
||||
|
||||
from mempalace.knowledge_graph import KnowledgeGraph
|
||||
|
||||
|
||||
class TestEntityOperations:
|
||||
def test_add_entity(self, kg):
|
||||
eid = kg.add_entity("Alice", entity_type="person")
|
||||
assert eid == "alice"
|
||||
|
||||
def test_add_entity_normalizes_id(self, kg):
|
||||
eid = kg.add_entity("Dr. Chen", entity_type="person")
|
||||
assert eid == "dr._chen"
|
||||
|
||||
def test_add_entity_upsert(self, kg):
|
||||
kg.add_entity("Alice", entity_type="person")
|
||||
kg.add_entity("Alice", entity_type="engineer")
|
||||
# Should not raise — INSERT OR REPLACE
|
||||
stats = kg.stats()
|
||||
assert stats["entities"] == 1
|
||||
|
||||
|
||||
class TestTripleOperations:
|
||||
def test_add_triple_creates_entities(self, kg):
|
||||
tid = kg.add_triple("Alice", "knows", "Bob")
|
||||
assert tid.startswith("t_alice_knows_bob_")
|
||||
stats = kg.stats()
|
||||
assert stats["entities"] == 2 # auto-created
|
||||
|
||||
def test_add_triple_with_dates(self, kg):
|
||||
tid = kg.add_triple("Max", "does", "swimming", valid_from="2025-01-01")
|
||||
assert tid.startswith("t_max_does_swimming_")
|
||||
|
||||
def test_duplicate_triple_returns_existing_id(self, kg):
|
||||
tid1 = kg.add_triple("Alice", "knows", "Bob")
|
||||
tid2 = kg.add_triple("Alice", "knows", "Bob")
|
||||
assert tid1 == tid2
|
||||
|
||||
def test_invalidated_triple_allows_re_add(self, kg):
|
||||
tid1 = kg.add_triple("Alice", "works_at", "Acme")
|
||||
kg.invalidate("Alice", "works_at", "Acme", ended="2025-01-01")
|
||||
tid2 = kg.add_triple("Alice", "works_at", "Acme")
|
||||
assert tid1 != tid2 # new triple since old one was closed
|
||||
|
||||
|
||||
class TestQueries:
|
||||
def test_query_outgoing(self, seeded_kg):
|
||||
results = seeded_kg.query_entity("Alice", direction="outgoing")
|
||||
predicates = {r["predicate"] for r in results}
|
||||
assert "parent_of" in predicates
|
||||
assert "works_at" in predicates
|
||||
|
||||
def test_query_incoming(self, seeded_kg):
|
||||
results = seeded_kg.query_entity("Max", direction="incoming")
|
||||
assert any(r["subject"] == "Alice" and r["predicate"] == "parent_of" for r in results)
|
||||
|
||||
def test_query_both_directions(self, seeded_kg):
|
||||
results = seeded_kg.query_entity("Max", direction="both")
|
||||
directions = {r["direction"] for r in results}
|
||||
assert "outgoing" in directions
|
||||
assert "incoming" in directions
|
||||
|
||||
def test_query_as_of_filters_expired(self, seeded_kg):
|
||||
results = seeded_kg.query_entity("Alice", as_of="2023-06-01", direction="outgoing")
|
||||
employers = [r["object"] for r in results if r["predicate"] == "works_at"]
|
||||
assert "Acme Corp" in employers
|
||||
assert "NewCo" not in employers
|
||||
|
||||
def test_query_as_of_shows_current(self, seeded_kg):
|
||||
results = seeded_kg.query_entity("Alice", as_of="2025-06-01", direction="outgoing")
|
||||
employers = [r["object"] for r in results if r["predicate"] == "works_at"]
|
||||
assert "NewCo" in employers
|
||||
assert "Acme Corp" not in employers
|
||||
|
||||
def test_query_relationship(self, seeded_kg):
|
||||
results = seeded_kg.query_relationship("does")
|
||||
assert len(results) == 2 # swimming + chess
|
||||
|
||||
|
||||
class TestInvalidation:
|
||||
def test_invalidate_sets_valid_to(self, seeded_kg):
|
||||
seeded_kg.invalidate("Max", "does", "chess", ended="2026-01-01")
|
||||
results = seeded_kg.query_entity("Max", direction="outgoing")
|
||||
chess = [r for r in results if r["object"] == "chess"]
|
||||
assert len(chess) == 1
|
||||
assert chess[0]["valid_to"] == "2026-01-01"
|
||||
assert chess[0]["current"] is False
|
||||
|
||||
|
||||
class TestTimeline:
|
||||
def test_timeline_all(self, seeded_kg):
|
||||
tl = seeded_kg.timeline()
|
||||
assert len(tl) >= 4
|
||||
|
||||
def test_timeline_entity(self, seeded_kg):
|
||||
tl = seeded_kg.timeline("Max")
|
||||
subjects_and_objects = {t["subject"] for t in tl} | {t["object"] for t in tl}
|
||||
assert "Max" in subjects_and_objects
|
||||
|
||||
def test_timeline_global_has_limit(self, kg):
|
||||
# Add > 100 triples
|
||||
for i in range(105):
|
||||
kg.add_triple(f"entity_{i}", "relates_to", f"entity_{i + 1}")
|
||||
tl = kg.timeline()
|
||||
assert len(tl) == 100 # LIMIT 100
|
||||
|
||||
|
||||
class TestStats:
|
||||
def test_stats_empty(self, kg):
|
||||
stats = kg.stats()
|
||||
assert stats["entities"] == 0
|
||||
assert stats["triples"] == 0
|
||||
|
||||
def test_stats_seeded(self, seeded_kg):
|
||||
stats = seeded_kg.stats()
|
||||
assert stats["entities"] >= 4
|
||||
assert stats["triples"] == 5
|
||||
assert stats["current_facts"] == 4 # 1 expired (Acme Corp)
|
||||
assert stats["expired_facts"] == 1
|
||||
Reference in New Issue
Block a user