"""Tests for Temporal Knowledge Graph implementation. Tests cover: - Temporal storage tests - Query operator tests (BEFORE, AFTER, DURING, OVERLAPS) - Historical summary tests - Integration with tools """ import pytest import tempfile import os from datetime import datetime, timedelta from agent.temporal_knowledge_graph import ( TemporalTripleStore, TemporalTriple, TemporalOperator ) from agent.temporal_reasoning import ( TemporalReasoner, ChangeType, HistoricalSummary ) from tools.temporal_kg_tool import ( store_fact_with_time, query_historical_state, get_fact_history, generate_temporal_summary, when_did_we_learn, how_has_it_changed, query_with_temporal_operator, get_worldview_at_time ) class TestTemporalTripleStore: """Tests for the TemporalTripleStore class.""" @pytest.fixture def store(self): """Create a temporary store for testing.""" with tempfile.NamedTemporaryFile(suffix='.db', delete=False) as f: db_path = f.name store = TemporalTripleStore(db_path) yield store # Cleanup os.unlink(db_path) def test_store_fact(self, store): """Test storing a basic fact.""" triple = store.store_fact("Timmy", "has_feature", "sovereignty") assert triple.subject == "Timmy" assert triple.predicate == "has_feature" assert triple.object == "sovereignty" assert triple.version == 1 assert triple.valid_until is None def test_store_fact_with_validity_period(self, store): """Test storing a fact with validity bounds.""" valid_from = "2026-01-01T00:00:00" valid_until = "2026-12-31T23:59:59" triple = store.store_fact( "Hermes", "status", "active", valid_from=valid_from, valid_until=valid_until ) assert triple.valid_from == valid_from assert triple.valid_until == valid_until def test_fact_versioning(self, store): """Test that facts are properly versioned.""" # Store initial fact triple1 = store.store_fact("Timmy", "version", "1.0") assert triple1.version == 1 # Store updated fact triple2 = store.store_fact("Timmy", "version", "2.0") assert triple2.version == 2 # Check that first fact was superseded history = store.get_fact_history("Timmy", "version") assert len(history) == 2 assert history[0].superseded_by == triple2.id def test_query_at_time(self, store): """Test querying facts at a specific time.""" # Store facts at different times store.store_fact("Timmy", "status", "alpha", valid_from="2026-01-01T00:00:00") store.store_fact("Timmy", "status", "beta", valid_from="2026-03-01T00:00:00") store.store_fact("Timmy", "status", "stable", valid_from="2026-06-01T00:00:00") # Query at different points feb_facts = store.query_at_time("2026-02-01T00:00:00", subject="Timmy") assert len(feb_facts) == 1 assert feb_facts[0].object == "alpha" may_facts = store.query_at_time("2026-05-01T00:00:00", subject="Timmy") assert len(may_facts) == 1 assert may_facts[0].object == "beta" jul_facts = store.query_at_time("2026-07-01T00:00:00", subject="Timmy") assert len(jul_facts) == 1 assert jul_facts[0].object == "stable" def test_query_temporal_operators(self, store): """Test temporal query operators.""" # Store some facts store.store_fact("A", "rel", "1", valid_from="2026-01-01T00:00:00") store.store_fact("B", "rel", "2", valid_from="2026-03-01T00:00:00") store.store_fact("C", "rel", "3", valid_from="2026-06-01T00:00:00") # Test BEFORE before_april = store.query_temporal( TemporalOperator.BEFORE, "2026-04-01T00:00:00" ) assert len(before_april) == 2 # A and B # Test AFTER after_feb = store.query_temporal( TemporalOperator.AFTER, "2026-02-01T00:00:00" ) assert len(after_feb) == 2 # B and C # Test DURING (at a specific time) during_may = store.query_temporal( TemporalOperator.DURING, "2026-05-01T00:00:00" ) assert len(during_may) == 1 # Only B is valid in May assert during_may[0].object == "2" def test_get_fact_history(self, store): """Test retrieving fact version history.""" # Create multiple versions store.store_fact("Feature", "status", "planned", valid_from="2026-01-01T00:00:00") store.store_fact("Feature", "status", "in_progress", valid_from="2026-02-01T00:00:00") store.store_fact("Feature", "status", "completed", valid_from="2026-03-01T00:00:00") history = store.get_fact_history("Feature", "status") assert len(history) == 3 assert history[0].object == "planned" assert history[1].object == "in_progress" assert history[2].object == "completed" # Check versions assert history[0].version == 1 assert history[1].version == 2 assert history[2].version == 3 def test_get_entity_changes(self, store): """Test getting entity changes in a time range.""" store.store_fact("Codebase", "feature", "auth", valid_from="2026-01-01T00:00:00") store.store_fact("Codebase", "feature", "logging", valid_from="2026-02-01T00:00:00") store.store_fact("Codebase", "feature", "metrics", valid_from="2026-03-01T00:00:00") changes = store.get_entity_changes( "Codebase", "2026-01-15T00:00:00", "2026-03-15T00:00:00" ) # Should include logging and metrics assert len(changes) >= 2 def test_export_import(self, store): """Test exporting and importing data.""" # Store some data store.store_fact("Test", "data", "value1") store.store_fact("Test", "data", "value2") # Export json_data = store.export_to_json() assert "Test" in json_data assert "value1" in json_data assert "value2" in json_data # Create new store and import with tempfile.NamedTemporaryFile(suffix='.db', delete=False) as f: db_path2 = f.name try: store2 = TemporalTripleStore(db_path2) store2.import_from_json(json_data) # Verify imported data facts = store2.query_at_time(datetime.now().isoformat(), subject="Test") assert len(facts) >= 1 finally: os.unlink(db_path2) class TestTemporalReasoner: """Tests for the TemporalReasoner class.""" @pytest.fixture def reasoner(self): """Create a temporary reasoner for testing.""" with tempfile.NamedTemporaryFile(suffix='.db', delete=False) as f: db_path = f.name store = TemporalTripleStore(db_path) reasoner = TemporalReasoner(store) yield reasoner os.unlink(db_path) def test_what_did_we_believe(self, reasoner): """Test "what did we believe" queries.""" # Set up facts reasoner.store.store_fact("Timmy", "view", "optimistic", valid_from="2026-01-01T00:00:00") reasoner.store.store_fact("Timmy", "view", "cautious", valid_from="2026-03-01T00:00:00") # Query before March beliefs = reasoner.what_did_we_believe("Timmy", "2026-02-15T00:00:00") assert len(beliefs) == 1 assert beliefs[0].object == "optimistic" def test_when_did_we_learn(self, reasoner): """Test "when did we learn" queries.""" timestamp = "2026-02-15T10:30:00" reasoner.store.store_fact( "MLX", "integrated_with", "Hermes", valid_from=timestamp ) when = reasoner.when_did_we_learn("MLX", "integrated_with") assert when == timestamp def test_how_has_it_changed(self, reasoner): """Test "how has it changed" queries.""" reasoner.store.store_fact("Security", "level", "low", valid_from="2026-01-01T00:00:00") reasoner.store.store_fact("Security", "level", "medium", valid_from="2026-02-01T00:00:00") reasoner.store.store_fact("Security", "level", "high", valid_from="2026-03-01T00:00:00") changes = reasoner.how_has_it_changed("Security", "2026-01-15T00:00:00") assert len(changes) >= 2 # Check that changes are properly categorized change_types = [c.change_type for c in changes] assert ChangeType.MODIFIED in change_types or ChangeType.ADDED in change_types def test_generate_temporal_summary(self, reasoner): """Test generating historical summaries.""" # Create a history of changes reasoner.store.store_fact("Project", "status", "planning", valid_from="2026-01-01T00:00:00") reasoner.store.store_fact("Project", "status", "development", valid_from="2026-02-01T00:00:00") reasoner.store.store_fact("Project", "milestone", "alpha", valid_from="2026-02-15T00:00:00") reasoner.store.store_fact("Project", "status", "testing", valid_from="2026-03-01T00:00:00") summary = reasoner.generate_temporal_summary( "Project", "2026-01-01T00:00:00", "2026-04-01T00:00:00" ) assert summary.entity == "Project" assert summary.total_changes >= 3 assert len(summary.evolution_timeline) >= 3 assert len(summary.current_state) >= 1 def test_get_worldview_at_time(self, reasoner): """Test getting complete worldview at a time.""" reasoner.store.store_fact("Timmy", "mood", "happy", valid_from="2026-01-01T00:00:00") reasoner.store.store_fact("Timmy", "task", "coding", valid_from="2026-01-01T00:00:00") reasoner.store.store_fact("Hermes", "status", "active", valid_from="2026-01-01T00:00:00") worldview = reasoner.get_worldview_at_time("2026-01-15T00:00:00") assert "Timmy" in worldview assert "Hermes" in worldview assert len(worldview["Timmy"]) == 2 def test_infer_temporal_relationship(self, reasoner): """Test temporal relationship inference.""" triple_a = reasoner.store.store_fact("A", "rel", "1", valid_from="2026-01-01T00:00:00") triple_a.valid_until = "2026-02-01T00:00:00" triple_b = reasoner.store.store_fact("B", "rel", "2", valid_from="2026-02-15T00:00:00") rel = reasoner.infer_temporal_relationship(triple_a, triple_b) assert "before" in rel.lower() class TestTemporalKGTools: """Tests for the temporal KG tool functions.""" @pytest.fixture(autouse=True) def reset_singleton(self): """Reset singleton instances before each test.""" import tools.temporal_kg_tool as tool_module tool_module._store = None tool_module._reasoner = None yield tool_module._store = None tool_module._reasoner = None def test_store_fact_with_time(self): """Test the store_fact_with_time tool function.""" result = store_fact_with_time( subject="Hermes Agent", predicate="has_feature", object="input_sanitizer", valid_from="2026-04-01T01:00:00" ) assert result["success"] is True assert result["triple"]["subject"] == "Hermes Agent" assert result["triple"]["predicate"] == "has_feature" assert result["triple"]["object"] == "input_sanitizer" def test_query_historical_state(self): """Test the query_historical_state tool function.""" # Store a fact first store_fact_with_time( subject="Timmy", predicate="view_on_sovereignty", object="strong", valid_from="2026-02-01T00:00:00" ) # Query it result = query_historical_state("Timmy", "2026-03-01T00:00:00") assert result["success"] is True assert result["subject"] == "Timmy" assert result["fact_count"] == 1 assert result["facts"][0]["object"] == "strong" def test_get_fact_history(self): """Test the get_fact_history tool function.""" # Create version history store_fact_with_time("Feature", "status", "planned", valid_from="2026-01-01T00:00:00") store_fact_with_time("Feature", "status", "done", valid_from="2026-02-01T00:00:00") result = get_fact_history("Feature", "status") assert result["success"] is True assert result["version_count"] == 2 assert len(result["versions"]) == 2 def test_when_did_we_learn(self): """Test the when_did_we_learn tool function.""" store_fact_with_time( "MLX", "integrated_with", "Hermes", valid_from="2026-03-15T12:00:00" ) result = when_did_we_learn("MLX", "integrated_with") assert result["success"] is True assert result["first_known"] == "2026-03-15T12:00:00" def test_how_has_it_changed(self): """Test the how_has_it_changed tool function.""" store_fact_with_time("Codebase", "feature_count", "10", valid_from="2026-01-01T00:00:00") store_fact_with_time("Codebase", "feature_count", "20", valid_from="2026-02-01T00:00:00") result = how_has_it_changed("Codebase", "2026-01-15T00:00:00") assert result["success"] is True assert result["change_count"] >= 1 def test_query_with_temporal_operator(self): """Test the query_with_temporal_operator tool function.""" store_fact_with_time("A", "rel", "1", valid_from="2026-01-01T00:00:00") store_fact_with_time("B", "rel", "2", valid_from="2026-03-01T00:00:00") result = query_with_temporal_operator("BEFORE", "2026-02-01T00:00:00") assert result["success"] is True assert result["fact_count"] == 1 assert result["facts"][0]["subject"] == "A" def test_get_worldview_at_time(self): """Test the get_worldview_at_time tool function.""" store_fact_with_time("Timmy", "mood", "good", valid_from="2026-01-01T00:00:00") store_fact_with_time("Hermes", "status", "running", valid_from="2026-01-01T00:00:00") result = get_worldview_at_time("2026-01-15T00:00:00") assert result["success"] is True assert result["entity_count"] == 2 def test_generate_temporal_summary(self): """Test the generate_temporal_summary tool function.""" store_fact_with_time("Security", "level", "low", valid_from="2026-01-01T00:00:00") store_fact_with_time("Security", "level", "high", valid_from="2026-03-01T00:00:00") result = generate_temporal_summary("Security", "2026-01-01T00:00:00", "2026-04-01T00:00:00") assert result["success"] is True assert result["entity"] == "Security" assert result["summary"]["total_changes"] >= 1 class TestIntegration: """Integration tests for the complete temporal KG system.""" @pytest.fixture def system(self): """Create a complete temporal KG system.""" with tempfile.NamedTemporaryFile(suffix='.db', delete=False) as f: db_path = f.name store = TemporalTripleStore(db_path) reasoner = TemporalReasoner(store) yield {"store": store, "reasoner": reasoner} os.unlink(db_path) def test_full_workflow(self, system): """Test a complete temporal knowledge workflow.""" store = system["store"] reasoner = system["reasoner"] # 1. Store initial facts about a security audit store.store_fact("SecurityAudit", "status", "scheduled", valid_from="2026-01-01T00:00:00") store.store_fact("SecurityAudit", "auditor", "ExternalFirm", valid_from="2026-01-01T00:00:00") # 2. Update as audit progresses store.store_fact("SecurityAudit", "status", "in_progress", valid_from="2026-02-01T00:00:00") store.store_fact("SecurityAudit", "findings", "none_yet", valid_from="2026-02-01T00:00:00") # 3. Complete audit store.store_fact("SecurityAudit", "status", "completed", valid_from="2026-03-01T00:00:00") store.store_fact("SecurityAudit", "findings", "5_minor_issues", valid_from="2026-03-01T00:00:00") store.store_fact("SecurityAudit", "recommendation", "address_within_30_days", valid_from="2026-03-01T00:00:00") # 4. Query historical state jan_state = reasoner.get_worldview_at_time("2026-01-15T00:00:00", ["SecurityAudit"]) assert jan_state["SecurityAudit"][0]["predicate"] == "status" assert jan_state["SecurityAudit"][0]["object"] == "scheduled" feb_state = reasoner.get_worldview_at_time("2026-02-15T00:00:00", ["SecurityAudit"]) status_fact = [f for f in feb_state["SecurityAudit"] if f["predicate"] == "status"][0] assert status_fact["object"] == "in_progress" # 5. Generate summary summary = reasoner.generate_temporal_summary( "SecurityAudit", "2026-01-01T00:00:00", "2026-04-01T00:00:00" ) assert summary.total_changes >= 5 assert any(f["predicate"] == "status" for f in summary.key_facts) # 6. Check when we learned about findings when = reasoner.when_did_we_learn("SecurityAudit", "findings") assert when is not None def test_temporal_inference(self, system): """Test temporal inference capabilities.""" store = system["store"] reasoner = system["reasoner"] # Store facts with temporal relationships triple_a = store.store_fact("EventA", "happened", "yes", valid_from="2026-01-01T00:00:00") triple_a.valid_until = "2026-01-31T23:59:59" triple_b = store.store_fact("EventB", "happened", "yes", valid_from="2026-02-01T00:00:00") # Infer relationship rel = reasoner.infer_temporal_relationship(triple_a, triple_b) assert "before" in rel.lower() if __name__ == "__main__": pytest.main([__file__, "-v"])