Spaces:
Sleeping
Sleeping
File size: 1,462 Bytes
14184e3 | 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 | """Tests for the lightweight local MemoryService (cluster + trajectory)."""
from app.services.memory_service import MemoryService
_DOC = "test_doc_hash_xyz"
def _clean(m: MemoryService):
m.flush_cluster(_DOC)
import os
p = m._traj_path(_DOC)
if os.path.exists(p):
os.remove(p)
def test_cluster_push_read_flush_roundtrip():
m = MemoryService()
_clean(m)
assert m.read_cluster(_DOC) == []
m.push_insights(_DOC, [{"concept": "A", "summary": "s"}])
m.push_insights(_DOC, [{"concept": "B", "summary": "t"}])
cluster = m.read_cluster(_DOC)
assert [c["concept"] for c in cluster] == ["A", "B"]
m.flush_cluster(_DOC)
assert m.read_cluster(_DOC) == []
def test_trajectory_is_append_only_and_persists():
m = MemoryService()
_clean(m)
m.append_trajectory(_DOC, {"node_id": "n1", "classification": "building_basics"})
m.append_trajectory(_DOC, {"node_id": "n1", "classification": "comfortable"})
traj = m.read_trajectory(_DOC)
assert len(traj) == 2
assert traj[0]["classification"] == "building_basics"
assert "ts" in traj[0]
# trajectory is NOT removed by a cluster flush
m.flush_cluster(_DOC)
assert len(m.read_trajectory(_DOC)) == 2
_clean(m)
def test_empty_document_id_is_noop():
m = MemoryService()
m.push_insights("", [{"x": 1}])
assert m.read_cluster("") == []
m.append_trajectory("", {"a": 1})
assert m.read_trajectory("") == []
|