Spaces:
Running
Running
File size: 5,596 Bytes
e8d7c11 | 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 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 | """Tests for memory API routes."""
import pytest
from fastapi.testclient import TestClient
class TestMemoryAPI:
"""Test memory API endpoints."""
def test_store_memory_entry(self, client: TestClient) -> None:
"""Test POST /api/memory/store creates new memory entry."""
payload = {
"memory_type": "short_term",
"content": {
"observation": "User clicked login button",
"action": "click",
},
"metadata": {"url": "https://example.com"},
"episode_id": "ep_001",
"agent_id": "agent_test",
}
response = client.post("/api/memory/store", json=payload)
assert response.status_code == 201
data = response.json()
assert "id" in data
assert data["memory_type"] == "short_term"
assert data["content"] == payload["content"]
assert data["episode_id"] == "ep_001"
assert "timestamp" in data
def test_store_memory_all_types(self, client: TestClient) -> None:
"""Test storing memory with all valid types."""
valid_types = ["short_term", "working", "long_term", "shared"]
for memory_type in valid_types:
payload = {
"memory_type": memory_type,
"content": {"test": f"data for {memory_type}"},
}
response = client.post("/api/memory/store", json=payload)
assert response.status_code == 201
data = response.json()
assert data["memory_type"] == memory_type
def test_store_memory_invalid_type(self, client: TestClient) -> None:
"""Test storing memory with invalid type."""
payload = {"memory_type": "invalid_type", "content": {"test": "data"}}
response = client.post("/api/memory/store", json=payload)
assert response.status_code == 422
def test_get_memory_entry(self, client: TestClient) -> None:
"""Test GET /api/memory/{entry_id}."""
# Store first
payload = {
"memory_type": "long_term",
"content": {"knowledge": "test data"},
}
store_response = client.post("/api/memory/store", json=payload)
assert store_response.status_code == 201
entry_id = store_response.json()["id"]
# Retrieve
response = client.get(f"/api/memory/{entry_id}")
assert response.status_code == 200
data = response.json()
assert data["id"] == entry_id
assert data["memory_type"] == "long_term"
def test_get_nonexistent_memory(self, client: TestClient) -> None:
"""Test GET /api/memory/{entry_id} for non-existent."""
response = client.get("/api/memory/nonexistent-id-12345")
assert response.status_code == 404
def test_delete_memory_entry(self, client: TestClient) -> None:
"""Test DELETE /api/memory/{entry_id}."""
# Store first
payload = {
"memory_type": "short_term",
"content": {"temporary": "data"},
}
store_response = client.post("/api/memory/store", json=payload)
assert store_response.status_code == 201
entry_id = store_response.json()["id"]
# Delete
response = client.delete(f"/api/memory/{entry_id}")
assert response.status_code == 204
# Verify deleted
get_response = client.get(f"/api/memory/{entry_id}")
assert get_response.status_code == 404
def test_query_memory(self, client: TestClient) -> None:
"""Test POST /api/memory/query."""
# Store some entries first
for i in range(3):
payload = {
"memory_type": "short_term",
"content": {"index": i, "data": f"test_{i}"},
}
client.post("/api/memory/store", json=payload)
# Query
query_payload = {"query": "test", "limit": 10}
response = client.post("/api/memory/query", json=query_payload)
assert response.status_code == 200
data = response.json()
assert "entries" in data
assert "total_found" in data
def test_get_memory_stats(self, client: TestClient) -> None:
"""Test GET /api/memory/stats/overview."""
response = client.get("/api/memory/stats/overview")
assert response.status_code == 200
data = response.json()
assert "short_term_count" in data
assert "working_count" in data
assert "long_term_count" in data
assert "shared_count" in data
assert "total_count" in data
def test_clear_memory_layer(self, client: TestClient) -> None:
"""Test DELETE /api/memory/clear/{memory_type}."""
# Store entries
payload = {
"memory_type": "short_term",
"content": {"test": "data"},
}
client.post("/api/memory/store", json=payload)
# Clear
response = client.delete("/api/memory/clear/short_term")
assert response.status_code == 204
def test_consolidate_memory(self, client: TestClient) -> None:
"""Test POST /api/memory/consolidate."""
# Store short-term entries
for i in range(3):
payload = {
"memory_type": "short_term",
"content": {"index": i},
}
client.post("/api/memory/store", json=payload)
# Consolidate
response = client.post("/api/memory/consolidate")
assert response.status_code == 200
data = response.json()
assert "consolidated_count" in data
|