wehub-resource-sync
c3749daf48
Tests / test-windows (push) Waiting to run
Tests / test-macos (push) Waiting to run
Tests / test-linux (3.13) (push) Failing after 0s
Tests / test-linux (3.11) (push) Failing after 1s
Tests / lint (push) Failing after 0s
Tests / test-linux (3.9) (push) Failing after 1s
Docker / build (push) Failing after 1s
Docker / build-gpu (push) Failing after 2s
393 lines
14 KiB
Python
393 lines
14 KiB
Python
"""
|
|
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).
|
|
"""
|
|
|
|
import pytest
|
|
import sqlite3
|
|
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
|
|
|
|
def test_add_triple_rejects_inverted_interval(self, kg):
|
|
# valid_to before valid_from would never satisfy
|
|
# `valid_from <= as_of AND valid_to >= as_of` — silently invisible
|
|
# to every query. Reject at write time instead.
|
|
with pytest.raises(ValueError, match="before valid_from"):
|
|
kg.add_triple(
|
|
"Alice",
|
|
"worked_at",
|
|
"Acme",
|
|
valid_from="2026-03-01",
|
|
valid_to="2026-02-01",
|
|
)
|
|
|
|
def test_add_triple_accepts_equal_dates(self, kg):
|
|
# Same-day intervals are valid (point-in-time facts).
|
|
tid = kg.add_triple(
|
|
"Alice",
|
|
"joined",
|
|
"Acme",
|
|
valid_from="2026-03-15",
|
|
valid_to="2026-03-15",
|
|
)
|
|
assert tid.startswith("t_alice_joined_acme_")
|
|
|
|
def test_add_triple_allows_only_one_bound(self, kg):
|
|
# The guard only fires when BOTH bounds are set.
|
|
tid1 = kg.add_triple("Alice", "knows", "Bob", valid_from="2026-01-01")
|
|
assert tid1.startswith("t_alice_knows_bob_")
|
|
kg.invalidate("Alice", "knows", "Bob", ended="2026-02-01")
|
|
tid2 = kg.add_triple("Alice", "knew", "Bob", valid_to="2026-03-01")
|
|
assert tid2.startswith("t_alice_knew_bob_")
|
|
|
|
|
|
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
|
|
|
|
def test_timeline_entity_has_limit(self, kg):
|
|
# Add > 100 triples all connected to a single entity
|
|
for i in range(105):
|
|
kg.add_triple(
|
|
"hub", "connects_to", f"spoke_{i}", valid_from=f"2025-01-{(i % 28) + 1:02d}"
|
|
)
|
|
tl = kg.timeline("hub")
|
|
assert len(tl) == 100 # LIMIT 100 on entity-filtered branch
|
|
|
|
|
|
class TestWALMode:
|
|
def test_wal_mode_enabled(self, kg):
|
|
conn = kg._conn()
|
|
mode = conn.execute("PRAGMA journal_mode").fetchone()[0]
|
|
conn.close()
|
|
assert mode == "wal"
|
|
|
|
|
|
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
|
|
|
|
|
|
class TestTemporalDateTimeCompatibility:
|
|
def test_datetime_query_matches_legacy_date_only_fact(self, kg):
|
|
kg.add_triple(
|
|
"Alice",
|
|
"ate_at",
|
|
"Cafe",
|
|
valid_from="2026-05-06",
|
|
valid_to="2026-05-06",
|
|
)
|
|
|
|
result = kg.query_entity("Alice", as_of="2026-05-06T15:00:00Z")
|
|
|
|
assert len(result) == 1
|
|
assert result[0]["object"] == "Cafe"
|
|
|
|
def test_datetime_query_before_legacy_date_only_fact_does_not_match(self, kg):
|
|
kg.add_triple(
|
|
"Alice",
|
|
"ate_at",
|
|
"Cafe",
|
|
valid_from="2026-05-06",
|
|
valid_to="2026-05-06",
|
|
)
|
|
|
|
result = kg.query_entity("Alice", as_of="2026-05-05T23:59:59Z")
|
|
|
|
assert result == []
|
|
|
|
def test_datetime_query_after_legacy_date_only_fact_does_not_match(self, kg):
|
|
kg.add_triple(
|
|
"Alice",
|
|
"ate_at",
|
|
"Cafe",
|
|
valid_from="2026-05-06",
|
|
valid_to="2026-05-06",
|
|
)
|
|
|
|
result = kg.query_entity("Alice", as_of="2026-05-07T00:00:00Z")
|
|
|
|
assert result == []
|
|
|
|
def test_rejects_timezone_offset_datetime_at_kg_layer(self, kg):
|
|
with pytest.raises(ValueError):
|
|
kg.add_triple(
|
|
"Bob",
|
|
"works_at",
|
|
"Globex",
|
|
valid_from="2026-05-06T20:30:00-05:00",
|
|
)
|
|
|
|
def test_rejects_naive_datetime_at_kg_layer(self, kg):
|
|
with pytest.raises(ValueError):
|
|
kg.add_triple(
|
|
"Carol",
|
|
"is_in",
|
|
"NYC",
|
|
valid_from="2026-05-07T01:23:00",
|
|
)
|
|
|
|
def test_rejects_space_separated_datetime_at_kg_layer(self, kg):
|
|
with pytest.raises(ValueError):
|
|
kg.add_triple(
|
|
"Eve",
|
|
"is_in",
|
|
"London",
|
|
valid_from="2026-05-06T15:00:00Z",
|
|
valid_to="2026-05-06 20:00:00",
|
|
)
|
|
|
|
def test_date_only_valid_to_is_end_of_day_for_interval_check(self, kg):
|
|
kg.add_triple(
|
|
"Eve",
|
|
"is_in",
|
|
"London",
|
|
valid_from="2026-05-06T15:00:00Z",
|
|
valid_to="2026-05-06",
|
|
)
|
|
|
|
result = kg.query_entity("Eve", as_of="2026-05-06T20:00:00Z")
|
|
|
|
assert len(result) == 1
|
|
assert result[0]["object"] == "London"
|
|
|
|
def test_rejects_interval_when_date_only_end_is_before_datetime_start(self, kg):
|
|
with pytest.raises(
|
|
ValueError,
|
|
match=r"valid_to='2026-05-06'.*valid_from='2026-05-07T01:00:00Z'",
|
|
):
|
|
kg.add_triple(
|
|
"Eve",
|
|
"is_in",
|
|
"London",
|
|
valid_from="2026-05-07T01:00:00Z",
|
|
valid_to="2026-05-06",
|
|
)
|
|
|
|
def test_query_relationship_uses_safe_temporal_comparison(self, kg):
|
|
kg.add_triple(
|
|
"Alice",
|
|
"visited",
|
|
"Cafe",
|
|
valid_from="2026-05-06",
|
|
valid_to="2026-05-06",
|
|
)
|
|
|
|
result = kg.query_relationship("visited", as_of="2026-05-06T15:00:00Z")
|
|
|
|
assert len(result) == 1
|
|
assert result[0]["subject"] == "Alice"
|
|
assert result[0]["object"] == "Cafe"
|
|
|
|
def test_invalidate_rejects_timezone_offset_ended(self, kg):
|
|
kg.add_triple(
|
|
"Alice",
|
|
"works_at",
|
|
"Acme",
|
|
valid_from="2026-05-06T14:00:00Z",
|
|
)
|
|
|
|
with pytest.raises(ValueError):
|
|
kg.invalidate(
|
|
"Alice",
|
|
"works_at",
|
|
"Acme",
|
|
ended="2026-05-06T20:30:00-05:00",
|
|
)
|
|
|
|
|
|
class TestKnowledgeGraphConnectionCleanup:
|
|
def test_close_closes_connection_and_resets_handle(self, tmp_path):
|
|
kg = KnowledgeGraph(str(tmp_path / "kg.sqlite3"))
|
|
conn = kg._conn()
|
|
|
|
kg.close()
|
|
|
|
assert kg._connection is None
|
|
with pytest.raises(sqlite3.ProgrammingError):
|
|
conn.execute("SELECT 1")
|
|
|
|
def test_context_manager_closes_connection(self, tmp_path):
|
|
with KnowledgeGraph(str(tmp_path / "kg.sqlite3")) as kg:
|
|
conn = kg._conn()
|
|
kg.add_entity("Alice")
|
|
|
|
assert kg._connection is None
|
|
with pytest.raises(sqlite3.ProgrammingError):
|
|
conn.execute("SELECT 1")
|
|
|
|
|
|
class TestSupersessionBoundary:
|
|
"""Regression coverage for the as-of boundary double-count (issue #1913):
|
|
an as-of query at the instant one fact ends and its successor begins must
|
|
return only the successor for a single-valued predicate."""
|
|
|
|
def _models(self, kg, as_of):
|
|
return sorted(
|
|
f["object"]
|
|
for f in kg.query_entity("Bot", as_of=as_of, direction="outgoing")
|
|
if f["predicate"] == "uses_model"
|
|
)
|
|
|
|
def test_exact_datetime_boundary_returns_only_successor(self, kg):
|
|
# Two facts sharing a precise instant: half-open upper bound (strict >)
|
|
# means the fact ending at T no longer matches at T.
|
|
kg.add_triple(
|
|
"Bot",
|
|
"uses_model",
|
|
"A",
|
|
valid_from="2026-05-01T00:00:00Z",
|
|
valid_to="2026-06-02T12:00:00Z",
|
|
)
|
|
kg.add_triple("Bot", "uses_model", "B", valid_from="2026-06-02T12:00:00Z")
|
|
|
|
assert self._models(kg, "2026-06-02T11:59:59Z") == ["A"]
|
|
assert self._models(kg, "2026-06-02T12:00:00Z") == ["B"]
|
|
assert self._models(kg, "2026-06-02T12:00:01Z") == ["B"]
|
|
|
|
def test_supersede_date_only_resolves_to_successor(self, kg):
|
|
kg.add_triple("Bot", "uses_model", "claude-opus-4-7", valid_from="2026-05-01")
|
|
kg.supersede("Bot", "uses_model", "claude-opus-4-7", "claude-opus-4-8", at="2026-06-02")
|
|
|
|
assert self._models(kg, "2026-06-01") == ["claude-opus-4-7"]
|
|
assert self._models(kg, "2026-06-02") == ["claude-opus-4-8"]
|
|
assert self._models(kg, "2026-06-03") == ["claude-opus-4-8"]
|
|
|
|
def test_supersede_datetime_boundary_resolves_to_successor(self, kg):
|
|
kg.add_triple("Bot", "uses_model", "A", valid_from="2026-05-01T00:00:00Z")
|
|
kg.supersede("Bot", "uses_model", "A", "B", at="2026-06-02T12:00:00Z")
|
|
|
|
assert self._models(kg, "2026-06-02T11:59:59Z") == ["A"]
|
|
assert self._models(kg, "2026-06-02T12:00:00Z") == ["B"]
|
|
|
|
def test_supersede_default_now_closes_old_and_opens_new(self, kg):
|
|
kg.add_triple("Bot", "uses_model", "A", valid_from="2026-05-01")
|
|
kg.supersede("Bot", "uses_model", "A", "B")
|
|
# A far-future as-of sees only the successor; the old fact was closed.
|
|
assert self._models(kg, "2099-01-01") == ["B"]
|
|
|
|
def test_supersede_degrades_to_add_when_no_open_old(self, kg):
|
|
tid = kg.supersede("Bot", "uses_model", "missing", "B", at="2026-01-01")
|
|
assert tid.startswith("t_bot_uses_model_b_")
|
|
assert self._models(kg, "2099-01-01") == ["B"]
|
|
|
|
def test_supersede_rejects_boundary_before_valid_from(self, kg):
|
|
kg.add_triple("Bot", "uses_model", "A", valid_from="2026-06-01")
|
|
with pytest.raises(ValueError, match="before valid_from"):
|
|
kg.supersede("Bot", "uses_model", "A", "B", at="2026-05-01")
|
|
|
|
def test_standalone_date_only_end_stays_valid_all_day(self, kg):
|
|
# Half-open change must NOT shrink a standalone date-only fact: it stays
|
|
# valid through the end of its final day (whole-day expansion retained).
|
|
kg.add_triple("Bot", "uses_model", "A", valid_from="2026-05-01", valid_to="2026-06-02")
|
|
assert self._models(kg, "2026-06-02") == ["A"]
|
|
assert self._models(kg, "2026-06-02T23:00:00Z") == ["A"]
|
|
assert self._models(kg, "2026-06-03") == []
|