Joseph Doherty
335 lines
11 KiB
Python
335 lines
11 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.world # registers handlers
|
|
from chat.state.world import (
|
|
active_scene,
|
|
find_container,
|
|
get_activity,
|
|
get_chat,
|
|
get_container,
|
|
get_scene,
|
|
list_chats,
|
|
)
|
|
|
|
|
|
def _chat_payload(**overrides):
|
|
payload = {
|
|
"id": "chat_bot_a",
|
|
"host_bot_id": "bot_a",
|
|
"guest_bot_id": None,
|
|
"initial_time": "2026-04-26T20:00:00+00:00",
|
|
"narrative_anchor": "Day 1 evening",
|
|
"weather": "clear",
|
|
}
|
|
payload.update(overrides)
|
|
return payload
|
|
|
|
|
|
def _container_payload(**overrides):
|
|
payload = {
|
|
"chat_id": "chat_bot_a",
|
|
"name": "office",
|
|
"type": "workplace",
|
|
"properties": {
|
|
"public": True,
|
|
"moving": False,
|
|
"audible_range": "normal",
|
|
"slots": [],
|
|
},
|
|
"parent_id": None,
|
|
}
|
|
payload.update(overrides)
|
|
return payload
|
|
|
|
|
|
def test_chat_created_initializes_chats_and_chat_state(tmp_path):
|
|
db = tmp_path / "t.db"
|
|
apply_migrations(db)
|
|
with open_db(db) as conn:
|
|
append_event(conn, kind="chat_created", payload=_chat_payload())
|
|
project(conn)
|
|
chat = get_chat(conn, "chat_bot_a")
|
|
assert chat is not None
|
|
assert chat["id"] == "chat_bot_a"
|
|
assert chat["host_bot_id"] == "bot_a"
|
|
assert chat["guest_bot_id"] is None
|
|
assert chat["time"] == "2026-04-26T20:00:00+00:00"
|
|
assert chat["weather"] == "clear"
|
|
assert chat["narrative_anchor"] == "Day 1 evening"
|
|
assert chat["active_scene_id"] is None
|
|
|
|
|
|
def test_get_chat_returns_none_for_missing_id(tmp_path):
|
|
db = tmp_path / "t.db"
|
|
apply_migrations(db)
|
|
with open_db(db) as conn:
|
|
assert get_chat(conn, "chat_missing") is None
|
|
|
|
|
|
def test_list_chats_returns_all_joined_rows(tmp_path):
|
|
db = tmp_path / "t.db"
|
|
apply_migrations(db)
|
|
with open_db(db) as conn:
|
|
append_event(conn, kind="chat_created", payload=_chat_payload())
|
|
append_event(conn, kind="chat_created", payload=_chat_payload(
|
|
id="chat_bot_b", host_bot_id="bot_b",
|
|
))
|
|
project(conn)
|
|
chats = list_chats(conn)
|
|
assert len(chats) == 2
|
|
ids = [c["id"] for c in chats]
|
|
assert ids == sorted(ids)
|
|
assert all("time" in c for c in chats)
|
|
|
|
|
|
def test_chat_created_with_optional_fields_defaulted(tmp_path):
|
|
db = tmp_path / "t.db"
|
|
apply_migrations(db)
|
|
with open_db(db) as conn:
|
|
append_event(conn, kind="chat_created", payload={
|
|
"id": "chat_bot_a",
|
|
"host_bot_id": "bot_a",
|
|
"initial_time": "2026-04-26T20:00:00+00:00",
|
|
})
|
|
project(conn)
|
|
chat = get_chat(conn, "chat_bot_a")
|
|
assert chat is not None
|
|
assert chat["guest_bot_id"] is None
|
|
assert chat["weather"] == ""
|
|
assert chat["narrative_anchor"] == ""
|
|
assert chat["active_scene_id"] is None
|
|
|
|
|
|
def test_container_created_inserts_and_findable_by_name(tmp_path):
|
|
db = tmp_path / "t.db"
|
|
apply_migrations(db)
|
|
with open_db(db) as conn:
|
|
append_event(conn, kind="chat_created", payload=_chat_payload())
|
|
append_event(conn, kind="container_created", payload=_container_payload())
|
|
project(conn)
|
|
c = find_container(conn, "chat_bot_a", "office")
|
|
assert c is not None
|
|
assert c["name"] == "office"
|
|
assert c["type"] == "workplace"
|
|
assert c["chat_id"] == "chat_bot_a"
|
|
assert c["parent_id"] is None
|
|
assert c["properties"] == {
|
|
"public": True,
|
|
"moving": False,
|
|
"audible_range": "normal",
|
|
"slots": [],
|
|
}
|
|
# get_container by id also works
|
|
c2 = get_container(conn, c["id"])
|
|
assert c2 is not None
|
|
assert c2["name"] == "office"
|
|
assert c2["properties"]["public"] is True
|
|
|
|
|
|
def test_find_container_returns_none_when_missing(tmp_path):
|
|
db = tmp_path / "t.db"
|
|
apply_migrations(db)
|
|
with open_db(db) as conn:
|
|
assert find_container(conn, "chat_bot_a", "nope") is None
|
|
assert get_container(conn, 999) is None
|
|
|
|
|
|
def test_activity_change_inserts_then_replaces(tmp_path):
|
|
db = tmp_path / "t.db"
|
|
apply_migrations(db)
|
|
with open_db(db) as conn:
|
|
append_event(conn, kind="chat_created", payload=_chat_payload())
|
|
append_event(conn, kind="container_created", payload=_container_payload())
|
|
# First activity event then a second for the same entity that supersedes it.
|
|
append_event(conn, kind="activity_change", payload={
|
|
"entity_id": "bot_a",
|
|
"container_id": 1,
|
|
"slot": "desk_chair",
|
|
"posture": "sitting",
|
|
"action": {
|
|
"verb": "writing email",
|
|
"interruptible": True,
|
|
"required_attention": "medium",
|
|
"expected_duration": "a few minutes",
|
|
"started_at": "2026-04-26T20:00:00+00:00",
|
|
},
|
|
"attention": "the screen",
|
|
"holding": [],
|
|
"status": {},
|
|
})
|
|
append_event(conn, kind="activity_change", payload={
|
|
"entity_id": "bot_a",
|
|
"container_id": 1,
|
|
"posture": "standing",
|
|
"action": {"verb": "pacing"},
|
|
"attention": "the window",
|
|
})
|
|
project(conn)
|
|
a = get_activity(conn, "bot_a")
|
|
assert a is not None
|
|
assert a["entity_id"] == "bot_a"
|
|
assert a["container_id"] == 1
|
|
# second event replaced first
|
|
assert a["posture"] == "standing"
|
|
assert a["action"]["verb"] == "pacing"
|
|
assert a["attention"] == "the window"
|
|
# only one row exists for that entity
|
|
count = conn.execute(
|
|
"SELECT COUNT(*) FROM activity WHERE entity_id = ?", ("bot_a",)
|
|
).fetchone()[0]
|
|
assert count == 1
|
|
|
|
|
|
def test_activity_change_initial_values_persist_when_only_one_event(tmp_path):
|
|
db = tmp_path / "t.db"
|
|
apply_migrations(db)
|
|
with open_db(db) as conn:
|
|
append_event(conn, kind="chat_created", payload=_chat_payload())
|
|
append_event(conn, kind="container_created", payload=_container_payload())
|
|
append_event(conn, kind="activity_change", payload={
|
|
"entity_id": "bot_a",
|
|
"container_id": 1,
|
|
"slot": "desk_chair",
|
|
"posture": "sitting",
|
|
"action": {"verb": "writing email"},
|
|
"attention": "the screen",
|
|
"holding": ["pen"],
|
|
"status": {"hungry": False},
|
|
})
|
|
project(conn)
|
|
a = get_activity(conn, "bot_a")
|
|
assert a is not None
|
|
assert a["slot"] == "desk_chair"
|
|
assert a["posture"] == "sitting"
|
|
assert a["action"]["verb"] == "writing email"
|
|
assert a["attention"] == "the screen"
|
|
assert a["holding"] == ["pen"]
|
|
assert a["status"] == {"hungry": False}
|
|
|
|
|
|
def test_activity_change_defaults_for_minimal_payload(tmp_path):
|
|
db = tmp_path / "t.db"
|
|
apply_migrations(db)
|
|
with open_db(db) as conn:
|
|
append_event(conn, kind="activity_change", payload={
|
|
"entity_id": "you",
|
|
})
|
|
project(conn)
|
|
a = get_activity(conn, "you")
|
|
assert a is not None
|
|
assert a["container_id"] is None
|
|
assert a["slot"] is None
|
|
assert a["posture"] == ""
|
|
assert a["action"] == {}
|
|
assert a["attention"] == ""
|
|
assert a["holding"] == []
|
|
assert a["status"] == {}
|
|
|
|
|
|
def test_get_activity_returns_none_for_missing(tmp_path):
|
|
db = tmp_path / "t.db"
|
|
apply_migrations(db)
|
|
with open_db(db) as conn:
|
|
assert get_activity(conn, "ghost") is None
|
|
|
|
|
|
def test_scene_opened_marks_active_scene_id(tmp_path):
|
|
db = tmp_path / "t.db"
|
|
apply_migrations(db)
|
|
with open_db(db) as conn:
|
|
append_event(conn, kind="chat_created", payload=_chat_payload())
|
|
append_event(conn, kind="container_created", payload=_container_payload())
|
|
append_event(conn, kind="scene_opened", payload={
|
|
"chat_id": "chat_bot_a",
|
|
"container_id": 1,
|
|
"started_at": "2026-04-26T20:00:00+00:00",
|
|
"participants": ["you", "bot_a"],
|
|
})
|
|
project(conn)
|
|
|
|
s = active_scene(conn, "chat_bot_a")
|
|
assert s is not None
|
|
assert s["chat_id"] == "chat_bot_a"
|
|
assert s["container_id"] == 1
|
|
assert s["started_at"] == "2026-04-26T20:00:00+00:00"
|
|
assert s["ended_at"] is None
|
|
assert s["significance"] == 0
|
|
assert s["participants"] == ["you", "bot_a"]
|
|
|
|
chat = get_chat(conn, "chat_bot_a")
|
|
assert chat["active_scene_id"] == s["id"]
|
|
|
|
s2 = get_scene(conn, s["id"])
|
|
assert s2 is not None
|
|
assert s2["participants"] == ["you", "bot_a"]
|
|
|
|
|
|
def test_scene_closed_clears_active_scene_id_and_records_end(tmp_path):
|
|
db = tmp_path / "t.db"
|
|
apply_migrations(db)
|
|
with open_db(db) as conn:
|
|
append_event(conn, kind="chat_created", payload=_chat_payload())
|
|
append_event(conn, kind="container_created", payload=_container_payload())
|
|
append_event(conn, kind="scene_opened", payload={
|
|
"chat_id": "chat_bot_a",
|
|
"container_id": 1,
|
|
"started_at": "2026-04-26T20:00:00+00:00",
|
|
"participants": ["you", "bot_a"],
|
|
})
|
|
# The first scene insert will be id=1 (first row in scenes).
|
|
append_event(conn, kind="scene_closed", payload={
|
|
"scene_id": 1,
|
|
"ended_at": "2026-04-26T21:00:00+00:00",
|
|
"significance": 2,
|
|
})
|
|
project(conn)
|
|
|
|
assert active_scene(conn, "chat_bot_a") is None
|
|
chat = get_chat(conn, "chat_bot_a")
|
|
assert chat["active_scene_id"] is None
|
|
s = get_scene(conn, 1)
|
|
assert s["ended_at"] == "2026-04-26T21:00:00+00:00"
|
|
assert s["significance"] == 2
|
|
|
|
|
|
def test_scene_closed_significance_defaults_to_zero(tmp_path):
|
|
db = tmp_path / "t.db"
|
|
apply_migrations(db)
|
|
with open_db(db) as conn:
|
|
append_event(conn, kind="chat_created", payload=_chat_payload())
|
|
append_event(conn, kind="scene_opened", payload={
|
|
"chat_id": "chat_bot_a",
|
|
"started_at": "2026-04-26T20:00:00+00:00",
|
|
"participants": ["you", "bot_a"],
|
|
})
|
|
append_event(conn, kind="scene_closed", payload={
|
|
"scene_id": 1,
|
|
"ended_at": "2026-04-26T21:00:00+00:00",
|
|
})
|
|
project(conn)
|
|
s = get_scene(conn, 1)
|
|
assert s["significance"] == 0
|
|
assert s["ended_at"] == "2026-04-26T21:00:00+00:00"
|
|
|
|
|
|
def test_get_scene_returns_none_for_missing(tmp_path):
|
|
db = tmp_path / "t.db"
|
|
apply_migrations(db)
|
|
with open_db(db) as conn:
|
|
assert get_scene(conn, 999) is None
|
|
assert active_scene(conn, "chat_missing") is None
|
|
|
|
|
|
def test_schema_version_after_migration_is_7(tmp_path):
|
|
db = tmp_path / "t.db"
|
|
apply_migrations(db)
|
|
with open_db(db) as conn:
|
|
row = conn.execute(
|
|
"SELECT value FROM meta WHERE key = 'schema_version'"
|
|
).fetchone()
|
|
assert int(row[0]) == 7
|