File size: 2,815 Bytes
8a87cbb
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
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