mirror of
https://github.com/Heretek-AI/AetherTable.git
synced 2026-11-09 09:49:33 -10:00
John SmithandClaude
fa18ebfff8
Closes audit A1 blockers F1-F4: /agent/turn gated to GM or bound-session participants; dynasty inject-lore and simulation tick GM-only; empirical- benchmark GM-only with 5/min bucket; intent/orchestrator/narrative routes authenticated in a 30/min llm bucket. Client stream/dynasty call sites send credentials with honest sign-in gating. Compose now passes overridable VITE_YSYNC_WS_URL to vtt-client. 33 red-first pytest tests, 6 vitest; pytest 651 passed / 22 skipped, vitest 102. Co-Authored-By: Claude <[email protected]>
121 lines
4.6 KiB
Python
121 lines
4.6 KiB
Python
"""Tests for SRD-grounded narration and the X-card safety rewind flow."""
|
|
|
|
import json
|
|
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
|
|
from vtt_orchestrator.server import _sign_token, app, extract_srd_context
|
|
|
|
client = TestClient(app)
|
|
|
|
|
|
def _auth_token(user_id: str, role: str = "player") -> str:
|
|
"""Valid HMAC session token for authenticated proxy calls."""
|
|
import time
|
|
|
|
return _sign_token({"user_id": user_id, "role": role, "exp": time.time() + 600})
|
|
|
|
|
|
def _stream_text(intent: str, action: str) -> str:
|
|
with client.stream(
|
|
"POST",
|
|
"/api/v1/orchestrator/narrative/stream",
|
|
headers={"Authorization": "Bearer " + _auth_token("usr_srd_gm", "gm")},
|
|
json={
|
|
"user_intent": intent,
|
|
"engine_execution_payload": {"action_name": action, "is_hit": True, "total_damage": 20},
|
|
},
|
|
) as resp:
|
|
assert resp.status_code == 200
|
|
raw = "".join(chunk for chunk in resp.iter_text())
|
|
return "".join(
|
|
json.loads(msg[6:])["token"]
|
|
for msg in raw.split("\n\n")
|
|
if msg.startswith("data: ") and json.loads(msg[6:]).get("token")
|
|
)
|
|
|
|
|
|
class TestSrdGrounding:
|
|
def test_extract_finds_monster_statblock(self):
|
|
facts = extract_srd_context("I charge the Adult Red Dragon")
|
|
assert any(f["type"] == "monster" and f["name"] == "Adult Red Dragon" for f in facts)
|
|
dragon = next(f for f in facts if f["name"] == "Adult Red Dragon")
|
|
assert dragon["ac"] == 19
|
|
assert "Multiattack" in dragon["action_names"]
|
|
|
|
def test_extract_finds_spell(self):
|
|
facts = extract_srd_context("I cast Fireball into the room")
|
|
fireball = next((f for f in facts if f.get("name") == "Fireball"), None)
|
|
assert fireball is not None
|
|
assert fireball["level_name"] == "Level 3"
|
|
assert fireball["school"].lower() == "evocation"
|
|
|
|
def test_no_reference_returns_empty(self):
|
|
assert extract_srd_context("I check the door for traps") == []
|
|
|
|
def test_lore_lookup_endpoint(self):
|
|
resp = client.get("/api/v1/compendium/lore-lookup", params={"q": "the Lich"})
|
|
assert resp.status_code == 200
|
|
assert resp.json()["facts"], "Lich must resolve from the bestiary"
|
|
|
|
def test_narration_contains_srd_fact(self):
|
|
text = _stream_text("I cast Fireball at the goblins", "Fireball")
|
|
assert "Level 3" in text
|
|
assert "Evocation" in text
|
|
|
|
|
|
class TestXCard:
|
|
def test_x_card_records_intervention_without_engine(self):
|
|
# X-card is state-affecting: an authenticated token is required, and a
|
|
# non-GM may only file the intervention under their own user id.
|
|
resp = client.post(
|
|
"/api/v1/safety/x-card",
|
|
params={"token": _auth_token("usr_test", "player")},
|
|
json={
|
|
"player_id": "usr_test",
|
|
"topic": "spiders",
|
|
"current_sequence_id": 7,
|
|
},
|
|
)
|
|
assert resp.status_code == 200
|
|
body = resp.json()
|
|
assert body["status"] == "SAFETY_INTERVENTION_ACTIVATED"
|
|
assert body["target_sequence_id"] == 6
|
|
assert "engine_rewind" not in body, "no rewind attempted without a bound session"
|
|
|
|
def test_anonymous_x_card_is_rejected(self):
|
|
resp = client.post(
|
|
"/api/v1/safety/x-card",
|
|
json={"player_id": "usr_test", "topic": "spiders", "current_sequence_id": 7},
|
|
)
|
|
assert resp.status_code == 401
|
|
|
|
def test_x_card_rewinds_engine_session_when_online(self):
|
|
# Create a session through the proxy; if the engine is down, skip.
|
|
# A GM token is used because out-of-band sessions have no lobby
|
|
# membership record a player could be validated against.
|
|
create = client.post(
|
|
"/api/v1/engine/session",
|
|
params={"token": _auth_token("safety-gm", "gm")},
|
|
json={"session_name": "safety-test"},
|
|
)
|
|
if create.status_code != 200:
|
|
pytest.skip("vtt-server engine not running")
|
|
|
|
session_id = create.json()["session_id"]
|
|
resp = client.post(
|
|
"/api/v1/safety/x-card",
|
|
params={"token": _auth_token("safety-gm", "gm")},
|
|
json={
|
|
"player_id": "safety-gm",
|
|
"topic": "torture",
|
|
"current_sequence_id": 5,
|
|
"engine_session_id": session_id,
|
|
},
|
|
)
|
|
assert resp.status_code == 200
|
|
engine_result = resp.json()["engine_rewind"]
|
|
# Either applied on the live ledger or explicitly unavailable offline.
|
|
assert engine_result["status"] in ("SAFETY_REWIND_SUCCESS", "ENGINE_UNAVAILABLE")
|