Joseph Doherty
350 lines
11 KiB
Python
350 lines
11 KiB
Python
from __future__ import annotations
|
|
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
|
|
from chat.app import app
|
|
from chat.db.connection import open_db
|
|
from chat.eventlog.log import append_event
|
|
from chat.eventlog.projector import project
|
|
|
|
|
|
@pytest.fixture
|
|
def client(tmp_path, monkeypatch):
|
|
cfg = tmp_path / "config.toml"
|
|
cfg.write_text('featherless_api_key = "test"\n')
|
|
monkeypatch.setenv("CHAT_CONFIG_PATH", str(cfg))
|
|
db = tmp_path / "test.db"
|
|
monkeypatch.setenv("CHAT_DB_PATH", str(db))
|
|
with TestClient(app) as c:
|
|
if hasattr(app.state, "background_worker"):
|
|
app.state.background_worker.enabled = False
|
|
yield c
|
|
|
|
|
|
def _seed_bot_with_state(db: Path) -> None:
|
|
"""Seed a bot plus a chat, container, scene, edge, memory, and activity row."""
|
|
with open_db(db) as conn:
|
|
append_event(
|
|
conn,
|
|
kind="bot_authored",
|
|
payload={
|
|
"id": "bot_a",
|
|
"name": "BotA",
|
|
"persona": "thoughtful, observant",
|
|
"voice_samples": [],
|
|
"traits": ["shy"],
|
|
"backstory": "",
|
|
"initial_relationship_to_you": "coworker",
|
|
"kickoff_prose": "you stay late",
|
|
},
|
|
)
|
|
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",
|
|
"narrative_anchor": "Day 1",
|
|
"weather": "",
|
|
},
|
|
)
|
|
append_event(
|
|
conn,
|
|
kind="container_created",
|
|
payload={
|
|
"chat_id": "chat_bot_a",
|
|
"name": "office",
|
|
"type": "workplace",
|
|
"properties": {},
|
|
},
|
|
)
|
|
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"],
|
|
},
|
|
)
|
|
append_event(
|
|
conn,
|
|
kind="edge_update",
|
|
payload={
|
|
"source_id": "bot_a",
|
|
"target_id": "you",
|
|
"chat_id": "chat_bot_a",
|
|
"affinity_delta": 5,
|
|
"trust_delta": 2,
|
|
},
|
|
)
|
|
append_event(
|
|
conn,
|
|
kind="memory_written",
|
|
payload={
|
|
"owner_id": "bot_a",
|
|
"chat_id": "chat_bot_a",
|
|
"pov_summary": "Talked about her sister",
|
|
"witness_you": 1,
|
|
"witness_host": 1,
|
|
"witness_guest": 0,
|
|
"significance": 2,
|
|
},
|
|
)
|
|
append_event(
|
|
conn,
|
|
kind="activity_change",
|
|
payload={
|
|
"entity_id": "bot_a",
|
|
"posture": "sitting",
|
|
"action": {"verb": "writing"},
|
|
},
|
|
)
|
|
project(conn)
|
|
|
|
|
|
def test_reset_purges_state_but_preserves_identity(client, tmp_path):
|
|
_seed_bot_with_state(tmp_path / "test.db")
|
|
response = client.post(
|
|
"/bots/bot_a/reset",
|
|
data={"confirm_name": "BotA"},
|
|
follow_redirects=False,
|
|
)
|
|
assert response.status_code == 303
|
|
assert response.headers["location"] == "/bots"
|
|
|
|
with open_db(tmp_path / "test.db") as conn:
|
|
# Identity preserved.
|
|
bot = conn.execute(
|
|
"SELECT id, name, kickoff_prose, initial_relationship_to_you "
|
|
"FROM bots WHERE id = 'bot_a'"
|
|
).fetchone()
|
|
assert bot is not None
|
|
assert bot[1] == "BotA"
|
|
assert bot[2] == "you stay late"
|
|
assert bot[3] == "coworker"
|
|
|
|
# State purged.
|
|
assert conn.execute(
|
|
"SELECT COUNT(*) FROM chats WHERE host_bot_id = 'bot_a'"
|
|
).fetchone()[0] == 0
|
|
assert conn.execute(
|
|
"SELECT COUNT(*) FROM scenes WHERE chat_id = 'chat_bot_a'"
|
|
).fetchone()[0] == 0
|
|
assert conn.execute(
|
|
"SELECT COUNT(*) FROM containers WHERE chat_id = 'chat_bot_a'"
|
|
).fetchone()[0] == 0
|
|
assert conn.execute(
|
|
"SELECT COUNT(*) FROM chat_state WHERE chat_id = 'chat_bot_a'"
|
|
).fetchone()[0] == 0
|
|
assert conn.execute(
|
|
"SELECT COUNT(*) FROM memories WHERE owner_id = 'bot_a'"
|
|
).fetchone()[0] == 0
|
|
assert conn.execute(
|
|
"SELECT COUNT(*) FROM edges WHERE source_id = 'bot_a' OR target_id = 'bot_a'"
|
|
).fetchone()[0] == 0
|
|
assert conn.execute(
|
|
"SELECT COUNT(*) FROM activity WHERE entity_id = 'bot_a'"
|
|
).fetchone()[0] == 0
|
|
|
|
# Event log records the bot_reset event.
|
|
assert conn.execute(
|
|
"SELECT COUNT(*) FROM event_log WHERE kind = 'bot_reset'"
|
|
).fetchone()[0] == 1
|
|
|
|
|
|
def test_reset_400_when_confirm_name_mismatch(client, tmp_path):
|
|
_seed_bot_with_state(tmp_path / "test.db")
|
|
response = client.post(
|
|
"/bots/bot_a/reset",
|
|
data={"confirm_name": "WrongName"},
|
|
follow_redirects=False,
|
|
)
|
|
assert response.status_code == 400
|
|
|
|
|
|
def test_reset_404_when_bot_missing(client):
|
|
response = client.post(
|
|
"/bots/no_such/reset",
|
|
data={"confirm_name": "Anything"},
|
|
follow_redirects=False,
|
|
)
|
|
assert response.status_code == 404
|
|
|
|
|
|
def test_bot_list_renders_reset_form(client, tmp_path):
|
|
_seed_bot_with_state(tmp_path / "test.db")
|
|
response = client.get("/bots")
|
|
assert response.status_code == 200
|
|
assert "Reset" in response.text
|
|
assert "confirm_name" in response.text
|
|
|
|
|
|
def _seed_two_bots_with_guest_link(
|
|
db: Path, *, extra_events: list[dict] | None = None
|
|
) -> None:
|
|
"""Seed bot_a + bot_b, each hosting their own chat, with bot_b a guest in chat_bot_a.
|
|
|
|
``extra_events`` is appended after the guest_added event and projected
|
|
together with the rest of the seed (so handlers run only once per event).
|
|
"""
|
|
with open_db(db) as conn:
|
|
# bot_a + its chat
|
|
append_event(
|
|
conn,
|
|
kind="bot_authored",
|
|
payload={
|
|
"id": "bot_a",
|
|
"name": "BotA",
|
|
"persona": "thoughtful",
|
|
"voice_samples": [],
|
|
"traits": [],
|
|
"backstory": "",
|
|
"initial_relationship_to_you": "coworker",
|
|
"kickoff_prose": "",
|
|
},
|
|
)
|
|
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",
|
|
"narrative_anchor": "Day 1",
|
|
"weather": "",
|
|
},
|
|
)
|
|
# bot_b + its own chat
|
|
append_event(
|
|
conn,
|
|
kind="bot_authored",
|
|
payload={
|
|
"id": "bot_b",
|
|
"name": "BotB",
|
|
"persona": "curious",
|
|
"voice_samples": [],
|
|
"traits": [],
|
|
"backstory": "",
|
|
"initial_relationship_to_you": "friend",
|
|
"kickoff_prose": "",
|
|
},
|
|
)
|
|
append_event(
|
|
conn,
|
|
kind="chat_created",
|
|
payload={
|
|
"id": "chat_bot_b",
|
|
"host_bot_id": "bot_b",
|
|
"initial_time": "2026-04-26T20:00:00+00:00",
|
|
"narrative_anchor": "Day 1",
|
|
"weather": "",
|
|
},
|
|
)
|
|
# bot_b joins chat_bot_a as a guest.
|
|
append_event(
|
|
conn,
|
|
kind="guest_added",
|
|
payload={
|
|
"chat_id": "chat_bot_a",
|
|
"guest_bot_id": "bot_b",
|
|
},
|
|
)
|
|
for ev in extra_events or []:
|
|
append_event(conn, kind=ev["kind"], payload=ev["payload"])
|
|
project(conn)
|
|
|
|
|
|
def test_reset_clears_guest_reference_in_other_chats(client, tmp_path):
|
|
db = tmp_path / "test.db"
|
|
_seed_two_bots_with_guest_link(db)
|
|
|
|
# Sanity-check the seed: bot_b is the guest in bot_a's chat.
|
|
from chat.state.world import get_chat
|
|
with open_db(db) as conn:
|
|
assert get_chat(conn, "chat_bot_a")["guest_bot_id"] == "bot_b"
|
|
assert get_chat(conn, "chat_bot_b") is not None
|
|
|
|
response = client.post(
|
|
"/bots/bot_b/reset",
|
|
data={"confirm_name": "BotB"},
|
|
follow_redirects=False,
|
|
)
|
|
assert response.status_code == 303
|
|
|
|
with open_db(db) as conn:
|
|
# The guest reference in bot_a's chat is cleared.
|
|
chat_a = get_chat(conn, "chat_bot_a")
|
|
assert chat_a is not None
|
|
assert chat_a["guest_bot_id"] is None
|
|
|
|
# bot_b's own chat is gone (Phase 1 host purge behavior).
|
|
assert get_chat(conn, "chat_bot_b") is None
|
|
|
|
# bot_a is untouched.
|
|
assert conn.execute(
|
|
"SELECT COUNT(*) FROM bots WHERE id = 'bot_a'"
|
|
).fetchone()[0] == 1
|
|
|
|
|
|
def test_reset_purges_guest_memories_from_other_chats(client, tmp_path):
|
|
db = tmp_path / "test.db"
|
|
_seed_two_bots_with_guest_link(
|
|
db,
|
|
extra_events=[
|
|
# bot_b is a guest in chat_bot_a and remembers things from there.
|
|
{
|
|
"kind": "memory_written",
|
|
"payload": {
|
|
"owner_id": "bot_b",
|
|
"chat_id": "chat_bot_a",
|
|
"pov_summary": "Met BotA; she was tense.",
|
|
"witness_you": 1,
|
|
"witness_host": 1,
|
|
"witness_guest": 1,
|
|
"significance": 3,
|
|
},
|
|
},
|
|
# And a memory from bot_b's own chat for good measure.
|
|
{
|
|
"kind": "memory_written",
|
|
"payload": {
|
|
"owner_id": "bot_b",
|
|
"chat_id": "chat_bot_b",
|
|
"pov_summary": "A quiet evening at home.",
|
|
"witness_you": 1,
|
|
"witness_host": 1,
|
|
"witness_guest": 0,
|
|
"significance": 1,
|
|
},
|
|
},
|
|
],
|
|
)
|
|
|
|
with open_db(db) as conn:
|
|
# Sanity: bot_b owns 2 memories pre-reset, one in each chat.
|
|
assert conn.execute(
|
|
"SELECT COUNT(*) FROM memories WHERE owner_id = 'bot_b'"
|
|
).fetchone()[0] == 2
|
|
|
|
response = client.post(
|
|
"/bots/bot_b/reset",
|
|
data={"confirm_name": "BotB"},
|
|
follow_redirects=False,
|
|
)
|
|
assert response.status_code == 303
|
|
|
|
with open_db(db) as conn:
|
|
# ALL of bot_b's memories are gone, including the cross-chat one in chat_bot_a.
|
|
assert conn.execute(
|
|
"SELECT COUNT(*) FROM memories WHERE owner_id = 'bot_b'"
|
|
).fetchone()[0] == 0
|
|
assert conn.execute(
|
|
"SELECT COUNT(*) FROM memories WHERE owner_id = 'bot_b' AND chat_id = 'chat_bot_a'"
|
|
).fetchone()[0] == 0
|