| import pytest |
|
|
| from aegis.core import ( |
| DuplicateMemoryKeyError, |
| MemoryEntry, |
| MemoryKeyNotFoundError, |
| MemoryStore, |
| ) |
|
|
|
|
| def test_memory_store_adds_and_gets_entry() -> None: |
| store = MemoryStore() |
| entry = MemoryEntry( |
| key="planner.output", |
| value={"steps": ["analyze", "execute"]}, |
| metadata={"agent": "planner"}, |
| ) |
|
|
| store.add(entry) |
|
|
| stored = store.get("planner.output") |
|
|
| assert stored.key == "planner.output" |
| assert stored.value == {"steps": ["analyze", "execute"]} |
| assert stored.metadata == {"agent": "planner"} |
| assert len(store) == 1 |
|
|
|
|
| def test_memory_store_initializes_with_entries() -> None: |
| store = MemoryStore( |
| [ |
| MemoryEntry(key="beta", value=2), |
| MemoryEntry(key="alpha", value=1), |
| ] |
| ) |
|
|
| assert store.keys() == ("alpha", "beta") |
| assert len(store) == 2 |
|
|
|
|
| def test_memory_store_rejects_duplicate_key() -> None: |
| store = MemoryStore([MemoryEntry(key="result", value=1)]) |
|
|
| with pytest.raises( |
| DuplicateMemoryKeyError, |
| match="Memory key already exists: result", |
| ): |
| store.add(MemoryEntry(key="result", value=2)) |
|
|
|
|
| def test_memory_store_set_replaces_existing_entry() -> None: |
| store = MemoryStore([MemoryEntry(key="result", value=1)]) |
|
|
| store.set(MemoryEntry(key="result", value=2)) |
|
|
| assert store.get("result").value == 2 |
| assert len(store) == 1 |
|
|
|
|
| def test_memory_store_raises_for_missing_key() -> None: |
| store = MemoryStore() |
|
|
| with pytest.raises( |
| MemoryKeyNotFoundError, |
| match="Memory key not found: missing", |
| ): |
| store.get("missing") |
|
|
|
|
| def test_memory_store_deletes_entry() -> None: |
| store = MemoryStore([MemoryEntry(key="result", value=1)]) |
|
|
| store.delete("result") |
|
|
| assert store.contains("result") is False |
| assert len(store) == 0 |
|
|
|
|
| def test_memory_store_rejects_delete_for_missing_key() -> None: |
| store = MemoryStore() |
|
|
| with pytest.raises( |
| MemoryKeyNotFoundError, |
| match="Memory key not found: missing", |
| ): |
| store.delete("missing") |
|
|
|
|
| def test_memory_store_normalizes_keys() -> None: |
| store = MemoryStore() |
|
|
| store.add(MemoryEntry(key=" planner.output ", value=1)) |
|
|
| assert store.contains("planner.output") is True |
| assert store.get(" planner.output ").key == "planner.output" |
|
|
|
|
| def test_memory_store_rejects_empty_key_lookup() -> None: |
| store = MemoryStore() |
|
|
| with pytest.raises(ValueError, match="Memory key must not be empty"): |
| store.get(" ") |
|
|
|
|
| def test_memory_store_clear_removes_all_entries() -> None: |
| store = MemoryStore( |
| [ |
| MemoryEntry(key="one", value=1), |
| MemoryEntry(key="two", value=2), |
| ] |
| ) |
|
|
| store.clear() |
|
|
| assert store.keys() == () |
| assert len(store) == 0 |
|
|
