wuyang
31 lines
1.2 KiB
Python
31 lines
1.2 KiB
Python
from __future__ import annotations
|
|
|
|
from agent_platform.store import RuntimeStore
|
|
|
|
|
|
async def test_conversation_mode_upgrade_is_one_way(settings) -> None:
|
|
store = RuntimeStore(settings.database_url)
|
|
await store.initialize()
|
|
try:
|
|
assert await store.select_mode("u1", "c1", "chat") == "chat"
|
|
assert await store.select_mode("u1", "c1", "work") == "work"
|
|
assert await store.select_mode("u1", "c1", "chat") == "work"
|
|
assert await store.select_mode("u2", "c1", "chat") == "chat"
|
|
finally:
|
|
await store.close()
|
|
|
|
|
|
async def test_memory_and_events_are_user_scoped(settings) -> None:
|
|
store = RuntimeStore(settings.database_url)
|
|
await store.initialize()
|
|
try:
|
|
memory_id = await store.remember("u1", "prefers compact answers")
|
|
assert [item["id"] for item in await store.recall("u1", "compact")] == [memory_id]
|
|
assert await store.recall("u2", "compact") == []
|
|
|
|
await store.append_event("r1", "u1", "c1", 1, "run.created", {"x": 1})
|
|
await store.append_event("r2", "u2", "c1", 1, "run.created", {"x": 2})
|
|
assert [event["payload"]["x"] for event in await store.events_for_chat("u1", "c1")] == [1]
|
|
finally:
|
|
await store.close()
|