from datetime import UTC, datetime from uuid import UUID, uuid7 from database.database import Database from memory import LongTermMemory, MemoryCategory from memory_store import LongTermMemoryStore def create_store(tmp_path): database = Database(tmp_path / "test.db") database.migrate() return LongTermMemoryStore(database) def test_get_returns_memory(tmp_path): store = create_store(tmp_path) memory = LongTermMemory.create( "The user prefers direct technical explanations.", MemoryCategory.SEMANTIC, ) store.add(memory) assert store.get(memory.id) == memory def test_created_memory_has_uuid7_id(): memory = LongTermMemory.create( "UUIDv7 teset", MemoryCategory.SEMANTIC, ) assert memory.id.version == 7 def test_created_memories_have_unique_ids(): first = LongTermMemory.create( "First memory.", MemoryCategory.SEMANTIC, ) second = LongTermMemory.create( "Second memory.", MemoryCategory.SEMANTIC, ) assert first.id != second.id def test_get_preserves_memory_id(tmp_path): store = create_store(tmp_path) memory = LongTermMemory.create( "ID round-trip test.", MemoryCategory.SEMANTIC, ) store.add(memory) result = store.get(memory.id) assert result is not None assert result.id == memory.id def test_get_missing_memory_returns_none(tmp_path): store = create_store(tmp_path) assert store.get(UUID("00000000-0000-7a3f-9c2e-1b4d8f6a0e37")) is None def test_list_returns_memories(tmp_path): store = create_store(tmp_path) first = LongTermMemory.create( "The user likes systems programming.", MemoryCategory.SEMANTIC, ) second = LongTermMemory.create( "The user completed Phase 7.1a.", MemoryCategory.EPISODIC, ) store.add(first) store.add(second) assert store.list() == [first, second] def test_delete_removes_memory(tmp_path): store = create_store(tmp_path) memory = LongTermMemory.create( "Temporary memory.", MemoryCategory.EPISODIC, ) store.add(memory) store.delete(memory.id) assert store.get(memory.id) is None def test_created_at_round_trips_as_utc(tmp_path): store = create_store(tmp_path) created_at = datetime( 2026, 9, 15, 12, 34, 56, 123456, tzinfo=UTC, ) memory = LongTermMemory( id=uuid7(), content="UTC timestamp test.", category=MemoryCategory.SEMANTIC, created_at=created_at, ) store.add(memory) result = store.get(memory.id) assert result is not None assert result.created_at == created_at assert result.created_at.tzinfo is not None assert result.created_at.utcoffset() == UTC.utcoffset(None) def test_search_matches_content(tmp_path): store = create_store(tmp_path) first = LongTermMemory.create( "The user prefers direct technical explanations.", MemoryCategory.SEMANTIC, ) second = LongTermMemory.create( "The user is working on a local personal agent.", MemoryCategory.SEMANTIC, ) store.add(first) store.add(second) assert store.search("technical") == [first] def test_search_returns_empty_list_when_no_memory_matches(tmp_path): store = create_store(tmp_path) memory = LongTermMemory.create( "The user likes systems programming.", MemoryCategory.SEMANTIC, ) store.add(memory) assert store.search("quantum physics") == [] def test_search_is_case_insensitive(tmp_path): store = create_store(tmp_path) memory = LongTermMemory.create( "The user prefers Rust.", MemoryCategory.SEMANTIC, ) store.add(memory) assert store.search("RUST") == [memory] def test_search_can_filter_by_category(tmp_path): store = create_store(tmp_path) semantic = LongTermMemory.create( "The user likes programming.", MemoryCategory.SEMANTIC, ) episodic = LongTermMemory.create( "The user discussed programming yesterday.", MemoryCategory.EPISODIC, ) store.add(semantic) store.add(episodic) assert store.search("programming", MemoryCategory.SEMANTIC) == [semantic] assert store.search("programming", MemoryCategory.EPISODIC) == [episodic] def test_search_respects_limit(tmp_path): store = create_store(tmp_path) memories = [ LongTermMemory.create( f"Programming memory {index}.", MemoryCategory.SEMANTIC, ) for index in range(3) ] for memory in memories: store.add(memory) assert store.search("programming", limit=2) == memories[:2]