new RPG system

tests/test_session_snapshots.py

@@ -0,0 +1,80 @@
+import asyncio
+import json
+
+import pytest
+
+import database.db as dbmod
+import services.memory as memory
+
+
+@pytest.fixture
+def snapshot_db(tmp_path, monkeypatch):
+    db_file = tmp_path / "snapshots.db"
+    monkeypatch.setenv("DB_PATH", str(db_file))
+    monkeypatch.setattr(dbmod, "DB_PATH", str(db_file))
+    monkeypatch.setattr(memory, "DB_PATH", str(db_file))
+    asyncio.run(dbmod.init_db())
+    return str(db_file)
+
+
+def test_snapshot_save_restore_and_delete_rollback(snapshot_db):
+    asyncio.run(_test_snapshot_save_restore_and_delete_rollback())
+
+
+async def _test_snapshot_save_restore_and_delete_rollback():
+    sid = "sess_snap_test"
+    await memory.get_or_create_session(sid, "default")
+    await memory.update_session_rpg(sid, True)
+    await memory.update_session_facts(sid, json.dumps([{"text": "Fact A", "rp_day": "день 1"}]))
+    await memory.update_session_plot_arc(
+        sid, json.dumps({"title": "Arc", "steps": [{"id": "s1", "title": "One"}], "current_step_index": 0})
+    )
+    await memory.set_session_affinity(sid, 3)
+    await memory.upsert_quest(sid, "Quest one", "active")
+
+    u1 = await memory.add_message(sid, "user", "hello")
+    await memory.save_state_snapshot(sid, u1)
+
+    await memory.update_session_facts(sid, json.dumps([{"text": "Fact B", "rp_day": "день 2"}]))
+    await memory.set_session_affinity(sid, 9)
+    a1 = await memory.add_message(sid, "assistant", "hi")
+    await memory.save_state_snapshot(sid, a1)
+
+    session = await memory.get_session(sid)
+    assert json.loads(session["facts_json"])[0]["text"] == "Fact B"
+    assert session["affinity"] == 9
+
+    await memory.delete_message_and_following(sid, a1)
+
+    session = await memory.get_session(sid)
+    assert json.loads(session["facts_json"])[0]["text"] == "Fact A"
+    assert session["affinity"] == 3
+    quests = await memory.get_quests(sid)
+    assert len(quests) == 1
+
+
+def test_fork_uses_snapshot_not_current_state(snapshot_db):
+    asyncio.run(_test_fork_uses_snapshot_not_current_state())
+
+
+async def _test_fork_uses_snapshot_not_current_state():
+    sid = "sess_fork_src"
+    await memory.get_or_create_session(sid, "default")
+    await memory.update_session_rpg(sid, True)
+
+    u1 = await memory.add_message(sid, "user", "start")
+    await memory.update_session_facts(sid, json.dumps([{"text": "Early", "rp_day": "1"}]))
+    await memory.save_state_snapshot(sid, u1)
+
+    a1 = await memory.add_message(sid, "assistant", "reply")
+    await memory.update_session_facts(sid, json.dumps([{"text": "Late", "rp_day": "2"}]))
+    await memory.save_state_snapshot(sid, a1)
+
+    new_id = await memory.fork_session(sid, u1)
+    assert new_id
+    forked = await memory.get_session(new_id)
+    assert json.loads(forked["facts_json"])[0]["text"] == "Early"
+
+    hist = await memory.get_history(new_id)
+    roles = [m["role"] for m in hist if m["role"] != "system"]
+    assert roles == ["user"]