Joseph Doherty
102 lines
3.3 KiB
Python
102 lines
3.3 KiB
Python
from __future__ import annotations
|
|
|
|
from chat.db.connection import open_db
|
|
from chat.db.migrate import apply_migrations
|
|
from chat.eventlog.log import append_event
|
|
from chat.eventlog.projector import project
|
|
import chat.state.entities # registers handlers
|
|
import chat.state.world # registers handlers
|
|
import chat.state.group_node # registers handlers
|
|
from chat.state.group_node import get_group_node
|
|
|
|
|
|
def _bot_payload(bot_id: str, name: str) -> dict:
|
|
return {
|
|
"id": bot_id,
|
|
"name": name,
|
|
"persona": "thoughtful, observant",
|
|
"voice_samples": [],
|
|
"traits": [],
|
|
"backstory": "",
|
|
"initial_relationship_to_you": "coworker",
|
|
"kickoff_prose": "",
|
|
}
|
|
|
|
|
|
def _chat_payload(chat_id: str = "chat_bot_a") -> dict:
|
|
return {
|
|
"id": chat_id,
|
|
"host_bot_id": "bot_a",
|
|
"guest_bot_id": "bot_b",
|
|
"initial_time": "2026-04-26T20:00:00+00:00",
|
|
"narrative_anchor": "Day 1 evening",
|
|
"weather": "clear",
|
|
}
|
|
|
|
|
|
def test_group_node_initialized_creates_row(tmp_path):
|
|
db = tmp_path / "t.db"
|
|
apply_migrations(db)
|
|
with open_db(db) as conn:
|
|
append_event(conn, kind="bot_authored", payload=_bot_payload("bot_a", "BotA"))
|
|
append_event(conn, kind="bot_authored", payload=_bot_payload("bot_b", "BotB"))
|
|
append_event(conn, kind="chat_created", payload=_chat_payload())
|
|
append_event(
|
|
conn,
|
|
kind="group_node_initialized",
|
|
payload={
|
|
"chat_id": "chat_bot_a",
|
|
"members": ["you", "bot_a", "bot_b"],
|
|
},
|
|
)
|
|
project(conn)
|
|
|
|
gn = get_group_node(conn, "chat_bot_a")
|
|
assert gn is not None
|
|
assert gn["chat_id"] == "chat_bot_a"
|
|
assert gn["members"] == ["you", "bot_a", "bot_b"]
|
|
assert gn["summary"] == ""
|
|
assert gn["dynamic"] == ""
|
|
assert gn["threads"] == []
|
|
|
|
|
|
def test_group_node_updated_changes_summary_and_dynamic(tmp_path):
|
|
db = tmp_path / "t.db"
|
|
apply_migrations(db)
|
|
with open_db(db) as conn:
|
|
append_event(conn, kind="bot_authored", payload=_bot_payload("bot_a", "BotA"))
|
|
append_event(conn, kind="bot_authored", payload=_bot_payload("bot_b", "BotB"))
|
|
append_event(conn, kind="chat_created", payload=_chat_payload())
|
|
append_event(
|
|
conn,
|
|
kind="group_node_initialized",
|
|
payload={
|
|
"chat_id": "chat_bot_a",
|
|
"members": ["you", "bot_a", "bot_b"],
|
|
},
|
|
)
|
|
append_event(
|
|
conn,
|
|
kind="group_node_updated",
|
|
payload={
|
|
"chat_id": "chat_bot_a",
|
|
"summary": "Three coworkers chatting about the project.",
|
|
"dynamic": "Tense but cordial.",
|
|
},
|
|
)
|
|
project(conn)
|
|
|
|
gn = get_group_node(conn, "chat_bot_a")
|
|
assert gn is not None
|
|
assert gn["summary"] == "Three coworkers chatting about the project."
|
|
assert gn["dynamic"] == "Tense but cordial."
|
|
# Members preserved across update
|
|
assert gn["members"] == ["you", "bot_a", "bot_b"]
|
|
|
|
|
|
def test_get_group_node_returns_none_for_missing_chat(tmp_path):
|
|
db = tmp_path / "t.db"
|
|
apply_migrations(db)
|
|
with open_db(db) as conn:
|
|
assert get_group_node(conn, "chat_missing") is None
|