Jesse.Markowitz/interactive-story
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

M2: cut the hosted product away from the local one

Browse Source 
94 files, +1,395 -6,578. Three files are new; twenty-four are gone. The
milestone is subtraction, and what is left is the single-user local
storyteller the specification describes.

Removed in full: campaign scripting and its QuickJS sandbox; multi-user
accounts, guest sessions, login, registration and the shared demo key;
the visitor-analytics tables, dashboard and page beacon; the access log
of sign-ins, addresses and devices; per-IP and per-user rate limiting
and quotas; Render deployment config; Postgres and psycopg; cloud
inference providers, the API-key field and the key encryption that
existed to store it; session-cookie signing. None of it was hidden
behind a flag — the routes are gone and answer 404.

Two things were kept that the brief allowed keeping. The `users` table
and its foreign keys stay as an internal ownership detail, because
rewriting them out means a migration across most of the schema to
delete a column that costs nothing; nothing creates a second user and
no request carries an identity. Five inert tables and four inert
columns stay for the same reason, so an M1 campaign database opens
unchanged.

The one addition is app/endpoints.py, which decides where a story may
be sent. Loopback, RFC1918, link-local, unique-local and CGNAT — an
explicit allowlist of networks, not a guess at what `ipaddress` means
by "private", which calls the documentation ranges private and IPv6
loopback reserved. Every address a hostname resolves to must be in it,
so a split answer does not squeak through, and the rule runs both when
the endpoint is saved and before every outbound request, because a name
that resolved to the LAN this morning can resolve elsewhere this
afternoon. Known cloud hosts are named in the refusal so the error says
why rather than looking like broken DNS. TLS is never traded against
it: M1's shared trust context is intact on all four clients and there
is no way to skip verification.

The hardcoded 120-second model timeout is now a setting. That was not
theoretical — on this GPU-less four-core host a cold load of
qwen2.5:3b-instruct took 648.9 seconds to produce the first turn, while
turns 2 to 5 of the same campaign took 3.6 to 13.1. Connect stays short
at 10s so a wrong address still fails fast; the read timeout defaults
to 300s and is bounded at 3600, because "wait longer" must stay a
number.

Two defects found while testing and fixed here. An unknown /api path
fell through the SPA catch-all and came back as HTML with status 200,
so a client asking for JSON parsed a web page instead of learning the
route was gone. And AIDND_CORS_ORIGINS accepted "*", which on an
unauthenticated loopback API would hand every page on the Internet a
write handle on the campaign database; it now refuses to start.

Verified rather than assumed. Offline, on a network with no route out
and no DNS: five turns, retry with both takes retained, restart with an
identical transcript digest, a failed model call leaving the accepted
AI-turn count untouched, and a capture with zero non-loopback unicast
packets. Against a real second machine on the LAN over HTTPS with a
private CA: four turns, restart, and a capture showing 289 packets to
the approved host, 344 loopback, zero anywhere else, zero DNS queries.
Cloud and public endpoints refused with their reasons; no API key
settable; every removed route 404.

604 backend tests pass, down from 648 by the fifteen retired with the
subsystems they tested and up by the twenty-nine added for the endpoint
policy and the removed surface. The scripting tests were not deleted:
eight files used a JavaScript counter as instrumentation for the state
snapshot and rollback machinery, which M2 does not touch, so the
counter moved to the world-state engine and those tests still assert
what they always did. Frontend lint and build are clean; the image
builds, and its wheel-building stage is gone with quickjs.

No M3 work. Undo is still destructive and there is still no Redo.
This commit is contained in:
JesseMarkowitz
2026-09-02 11:27:14 -04:00
parent 1a28a9a708
commit 8c65ae99de
94 changed files with 1384 additions and 6567 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/tests/fakes.py
+41
View File
@@ -39,3 +39,44 @@ class ScriptedProvider:
if isinstance(reply, Exception):
raise reply
yield ("text", reply)
# ---------------------------------------------------------------------------
# Deterministic per-turn state instrumentation
# ---------------------------------------------------------------------------
# Several tests need a value that changes by a fixed amount on every turn, so
# that a rollback failure is arithmetic rather than a judgement call: if a take
# stacks instead of replacing, the total is off by exactly one turn's worth.
#
# That instrumentation used to be a JavaScript `output` hook doing
# `state.gold += 10` in the QuickJS sandbox. M2 removed campaign scripting, and
# the tests below are not about scripting — they are about the state snapshot,
# rollback, and branch-isolation machinery in `attempts.py` and `tree.py`,
# which is unchanged.
#
# The counter therefore moved to the world-state engine, which is a real
# remaining product path: the model emits a ```state delta block, the referee
# applies it, and the result lands in `adventure.world_state`. The fake
# provider decides what the model "emits", so it is exactly as deterministic as
# the script was, and it exercises production code rather than a test hook.
#: A schema with a plain unbounded counter. No `max_delta_per_turn` and no
#: `cooldown`, so every +10 is applied in full, every turn.
GOLD_SCHEMA = {
"player": {
"hp": {"min": 0, "max": 100, "initial": 100},
"gold": {"min": 0, "max": 1_000_000, "initial": 0},
}
}
GOLD_PER_TURN = 10
def gold_reply(text: str, amount: int = GOLD_PER_TURN) -> str:
"""A model reply that narrates `text` and banks `amount` gold."""
return f'{text}\n```state\n{{"player.gold": {amount}}}\n```'
def gold_replies(prefix: str = "Take", count: int = 40) -> list[str]:
"""`count` numbered replies, each banking one turn's worth of gold."""
return [gold_reply(f"{prefix} {n}.") for n in range(1, count + 1)]
backend/tests/test_accesslog.py
-208
View File
@@ -1,208 +0,0 @@
"""The access log: app/accesslog.py and GET /api/analytics/access.
This is the half of the analytics work that identifies people on purpose,
so these tests pin the details that would quietly make it wrong. The
address recorded must be the hardened one, not a header a client chose.
Session rows must be thinned instead of written on every page load. And a
row must outlive the account it describes, because guest cleanup deletes
accounts on a schedule, and a log that deletes itself is not a log.
python -m pytest tests/test_accesslog.py -v
"""
import pytest
from fastapi import Depends
from fastapi.testclient import TestClient
from app import accesslog, auth, limits, models, security
from app.database import Base, SessionLocal, engine, get_db
from app.main import app
EDGE = "198.51.100.77" # what the trusted proxy appended
SPOOF = "10.0.0.1" # what a client put in front of it
@pytest.fixture(autouse=True)
def clean_state():
accesslog._last_session.clear()
yield
accesslog._last_session.clear()
@pytest.fixture()
def client(monkeypatch):
Base.metadata.create_all(bind=engine)
setup = SessionLocal()
owner = models.User(is_guest=False, email="owner@example.com")
member = models.User(
is_guest=False, email="player@example.com",
password_hash=security.hash_password("hunter2long"),
)
setup.add_all([owner, member])
setup.commit()
ids = {"owner": owner.id, "member": member.id}
setup.close()
monkeypatch.setattr(limits, "rate_limit", lambda *a, **k: None)
monkeypatch.setattr(limits, "check_login_allowed", lambda *a, **k: None)
monkeypatch.setattr(auth, "MULTI_USER", True)
monkeypatch.setattr(auth, "ANALYTICS_EMAILS", {"owner@example.com"})
# /auth/me resolves its own session, so the cookie flow below is the real
# one. Every other endpoint goes through get_current_user, and `act_as`
# decides who that is.
acting = {"id": ids["owner"]}
def _current(db=Depends(get_db)):
return db.get(models.User, acting["id"])
app.dependency_overrides[auth.get_current_user] = _current
try:
test_client = TestClient(app)
test_client.ids = ids
test_client.act_as = lambda user_id: acting.update(id=user_id)
yield test_client
finally:
app.dependency_overrides.clear()
Base.metadata.drop_all(bind=engine)
def visit(client, ip=EDGE, ua="Mozilla/5.0 (Windows NT 10.0; Win64; x64)"):
return client.get(
"/api/auth/me",
headers={"x-forwarded-for": f"{SPOOF}, {ip}", "user-agent": ua},
)
def rows(kind=None):
db = SessionLocal()
try:
query = db.query(models.AccessEvent).order_by(models.AccessEvent.id)
if kind:
query = query.filter_by(kind=kind)
return query.all()
finally:
db.close()
def read_log(client, **params):
return client.get("/api/analytics/access", params=params)
# ---------- Writing ----------
def test_a_new_session_is_logged(client):
visit(client)
logged = rows()
assert len(logged) == 1
entry = logged[0]
assert entry.kind == accesslog.SESSION
assert entry.is_guest and entry.who.startswith("Guest #")
assert entry.device == "desktop"
def test_the_address_is_the_hardened_one_not_the_clients(client):
visit(client)
# The client prepended its own value. Only the hop the edge appended counts.
# Recording the leftmost value would make every row forgeable, which is
# worse for a log than having no log at all.
assert rows()[0].ip == EDGE
def test_session_rows_are_thinned_to_one_per_day_per_address(client):
for _ in range(4):
visit(client)
assert len(rows(accesslog.SESSION)) == 1
def test_a_changed_address_writes_a_new_row(client):
visit(client)
visit(client, ip="203.0.113.9")
logged = rows(accesslog.SESSION)
assert [entry.ip for entry in logged] == [EDGE, "203.0.113.9"]
# Same session throughout, so both rows name the same visitor.
assert logged[0].who == logged[1].who
def test_sign_in_and_failure_are_both_logged(client):
client.post("/api/auth/login", json={"email": "player@example.com", "password": "wrong"},
headers={"x-forwarded-for": EDGE})
client.post("/api/auth/login", json={"email": "player@example.com", "password": "hunter2long"},
headers={"x-forwarded-for": EDGE})
kinds = [entry.kind for entry in rows()]
assert accesslog.LOGIN_FAILED in kinds and accesslog.LOGIN in kinds
failure = rows(accesslog.LOGIN_FAILED)[0]
# This records the address that was tried, not the account it belongs to.
# A failed attempt against an address with no matching account is
# exactly what this row exists to capture.
assert failure.who == "player@example.com"
assert failure.user_id is None
assert rows(accesslog.LOGIN)[0].user_id == client.ids["member"]
def test_registering_is_logged_against_the_upgraded_account(client):
visit(client) # creates the guest whose session then registers
client.act_as(rows()[0].user_id)
client.post("/api/auth/register", json={"email": "new@example.com", "password": "hunter2long"})
entry = rows(accesslog.REGISTER)[0]
assert entry.who == "new@example.com" and not entry.is_guest
def test_a_row_outlives_the_account_it_describes(client):
visit(client)
entry = rows()[0]
db = SessionLocal()
try:
db.delete(db.get(models.User, entry.user_id))
db.commit()
finally:
db.close()
# There is no foreign key, and `who` is a snapshot. Guest cleanup deletes
# accounts on a schedule, and a log that vanishes along with them is not
# a log.
survivor = rows()[0]
assert survivor.who == entry.who and survivor.ip == EDGE
def test_a_long_user_agent_is_truncated(client):
visit(client, ua="Mozilla/" + "x" * 500)
assert len(rows()[0].user_agent) == accesslog.MAX_UA
def test_a_logging_failure_does_not_break_the_request(client, monkeypatch):
monkeypatch.setattr(accesslog, "_client_ip", lambda request: 1 / 0)
# The log observes sign-in. A logging failure must not block the request.
assert visit(client).status_code == 200
# ---------- Reading ----------
def test_the_log_is_invisible_to_everyone_but_the_owner(client):
visit(client)
assert read_log(client).status_code == 200
client.act_as(client.ids["member"])
assert read_log(client).status_code == 404
def test_the_log_reads_newest_first_and_pages_backwards(client):
for index in range(5):
visit(client, ip=f"203.0.113.{index}")
first = read_log(client, limit=2).json()
assert [event["ip"] for event in first["events"]] == ["203.0.113.4", "203.0.113.3"]
assert first["has_more"]
older = read_log(client, limit=2, before_id=first["events"][-1]["id"]).json()
assert [event["ip"] for event in older["events"]] == ["203.0.113.2", "203.0.113.1"]
def test_the_log_filters_by_kind_and_searches(client):
visit(client)
client.post("/api/auth/login", json={"email": "player@example.com", "password": "hunter2long"},
headers={"x-forwarded-for": "203.0.113.44"})
assert len(read_log(client, kind="login").json()["events"]) == 1
by_email = read_log(client, q="player@example.com").json()["events"]
assert len(by_email) == 1 and by_email[0]["kind"] == "login"
by_ip = read_log(client, q="203.0.113.44").json()["events"]
assert len(by_ip) == 1
assert read_log(client, q="nobody@example.com").json()["events"] == []
backend/tests/test_action_paging.py
-1
View File
@@ -48,7 +48,6 @@ def client(monkeypatch):
adv_id, user_id = adventure.id, user.id
setup.close()
monkeypatch.setattr(limits, "rate_limit", lambda *a, **k: None)
monkeypatch.setattr(limits, "check_row_cap", lambda *a, **k: None)
def _current_user(db=Depends(get_db)):
backend/tests/test_analytics.py
-385
View File
@@ -1,385 +0,0 @@
"""Visit analytics: app/analytics.py and the two endpoints in front of it.
This file tests three things, and the rest is arithmetic. The counters must
survive the buffer/UPSERT round trip: a flush adds to what is already
stored instead of replacing it, or every number would show only the last
minute. The funnel counts people rather than clicks, which is the only
reason the visitor-day table exists. The gate holds: a stranger cannot read
the dashboard, and cannot inflate what it reports beyond hitting the page.
python -m pytest tests/test_analytics.py -v
"""
from datetime import timedelta
import pytest
from fastapi import Depends
from fastapi.testclient import TestClient
from app import analytics, auth, limits, models
from app.database import Base, SessionLocal, engine, get_db
from app.main import app
@pytest.fixture(autouse=True)
def clean_buffer():
"""The buffer is process-wide, so a test that leaves counts in it would
show up inside the next one's flush."""
analytics._counts.clear()
analytics._visits.clear()
analytics._labels_seen.clear()
yield
analytics._counts.clear()
analytics._visits.clear()
analytics._labels_seen.clear()
@pytest.fixture()
def db():
Base.metadata.create_all(bind=engine)
session = SessionLocal()
try:
yield session
finally:
session.close()
Base.metadata.drop_all(bind=engine)
def counter(db, metric, label):
row = (
db.query(models.AnalyticsDaily)
.filter_by(metric=metric, label=label)
.one_or_none()
)
return row.hits if row else 0
def make_user(db, email=None):
user = models.User(is_guest=email is None, email=email)
db.add(user)
db.commit()
return user
# ---------- The buffer and its flush ----------
def test_counts_accumulate_across_flushes(db):
analytics.record(analytics.M_PAGE, "/")
analytics.record(analytics.M_PAGE, "/")
analytics.flush(db)
analytics.record(analytics.M_PAGE, "/")
analytics.flush(db)
# The second flush has to find the existing row and add to it. Replacing it
# would leave every counter showing only the newest minute of traffic.
assert counter(db, analytics.M_PAGE, "/") == 3
def test_flush_is_a_no_op_when_nothing_happened(db):
analytics.flush(db)
assert db.query(models.AnalyticsDaily).count() == 0
def test_a_failed_flush_keeps_the_counts(db, monkeypatch):
analytics.record(analytics.M_PAGE, "/")
monkeypatch.setattr(analytics, "_write_counts", lambda *a: 1 / 0)
analytics.flush(db) # must not raise
monkeypatch.undo()
analytics.flush(db)
assert counter(db, analytics.M_PAGE, "/") == 1
def test_label_cardinality_is_capped(db):
for i in range(analytics.MAX_LABELS_PER_METRIC + 25):
analytics.record(analytics.M_REFERRER, f"host{i}.example")
analytics.flush(db)
labels = db.query(models.AnalyticsDaily).filter_by(metric=analytics.M_REFERRER).count()
# Everything past the cap is folded into one bucket, so a referrer flood
# cannot create unlimited rows.
assert labels == analytics.MAX_LABELS_PER_METRIC + 1
assert counter(db, analytics.M_REFERRER, analytics.OTHER) == 25
# ---------- Visitors ----------
def test_visitor_id_is_stable_and_keyed(db, monkeypatch):
user = make_user(db)
handle = analytics.visitor_id(user)
assert handle == analytics.visitor_id(user) # a returning visitor
assert handle != analytics.visitor_id(make_user(db)) # is still one visitor
assert len(handle) == 32 and int(handle, 16) >= 0 # opaque hex, not an id
# Keyed on the app secret, not a bare hash of the user id. Otherwise
# anyone holding this table could rebuild the mapping by hashing
# sequential ids.
monkeypatch.setattr(analytics.security, "SECRET_KEY", b"a-different-secret")
assert analytics.visitor_id(user) != handle
def test_a_repeat_visitor_is_new_only_once(db):
user = make_user(db)
analytics.record_visit(user)
analytics.flush(db)
rows = db.query(models.AnalyticsVisitorDay).all()
assert len(rows) == 1 and rows[0].is_new
# Same visitor, a later day: seen before, so not new. The row is not
# merged into the first day's row either.
tomorrow = (models.utcnow().date() + timedelta(days=1)).isoformat()
analytics._visits[(tomorrow, analytics.visitor_id(user))] = set()
analytics.flush(db)
rows = db.query(models.AnalyticsVisitorDay).order_by(models.AnalyticsVisitorDay.day).all()
assert [row.is_new for row in rows] == [True, False]
def test_one_row_per_visitor_per_day_however_much_they_do(db):
user = make_user(db)
for _ in range(5):
analytics.record_event(analytics.EV_ADVENTURE, user)
analytics.flush(db)
assert db.query(models.AnalyticsVisitorDay).count() == 1
assert counter(db, analytics.M_EVENT, analytics.EV_ADVENTURE) == 5
def test_funnel_flags_only_ever_turn_on(db):
user = make_user(db)
analytics.record_event(analytics.EV_TURN, user)
analytics.flush(db)
# A later visit that reaches no funnel step must not clear the earlier one.
analytics.record_visit(user)
analytics.flush(db)
row = db.query(models.AnalyticsVisitorDay).one()
assert row.played and not row.created
def test_purge_drops_only_rows_past_the_horizon(db):
old = (models.utcnow().date() - timedelta(days=analytics.RETENTION_DAYS + 1)).isoformat()
db.add(models.AnalyticsVisitorDay(day=old, visitor="a" * 32))
db.add(models.AnalyticsVisitorDay(day=analytics._today(), visitor="b" * 32))
db.commit()
assert analytics.purge_old_visitor_days(db) == 1
assert [r.visitor for r in db.query(models.AnalyticsVisitorDay)] == ["b" * 32]
# ---------- Normalizing what a browser claims ----------
@pytest.mark.parametrize("path, expected", [
("/", "/"),
("/adventures", "/adventures"),
("/adventures/", "/adventures"),
("/play/12?x=1", "/play/:id"),
("/scenarios/9#top", "/scenarios/:id"),
("/wp-admin", "(other)"),
("/play/../../etc", "(other)"),
("", "/"),
])
def test_route_normalization(path, expected):
assert analytics.normalize_route(path) == expected
@pytest.mark.parametrize("referrer, expected", [
("", "(direct)"),
("https://news.ycombinator.com/item?id=1", "news.ycombinator.com"),
("https://www.google.com/", "google.com"),
("https://ai-dnd.example/scenarios", ""), # our own host: not a referral
("javascript:alert(1)", "(other)"),
("https://" + "x" * 200 + ".com", "(other)"),
])
def test_referrer_normalization(referrer, expected):
assert analytics.normalize_referrer(referrer, "ai-dnd.example") == expected
@pytest.mark.parametrize("ua, expected", [
("Mozilla/5.0 (iPhone; CPU iPhone OS 17_0) AppleWebKit", "mobile"),
("Mozilla/5.0 (iPad; CPU OS 17_0) AppleWebKit", "tablet"),
("Mozilla/5.0 (Windows NT 10.0; Win64; x64)", "desktop"),
("Googlebot/2.1", "bot"),
("", "(unknown)"),
])
def test_device_detection(ua, expected):
assert analytics.device_of(ua) == expected
def test_only_iso_looking_country_headers_are_trusted():
assert analytics.country_of({"cf-ipcountry": "de"}) == "DE"
assert analytics.country_of({"cf-ipcountry": "Norway"}) == analytics.UNKNOWN
assert analytics.country_of({"cf-ipcountry": "XX"}) == analytics.UNKNOWN
assert analytics.country_of({}) == analytics.UNKNOWN
def test_error_labels_use_the_route_not_the_path():
class Route:
path = "/api/adventures/{adventure_id}"
assert analytics.api_route_label({"route": Route()}, 500) == "500 /api/adventures/{adventure_id}"
# An unmatched path is entirely attacker-chosen, so it never becomes a label.
assert analytics.api_route_label({}, 404) == "404 (unmatched)"
# ---------- The summary ----------
def test_summary_counts_people_once_per_step(db):
one, two = make_user(db), make_user(db)
for _ in range(3):
analytics.record_event(analytics.EV_SCENARIO_OPEN, one)
analytics.record_event(analytics.EV_TURN, one)
analytics.record_event(analytics.EV_SCENARIO_OPEN, two)
result = analytics.summary(db, days=7)
steps = {row["step"]: row["count"] for row in result["funnel"]}
assert steps["Visited"] == 2
assert steps["Opened a scenario"] == 2
assert steps["Played a turn"] == 1 # not 3, because one person made three turns
assert steps["Signed up"] == 0
# Raw event totals still count every occurrence.
assert result["totals"]["turns"] == 3
assert result["totals"]["visitors"] == 2
def test_summary_series_covers_every_day_including_empty_ones(db):
analytics.record(analytics.M_PAGE, "/")
result = analytics.summary(db, days=7)
assert len(result["series"]) == 7
assert result["series"][-1]["day"] == models.utcnow().date().isoformat()
assert result["series"][-1]["pageviews"] == 1
assert result["series"][0]["pageviews"] == 0
def test_summary_flushes_before_reading(db):
analytics.record(analytics.M_EVENT, analytics.EV_TURN)
# Never flushed by hand: the dashboard must not be up to a minute stale.
assert analytics.summary(db, days=1)["totals"]["turns"] == 1
def test_summary_reports_pages_referrers_and_errors(db):
analytics.record(analytics.M_PAGE, "/play/:id", n=4)
analytics.record(analytics.M_REFERRER, "news.ycombinator.com", n=2)
analytics.record(analytics.M_ERROR, "500 /api/adventures/{adventure_id}")
result = analytics.summary(db, days=30)
assert result["pages"][0] == {"label": "/play/:id", "hits": 4}
assert result["referrers"][0]["label"] == "news.ycombinator.com"
assert result["totals"]["errors"] == 1
# ---------- The endpoints ----------
@pytest.fixture()
def client(monkeypatch):
Base.metadata.create_all(bind=engine)
setup = SessionLocal()
visitor = models.User(is_guest=True)
owner = models.User(is_guest=False, email="owner@example.com")
setup.add_all([visitor, owner])
setup.commit()
ids = {"visitor": visitor.id, "owner": owner.id}
setup.close()
monkeypatch.setattr(limits, "rate_limit", lambda *a, **k: None)
# Multi-user is what makes the gate mean anything: local mode trusts
# whoever is at the keyboard, because it is the operator's own machine.
monkeypatch.setattr(auth, "MULTI_USER", True)
monkeypatch.setattr(auth, "ANALYTICS_EMAILS", {"owner@example.com"})
current = {"id": ids["visitor"]}
def _current_user(db=Depends(get_db)):
return db.get(models.User, current["id"])
app.dependency_overrides[auth.get_current_user] = _current_user
monkeypatch.setattr(
auth, "resolve_session_user", lambda request, db: db.get(models.User, current["id"])
)
try:
client = TestClient(app)
client.ids, client.current = ids, current
yield client
finally:
app.dependency_overrides.clear()
Base.metadata.drop_all(bind=engine)
def read_summary(client, days=30):
return client.get(f"/api/analytics/summary?days={days}")
def test_dashboard_is_invisible_to_everyone_but_the_owner(client):
assert read_summary(client).status_code == 404
client.current["id"] = client.ids["owner"]
assert read_summary(client).status_code == 200
def test_collect_records_a_pageview_and_the_visit(client):
resp = client.post("/api/analytics/collect", json={"path": "/play/7", "first": True,
"referrer": "https://news.ycombinator.com/"})
assert resp.status_code == 204
client.current["id"] = client.ids["owner"]
body = read_summary(client).json()
assert body["pages"][0] == {"label": "/play/:id", "hits": 1}
assert body["referrers"][0]["label"] == "news.ycombinator.com"
assert body["totals"]["visitors"] == 1
def test_referrer_and_device_are_recorded_once_per_visit_not_per_view(client):
for path in ("/", "/scenarios", "/adventures"):
client.post("/api/analytics/collect", json={"path": path, "first": path == "/"})
client.current["id"] = client.ids["owner"]
body = read_summary(client).json()
assert body["totals"]["pageviews"] == 3
# Three views, one visit: the referral and the device are facts about the
# visit, so counting them per view would multiply every one of them.
assert sum(row["hits"] for row in body["devices"]) == 1
def test_the_owners_own_visits_are_not_traffic(client):
client.current["id"] = client.ids["owner"]
client.post("/api/analytics/collect", json={"path": "/", "first": True})
assert read_summary(client).json()["totals"]["pageviews"] == 0
def test_a_client_cannot_invent_pages_or_events(client):
client.post("/api/analytics/collect", json={"path": "/../../admin", "first": True})
# There is no field for it, so a made-up event is not even expressible.
client.post("/api/analytics/collect", json={"path": "/", "event": "signup"})
client.current["id"] = client.ids["owner"]
body = read_summary(client).json()
assert {row["label"] for row in body["pages"]} == {"(other)", "/"}
assert body["totals"]["signups"] == 0
def test_api_errors_are_counted_by_route(client):
client.get("/api/adventures/999999")
client.current["id"] = client.ids["owner"]
errors = read_summary(client).json()["errors"]
assert errors and errors[0]["label"].startswith("404 /api/adventures/")
# ---------- The dialect the tests never run on ----------
def test_the_upserts_compile_for_postgres():
"""Prod runs on Neon, but these tests run on SQLite, and a failed flush
is caught and logged instead of raised. A dialect mistake would
therefore stay invisible until the dashboard quietly stayed empty. This
test compiles both statements against Postgres without connecting to
one.
"""
from sqlalchemy import create_engine
from sqlalchemy.dialects import postgresql
from sqlalchemy.orm import sessionmaker
session = sessionmaker(bind=create_engine("postgresql+psycopg://u:p@localhost/db"))()
compiled = []
def capture(statement, *args, **kwargs):
compiled.append(str(statement.compile(dialect=postgresql.dialect())))
session.execute = capture
session.scalars = lambda *a, **k: []
analytics._write_counts(session, {("2026-01-01", "pageview", "/"): 2})
analytics._write_visits(session, {("2026-01-01", "f" * 32): {"played"}})
counts, visits = compiled
assert "ON CONFLICT (day, metric, label) DO UPDATE" in counts
assert "analytics_daily.hits + excluded.hits" in counts
assert "ON CONFLICT (day, visitor) DO UPDATE" in visits
assert "analytics_visitor_days.played OR excluded.played" in visits
# is_new is settled by the first write of a visitor's first day and must
# not be in the update clause at all.
assert "is_new" not in visits.split("DO UPDATE")[1]
backend/tests/test_attempt_siblings.py
+3 -16
View File
@@ -19,17 +19,10 @@ from app.database import Base, SessionLocal, engine, get_db
from app.main import app
from app.routers import adventures
from fakes import ScriptedProvider
from fakes import GOLD_SCHEMA, ScriptedProvider, gold_replies, gold_reply
SCHEMA = {"player": {"hp": {"min": 0, "max": 100, "initial": 100}}}
SCHEMA = GOLD_SCHEMA
GOLD_SCRIPT = """
const modifier = (text) => {
state.gold = (state.gold || 0) + 10;
return { text };
};
modifier(text);
"""
@pytest.fixture()
@@ -45,14 +38,11 @@ def client(monkeypatch):
setup.flush()
adv = models.Adventure(
user_id=user.id, title="Cave", scenario_id=scenario.id,
script_state={}, world_state={"player": {"hp": 100}},
world_state={"player": {"hp": 100, "gold": 0}},
)
setup.add(adv)
setup.flush()
setup.add(models.Action(adventure_id=adv.id, type="start", text="You enter a cave."))
setup.add(models.AdventureScript(
adventure_id=adv.id, position=0, enabled=True, name="Gold", output_js=GOLD_SCRIPT,
))
setup.commit()
adv_id, user_id = adv.id, user.id
setup.close()
@@ -61,9 +51,6 @@ def client(monkeypatch):
ScriptedProvider.calls = 0
ScriptedProvider.prompts = []
monkeypatch.setattr(adventures.turns, "OpenAICompatibleProvider", ScriptedProvider)
monkeypatch.setattr(auth, "resolve_provider_config", lambda s: auth.ProviderConfig(
"http://fake", "k", "test-model", False))
monkeypatch.setattr(limits, "rate_limit", lambda *a, **k: None)
monkeypatch.setattr(limits, "check_row_cap", lambda *a, **k: None)
def _current_user(db=Depends(get_db)):
backend/tests/test_branch_clause.py
-14
View File
@@ -30,7 +30,6 @@ from app import auth, limits, models, tree
from app.context import history, lineage
from app.database import Base, SessionLocal, engine, get_db
from app.main import app
from app.scripting import ScriptPipeline
from tools import dbmeter
@@ -224,24 +223,11 @@ def test_an_already_loaded_collection_is_cut_down_to_the_path(forked):
assert labels(history.tail(adventure, 3)) == ["B5", "C6", "C7"]
def test_user_scripts_are_handed_the_path(forked):
"""The same risk one layer up, in code visible to users:
`pipeline._history()` is the documented scripting history API."""
db, adventure, _ = forked
list(adventure.actions) # the pipeline's caller has usually loaded these
pipeline = ScriptPipeline(adventure, db)
assert [h["text"] for h in pipeline._history()] == [
"A0", "A1", "A2", "A3", "B4", "B5", "C6", "C7"
]
assert pipeline._info()["actionCount"] == 8
# ------------------------------------------------------------ over the wire
@pytest.fixture()
def client(forked, monkeypatch):
db, adventure, ids = forked
monkeypatch.setattr(limits, "rate_limit", lambda *a, **k: None)
monkeypatch.setattr(limits, "check_row_cap", lambda *a, **k: None)
def _current_user(session=Depends(get_db)):
backend/tests/test_branch_forking.py
+15 -22
View File
@@ -22,7 +22,7 @@ from app.database import Base, SessionLocal, engine, get_db
from app.main import app
from app.routers import adventures
from fakes import ScriptedProvider
from fakes import GOLD_SCHEMA, ScriptedProvider, gold_replies, gold_reply
# `hp` moves freely. `mana` has a cooldown of 2 turns, so an incorrect
# advance shows up as a change the referee should have rejected.
@@ -30,16 +30,13 @@ SCHEMA = {
"player": {
"hp": {"min": 0, "max": 100, "initial": 100},
"mana": {"min": 0, "max": 50, "initial": 50, "cooldown": 2},
# The per-turn counter these tests measure rollbacks with. Unbounded and
# uncapped on purpose, so every turn's +10 lands in full. See
# `fakes.gold_reply`.
"gold": {"min": 0, "max": 1_000_000, "initial": 0},
}
}
GOLD_SCRIPT = """
const modifier = (text) => {
state.gold = (state.gold || 0) + 10;
return { text };
};
modifier(text);
"""
@pytest.fixture()
@@ -55,14 +52,11 @@ def client(monkeypatch):
setup.flush()
adv = models.Adventure(
user_id=user.id, title="Cave", scenario_id=scenario.id,
script_state={}, world_state={"player": {"hp": 100, "mana": 50}},
world_state={"player": {"hp": 100, "mana": 50, "gold": 0}},
)
setup.add(adv)
setup.flush()
setup.add(models.Action(adventure_id=adv.id, type="start", text="You enter a cave."))
setup.add(models.AdventureScript(
adventure_id=adv.id, position=0, enabled=True, name="Gold", output_js=GOLD_SCRIPT,
))
setup.commit()
adv_id, user_id = adv.id, user.id
setup.close()
@@ -71,9 +65,6 @@ def client(monkeypatch):
ScriptedProvider.calls = 0
ScriptedProvider.prompts = []
monkeypatch.setattr(adventures.turns, "OpenAICompatibleProvider", ScriptedProvider)
monkeypatch.setattr(auth, "resolve_provider_config", lambda s: auth.ProviderConfig(
"http://fake", "k", "test-model", False))
monkeypatch.setattr(limits, "rate_limit", lambda *a, **k: None)
monkeypatch.setattr(limits, "check_row_cap", lambda *a, **k: None)
def _current_user(db=Depends(get_db)):
@@ -121,7 +112,7 @@ def _state(adv_id):
db = SessionLocal()
try:
adv = db.get(models.Adventure, adv_id)
return adv.script_state, adv.world_state
return (adv.world_state or {}).get("player", {}), adv.world_state
finally:
db.close()
@@ -322,11 +313,13 @@ def test_forking_a_live_node_on_another_branch_is_refused(client):
# -------------------------------------------------------------- the state
def test_switching_restores_the_script_and_world_state(client):
def test_switching_restores_the_state_a_branch_left_behind(client):
# Two stats move: hp differs per attempt, and gold counts turns. Between
# them, a switch that restored the wrong snapshot is visible either way.
ScriptedProvider.replies = [
"A scratch.\n```state\n{\"player.hp\": -5}\n```",
"A beating.\n```state\n{\"player.hp\": -40}\n```",
"Onward.",
'A scratch.\n```state\n{"player.hp": -5, "player.gold": 10}\n```',
'A beating.\n```state\n{"player.hp": -40, "player.gold": 10}\n```',
gold_reply("Onward."),
]
_play(client)
_retry(client)
@@ -336,9 +329,9 @@ def test_switching_restores_the_script_and_world_state(client):
discarded = [a.id for a in _rows(client.adv_id) if a.type == "ai" and not a.live][0]
_fork(client, discarded)
script_state, world_state = _state(client.adv_id)
player, world_state = _state(client.adv_id)
assert world_state["player"]["hp"] == 95, "the attempt this branch tells"
assert script_state == {"gold": 10}, "one turn of gold, not three"
assert player["gold"] == 10, "one turn of gold, not three"
client.post(f"/api/adventures/{client.adv_id}/branches/{parent}/switch")
assert _state(client.adv_id) == on_parent
backend/tests/test_branch_management.py
-3
View File
@@ -55,9 +55,6 @@ def client(monkeypatch):
ScriptedProvider.replies = ["Attempt one.", "Attempt two.", "Next turn."]
ScriptedProvider.calls = 0
monkeypatch.setattr(adventures.turns, "OpenAICompatibleProvider", ScriptedProvider)
monkeypatch.setattr(auth, "resolve_provider_config", lambda s: auth.ProviderConfig(
"http://fake", "k", "test-model", False))
monkeypatch.setattr(limits, "rate_limit", lambda *a, **k: None)
monkeypatch.setattr(limits, "check_row_cap", lambda *a, **k: None)
def _current_user(db=Depends(get_db)):
backend/tests/test_bundle_v2.py
+5 -20
View File
@@ -33,19 +33,12 @@ from app.database import Base, SessionLocal, engine, get_db
from app.main import app
from app.routers import adventures
from fakes import ScriptedProvider
from fakes import GOLD_SCHEMA, ScriptedProvider, gold_replies, gold_reply
SCHEMA = {"player": {"hp": {"min": 0, "max": 100, "initial": 100}}}
SCHEMA = GOLD_SCHEMA
# Ten gold a turn, so the stored gold total tells how many turns the
# story behind it played. This makes an after-snapshot visible from outside.
GOLD_SCRIPT = """
const modifier = (text) => {
state.gold = (state.gold || 0) + 10;
return { text };
};
modifier(text);
"""
OPENING = "You enter a cave."
@@ -63,14 +56,11 @@ def client(monkeypatch):
setup.flush()
adv = models.Adventure(
user_id=user.id, title="Cave", scenario_id=scenario.id,
script_state={}, world_state={"player": {"hp": 100}},
world_state={"player": {"hp": 100, "gold": 0}},
)
setup.add(adv)
setup.flush()
setup.add(models.Action(adventure_id=adv.id, type="start", text=OPENING))
setup.add(models.AdventureScript(
adventure_id=adv.id, position=0, enabled=True, name="Gold", output_js=GOLD_SCRIPT,
))
setup.commit()
adv_id, user_id = adv.id, user.id
setup.close()
@@ -78,9 +68,6 @@ def client(monkeypatch):
ScriptedProvider.replies = ["A reply."]
ScriptedProvider.calls = 0
monkeypatch.setattr(adventures.turns, "OpenAICompatibleProvider", ScriptedProvider)
monkeypatch.setattr(auth, "resolve_provider_config", lambda s: auth.ProviderConfig(
"http://fake", "k", "test-model", False))
monkeypatch.setattr(limits, "rate_limit", lambda *a, **k: None)
monkeypatch.setattr(limits, "check_row_cap", lambda *a, **k: None)
def _current_user(db=Depends(get_db)):
@@ -186,7 +173,7 @@ def _branch_rows(adv_id) -> list[models.Branch]:
def _script_state(adv_id) -> dict:
db = SessionLocal()
try:
return db.get(models.Adventure, adv_id).script_state
return (db.get(models.Adventure, adv_id).world_state or {}).get("player", {})
finally:
db.close()
@@ -272,7 +259,7 @@ def test_a_switch_in_the_copy_restores_what_that_branch_left_behind(client):
# Play one more turn on the fork, so the two tips end up at
# genuinely different totals. Turn for turn, both branches earn the
# same gold, so a switch that restored nothing would still look right.
ScriptedProvider.replies = ["Further still."]
ScriptedProvider.replies = [gold_reply("Further still.")]
_play(client, original, "press on")
per_branch = []
@@ -398,7 +385,6 @@ def test_a_fork_with_no_depth_is_refused(client):
def test_more_branches_than_the_cap_is_refused(client, monkeypatch):
monkeypatch.setattr(auth, "MULTI_USER", True)
payload = {
"format": bundle.FORMAT, "title": "Too many",
"branches": [{"parent": None, "forkDepth": None}]
@@ -558,7 +544,6 @@ def test_the_action_cap_counts_the_rows_a_v1_file_expands_into(client, monkeypat
multiple of it. The body-size limit does not help here: the text is
tiny, and the row count is the actual cost.
"""
monkeypatch.setattr(auth, "MULTI_USER", True)
monkeypatch.setattr(limits, "MAX_ACTIONS_PER_ADVENTURE", 6)
monkeypatch.setattr(limits, "_BUNDLE_LIST_CAPS",
{**limits._BUNDLE_LIST_CAPS, "actions": 6})
backend/tests/test_chat.py
+43 -165
View File
@@ -1,8 +1,14 @@
"""HTTP tests for the AI Chat scratchpad (power users only).
"""HTTP tests for the AI Chat scratchpad.
Covers the access gate, the streamed reply, and the demo-key model pinning.
This pinning must not let a public visitor reach paid models through this
page.
Most of this file used to be about the shared demo key: an access gate on a
"power user" email allowlist, and a pinning rule that stopped a public visitor
reaching paid models on a server-funded key. M2 removed the hosted deployment
those defended, so the rules they tested no longer exist to be tested. See
`planning/reports/M2-*` for the accounting.
What remains is what the page still does: stream a reply from the configured
model, honour a system prompt and a per-request model override, and refuse a
conversation too large to send.
python -m pytest tests/test_chat.py -v
"""
@@ -10,25 +16,27 @@ import pytest
from fastapi import Depends
from fastapi.testclient import TestClient
from app import auth, limits, models
from app import auth, models
from app.database import Base, SessionLocal, engine, get_db
from app.main import app
from app.routers import chat
class FakeProvider:
"""Records what it was constructed with, then streams a fixed reply. Stands
in for the real egress point, so asserting on last_key/last_model is
asserting on exactly what would have gone over the wire."""
"""Records what it was constructed with, then streams a fixed reply.
It stands in for the real egress point, so asserting on `last_endpoint` and
`last_model` is asserting on exactly what would have gone over the wire.
There is no `last_key` any more: the provider takes no API key, because
Ollama does not use one.
"""
last_usage = None
last_model = None
last_key = None
last_endpoint = None
last_messages = None
def __init__(self, endpoint_url, api_key, model, api_mode="chat", reasoning_max_tokens=0):
def __init__(self, endpoint_url, model, api_mode="chat", read_timeout=None):
FakeProvider.last_model = model
FakeProvider.last_key = api_key
FakeProvider.last_endpoint = endpoint_url
async def chat(self, messages, *, temperature, max_tokens):
@@ -41,68 +49,35 @@ class FakeProvider:
def client(monkeypatch):
Base.metadata.create_all(bind=engine)
setup = SessionLocal()
user = models.User(is_guest=False, email="power@example.com")
user = models.User(is_guest=False)
setup.add(user)
setup.flush()
setup.add(models.Settings(user_id=user.id, api_key="enc:dummy", model="test-model"))
setup.add(models.Settings(user_id=user.id, model="test-model"))
setup.commit()
user_id = user.id
setup.close()
monkeypatch.setattr(chat, "OpenAICompatibleProvider", FakeProvider)
monkeypatch.setattr(limits, "rate_limit", lambda *a, **k: None)
# Multi-user mode is what makes the power-user gate meaningful, because
# local mode trusts everyone. The allowlist is set per test.
monkeypatch.setattr(auth, "MULTI_USER", True)
monkeypatch.setattr(auth, "POWER_USERS", {"power@example.com"})
# These tests deliberately do not stub resolve_provider_config. The
# point is to exercise the real BYOK-vs-demo decision, since that
# decision is what keeps the shared key off paid models. Each test
# picks a mode with _byok/_demo below.
def _current_user(db=Depends(get_db)):
return db.get(models.User, user_id)
app.dependency_overrides[auth.get_current_user] = _current_user
c = TestClient(app)
try:
yield TestClient(app)
yield c
finally:
app.dependency_overrides.clear()
Base.metadata.drop_all(bind=engine)
def _send(client, **extra):
return client.post("/api/chat/stream", json={"messages": [{"role": "user", "content": "hi"}], **extra})
def _send(client, **body):
payload = {"messages": [{"role": "user", "content": "hi"}]}
payload.update(body)
return client.post("/api/chat/stream", json=payload)
def _byok(monkeypatch):
"""The user brought their own key: no demo key in play, any model allowed."""
monkeypatch.setattr(auth, "demo_enabled", lambda: False)
db = SessionLocal()
try:
settings = db.query(models.Settings).first()
settings.api_key = "sk-my-own-key" # legacy-plaintext path: used as-is
db.commit()
finally:
db.close()
def _demo(monkeypatch, whitelist=("free/allowed",)):
"""The user has no key, so turns run on the server-funded demo key."""
monkeypatch.setattr(auth, "demo_enabled", lambda: True)
monkeypatch.setattr(auth, "DEMO_API_KEY", "demo-key")
monkeypatch.setattr(auth, "DEMO_ENDPOINT_URL", "http://demo")
monkeypatch.setattr(auth, "DEMO_MODELS", list(whitelist))
def test_non_power_user_gets_404(client, monkeypatch):
monkeypatch.setattr(auth, "POWER_USERS", set())
assert _send(client).status_code == 404
assert client.get("/api/chat/config").status_code == 404
def test_power_user_streams_a_reply(client, monkeypatch):
_byok(monkeypatch)
def test_the_page_streams_a_reply(client):
resp = _send(client)
assert resp.status_code == 200, resp.text
assert '"type": "reasoning"' in resp.text
@@ -111,136 +86,39 @@ def test_power_user_streams_a_reply(client, monkeypatch):
assert FakeProvider.last_messages == [{"role": "user", "content": "hi"}]
def test_system_prompt_and_model_override_are_honoured(client, monkeypatch):
_byok(monkeypatch)
def test_it_uses_the_configured_endpoint_and_model(client):
_send(client)
assert FakeProvider.last_model == "test-model"
# The default from `models.Settings`, and the only kind of address the
# endpoint policy allows without configuration.
assert FakeProvider.last_endpoint == "http://localhost:11434/v1"
def test_system_prompt_and_model_override_are_honoured(client):
resp = client.post("/api/chat/stream", json={
"messages": [
{"role": "system", "content": "Be terse."},
{"role": "user", "content": "hi"},
],
"model": "some/other-model",
"model": "some-other-model",
})
assert resp.status_code == 200, resp.text
# BYOK: any model the user names is passed straight through, on their key.
assert FakeProvider.last_model == "some/other-model"
assert FakeProvider.last_key == "sk-my-own-key"
assert FakeProvider.last_model == "some-other-model"
assert FakeProvider.last_messages[0] == {"role": "system", "content": "Be terse."}
def test_demo_key_pins_model_to_whitelist(client, monkeypatch):
_demo(monkeypatch)
resp = _send(client, model="expensive/paid-model")
assert resp.status_code == 200, resp.text
# Refused visibly: the whitelisted model runs instead, with a note. The
# paid slug must never reach the wire alongside the server-funded key.
assert FakeProvider.last_model == "free/allowed"
assert FakeProvider.last_key == "demo-key"
assert '"type": "note"' in resp.text
# A whitelisted model is still selectable on the demo key.
_demo(monkeypatch, ["free/allowed", "free/second"])
resp = _send(client, model="free/second")
assert resp.status_code == 200, resp.text
assert FakeProvider.last_model == "free/second"
def test_demo_key_ignores_an_off_whitelist_settings_model(client, monkeypatch):
"""The override is not the only untrusted input. `Settings.model` is
also user-set, and it must be pinned the same way when there is no
BYOK key."""
_demo(monkeypatch)
db = SessionLocal()
try:
db.query(models.Settings).first().model = "expensive/paid-model"
db.commit()
finally:
db.close()
resp = _send(client)
assert resp.status_code == 200, resp.text
assert FakeProvider.last_model == "free/allowed"
def test_demo_key_endpoint_cannot_be_redirected(client, monkeypatch):
"""A user-controlled `endpoint_url` would leak the key itself, which is
worse than spending it. The demo branch pins the URL too."""
_demo(monkeypatch)
db = SessionLocal()
try:
db.query(models.Settings).first().endpoint_url = "http://attacker.example/v1"
db.commit()
finally:
db.close()
assert _send(client).status_code == 200
assert FakeProvider.last_endpoint == "http://demo"
assert FakeProvider.last_key == "demo-key"
def test_provider_config_refuses_server_funded_paid_model(monkeypatch):
"""The structural backstop: a hand-built config (a future code path that
forgets to go through resolve_provider_config) cannot run a
server-funded turn on an off-whitelist model."""
monkeypatch.setattr(auth, "DEMO_API_KEY", "demo-key")
monkeypatch.setattr(auth, "DEMO_MODELS", ["free/allowed"])
with pytest.raises(ValueError):
auth.ProviderConfig("http://demo", "demo-key", "expensive/paid-model", True)
auth.ProviderConfig("http://demo", "demo-key", "free/allowed", True) # whitelisted: fine
# The user's own key with any model stays fine.
auth.ProviderConfig("http://any", "sk-mine", "expensive/paid-model", False)
def test_byok_user_may_reuse_the_demo_keys_value(client, monkeypatch):
"""Regression: the demo key is just an OpenRouter key, so a user can paste
that same value into their own Settings. That is still BYOK, because
the user is paying, and it must not trip the guard. It used to raise
on every resolution, which returned a 500 from `GET /auth/me` and
broke the entire SPA (no nav, no chat)."""
monkeypatch.setattr(auth, "demo_enabled", lambda: True)
monkeypatch.setattr(auth, "DEMO_API_KEY", "shared-key")
monkeypatch.setattr(auth, "DEMO_ENDPOINT_URL", "http://demo")
monkeypatch.setattr(auth, "DEMO_MODELS", ["free/allowed"])
def test_a_request_with_no_model_anywhere_is_refused(client):
db = SessionLocal()
try:
settings = db.query(models.Settings).first()
settings.api_key = "shared-key" # same value, but supplied by the user
settings.model = "expensive/paid-model" # their spend, their choice
settings.model = ""
db.commit()
finally:
db.close()
assert client.get("/api/auth/me").status_code == 200
assert client.get("/api/chat/config").status_code == 200
resp = _send(client)
assert resp.status_code == 200, resp.text
assert FakeProvider.last_model == "expensive/paid-model"
assert FakeProvider.last_key == "shared-key"
assert _send(client).status_code == 400
def test_resolve_provider_config_is_the_single_choke_point(monkeypatch):
"""Turns, AI Chat, and the connection test all resolve through this one
function, so pinning it here pins every caller. No DB or HTTP needed."""
monkeypatch.setattr(auth, "demo_enabled", lambda: True)
monkeypatch.setattr(auth, "DEMO_API_KEY", "demo-key")
monkeypatch.setattr(auth, "DEMO_ENDPOINT_URL", "http://demo")
monkeypatch.setattr(auth, "DEMO_MODELS", ["free/allowed"])
# No key of their own: both endpoint and model are pinned, regardless
# of what they set.
no_key = models.Settings(endpoint_url="http://mine/v1", api_key="", model="expensive/paid")
assert auth.resolve_provider_config(no_key) == auth.ProviderConfig(
"http://demo", "demo-key", "free/allowed", True)
assert auth.resolve_provider_config(
no_key, model_override="expensive/paid").model == "free/allowed"
assert auth.resolve_provider_config(
no_key, model_override="free/allowed").model == "free/allowed"
# Their own key: their endpoint, their key, their choice of model.
byok = models.Settings(endpoint_url="http://mine/v1", api_key="sk-mine", model="expensive/paid")
assert auth.resolve_provider_config(byok) == auth.ProviderConfig(
"http://mine/v1", "sk-mine", "expensive/paid", False)
def test_oversized_conversation_is_refused(client, monkeypatch):
_byok(monkeypatch)
def test_oversized_conversation_is_refused(client):
huge = "x" * 90_000
resp = client.post("/api/chat/stream", json={
"messages": [{"role": "user", "content": huge} for _ in range(5)],
backend/tests/test_delete_state.py
+18 -25
View File
@@ -1,6 +1,6 @@
"""Deleting a turn puts the shared state back.
`script_state` and `world_state` belong to the adventure, not to the node
`world_state` belongs to the adventure, not to the node
that changed them. Undo, retry, a take and a branch switch all restore them;
the delete endpoint did not. Deleting an AI turn removed the text and left
everything the turn did to the numbers standing.
@@ -23,7 +23,7 @@ from app import auth, limits, models
from app.database import Base, SessionLocal, engine, get_db
from app.main import app
from app.routers import adventures
from fakes import ScriptedProvider
from fakes import GOLD_SCHEMA, ScriptedProvider, gold_replies, gold_reply
# `mana` carries a cooldown, so a clock that was not rolled back shows up as
# a refusal rather than as a number that is merely off.
@@ -31,20 +31,19 @@ SCHEMA = {
"player": {
"hp": {"min": 0, "max": 100, "initial": 100},
"mana": {"min": 0, "max": 50, "initial": 50, "cooldown": 2},
# The per-turn counter these tests measure rollbacks with. Unbounded and
# uncapped on purpose, so every turn's +10 lands in full. See
# `fakes.gold_reply`.
"gold": {"min": 0, "max": 1_000_000, "initial": 0},
}
}
# Ten gold a turn. A total that only ever climbs makes a missing rollback
# obvious: it is off by exactly one turn's worth.
GOLD_SCRIPT = """
const modifier = (text) => {
state.gold = (state.gold || 0) + 10;
return { text };
};
modifier(text);
"""
DRAIN = 'Drained.\n```state\n{"player.mana": -10}\n```'
# The same turn, also banking the per-turn counter the rollback tests measure.
DRAIN_AND_GOLD = 'Drained.\n```state\n{"player.mana": -10, "player.gold": 10}\n```'
@pytest.fixture()
@@ -60,24 +59,18 @@ def client(monkeypatch):
setup.flush()
adv = models.Adventure(
user_id=user.id, title="Tower", scenario_id=scenario.id,
script_state={}, world_state={"player": {"hp": 100, "mana": 50}},
world_state={"player": {"hp": 100, "mana": 50, "gold": 0}},
)
setup.add(adv)
setup.flush()
setup.add(models.Action(adventure_id=adv.id, type="start", text="You begin."))
setup.add(models.AdventureScript(
adventure_id=adv.id, position=0, enabled=True, name="Gold", output_js=GOLD_SCRIPT,
))
setup.commit()
adv_id, user_id = adv.id, user.id
setup.close()
ScriptedProvider.replies = [DRAIN]
ScriptedProvider.replies = [DRAIN_AND_GOLD]
ScriptedProvider.calls = 0
monkeypatch.setattr(adventures.turns, "OpenAICompatibleProvider", ScriptedProvider)
monkeypatch.setattr(auth, "resolve_provider_config", lambda s: auth.ProviderConfig(
"http://fake", "k", "test-model", False))
monkeypatch.setattr(limits, "rate_limit", lambda *a, **k: None)
monkeypatch.setattr(limits, "check_row_cap", lambda *a, **k: None)
def _current_user(db=Depends(get_db)):
@@ -116,7 +109,7 @@ def _state(adv_id):
db = SessionLocal()
try:
adv = db.get(models.Adventure, adv_id)
return adv.script_state, adv.world_state
return (adv.world_state or {}).get("player", {}), adv.world_state
finally:
db.close()
@@ -168,17 +161,17 @@ def test_the_next_turn_is_not_refused_for_a_deleted_turn_s_cooldown(client):
assert [c for c in _last_changes(client.adv_id) if c["kind"] == "rejected"] == []
def test_deleting_the_ai_turn_rewinds_the_script_state(client):
"""The same restore, on the other half of the shared state. Without it a
replayed turn stacks its script run on top of the deleted one's."""
def test_deleting_the_ai_turn_rewinds_the_counter(client):
"""The same restore, seen through a value that only ever climbs. Without it
a replayed turn stacks its effects on top of the deleted one's."""
_play(client)
assert _state(client.adv_id)[0] == {"gold": 10}
assert _state(client.adv_id)[0]["gold"] == 10
_delete(client, _ai_rows(client.adv_id)[-1].id)
assert _state(client.adv_id)[0] == {}
assert _state(client.adv_id)[0]["gold"] == 0
_continue(client)
assert _state(client.adv_id)[0] == {"gold": 10}, "one turn of gold, not two"
assert _state(client.adv_id)[0]["gold"] == 10, "one turn of gold, not two"
# ------------------------------------------------- deleting further back
@@ -190,7 +183,7 @@ def test_deleting_a_turn_the_story_moved_past_leaves_the_tip_alone(client):
_play(client)
_play(client, "press on")
before = _state(client.adv_id)
assert before[0] == {"gold": 20}
assert before[0]["gold"] == 20
first_ai = _ai_rows(client.adv_id)[0]
assert _delete(client, first_ai.id).status_code == 204
backend/tests/test_egress.py
-1
View File
@@ -99,7 +99,6 @@ def client(monkeypatch):
adv_id, user_id = adventure.id, user.id
setup.close()
monkeypatch.setattr(limits, "rate_limit", lambda *a, **k: None)
monkeypatch.setattr(limits, "check_row_cap", lambda *a, **k: None)
def _current_user(db=Depends(get_db)):
backend/tests/test_embedding_model_switch.py
-1
View File
@@ -57,7 +57,6 @@ def client(monkeypatch):
adv_id, user_id = adventure.id, user.id
setup.close()
monkeypatch.setattr(limits, "rate_limit", lambda *a, **k: None)
monkeypatch.setattr(limits, "check_row_cap", lambda *a, **k: None)
def _current_user(db=Depends(get_db)):
backend/tests/test_endpoint_policy.py
+224
View File
@@ -0,0 +1,224 @@
"""Which inference endpoints the product will and will not talk to.
The endpoint is where the whole campaign goes: prompts, narration, retrieved
memories, embedding inputs. `app/endpoints.py` is the rule that keeps that on
the user's own infrastructure, and these tests are what stops the rule quietly
widening.
Nothing here touches the network. Hostnames are resolved through a stub, so the
tests are about the *policy*, run identically offline, and cannot be made to
pass or fail by whatever DNS the machine happens to have.
python -m pytest tests/test_endpoint_policy.py -v
"""
import ipaddress
import socket
import pytest
from fastapi import Depends
from fastapi.testclient import TestClient
from app import auth, endpoints, models
from app.database import Base, SessionLocal, engine, get_db
from app.main import app
from app.providers import OpenAICompatibleProvider, ProviderError
@pytest.fixture()
def resolves(monkeypatch):
"""Points every hostname at addresses the test names."""
table: dict[str, list[str]] = {}
def fake_getaddrinfo(host, port, *a, **k):
# An address literal resolves to itself, as the real resolver does.
try:
ipaddress.ip_address(host)
except ValueError:
if host not in table:
raise socket.gaierror(-2, "Name or service not known")
answers = table[host]
else:
answers = [host]
return [
(
socket.AF_INET6 if ":" in ip else socket.AF_INET,
socket.SOCK_STREAM, 6, "", (ip, port or 0),
)
for ip in answers
]
monkeypatch.setattr(socket, "getaddrinfo", fake_getaddrinfo)
return table
# --- allowed: the user's own machine, and the user's own network ----------
@pytest.mark.parametrize("url", [
"http://127.0.0.1:11434/v1",
"http://[::1]:11434/v1",
"http://192.168.1.50:11434/v1", # RFC1918
"http://10.0.0.7:11434/v1", # RFC1918
"http://172.16.4.4:11434/v1", # RFC1918
"https://192.168.1.50:8443/v1", # TLS on the LAN
"http://100.100.5.6:11434/v1", # CGNAT, which is what a mesh VPN hands out
"http://[fd00::1]:11434/v1", # IPv6 unique-local
])
def test_local_and_lan_addresses_are_allowed(url, resolves):
assert endpoints.rejection_reason(url) is None
def test_a_hostname_resolving_to_the_lan_is_allowed(resolves):
resolves["ollama.home.arpa"] = ["192.168.1.50"]
assert endpoints.rejection_reason("https://ollama.home.arpa:8443/v1") is None
# --- refused: anywhere else ----------------------------------------------
@pytest.mark.parametrize("url", [
"http://8.8.8.8:11434/v1",
"http://1.1.1.1:11434/v1",
"http://[2001:4860:4860::8888]:11434/v1",
])
def test_public_addresses_are_refused(url, resolves):
reason = endpoints.rejection_reason(url)
assert reason is not None
assert "public Internet address" in reason
@pytest.mark.parametrize("url", [
"https://198.51.100.9/v1", # TEST-NET-2, a documentation range
"http://0.0.0.0:11434/v1", # "this host", not an address to dial
])
def test_addresses_that_are_neither_local_nor_public_are_refused(url, resolves):
"""Deny by default. `ipaddress` calls both of these "private", which is why
the policy names the networks it allows instead of asking that question."""
assert endpoints.rejection_reason(url) is not None
def test_a_hostname_resolving_to_the_public_internet_is_refused(resolves):
resolves["ollama.example.com"] = ["93.184.216.34"]
reason = endpoints.rejection_reason("https://ollama.example.com/v1")
assert reason is not None and "public Internet address" in reason
def test_a_split_horizon_answer_is_refused(resolves):
"""One public address among the answers is enough. A name that resolves to
both a LAN address and a public one must not be usable: which one the
request actually reaches is not ours to decide."""
resolves["sneaky.example"] = ["192.168.1.50", "93.184.216.34"]
assert endpoints.rejection_reason("http://sneaky.example:11434/v1") is not None
@pytest.mark.parametrize("host", [
"openrouter.ai", "api.openai.com", "api.anthropic.com", "api.groq.com",
])
def test_known_cloud_providers_are_named_in_the_refusal(host, resolves):
"""They would be refused by address anyway — every one resolves publicly.
Naming them turns 'your DNS might be broken' into 'this build has no cloud
provider support'."""
resolves[host] = ["192.168.1.50"] # even if DNS said otherwise
reason = endpoints.rejection_reason(f"https://{host}/v1")
assert reason is not None
assert "cloud inference service" in reason
def test_a_subdomain_of_a_cloud_provider_is_refused(resolves):
resolves["eu.api.openai.com"] = ["192.168.1.50"]
assert endpoints.rejection_reason("https://eu.api.openai.com/v1") is not None
@pytest.mark.parametrize("url", [
"ftp://127.0.0.1/v1",
"file:///etc/passwd",
"127.0.0.1:11434", # no scheme
"",
"http://",
])
def test_things_that_are_not_usable_urls_are_refused(url, resolves):
assert endpoints.rejection_reason(url) is not None
def test_an_unresolvable_host_is_refused_with_advice(resolves):
reason = endpoints.rejection_reason("http://nope.invalid:11434/v1")
assert reason is not None and "could not be resolved" in reason
# --- the rule is applied, not merely available ---------------------------
@pytest.fixture()
def client(monkeypatch):
Base.metadata.create_all(bind=engine)
setup = SessionLocal()
user = models.User(is_guest=False)
setup.add(user)
setup.flush()
setup.add(models.Settings(user_id=user.id, model="test-model"))
setup.commit()
user_id = user.id
setup.close()
def _current_user(db=Depends(get_db)):
return db.get(models.User, user_id)
app.dependency_overrides[auth.get_current_user] = _current_user
c = TestClient(app)
try:
yield c
finally:
app.dependency_overrides.clear()
Base.metadata.drop_all(bind=engine)
def test_saving_a_public_endpoint_is_refused(client, resolves):
resolves["openrouter.ai"] = ["93.184.216.34"]
r = client.put("/api/settings", json={"endpoint_url": "https://openrouter.ai/api/v1"})
assert r.status_code == 400, r.text
assert "can't be used" in r.json()["detail"]
# And it was not written.
assert client.get("/api/settings").json()["endpoint_url"] != "https://openrouter.ai/api/v1"
def test_saving_a_lan_endpoint_is_allowed(client, resolves):
r = client.put("/api/settings", json={"endpoint_url": "http://192.168.1.50:11434/v1"})
assert r.status_code == 200, r.text
assert client.get("/api/settings").json()["endpoint_url"] == "http://192.168.1.50:11434/v1"
def test_the_connection_test_reports_a_refused_endpoint_as_such(client, resolves):
"""A row edited by hand, or a name that has started resolving elsewhere,
must not simply look 'unreachable'."""
db = SessionLocal()
try:
db.query(models.Settings).first().endpoint_url = "https://openrouter.ai/api/v1"
db.commit()
finally:
db.close()
resolves["openrouter.ai"] = ["93.184.216.34"]
body = client.post("/api/settings/test").json()
assert body["ok"] is False
assert body["kind"] == "rejected"
@pytest.mark.anyio
async def test_the_provider_refuses_before_it_sends_anything(resolves):
"""The check that actually matters. Whatever is stored, no request leaves
for an address outside the policy — so a database edited behind the app's
back cannot turn into an exfiltration path."""
resolves["openrouter.ai"] = ["93.184.216.34"]
provider = OpenAICompatibleProvider("https://openrouter.ai/api/v1", "m")
with pytest.raises(ProviderError) as exc:
async for _ in provider.generate(
_parts(), temperature=0.8, max_tokens=10
):
pass
assert "can't be used" in str(exc.value)
def _parts():
from app.providers.base import PromptParts
return PromptParts(system="s", story="t")
@pytest.fixture
def anyio_backend():
return "asyncio"
backend/tests/test_guest_cleanup.py
-205
View File
@@ -1,205 +0,0 @@
"""Guest retention policy: app/cleanup.py.
These tests cover the two things that matter. Idle guests and their whole
data graph must actually be deleted, and nothing else must ever be
deleted.
python -m pytest tests/test_guest_cleanup.py -v
"""
from datetime import timedelta
import pytest
from sqlalchemy import create_engine, event
from sqlalchemy.orm import sessionmaker
from app import cleanup, models
from app.database import Base
from app.migrations import bootstrap
@pytest.fixture()
def db(tmp_path):
engine = create_engine(f"sqlite:///{tmp_path/'t.db'}", connect_args={"check_same_thread": False})
@event.listens_for(engine, "connect")
def _fk(dbapi_connection, _record):
# The whole policy relies on ON DELETE CASCADE. SQLite ignores every
# one of them unless this is set, the same as database.py does.
cur = dbapi_connection.cursor()
cur.execute("PRAGMA foreign_keys=ON")
cur.close()
bootstrap(engine)
Base.metadata.create_all(bind=engine)
session = sessionmaker(bind=engine, autoflush=False, expire_on_commit=False)()
yield session
session.close()
NOW = models.utcnow().replace(tzinfo=None)
def make_user(db, *, days_idle=None, days_old=0, guest=True, email=None):
"""A user last seen `days_idle` ago (None = never seen, only created)."""
user = models.User(
is_guest=guest,
email=email,
password_hash=None if email is None else "x",
created_at=NOW - timedelta(days=days_old),
last_seen_at=None if days_idle is None else NOW - timedelta(days=days_idle),
)
db.add(user)
db.commit()
return user
def sweep(db):
return cleanup.delete_stale_guests(db, now=NOW)
def alive(db, user_id):
# A count, not db.get: the sweep deletes with synchronize_session=False, so
# the session's identity map still holds the object and db.get would answer
# from memory without ever asking the database.
return db.query(models.User).filter(models.User.id == user_id).count() == 1
# ---------- what goes ----------
def test_deletes_guest_idle_past_the_window(db):
user = make_user(db, days_idle=6)
assert sweep(db) == 1
assert not alive(db, user.id)
def test_keeps_guest_inside_the_window(db):
user = make_user(db, days_idle=4)
assert sweep(db) == 0
assert alive(db, user.id)
def test_boundary_is_not_yet_stale(db):
# Exactly 5 days survives. The comparison is strict.
user = make_user(db, days_idle=cleanup.RETENTION_DAYS)
assert sweep(db) == 0
assert alive(db, user.id)
def test_never_seen_guest_falls_back_to_created_at(db):
"""last_seen_at is NULL until a guest's second request (auth._touch runs
hourly), so a coalesce-less query would delete brand-new visitors."""
fresh = make_user(db, days_idle=None, days_old=0)
stale = make_user(db, days_idle=None, days_old=9)
assert sweep(db) == 1
assert alive(db, fresh.id)
assert not alive(db, stale.id)
def test_recent_visit_beats_an_old_created_at(db):
# A long-standing guest who came back yesterday stays.
user = make_user(db, days_idle=1, days_old=90)
assert sweep(db) == 0
assert alive(db, user.id)
# ---------- what must never go ----------
def test_spares_registered_users(db):
"""Registering upgrades the guest row in place. An idle account here is
a real user with real data, which is what signing up is meant to
protect."""
user = make_user(db, days_idle=400, guest=False, email="a@b.com")
assert sweep(db) == 0
assert alive(db, user.id)
def test_spares_the_local_mode_user(db):
# email NULL but is_guest False: local mode's implicit owner of everything.
user = make_user(db, days_idle=400, guest=False)
assert sweep(db) == 0
assert alive(db, user.id)
def test_spares_a_guest_flagged_row_that_has_an_email(db):
# This row should not exist, but both clauses are checked so it cannot be collected.
user = make_user(db, days_idle=400, guest=True, email="odd@b.com")
assert sweep(db) == 0
assert alive(db, user.id)
def test_leaves_seeded_public_scenarios_alone(db):
"""Seeded demo content has user_id NULL, so it is outside the filter."""
seeded = models.Scenario(user_id=None, is_public=True, title="Demo")
db.add(seeded)
make_user(db, days_idle=30)
db.commit()
assert sweep(db) == 1
assert db.query(models.Scenario).filter(models.Scenario.id == seeded.id).count() == 1
def test_disabled_when_retention_is_zero(db, monkeypatch):
monkeypatch.setattr(cleanup, "RETENTION_DAYS", 0)
user = make_user(db, days_idle=999)
assert sweep(db) == 0
assert alive(db, user.id)
def test_enabled_requires_multi_user(monkeypatch):
from app import auth
monkeypatch.setattr(auth, "MULTI_USER", False)
assert cleanup.enabled() is False
monkeypatch.setattr(auth, "MULTI_USER", True)
monkeypatch.setattr(cleanup, "RETENTION_DAYS", 5)
assert cleanup.enabled() is True
monkeypatch.setattr(cleanup, "RETENTION_DAYS", 0)
assert cleanup.enabled() is False
# ---------- the cascade ----------
def test_deletes_the_whole_data_graph(db):
"""One DELETE must remove the adventure, its actions and memories, the
story cards, and the settings row. Nothing is loaded into Python, so if
the FK cascade does not reach a table, its rows are silently orphaned,
or the statement fails, instead of being removed."""
user = make_user(db, days_idle=30)
scenario = models.Scenario(user_id=user.id, title="S")
db.add(scenario)
db.commit()
adventure = models.Adventure(user_id=user.id, scenario_id=scenario.id, title="A")
db.add(adventure)
db.commit()
db.add_all([
models.Action(adventure_id=adventure.id, type="ai", text="t"),
models.Memory(adventure_id=adventure.id, text="m", source_start=0, source_end=0),
models.StoryCard(adventure_id=adventure.id, name="c"),
models.Settings(user_id=user.id),
])
db.commit()
assert sweep(db) == 1
for model in (models.Scenario, models.Adventure, models.Action,
models.Memory, models.StoryCard, models.Settings):
assert db.query(model).count() == 0, f"{model.__name__} rows survived"
def test_one_users_cleanup_does_not_touch_another(db):
keeper = make_user(db, days_idle=1)
keep_adv = models.Adventure(user_id=keeper.id, title="mine")
goner = make_user(db, days_idle=30)
db.add_all([keep_adv, models.Adventure(user_id=goner.id, title="theirs")])
db.commit()
assert sweep(db) == 1
remaining = db.query(models.Adventure).all()
assert [a.title for a in remaining] == ["mine"]
def test_sweep_swallows_errors(monkeypatch):
"""A broken cleanup must not take the app down (same rule as seeding)."""
monkeypatch.setattr(cleanup.auth, "MULTI_USER", True)
monkeypatch.setattr(cleanup, "delete_stale_guests",
lambda *a, **k: (_ for _ in ()).throw(RuntimeError("boom")))
assert cleanup.sweep() == 0
backend/tests/test_local_only_surface.py
+181
View File
@@ -0,0 +1,181 @@
"""What M2 removed stays removed, and what it must not break stays working.
A subtractive milestone needs tests that fail if the surface grows back. These
are cheap, blunt, and deliberately not clever: they assert against the running
app's route table, the shipped configuration files, and the settings API.
python -m pytest tests/test_local_only_surface.py -v
"""
import re
from pathlib import Path
import pytest
from fastapi import Depends
from fastapi.testclient import TestClient
from app import auth, models
from app.database import Base, SessionLocal, engine, get_db
from app.main import app
from app.providers.openai_compatible import (
CONNECT_TIMEOUT, DEFAULT_READ_TIMEOUT, OpenAICompatibleProvider,
)
REPO = Path(__file__).resolve().parents[2]
def _paths() -> set[str]:
"""Every path the app serves, including those inside included routers."""
found = set()
def walk(routes):
for r in routes:
path = getattr(r, "path", None)
if path:
found.add(path)
walk(getattr(r, "routes", []) or [])
walk(app.routes)
return found
@pytest.fixture()
def client():
Base.metadata.create_all(bind=engine)
setup = SessionLocal()
user = models.User(is_guest=False)
setup.add(user)
setup.flush()
setup.add(models.Settings(user_id=user.id, model="test-model"))
setup.commit()
user_id = user.id
setup.close()
def _current_user(db=Depends(get_db)):
return db.get(models.User, user_id)
app.dependency_overrides[auth.get_current_user] = _current_user
c = TestClient(app)
try:
yield c
finally:
app.dependency_overrides.clear()
Base.metadata.drop_all(bind=engine)
# --- the removed surfaces ------------------------------------------------
@pytest.mark.parametrize("prefix", ["/api/auth", "/api/analytics", "/api/scripts"])
def test_no_route_serves_a_removed_subsystem(prefix):
"""Accounts, the visitor dashboard, and campaign scripting are gone as
routes, not merely hidden behind a flag."""
assert not [p for p in _paths() if p.startswith(prefix)], prefix
@pytest.mark.parametrize("path", [
"/api/auth/me", "/api/auth/login", "/api/auth/register", "/api/auth/logout",
"/api/analytics/summary", "/api/analytics/collect", "/api/analytics/access",
"/api/scripts", "/api/adventures/1/scripts", "/api/adventures/1/script-state",
])
def test_a_removed_endpoint_answers_404(client, path):
assert client.get(path).status_code == 404, path
def test_the_application_has_no_scripting_engine():
with pytest.raises(ImportError):
__import__("app.scripting")
def test_no_module_imports_quickjs():
"""The dependency is gone from requirements; this catches an import that
would put it back."""
for py in (REPO / "backend" / "app").rglob("*.py"):
assert "import quickjs" not in py.read_text(), py
def test_requirements_carry_no_hosted_dependencies():
text = (REPO / "backend" / "requirements.txt").read_text()
for gone in ("quickjs", "psycopg", "cryptography"):
assert gone not in text, gone
def test_no_render_deployment_config():
assert not (REPO / "render.yaml").exists()
# --- no cloud provider, no key ------------------------------------------
def test_settings_expose_no_api_key_field(client):
body = client.get("/api/settings").json()
assert "api_key" not in body
assert "has_api_key" not in body
def test_an_api_key_cannot_be_set_through_the_api(client):
"""Pydantic ignores unknown fields, so this asserts the value does not
land rather than that the request is refused."""
client.put("/api/settings", json={"api_key": "sk-should-not-stick"})
db = SessionLocal()
try:
assert db.query(models.Settings).first().api_key == ""
finally:
db.close()
def test_the_provider_sends_no_authorization_header():
provider = OpenAICompatibleProvider("http://127.0.0.1:11434/v1", "m")
assert "Authorization" not in provider._headers()
# --- the model timeout ---------------------------------------------------
def test_the_default_timeout_is_generous_but_finite():
"""M1 measured a cold model load exceeding the inherited hardcoded 120s on
a CPU-only host. It must be longer than that, and it must be a number."""
assert DEFAULT_READ_TIMEOUT > 120
assert DEFAULT_READ_TIMEOUT <= 3600
def test_connect_stays_short_while_reading_stays_patient():
"""A wrong address should fail in seconds; a loading model should not."""
provider = OpenAICompatibleProvider("http://127.0.0.1:11434/v1", "m")
timeout = provider._timeout()
assert timeout.connect == CONNECT_TIMEOUT <= 30
assert timeout.read == DEFAULT_READ_TIMEOUT
def test_the_timeout_is_configurable(client):
r = client.put("/api/settings", json={"model_timeout_seconds": 900})
assert r.status_code == 200, r.text
assert client.get("/api/settings").json()["model_timeout_seconds"] == 900
@pytest.mark.parametrize("value", [0, 29, 3601, -1])
def test_an_unusable_timeout_is_refused(client, value):
"""Not zero, not negative, and not "wait forever" spelled as a big number."""
assert client.put(
"/api/settings", json={"model_timeout_seconds": value}
).status_code == 422
def test_the_provider_honours_the_configured_timeout():
provider = OpenAICompatibleProvider("http://127.0.0.1:11434/v1", "m", read_timeout=45)
assert provider._timeout().read == 45
# --- the storyteller stays on loopback -----------------------------------
def test_the_native_start_scripts_bind_loopback():
for script in ("start.sh", "start.ps1"):
text = (REPO / script).read_text(errors="ignore")
assert "--host 127.0.0.1" in text, script
assert "--host 0.0.0.0" not in text, script
def test_compose_publishes_to_loopback_only():
"""The container listens on 0.0.0.0 because a published port cannot reach
anything else. What must stay loopback is the *host* side of the mapping."""
text = (REPO / "docker-compose.yml").read_text()
published = re.findall(r'^\s*-\s*"([^"]+)"', text, re.M)
assert published, "no published ports found — has the file moved?"
for mapping in published:
assert mapping.startswith("127.0.0.1:"), mapping
backend/tests/test_memory_rewrite.py
+24 -27
View File
@@ -128,7 +128,7 @@ def fill_bank(db, adventure, *, count=2):
def options(**overrides):
args = dict(write=False, adventure=None, email=None, limit=None,
include_forgotten=False, embed=False, endpoint=None,
model=None, api_key=None)
model=None)
args.update(overrides)
return argparse.Namespace(**args)
@@ -248,23 +248,6 @@ def test_the_rewrite_prompt_is_the_one_the_app_sends(db):
# --------------------------------------------------------------------- the tool
def test_the_database_line_carries_no_password():
"""The report names the database it is about to rewrite. That line ends up
in a console, a screenshot or a pasted bug report."""
shown = rewrite_memories.safe_dsn(
"postgresql://parth:hunter2@ep-cool-frost.us-east-1.aws.neon.tech/aidnd"
"?sslmode=require")
assert "hunter2" not in shown
assert "sslmode" not in shown # a password can be passed there too
assert shown == ("postgresql://parth@ep-cool-frost.us-east-1.aws.neon.tech"
"/aidnd")
def test_an_unparseable_database_url_shows_nothing_at_all():
assert rewrite_memories.safe_dsn("not-a-url") == "(configured)"
def test_without_write_nothing_changes(db, monkeypatch):
adventure = make_adventure(db)
first, _ = fill_bank(db, adventure)
@@ -316,9 +299,12 @@ def test_a_hand_written_memory_is_left_alone(db, monkeypatch):
assert typed.text == "Kaelen owes a debt."
def test_an_owner_with_no_api_key_is_skipped(db, monkeypatch):
"""Summarization spends the user's own key and never the shared demo key."""
adventure = make_adventure(db, key="")
def test_an_adventure_with_no_model_configured_is_skipped(db, monkeypatch):
"""Nothing to summarize with, so the memory is reported rather than
silently left as it was. This used to be about the owner's API key; after
M2 the thing that can be missing is the model name."""
adventure = make_adventure(db)
_blank_model(db, adventure)
first, _ = fill_bank(db, adventure)
stub = StubSummarizer()
monkeypatch.setattr(memorybank, "summary_provider", lambda s: stub)
@@ -328,10 +314,10 @@ def test_an_owner_with_no_api_key_is_skipped(db, monkeypatch):
assert first.text == "You entered the crypt 1."
def test_an_api_key_on_the_command_line_covers_that_owner(db, monkeypatch):
"""The way to summarize for an owner who has no key of their own, and the
way to point a run at the local Claude shim instead of a paid endpoint."""
adventure = make_adventure(db, key="")
def test_a_model_on_the_command_line_covers_that_adventure(db, monkeypatch):
"""How to run the rewrite against a model other than the configured one."""
adventure = make_adventure(db)
_blank_model(db, adventure)
first, _ = fill_bank(db, adventure)
built: list[tuple] = []
@@ -340,11 +326,22 @@ def test_an_api_key_on_the_command_line_covers_that_owner(db, monkeypatch):
return StubSummarizer()
monkeypatch.setattr("app.providers.OpenAICompatibleProvider", build)
assert run_tool(options(write=True, api_key="sk-cli", model="sonnet",
assert run_tool(options(write=True, model="other-model",
endpoint="http://127.0.0.1:8787/v1")) == 0
db.expire_all()
assert first.text == "Kaelen entered the crypt 1."
assert built[0][:3] == ("http://127.0.0.1:8787/v1", "sk-cli", "sonnet")
assert built[0][:2] == ("http://127.0.0.1:8787/v1", "other-model")
def _blank_model(db, adventure) -> None:
settings = (
db.query(models.Settings)
.filter(models.Settings.user_id == adventure.user_id)
.first()
)
settings.model = ""
settings.summary_model = ""
db.commit()
def test_limit_stops_early(db, monkeypatch):
backend/tests/test_netguard.py
-60
View File
@@ -1,60 +0,0 @@
"""Tests for the SSRF guard on the user-supplied BYOK endpoint_url.
python -m pytest tests/test_netguard.py -v
"""
import pytest
from app import auth, netguard
@pytest.fixture
def hosted(monkeypatch):
monkeypatch.setattr(auth, "MULTI_USER", True)
def _resolves_to(monkeypatch, ip: str):
"""Pin `getaddrinfo` so the test exercises the address decision, not real DNS."""
monkeypatch.setattr(
netguard.socket, "getaddrinfo",
lambda *a, **k: [(2, 1, 6, "", (ip, 443))],
)
@pytest.mark.parametrize("ip", [
"127.0.0.1", # loopback
"169.254.169.254", # cloud metadata (link-local)
"10.0.0.5", # RFC1918
"192.168.1.1", # RFC1918
"172.16.0.9", # RFC1918
"0.0.0.0", # unspecified
"100.64.0.1", # carrier-grade NAT
"::1", # IPv6 loopback
"fd00::1", # IPv6 unique-local
])
def test_blocks_non_public_addresses(hosted, monkeypatch, ip):
_resolves_to(monkeypatch, ip)
assert netguard.endpoint_block_reason("https://evil.example.com/v1") is not None
def test_allows_public_address(hosted, monkeypatch):
_resolves_to(monkeypatch, "104.18.0.1") # a public IP
assert netguard.endpoint_block_reason("https://openrouter.ai/api/v1") is None
def test_rejects_non_http_scheme(hosted):
assert netguard.endpoint_block_reason("file:///etc/passwd") is not None
assert netguard.endpoint_block_reason("gopher://x/") is not None
def test_unresolvable_host_is_blocked(hosted, monkeypatch):
def boom(*a, **k):
raise netguard.socket.gaierror("no such host")
monkeypatch.setattr(netguard.socket, "getaddrinfo", boom)
assert netguard.endpoint_block_reason("https://nope.invalid/v1") is not None
def test_noop_in_local_mode(monkeypatch):
monkeypatch.setattr(auth, "MULTI_USER", False)
# Local installs must reach localhost (Ollama). The guard never blocks local mode.
assert netguard.endpoint_block_reason("http://localhost:11434/v1") is None
assert netguard.endpoint_block_reason("http://127.0.0.1:11434/v1") is None
backend/tests/test_persona.py
-1
View File
@@ -275,7 +275,6 @@ def client(monkeypatch):
user_id = user.id
setup.close()
monkeypatch.setattr(limits, "rate_limit", lambda *a, **k: None)
monkeypatch.setattr(limits, "check_row_cap", lambda *a, **k: None)
def _current_user(db=Depends(get_db)):
backend/tests/test_prompt_caching.py
+6 -45
View File
@@ -17,10 +17,10 @@ Three things must hold, and each is easy to undo by accident:
* Moving a section out of the system block does not drop it from the
token budget. It is still in the prompt.
This file also covers two request-level concerns: preferring one
OpenRouter upstream, since each upstream holds its own cache, and reading
back the usage the endpoint reports, so the hit rate is measurable rather
than assumed.
This file also covers one request-level concern: reading back the usage
the endpoint reports, so the hit rate is measurable rather than assumed.
It covered OpenRouter upstream pinning too, until M2 removed cloud
provider support.
python -m pytest tests/test_prompt_caching.py -v
"""
@@ -40,50 +40,11 @@ SCHEMA = {
}
# ------------------------------------------------- preferring one upstream
def _routed(endpoint, model):
provider = OpenAICompatibleProvider(endpoint, "k", model, "chat", 0)
body = {"max_tokens": 100}
provider._apply_provider_routing(body)
return body
def test_openrouter_deepseek_pins_the_upstream():
"""Each upstream has its own cache, so routing has to be deterministic."""
body = _routed("https://openrouter.ai/api/v1", "deepseek/deepseek-v4-flash-0731")
assert body["provider"] == {"order": ["deepseek"]}
def test_fallbacks_stay_on():
"""A preference, not a restriction: if deepseek is down the turn still runs
somewhere else and merely misses the cache."""
body = _routed("https://openrouter.ai/api/v1", "deepseek/deepseek-v4-flash-0731")
assert "allow_fallbacks" not in body["provider"]
def test_non_openrouter_endpoints_get_no_provider_field():
"""Ollama and other providers reject fields they do not know. This is
the same problem the `reasoning` param works around."""
body = _routed("http://localhost:11434/v1", "deepseek/deepseek-v4-flash-0731")
assert "provider" not in body
def test_unknown_vendors_are_left_alone():
"""The vendor half of a slug is not reliably a provider slug: Google's
models are served by "google-ai-studio", and there is no "google". A guess
would be a routing preference naming an upstream that does not exist."""
body = _routed("https://openrouter.ai/api/v1", "google/gemma-4-26b-a4b-it:free")
assert "provider" not in body
# ------------------------------------------------------ reading usage back
def test_usage_is_recorded_from_a_final_chunk():
"""In a stream, the usage block arrives in a final chunk that carries
no choices, which is why it is read separately from the text
extraction."""
provider = OpenAICompatibleProvider("https://openrouter.ai/api/v1", "k", "m")
provider = OpenAICompatibleProvider("http://127.0.0.1:11434/v1", "m")
assert provider.last_usage is None
provider._record_usage({"choices": [{"delta": {"content": "hi"}}]})
assert provider.last_usage is None
@@ -95,7 +56,7 @@ def test_usage_is_recorded_from_a_final_chunk():
def test_a_later_chunk_without_usage_does_not_erase_it():
provider = OpenAICompatibleProvider("https://openrouter.ai/api/v1", "k", "m")
provider = OpenAICompatibleProvider("http://127.0.0.1:11434/v1", "m")
provider._record_usage({"usage": {"prompt_tokens": 5}})
provider._record_usage({"choices": [{"delta": {"content": "x"}}]})
provider._record_usage({"usage": {}})
backend/tests/test_ratelimit_hardening.py
-99
View File
@@ -1,99 +0,0 @@
"""Regression tests for the X-Forwarded-For rate-limit bypass and the
per-account login throttle added to close it.
Background: uvicorn's `--forwarded-allow-ips "*"` trusted the leftmost
`X-Forwarded-For` entry. The client controls that entry, so rotating the
header issued a fresh rate-limit bucket on every request. `client_ip` now
reads the hop the trusted edge appends, which is the rightmost one. Login
also has an email-keyed throttle that no IP trick can weaken.
python -m pytest tests/test_ratelimit_hardening.py -v
"""
import pytest
from app import auth, limits
class _Req:
"""Minimal stand-in for starlette's Request: a header lookup and a peer."""
def __init__(self, xff: str | None, peer: str | None = "10.0.0.1"):
self.headers = {} if xff is None else {"x-forwarded-for": xff}
self.client = None if peer is None else type("C", (), {"host": peer})()
# ---------- client_ip: the spoof-resistant hop ----------
def test_client_ip_takes_appended_rightmost_hop(monkeypatch):
monkeypatch.setattr(limits, "TRUSTED_PROXY_HOPS", 1)
# An attacker prepends a fake IP. The edge appends the real one on the right.
req = _Req("203.0.113.9, 198.51.100.77")
assert limits.client_ip(req) == "198.51.100.77"
def test_client_ip_ignores_spoofed_leftmost(monkeypatch):
monkeypatch.setattr(limits, "TRUSTED_PROXY_HOPS", 1)
# The keyed IP stays the real hop regardless of what the client adds on
# the left, so rotating that value no longer creates a new bucket.
a = limits.client_ip(_Req("1.1.1.1, 198.51.100.77"))
b = limits.client_ip(_Req("2.2.2.2, 198.51.100.77"))
c = limits.client_ip(_Req("evil, junk, 198.51.100.77"))
assert a == b == c == "198.51.100.77"
def test_client_ip_honours_extra_trusted_hops(monkeypatch):
monkeypatch.setattr(limits, "TRUSTED_PROXY_HOPS", 2)
# Two trusted hops: real client is second from the right.
req = _Req("9.9.9.9, 203.0.113.5, 198.51.100.77")
assert limits.client_ip(req) == "203.0.113.5"
def test_client_ip_falls_back_to_socket_peer():
assert limits.client_ip(_Req(None, peer="172.16.0.4")) == "172.16.0.4"
assert limits.client_ip(_Req(None, peer=None)) == "unknown"
# ---------- per-account login throttle ----------
@pytest.fixture(autouse=True)
def _multi_user(monkeypatch):
monkeypatch.setattr(auth, "MULTI_USER", True)
# Isolate the module-level failure map for each test.
from collections import defaultdict, deque
monkeypatch.setattr(limits, "_login_fails", defaultdict(deque))
def test_login_throttle_blocks_after_limit():
email = "victim@example.com"
# Each attempt up to the limit is allowed and recorded as a failure.
for _ in range(limits.LOGIN_FAIL_LIMIT):
limits.check_login_allowed(email) # does not raise
limits.note_login_failure(email)
# One more failure exceeds the limit.
with pytest.raises(limits.HTTPException) as exc:
limits.check_login_allowed(email)
assert exc.value.status_code == 429
def test_login_throttle_is_per_account():
for _ in range(limits.LOGIN_FAIL_LIMIT):
limits.note_login_failure("a@example.com")
with pytest.raises(limits.HTTPException):
limits.check_login_allowed("a@example.com")
# A different account is unaffected because the throttle keys on email, not IP.
limits.check_login_allowed("b@example.com") # must not raise
def test_successful_login_clears_the_streak():
email = "typo@example.com"
for _ in range(limits.LOGIN_FAIL_LIMIT):
limits.note_login_failure(email)
limits.note_login_success(email)
limits.check_login_allowed(email) # failure streak cleared, must not raise
def test_throttle_is_noop_in_local_mode(monkeypatch):
monkeypatch.setattr(auth, "MULTI_USER", False)
for _ in range(limits.LOGIN_FAIL_LIMIT * 3):
limits.note_login_failure("solo@example.com")
limits.check_login_allowed("solo@example.com") # never throttled locally
backend/tests/test_reasoning_param.py
-45
View File
@@ -1,45 +0,0 @@
"""What the provider puts in the `reasoning` request field for each budget
setting: a positive budget asks for thinking, 0 stays silent, -1 turns it off.
python -m pytest tests/test_reasoning_param.py -v
"""
from app.providers.openai_compatible import OpenAICompatibleProvider
def _body(reasoning_max_tokens, api_mode="chat", max_tokens=1000):
provider = OpenAICompatibleProvider(
"https://openrouter.ai/api/v1", "k", "deepseek/deepseek-v4-flash-0731",
api_mode, reasoning_max_tokens,
)
body = {"max_tokens": max_tokens}
provider._apply_reasoning_budget(body)
return body
def test_zero_sends_nothing():
"""Ollama and other providers reject unknown fields. Sending 0 must not add a `reasoning` field."""
assert "reasoning" not in _body(0)
def test_positive_budget_adds_thinking_tokens():
body = _body(500)
assert body["reasoning"] == {"max_tokens": 500}
# the story output keeps its own full budget on top of the thinking budget
assert body["max_tokens"] == 1500
def test_negative_turns_reasoning_off():
body = _body(-1)
assert body["reasoning"] == {"effort": "none"}
# "off" must not inflate the output budget
assert body["max_tokens"] == 1000
def test_off_is_not_merely_excluded():
"""`exclude: true` still generates and bills for reasoning tokens. The off setting must omit the field entirely instead of relying on `exclude`."""
assert _body(-1)["reasoning"].get("exclude") is None
def test_completion_mode_never_sends_reasoning():
for budget in (-1, 0, 500):
assert "reasoning" not in _body(budget, api_mode="completion")
backend/tests/test_retry_variants.py
+15 -28
View File
@@ -14,18 +14,11 @@ from app.main import app
from app.providers import ProviderError
from app.routers import adventures
from fakes import ScriptedProvider
from fakes import GOLD_SCHEMA, ScriptedProvider, gold_replies, gold_reply
SCHEMA = {"player": {"hp": {"min": 0, "max": 100, "initial": 100}}}
SCHEMA = GOLD_SCHEMA
# Each turn spends 10 gold, so a double-applied or un-rolled-back attempt shows.
GOLD_SCRIPT = """
const modifier = (text) => {
state.gold = (state.gold || 0) + 10;
return { text };
};
modifier(text);
"""
@pytest.fixture()
@@ -41,14 +34,11 @@ def client(monkeypatch):
setup.flush()
adv = models.Adventure(
user_id=user.id, title="Cave", scenario_id=scenario.id,
script_state={}, world_state={"player": {"hp": 100}},
world_state={"player": {"hp": 100, "gold": 0}},
)
setup.add(adv)
setup.flush()
setup.add(models.Action(adventure_id=adv.id, type="start", text="You enter a cave."))
setup.add(models.AdventureScript(
adventure_id=adv.id, position=0, enabled=True, name="Gold", output_js=GOLD_SCRIPT,
))
setup.commit()
adv_id, user_id = adv.id, user.id
setup.close()
@@ -57,9 +47,6 @@ def client(monkeypatch):
ScriptedProvider.calls = 0
ScriptedProvider.prompts = []
monkeypatch.setattr(adventures.turns, "OpenAICompatibleProvider", ScriptedProvider)
monkeypatch.setattr(auth, "resolve_provider_config", lambda s: auth.ProviderConfig(
"http://fake", "k", "test-model", False))
monkeypatch.setattr(limits, "rate_limit", lambda *a, **k: None)
monkeypatch.setattr(limits, "check_row_cap", lambda *a, **k: None)
def _current_user(db=Depends(get_db)):
@@ -80,7 +67,7 @@ def _adv(adv_id):
db = SessionLocal()
try:
adv = db.get(models.Adventure, adv_id)
return adv.script_state, adv.world_state
return (adv.world_state or {}).get("player", {}), adv.world_state
finally:
db.close()
@@ -177,15 +164,15 @@ def test_three_attempts_all_kept_in_order(client):
def test_switching_back_restores_that_attempt_state(client):
ScriptedProvider.replies = [
"You take a scratch.\n```state\n{\"player.hp\": -5}\n```",
"You take a beating.\n```state\n{\"player.hp\": -40}\n```",
'You take a scratch.\n```state\n{"player.hp": -5, "player.gold": 10}\n```',
'You take a beating.\n```state\n{"player.hp": -40, "player.gold": 10}\n```',
]
_play(client)
assert _adv(client.adv_id)[1]["player"]["hp"] == 95
_retry(client)
script_state, world_state = _adv(client.adv_id)
player, world_state = _adv(client.adv_id)
assert world_state["player"]["hp"] == 60
assert script_state == {"gold": 10} # rolled back, not stacked to 20
assert player["gold"] == 10 # rolled back, not stacked to 20
last = _actions(client)[-1]
r = client.post(
@@ -194,9 +181,9 @@ def test_switching_back_restores_that_attempt_state(client):
assert r.json()["text"].startswith("You take a scratch")
assert r.json()["take_index"] == 0
# The stats follow the narration back.
script_state, world_state = _adv(client.adv_id)
player, world_state = _adv(client.adv_id)
assert world_state["player"]["hp"] == 95
assert script_state == {"gold": 10}
assert player["gold"] == 10
# And forward again.
client.post(f"/api/adventures/{client.adv_id}/actions/{last['id']}/variant",
@@ -251,26 +238,26 @@ def test_switching_to_a_missing_index_is_rejected(client):
def test_failed_retry_leaves_the_previous_attempt_in_charge(client):
"""A provider error mid-retry must undo the rollback, or the stats on
screen would silently disagree with the text still shown."""
ScriptedProvider.replies = ["Attempt one.", ProviderError("upstream is down")]
ScriptedProvider.replies = [gold_reply("Attempt one."), ProviderError("upstream is down")]
_play(client)
assert _adv(client.adv_id)[0] == {"gold": 10}
assert _adv(client.adv_id)[0]["gold"] == 10
client.post(f"/api/adventures/{client.adv_id}/retry")
actions = _actions(client)
assert actions[-1]["text"] == "Attempt one." # text never lost
assert _adv(client.adv_id)[0] == {"gold": 10} # and the state still matches it
assert _adv(client.adv_id)[0]["gold"] == 10 # and the state still matches it
def test_undo_removes_the_action_and_its_history(client):
ScriptedProvider.replies = ["One.", "Two."]
ScriptedProvider.replies = [gold_reply(t) for t in ["One.", "Two."]]
_play(client)
_retry(client)
r = client.post(f"/api/adventures/{client.adv_id}/undo")
assert r.status_code == 200, r.text
# Undo returns the newest window now, not the whole story.
assert [a["type"] for a in r.json()["actions"]] == ["start"]
assert _adv(client.adv_id)[0] == {}
assert _adv(client.adv_id)[0]["gold"] == 0
def test_editing_the_text_updates_the_live_variant(client):
backend/tests/test_scenario_art.py
-1
View File
@@ -125,7 +125,6 @@ def client(monkeypatch):
user_id = user.id
setup.close()
monkeypatch.setattr(limits, "rate_limit", lambda *a, **k: None)
monkeypatch.setattr(limits, "check_row_cap", lambda *a, **k: None)
def _current_user(db=Depends(get_db)):
backend/tests/test_scenario_refresh.py
-1
View File
@@ -34,7 +34,6 @@ def client(monkeypatch):
user_id = user.id
setup.close()
monkeypatch.setattr(limits, "rate_limit", lambda *a, **k: None)
monkeypatch.setattr(limits, "check_row_cap", lambda *a, **k: None)
def _current_user(db=Depends(get_db)):
backend/tests/test_state_revert.py
+22 -18
View File
@@ -1,6 +1,12 @@
"""Tests for undo and retry rolling back the shared `script_state`
"""Tests for undo and retry rolling back the shared `world_state`
(plan/11-state-revert-and-retry-fix.md).
The state being rolled back was the scripting engine's `script_state` until
M2 removed campaign scripting. The machinery under test — `attempts.restore_state`,
`roll_back_before`, and the per-node outcome snapshot — is unchanged; only the
column it moves has. `world_state`/`world_state_after` is now the shared state
an adventure carries, so that is what these tests exercise.
Phase 14 SP4 reversed the snapshots. An action used to carry the state as
it stood before it ran, and rolling back read the snapshot off the action
being removed. Now it carries the state it left behind, and rolling back
@@ -36,11 +42,11 @@ def db():
adventures.turns._active_turns.clear()
def _make_adventure(db, script_state):
def _make_adventure(db, world_state):
user = models.User(is_guest=False)
db.add(user)
db.flush()
adv = models.Adventure(user_id=user.id, title="T", script_state=script_state)
adv = models.Adventure(user_id=user.id, title="T", world_state=world_state)
db.add(adv)
db.flush()
return user, adv
@@ -49,7 +55,7 @@ def _make_adventure(db, script_state):
def _add(db, adv, index, type_, text="x", state_after=None):
a = models.Action(
adventure_id=adv.id, type=type_, text=text,
state_after=state_after,
world_state_after=state_after,
)
db.add(a)
db.flush()
@@ -72,7 +78,7 @@ def _forget_snapshots(db, adv):
# ---------------------------------------------------------------- undo
def test_undo_reverts_state_to_before_the_turn(db):
# A turn moved script_state from {gold:0} to {gold:10}. The node in
# A turn moved world_state from {gold:0} to {gold:10}. The node in
# front of the turn records where it started. The current state is
# the mutated one.
user, adv = _make_adventure(db, {"gold": 10})
@@ -83,7 +89,7 @@ def test_undo_reverts_state_to_before_the_turn(db):
adventures.undo_turn(adv.id, db=db, adventure=adv)
assert adv.script_state == {"gold": 0}
assert adv.world_state == {"gold": 0}
assert [a.type for a in adv.actions] == ["start"]
@@ -97,7 +103,7 @@ def test_undo_of_bare_continue_uses_the_node_in_front(db):
adventures.undo_turn(adv.id, db=db, adventure=adv)
assert adv.script_state == {"gold": 0}
assert adv.world_state == {"gold": 0}
assert [a.type for a in adv.actions] == ["start"]
@@ -112,7 +118,7 @@ def test_undo_leaves_state_untouched_when_snapshot_missing(db):
adventures.undo_turn(adv.id, db=db, adventure=adv)
assert adv.script_state == {"gold": 10}
assert adv.world_state == {"gold": 10}
def test_undo_raises_when_nothing_to_undo(db):
@@ -188,24 +194,24 @@ def test_snapshot_outcome_is_an_independent_deep_copy(db):
_, adv = _make_adventure(db, {"nested": {"n": 1}})
node = models.Action(adventure_id=adv.id, type="ai", text="x")
attempts.snapshot_outcome(adv, node)
adv.script_state["nested"]["n"] = 99
assert node.state_after == {"nested": {"n": 1}} # unaffected by later mutation
adv.world_state["nested"]["n"] = 99
assert node.world_state_after == {"nested": {"n": 1}} # unaffected by later mutation
def test_snapshot_outcome_handles_non_dict(db):
_, adv = _make_adventure(db, {})
adv.script_state = None
adv.world_state = None
node = models.Action(adventure_id=adv.id, type="ai", text="x")
attempts.snapshot_outcome(adv, node)
assert node.state_after == {}
assert node.world_state_after == {}
def test_restore_state_ignores_a_node_with_no_outcome(db):
_, adv = _make_adventure(db, {"gold": 7})
attempts.restore_state(adv, models.Action(adventure_id=adv.id, type="ai"))
assert adv.script_state == {"gold": 7}
assert adv.world_state == {"gold": 7}
attempts.restore_state(adv, None)
assert adv.script_state == {"gold": 7}
assert adv.world_state == {"gold": 7}
# ---------------------------------------------------------------- retry
@@ -220,8 +226,6 @@ def test_retry_restores_the_state_the_turn_started_from(db, monkeypatch):
_add(db, adv, 2, "ai", state_after={"gold": 20})
db.commit()
monkeypatch.setattr(adventures.limits, "rate_limit", lambda *a, **k: None)
monkeypatch.setattr(adventures.turns, "check_demo_cap", lambda *a, **k: None)
async def _noop(*a, **k):
if False:
@@ -230,12 +234,12 @@ def test_retry_restores_the_state_the_turn_started_from(db, monkeypatch):
adventures.retry_action(adv.id, request=None, db=db, user=user, adventure=adv)
assert adv.script_state == {"gold": 10}
assert adv.world_state == {"gold": 10}
# Nothing is written until a replacement actually arrives: the attempt on
# screen is left exactly as it was, and stays the live one.
assert [a.type for a in adv.actions] == ["start", "do", "ai"]
last = adv.actions[-1]
assert last.live is True
assert len(attempts.group(db, last)) == 1 # no sibling was filed
assert last.state_after == {"gold": 20} # its own outcome, untouched
assert last.world_state_after == {"gold": 20} # its own outcome, untouched
adventures.turns._active_turns.discard(adv.id)
backend/tests/test_story_tree_baseline.py
+21 -34
View File
@@ -26,20 +26,13 @@ from app.database import Base, SessionLocal, engine, get_db
from app.main import app
from app.routers import adventures
from fakes import ScriptedProvider
from fakes import GOLD_SCHEMA, ScriptedProvider, gold_replies, gold_reply
# A world-state schema, so the RPG layer is exercised rather than skipped.
SCHEMA = {"player": {"hp": {"min": 0, "max": 100, "initial": 100}}}
SCHEMA = GOLD_SCHEMA
# Ten gold a turn. A turn that double-applies its hooks, or one that fails to
# roll back on retry, shows up here as a wrong total rather than as nothing.
GOLD_SCRIPT = """
const modifier = (text) => {
state.gold = (state.gold || 0) + 10;
return { text };
};
modifier(text);
"""
OPENING = "You enter a cave."
@@ -59,7 +52,7 @@ def _make_world(monkeypatch, *, seeded_actions: int = 0):
setup.flush()
adv = models.Adventure(
user_id=user.id, title="Cave", scenario_id=scenario.id,
script_state={}, world_state={"player": {"hp": 100}},
world_state={"player": {"hp": 100, "gold": 0}},
)
setup.add(adv)
setup.flush()
@@ -70,20 +63,14 @@ def _make_world(monkeypatch, *, seeded_actions: int = 0):
type="ai" if i % 2 else "do",
text=f"Seeded turn {i}.",
))
setup.add(models.AdventureScript(
adventure_id=adv.id, position=0, enabled=True, name="Gold", output_js=GOLD_SCRIPT,
))
setup.commit()
adv_id, user_id = adv.id, user.id
setup.close()
ScriptedProvider.replies = ["A reply."]
ScriptedProvider.replies = [gold_reply(t) for t in ["A reply."]]
ScriptedProvider.calls = 0
ScriptedProvider.prompts = []
monkeypatch.setattr(adventures.turns, "OpenAICompatibleProvider", ScriptedProvider)
monkeypatch.setattr(auth, "resolve_provider_config", lambda s: auth.ProviderConfig(
"http://fake", "k", "test-model", False))
monkeypatch.setattr(limits, "rate_limit", lambda *a, **k: None)
monkeypatch.setattr(limits, "check_row_cap", lambda *a, **k: None)
def _current_user(db=Depends(get_db)):
@@ -145,7 +132,7 @@ def _state(adv_id):
db = SessionLocal()
try:
adv = db.get(models.Adventure, adv_id)
return adv.script_state, adv.world_state
return (adv.world_state or {}).get("player", {}), adv.world_state
finally:
db.close()
@@ -172,7 +159,7 @@ def test_short_adventure_returns_everything(client):
# --------------------------------------------------------------- the turn
def test_a_turn_appends_the_player_action_then_the_ai_action(client):
ScriptedProvider.replies = ["The dark presses in."]
ScriptedProvider.replies = [gold_reply(t) for t in ["The dark presses in."]]
_play(client, "light a torch")
actions = _actions(client)
assert [a["type"] for a in actions] == ["start", "do", "ai"]
@@ -181,7 +168,7 @@ def test_a_turn_appends_the_player_action_then_the_ai_action(client):
def test_say_and_story_and_continue_all_work(client):
ScriptedProvider.replies = ["One.", "Two.", "Three."]
ScriptedProvider.replies = [gold_reply(t) for t in ["One.", "Two.", "Three."]]
_play(client, "hello", type="say")
_play(client, "The wind rises.", type="story")
_play(client, "", type="continue")
@@ -190,7 +177,7 @@ def test_say_and_story_and_continue_all_work(client):
def test_the_story_so_far_is_replayed_into_the_prompt(client):
ScriptedProvider.replies = ["First.", "Second."]
ScriptedProvider.replies = [gold_reply(t) for t in ["First.", "Second."]]
_play(client, "go north")
_play(client, "go south")
story = ScriptedProvider.prompts[-1][1]
@@ -201,7 +188,7 @@ def test_the_story_so_far_is_replayed_into_the_prompt(client):
def test_scripts_run_once_per_turn(client):
"""The gold script adds ten a turn. Two turns is twenty — not forty."""
ScriptedProvider.replies = ["One.", "Two."]
ScriptedProvider.replies = [gold_reply(t) for t in ["One.", "Two."]]
_play(client)
_play(client)
script_state, _ = _state(client.adv_id)
@@ -211,7 +198,7 @@ def test_scripts_run_once_per_turn(client):
# ------------------------------------------------------------------ retry
def test_retry_replaces_the_text_and_keeps_the_attempt(client):
ScriptedProvider.replies = ["Attempt one.", "Attempt two."]
ScriptedProvider.replies = [gold_reply(t) for t in ["Attempt one.", "Attempt two."]]
_play(client)
assert _texts(client)[-1] == "Attempt one."
@@ -233,7 +220,7 @@ def test_retry_replaces_the_text_and_keeps_the_attempt(client):
def test_retry_does_not_stack_script_effects(client):
"""The discarded attempt's ten gold is rolled back, so one turn plus one
retry is still ten, not twenty."""
ScriptedProvider.replies = ["Attempt one.", "Attempt two."]
ScriptedProvider.replies = [gold_reply(t) for t in ["Attempt one.", "Attempt two."]]
_play(client)
client.post(f"/api/adventures/{client.adv_id}/retry")
script_state, _ = _state(client.adv_id)
@@ -241,7 +228,7 @@ def test_retry_does_not_stack_script_effects(client):
def test_switching_back_to_an_earlier_attempt_restores_it(client):
ScriptedProvider.replies = ["Attempt one.", "Attempt two."]
ScriptedProvider.replies = [gold_reply(t) for t in ["Attempt one.", "Attempt two."]]
_play(client)
client.post(f"/api/adventures/{client.adv_id}/retry")
action_id = _actions(client)[-1]["id"]
@@ -259,7 +246,7 @@ def test_switching_back_to_an_earlier_attempt_restores_it(client):
def test_only_the_newest_turn_can_be_switched(client):
"""An older turn's alternatives stay readable but not selectable. The
story after it continues from what is live."""
ScriptedProvider.replies = ["One.", "Again.", "Two."]
ScriptedProvider.replies = [gold_reply(t) for t in ["One.", "Again.", "Two."]]
_play(client)
client.post(f"/api/adventures/{client.adv_id}/retry")
older_id = _actions(client)[-1]["id"]
@@ -274,7 +261,7 @@ def test_only_the_newest_turn_can_be_switched(client):
# ------------------------------------------------------------------- undo
def test_undo_removes_the_whole_turn_and_rolls_state_back(client):
ScriptedProvider.replies = ["One.", "Two."]
ScriptedProvider.replies = [gold_reply(t) for t in ["One.", "Two."]]
_play(client, "go north")
_play(client, "go south")
assert len(_actions(client)) == 5
@@ -365,7 +352,7 @@ def test_paging_past_the_start_reports_the_end(long_client):
# ------------------------------------------------------- editing and deleting
def test_editing_an_action_sticks(client):
ScriptedProvider.replies = ["Original."]
ScriptedProvider.replies = [gold_reply(t) for t in ["Original."]]
_play(client)
action_id = _actions(client)[-1]["id"]
r = client.patch(f"/api/adventures/{client.adv_id}/actions/{action_id}",
@@ -380,7 +367,7 @@ def test_editing_an_action_sticks(client):
def test_editing_a_retried_action_survives_a_reload(client):
"""The edit has to reach the live attempt too, or paging away and back
reverts it."""
ScriptedProvider.replies = ["One.", "Two."]
ScriptedProvider.replies = [gold_reply(t) for t in ["One.", "Two."]]
_play(client)
client.post(f"/api/adventures/{client.adv_id}/retry")
action_id = _actions(client)[-1]["id"]
@@ -390,7 +377,7 @@ def test_editing_a_retried_action_survives_a_reload(client):
def test_deleting_an_action_removes_it(client):
ScriptedProvider.replies = ["One."]
ScriptedProvider.replies = [gold_reply(t) for t in ["One."]]
_play(client)
action_id = _actions(client)[-1]["id"]
r = client.delete(f"/api/adventures/{client.adv_id}/actions/{action_id}")
@@ -431,7 +418,7 @@ def test_export_carries_the_whole_story(client):
same fact: a bundle that stores coordinates has no use for a repeating
group. Everything else here still passes unmodified.
"""
ScriptedProvider.replies = ["One.", "Two."]
ScriptedProvider.replies = [gold_reply(t) for t in ["One.", "Two."]]
_play(client, "go north")
_play(client, "go south")
@@ -446,7 +433,7 @@ def test_export_carries_the_whole_story(client):
def test_export_round_trips_through_import(client):
ScriptedProvider.replies = ["One."]
ScriptedProvider.replies = [gold_reply(t) for t in ["One."]]
_play(client, "go north")
bundle = client.get(f"/api/adventures/{client.adv_id}/export").json()
@@ -461,7 +448,7 @@ def test_export_round_trips_through_import(client):
def test_export_keeps_retry_attempts(client):
ScriptedProvider.replies = ["Attempt one.", "Attempt two."]
ScriptedProvider.replies = [gold_reply(t) for t in ["Attempt one.", "Attempt two."]]
_play(client)
client.post(f"/api/adventures/{client.adv_id}/retry")
@@ -474,7 +461,7 @@ def test_export_keeps_retry_attempts(client):
# -------------------------------------------------------------- world state
def test_world_state_is_readable_and_survives_a_turn(client):
ScriptedProvider.replies = ["Nothing changes."]
ScriptedProvider.replies = [gold_reply(t) for t in ["Nothing changes."]]
_play(client)
r = client.get(f"/api/adventures/{client.adv_id}/world-state")
assert r.status_code == 200, r.text
backend/tests/test_take_edit.py
-3
View File
@@ -57,9 +57,6 @@ def client(monkeypatch):
ScriptedProvider.replies = [f"Take {n}." for n in range(1, 40)]
ScriptedProvider.calls = 0
monkeypatch.setattr(adventures.turns, "OpenAICompatibleProvider", ScriptedProvider)
monkeypatch.setattr(auth, "resolve_provider_config", lambda s: auth.ProviderConfig(
"http://fake", "k", "test-model", False))
monkeypatch.setattr(limits, "rate_limit", lambda *a, **k: None)
monkeypatch.setattr(limits, "check_row_cap", lambda *a, **k: None)
def _current_user(db=Depends(get_db)):
backend/tests/test_take_parentage.py
-6
View File
@@ -58,12 +58,6 @@ def client(monkeypatch):
ScriptedProvider.replies = [f"Take {n}." for n in range(1, 40)]
ScriptedProvider.calls = 0
monkeypatch.setattr(adventures.turns, "OpenAICompatibleProvider", ScriptedProvider)
monkeypatch.setattr(
auth,
"resolve_provider_config",
lambda s: auth.ProviderConfig("http://fake", "k", "test-model", False),
)
monkeypatch.setattr(limits, "rate_limit", lambda *a, **k: None)
monkeypatch.setattr(limits, "check_row_cap", lambda *a, **k: None)
def _current_user(db=Depends(get_db)):
backend/tests/test_take_state.py
+9 -21
View File
@@ -24,20 +24,14 @@ from app.database import Base, SessionLocal, engine, get_db
from app.main import app
from app.routers import adventures
from fakes import ScriptedProvider
from fakes import GOLD_SCHEMA, ScriptedProvider, gold_replies, gold_reply
SCHEMA = {"player": {"hp": {"min": 0, "max": 100, "initial": 100}}}
SCHEMA = GOLD_SCHEMA
# Ten gold a turn, every turn. A number that only ever increases makes a
# rollback failure obvious: if a take stacks instead of replacing, the gold
# total is off by exactly one turn's worth.
GOLD_SCRIPT = """
const modifier = (text) => {
state.gold = (state.gold || 0) + 10;
return { text };
};
modifier(text);
"""
# Ten gold a turn, every turn, banked through the world-state engine. A number
# that only ever increases makes a rollback failure obvious: if a take stacks
# instead of replacing, the gold total is off by exactly one turn's worth. See
# `fakes.gold_reply` for why this is not a script any more.
@pytest.fixture()
@@ -53,24 +47,18 @@ def client(monkeypatch):
setup.flush()
adv = models.Adventure(
user_id=user.id, title="Vault", scenario_id=scenario.id,
script_state={}, world_state={"player": {"hp": 100}},
world_state={"player": {"hp": 100, "gold": 0}},
)
setup.add(adv)
setup.flush()
setup.add(models.Action(adventure_id=adv.id, type="start", text="You begin."))
setup.add(models.AdventureScript(
adventure_id=adv.id, position=0, enabled=True, name="Gold", output_js=GOLD_SCRIPT,
))
setup.commit()
adv_id, user_id = adv.id, user.id
setup.close()
ScriptedProvider.replies = [f"Take {n}." for n in range(1, 40)]
ScriptedProvider.replies = gold_replies("Take", 39)
ScriptedProvider.calls = 0
monkeypatch.setattr(adventures.turns, "OpenAICompatibleProvider", ScriptedProvider)
monkeypatch.setattr(auth, "resolve_provider_config", lambda s: auth.ProviderConfig(
"http://fake", "k", "test-model", False))
monkeypatch.setattr(limits, "rate_limit", lambda *a, **k: None)
monkeypatch.setattr(limits, "check_row_cap", lambda *a, **k: None)
def _current_user(db=Depends(get_db)):
@@ -107,7 +95,7 @@ def _take(client, action_id, text=""):
def _gold(adv_id) -> int:
db = SessionLocal()
try:
return (db.get(models.Adventure, adv_id).script_state or {}).get("gold", 0)
return (db.get(models.Adventure, adv_id).world_state or {}).get("player", {}).get("gold", 0)
finally:
db.close()
backend/tests/test_tree_migration.py
-1
View File
@@ -397,7 +397,6 @@ def client(monkeypatch):
user_id, scenario_id = user.id, scenario.id
setup.close()
monkeypatch.setattr(limits, "rate_limit", lambda *a, **k: None)
monkeypatch.setattr(limits, "check_row_cap", lambda *a, **k: None)
def _current_user(db=Depends(get_db)):
backend/tests/test_turn_flow_integration.py
+28 -33
View File
@@ -1,9 +1,11 @@
"""End-to-end HTTP tests for undo and retry state revert. These tests drive
real turns through the actual routes and scripting engine, with only the LLM
provider mocked.
real turns through the actual routes and the world-state engine, with only the
LLM provider mocked.
A script's output hook adds 10 gold each turn. These tests confirm that the
adventure's stored gold total stays correct across play, undo, and retry.
The model banks 10 gold each turn through its ```state delta block. These tests
confirm that the adventure's stored gold total stays correct across play, undo,
and retry. It was a JavaScript output hook that added the gold until M2 removed
campaign scripting; the state machinery under test is the same either way.
python -m pytest tests/test_turn_flow_integration.py -v
"""
@@ -16,15 +18,8 @@ from app.database import Base, SessionLocal, engine, get_db
from app.main import app
from app.routers import adventures
from fakes import ScriptedProvider
from fakes import GOLD_SCHEMA, ScriptedProvider, gold_replies, gold_reply
GOLD_SCRIPT = """
const modifier = (text) => {
state.gold = (state.gold || 0) + 10;
return { text };
};
modifier(text);
"""
AI_REPLY = "The torch flickers as you press onward."
@@ -38,24 +33,21 @@ def client(monkeypatch):
setup.add(user)
setup.flush()
setup.add(models.Settings(user_id=user.id, api_key="enc:dummy", model="test-model"))
adv = models.Adventure(user_id=user.id, title="Cave", script_state={})
scenario = models.Scenario(user_id=user.id, title="S", stat_schema=GOLD_SCHEMA)
setup.add(scenario)
setup.flush()
adv = models.Adventure(user_id=user.id, title="Cave", scenario_id=scenario.id,
world_state={"player": {"hp": 100, "gold": 0}})
setup.add(adv)
setup.flush()
setup.add(models.Action(adventure_id=adv.id, type="start", text="You enter a cave."))
setup.add(models.AdventureScript(
adventure_id=adv.id, position=0, enabled=True, name="Gold",
output_js=GOLD_SCRIPT,
))
setup.commit()
adv_id, user_id = adv.id, user.id
setup.close()
# Force a real, non-demo turn that uses the fake provider.
ScriptedProvider.replies = [AI_REPLY]
ScriptedProvider.replies = [gold_reply(AI_REPLY)]
monkeypatch.setattr(adventures.turns, "OpenAICompatibleProvider", ScriptedProvider)
monkeypatch.setattr(auth, "resolve_provider_config", lambda s: auth.ProviderConfig(
"http://fake", "k", "test-model", False))
monkeypatch.setattr(limits, "rate_limit", lambda *a, **k: None)
monkeypatch.setattr(limits, "check_row_cap", lambda *a, **k: None)
def _current_user(db=Depends(get_db)):
@@ -73,10 +65,13 @@ def client(monkeypatch):
Base.metadata.drop_all(bind=engine)
def _state(adv_id):
def _gold(adv_id) -> int:
"""The banked total. Reading one stat rather than the whole section keeps
these assertions about the rollback, not about the schema's shape."""
db = SessionLocal()
try:
return db.get(models.Adventure, adv_id).script_state
player = (db.get(models.Adventure, adv_id).world_state or {}).get("player", {})
return player.get("gold", 0)
finally:
db.close()
@@ -88,38 +83,38 @@ def _play(client, type_="do", text="look around"):
def test_play_then_undo_reverts_gold(client):
assert _state(client.adv_id) == {}
assert _gold(client.adv_id) == 0
_play(client)
assert _state(client.adv_id) == {"gold": 10}
assert _gold(client.adv_id) == 10
r = client.post(f"/api/adventures/{client.adv_id}/undo")
assert r.status_code == 200, r.text
assert _state(client.adv_id) == {} # gold reverted to zero
assert _gold(client.adv_id) == 0 # gold reverted to zero
def test_two_turns_then_undo_reverts_only_last(client):
_play(client)
_play(client)
assert _state(client.adv_id) == {"gold": 20}
assert _gold(client.adv_id) == 20
client.post(f"/api/adventures/{client.adv_id}/undo")
assert _state(client.adv_id) == {"gold": 10} # back to after turn 1, not 0
assert _gold(client.adv_id) == 10 # back to after turn 1, not 0
def test_retry_does_not_double_apply_gold(client):
_play(client)
assert _state(client.adv_id) == {"gold": 10}
assert _gold(client.adv_id) == 10
# Before the fix, this produced 20 because the output hook ran twice.
# Before the fix, this produced 20 because the turn's effects ran twice.
# Now it stays 10.
r = client.post(f"/api/adventures/{client.adv_id}/retry")
assert r.status_code == 200, r.text
assert _state(client.adv_id) == {"gold": 10}
assert _gold(client.adv_id) == 10
def test_retry_then_undo_still_clean(client):
_play(client)
client.post(f"/api/adventures/{client.adv_id}/retry")
assert _state(client.adv_id) == {"gold": 10}
assert _gold(client.adv_id) == 10
client.post(f"/api/adventures/{client.adv_id}/undo")
assert _state(client.adv_id) == {}
assert _gold(client.adv_id) == 0
backend/tests/test_worldstate_integration.py
-3
View File
@@ -63,9 +63,6 @@ def client(monkeypatch):
ScriptedProvider.replies = [AI_REPLY]
monkeypatch.setattr(adventures.turns, "OpenAICompatibleProvider", ScriptedProvider)
monkeypatch.setattr(auth, "resolve_provider_config", lambda s: auth.ProviderConfig(
"http://fake", "k", "test-model", False))
monkeypatch.setattr(limits, "rate_limit", lambda *a, **k: None)
monkeypatch.setattr(limits, "check_row_cap", lambda *a, **k: None)
def _current_user(db=Depends(get_db)):