tests/test_tarot_progression.py
Rama master · réplica de lectura del repositorio original.
"""Tarot reading-progression tests (WebKit / playwright).
The /tarot reader is a client-side state machine: a turn streams from
/api/tarot/chat, a typewriter reveals it (silent or audio-paced), and the
querent's input must come back when the turn ends. The invariant these tests
pin down is that the reading ALWAYS PROGRESSES — whatever the server, network,
SSE framing, or voice layer does, the turn settles (streaming clears, the input
bar is usable again) and never freezes the page.
They drive the real page in WebKit (iOS-engine parity — see the project memory),
mocking only the boundaries: /api/tarot/chat (the SSE turn), the vendored marked (a
no-op shim so reveal text == buffered text), /tarot-voice.js (a controllable
fake voice, to exercise the audio-failure / stall fallbacks without a TTS box),
and the two canned-opening endpoints — /api/tarot/opening (which answers "no
clip" by default, so the turn under test is the LIVE one) and /api/tarot/warm
(which must never reach the home GPU box from a test). Nothing here hits the
LLM or the hosaka WebSocket.
Marked `browser` so the general smoke step skips them (`-m "not browser"`) —
playwright + WebKit are heavier than the HTTP suite. They DO run in pre-commit
as a dedicated step whenever the reader changes (its JS / template / api / this
test), so a hang/freeze can't ship. Run by hand with:
.venv/bin/playwright install webkit # once
.venv/bin/pytest tests/test_tarot_progression.py -q
Skips cleanly (like the smoke suite) when playwright isn't installed, the WebKit
browser is absent, or the app isn't reachable.
"""
import io
import json
import re
import wave
import pytest
playwright_api = pytest.importorskip("playwright.sync_api")
PWError = playwright_api.Error # used by open_tarot's wait_for_selector guard
pytestmark = pytest.mark.browser
# ── SSE / route helpers ────────────────────────────────────────────────────
def sse(*events: dict) -> str:
"""Frame events as the SSE the reader stream emits: `data: <json>\\n\\n`."""
return "".join(f"data: {json.dumps(e)}\n\n" for e in events)
def txt(delta: str) -> dict:
return {"type": "text", "delta": delta}
def tool(name: str, **input_) -> dict:
return {"type": "tool_call", "name": name, "input": input_, "count": 1}
def fulfill_sse(body: str):
"""A /api/tarot/chat route handler that streams `body` as 200 event-stream."""
def handler(route):
route.fulfill(status=200, content_type="text/event-stream", body=body)
return handler
def fulfill_json(payload: dict):
def handler(route):
route.fulfill(status=200, content_type="application/json", body=json.dumps(payload))
return handler
def wav_bytes(seconds: float = 0.4) -> bytes:
"""A real (silent) 24 kHz mono WAV — enough for the page to fetch and hand
to the voice layer, without a TTS box in the loop."""
buf = io.BytesIO()
with wave.open(buf, "wb") as w:
w.setnchannels(1)
w.setsampwidth(2)
w.setframerate(24000)
w.writeframes(b"\x00\x00" * int(24000 * seconds))
return buf.getvalue()
def fulfill_js(js: str):
# Single-arg handler: playwright invokes route handlers as (route, request),
# so a `js=...` default param would be clobbered by the request positional.
def handler(route):
route.fulfill(status=200, content_type="application/javascript", body=js)
return handler
# Offline, deterministic marked: parse(x) == x, so revealed innerText is exactly
# the buffered stream text (no markdown rewriting to assert around).
_marked = fulfill_js("window.marked={use(){},parse:s=>s};")
# The stub voices below REPLACE tarot-voice.js wholesale, so each one has to
# carry the module's whole surface — startOpeningTurn calls probeHome() and
# homeDown() before a turn even begins, and a missing method throws there, not
# where it is used. Add any new tarotVoice export to all three.
# A fake voice whose narration FAILS immediately: streamResponse takes the audio
# branch, the typewriter bails to the guessed pace, the reading still finishes.
VOICE_FAIL = """
window.tarotVoice = {
ready: () => true,
wantsDeferredOpening: () => false,
armOpeningUnlock: () => {},
armPersistedUnlock: () => {},
isOn: () => true,
probeHome: () => Promise.resolve(true),
homeDown: () => false,
speak: () => ({ ok:false, ended:true, error:'stub voice fail',
elapsed:()=>0, duration:()=>0 }),
speakClip: () => ({ ok:false, ended:true, error:'stub clip fail',
elapsed:()=>0, duration:()=>0 }),
};
"""
# A fake voice that connects but never streams a single PCM chunk (duration
# stays 0) — the "no audio at all" case the first-audio grace must catch and
# finish at the guessed pace after the window elapses.
VOICE_STALL = """
window.tarotVoice = {
ready: () => true,
wantsDeferredOpening: () => false,
armOpeningUnlock: () => {},
armPersistedUnlock: () => {},
isOn: () => true,
probeHome: () => Promise.resolve(true),
homeDown: () => false,
speak: () => ({ ok:true, ended:false, error:null,
elapsed:()=>0, duration:()=>0 }),
speakClip: () => ({ ok:true, ended:false, error:null,
elapsed:()=>0, duration:()=>0 }),
};
"""
# A fake voice that buffers audio (duration>0) but whose playback clock never
# advances — the mid-stream freeze the 2.5s stall watchdog must catch once audio
# is flowing. Some audio buffered, so this is NOT flagged "voice unavailable".
VOICE_MIDSTALL = """
window.tarotVoice = {
ready: () => true,
wantsDeferredOpening: () => false,
armOpeningUnlock: () => {},
armPersistedUnlock: () => {},
isOn: () => true,
probeHome: () => Promise.resolve(true),
homeDown: () => false,
speak: () => ({ ok:true, ended:false, error:null,
elapsed:()=>0, duration:()=>1.5 }),
speakClip: () => ({ ok:true, ended:false, error:null,
elapsed:()=>0, duration:()=>1.5 }),
};
"""
# The reading has SETTLED: not streaming, and the turn left a result behind
# (revealed reader text, or a sys note for a prose-less / errored turn).
SETTLED = """() => {
if (typeof streaming === 'undefined' || streaming) return false;
const t = document.getElementById('terminal');
if (!t) return false;
const reader = t.querySelector('.msg.assistant .msg-body');
const readerDone = !!(reader && reader.innerText.trim().length > 0);
const status = document.getElementById('tarot-statusline');
const statusDone = !!(status && status.innerText.trim().length > 0);
return readerDone || statusDone;
}"""
# ── fixtures ───────────────────────────────────────────────────────────────
# The session-scoped WebKit `browser` fixture lives in conftest.py, shared with
# the other browser-marked files (one sync_playwright per session — see the note
# there). This module only adds the per-test page wiring.
@pytest.fixture
def open_tarot(browser, base_url, admin_headers):
"""Open /tarot in a fresh context with the boundaries mocked.
`chat_handler` answers /api/tarot/chat (this turn). Reader narration is
audible by default, so the opening is eager-generated on load but HELD for
the first gesture (browsers won't autoplay audio) — install routes BEFORE
goto, then tap to begin the reading. The stub voices (voice_js) report no
deferral, so for them the opening auto-fires and the tap is a harmless no-op.
`home_voice` is what /api/hosaka/health reports for the home GPU box —
False is "the box that renders the reader's voice is unreachable".
`clip` is what /api/tarot/opening answers. None (the default) means "this
hour has no pre-generated opening", which is what puts the opening turn back
on the LIVE path every test below exercises; pass a dict to exercise the
canned path instead. `audio` is the body served for that clip's .wav.
"""
contexts = []
def _open(chat_handler, *, voice_js=None, init_script=None, clip=None, audio=None,
home_voice=True):
ctx = browser.new_context(
extra_http_headers={"Authorization": admin_headers["Authorization"]})
contexts.append(ctx)
pg = ctx.new_page()
pg.route("**/vendor/marked-*.min.js", _marked)
if voice_js is not None:
pg.route("**/tarot-voice.js*", fulfill_js(voice_js))
# Warming loads models on a home GPU box — never from a test.
pg.route("**/api/tarot/warm", fulfill_json({"ok": True, "skipped": "test"}))
# The reader's-voice probe. Default UP: these run against the live app,
# whose home box may genuinely be down, and the "voice offline" note is
# latest-wins in the same status bar the other tests read.
pg.route("**/api/hosaka/health",
fulfill_json({"ok": True, "home": home_voice, "piper": True}))
pg.route(re.compile(r"/api/tarot/opening(\?|$)"), fulfill_json({"clip": clip}))
if audio is not None:
pg.route(re.compile(r"/api/tarot/opening/.*\.wav$"),
lambda r: r.fulfill(status=200, content_type="audio/wav", body=audio))
pg.route("**/api/tarot/chat", chat_handler)
if init_script:
pg.add_init_script(init_script)
pg.goto(f"{base_url}/tarot", wait_until="domcontentloaded")
# Audible-by-default opening holds for the first gesture; tap to start it.
try:
pg.wait_for_selector("body.opening-pending", timeout=3000)
pg.mouse.click(200, 400)
except PWError:
pass # not held (stub voice) -> already auto-fired
return pg
yield _open
for c in contexts:
try:
c.close()
except Exception:
pass
# GENEROUS BY DEFAULT. These tests assert that a turn SETTLES, never how fast:
# the box is 1967MB with two Claude sessions on it, and a second headless WebKit
# elsewhere on the machine is enough to push a reveal past 10s. A tight timeout
# here does not pin anything -- it just fails a healthy page under load, which
# it did on three separate runs (happy_path, malformed_sse, the canned opening),
# each passing on its own moments later.
def settle(pg, timeout=20000):
pg.wait_for_function(SETTLED, timeout=timeout)
def reader_text(pg) -> str:
el = pg.query_selector(".msg.assistant .msg-body")
return el.inner_text() if el else ""
def sys_texts(pg) -> list[str]:
# Sys notes now live in the top status bar (latest-wins), not the chat.
el = pg.query_selector("#tarot-statusline")
t = el.inner_text() if el else ""
return [t] if t.strip() else []
def assert_recovered(pg):
"""The querent can act again: not streaming, input unblocked, nothing held."""
assert pg.evaluate("typeof streaming !== 'undefined' && streaming === false")
assert not pg.evaluate("document.body.classList.contains('no-input')")
assert not pg.evaluate("document.body.classList.contains('opening-pending')")
assert not pg.evaluate("document.body.classList.contains('reader-speaking')")
# ── tests ──────────────────────────────────────────────────────────────────
def test_happy_path_reveals_full_text(open_tarot):
pg = open_tarot(fulfill_sse(sse(txt("The cards "), txt("stir."))))
settle(pg)
assert reader_text(pg) == "The cards stir."
assert_recovered(pg)
def test_server_500_logs_error_and_recovers(open_tarot):
pg = open_tarot(lambda r: r.fulfill(status=500, content_type="text/plain", body="boom"))
settle(pg)
assert any("error" in n.lower() for n in sys_texts(pg))
assert_recovered(pg)
def test_network_abort_logs_error_and_recovers(open_tarot):
pg = open_tarot(lambda r: r.abort("failed"))
settle(pg)
assert any("error" in n.lower() for n in sys_texts(pg))
assert_recovered(pg)
def test_malformed_sse_skips_bad_lines(open_tarot):
# bad JSON and a truncated event between two valid text deltas: the parser
# skips the junk (try/catch continue) and the good text still reveals whole.
body = (
'data: {"type":"text","delta":"Hello"}\n\n'
'data: not-json\n\n'
'data: {"type":"text","delta":" world."}\n\n'
'data: {"type":"text","delta":\n\n'
)
pg = open_tarot(fulfill_sse(body))
settle(pg)
assert reader_text(pg) == "Hello world."
assert_recovered(pg)
def test_textless_tool_turn_still_progresses(open_tarot):
# A turn that emits only a tool_call (no prose) must not hang the reveal
# loop waiting on text that never comes — it settles and notes the lookup.
pg = open_tarot(fulfill_sse(sse(tool("lookup_card_meaning", card_id="the_fool"))))
settle(pg)
assert any("the_fool" in n for n in sys_texts(pg))
assert_recovered(pg)
def test_voice_failure_falls_back_to_text(open_tarot):
pg = open_tarot(fulfill_sse(sse(txt("Reading anyway."))), voice_js=VOICE_FAIL)
settle(pg)
assert reader_text(pg) == "Reading anyway."
assert any("voice unavailable" in n.lower() for n in sys_texts(pg))
assert_recovered(pg)
def test_voice_stall_watchdog_unblocks(open_tarot):
# Upstream connects but never streams a chunk (duration stays 0); the
# first-audio grace bails to the guessed pace so the reveal still completes
# and, since zero audio ever played, flags the voice unavailable. The
# timeout has to clear typewriter.js's FIRST_AUDIO_MS (12s -- a cold kokoro
# measured 10.3s to first audio) plus the reveal that follows the bail; at
# the old 12000 this test timed out on the grace window itself.
pg = open_tarot(fulfill_sse(sse(txt("Stalled but finishes."))), voice_js=VOICE_STALL)
settle(pg, timeout=20000)
assert reader_text(pg) == "Stalled but finishes."
assert any("voice unavailable" in n.lower() for n in sys_texts(pg))
assert_recovered(pg)
def test_voice_midstream_stall_unblocks(open_tarot):
# Audio buffers (duration>0) but the playback clock never advances — the
# mid-stream freeze. The 2.5s stall watchdog bails to the guessed pace so the
# reveal completes. Some audio buffered, so it is NOT flagged unavailable.
pg = open_tarot(fulfill_sse(sse(txt("Frozen mid-stream."))), voice_js=VOICE_MIDSTALL)
settle(pg, timeout=20000)
assert reader_text(pg) == "Frozen mid-stream."
assert not any("voice unavailable" in n.lower() for n in sys_texts(pg))
assert_recovered(pg)
# ── the spread: turning a card ─────────────────────────────────────────────
# A dealt three-card spread, straight into localStorage, with the narrator OFF
# ("0") so the turn under test is the silent one and the reveal is not waiting
# on a voice. The page reads exactly these three keys on load.
def seeded_spread(flipped=()) -> str:
cards = [
{"position": "past", "card_id": "the_fool", "name": "The Fool",
"image": "/tarot/cards/the_fool.jpg", "reversed": False},
{"position": "present", "card_id": "the_tower", "name": "The Tower",
"image": "/tarot/cards/the_tower.jpg", "reversed": True},
{"position": "future", "card_id": "strength", "name": "Strength",
"image": "/tarot/cards/strength.jpg", "reversed": False},
]
for c in cards:
c["flipped"] = c["position"] in flipped
spread = {"type": "three", "frame": "past_present_future", "cards": cards}
sig = {"card_id": "queen_of_cups", "name": "Queen of Cups",
"image": "/tarot/cards/queen_of_cups.jpg"}
return (f"localStorage.setItem('tarot.spread', {json.dumps(json.dumps(spread))});"
f"localStorage.setItem('tarot.significator', {json.dumps(json.dumps(sig))});"
"localStorage.setItem('tarot.voice', '0');")
def test_a_turned_card_shows_its_face_before_the_turn_ends(open_tarot):
"""The zoom is dismissed mid-reading, and the spread behind it must have
moved: the card paints face-up on the tap, while `flipped` is still
uncommitted (the commit waits for the reader to actually narrate it)."""
pg = open_tarot(fulfill_sse(sse(txt("The Fool steps off."))),
init_script=seeded_spread())
settle(pg) # the opening turn
pg.click(".tarot-card[data-position='past']")
pg.wait_for_function(
"""() => {
const el = document.querySelector(".tarot-card[data-position='past']");
return el && el.dataset.flipped === 'true'
&& spread.cards[0].flipped === false;
}""",
timeout=4000)
def test_a_turned_card_zooms_again_while_its_turn_is_still_running(open_tarot):
"""Flip, dismiss the zoom, tap the card again mid-turn: it must zoom. The
card is only PAINTED until the turn commits, and the tap used to fall into
the `streaming` guard and do nothing (the last card most of all)."""
calls = []
def handler(route):
calls.append(route)
if len(calls) == 1: # the opening turn
fulfill_sse(sse(txt("Sit.")))(route)
# the card's turn is left hanging: streaming stays true
pg = open_tarot(handler, init_script=seeded_spread(flipped=("past", "present")))
settle(pg)
pg.click(".tarot-card[data-position='future']")
pg.wait_for_selector("#card-zoom.open", timeout=4000)
pg.click("#card-zoom")
pg.wait_for_function("() => !document.querySelector('#card-zoom').classList.contains('open')"
" && streaming === true", timeout=4000)
pg.click(".tarot-card[data-position='future']")
pg.wait_for_selector("#card-zoom.open", timeout=4000)
def test_a_failed_turn_puts_the_card_back(open_tarot):
"""The paint is optimistic, so a turn that never narrates has to take it
back — otherwise the position reads as revealed and can never be retried."""
pg = open_tarot(lambda r: r.fulfill(status=500, content_type="text/plain", body="boom"),
init_script=seeded_spread())
settle(pg)
pg.click(".tarot-card[data-position='past']")
pg.wait_for_function(
"""() => {
const el = document.querySelector(".tarot-card[data-position='past']");
return el && el.dataset.flipped === 'false' && streaming === false;
}""",
timeout=8000)
# ── the canned opening (the pre-generated first turn) ───────────────────────
CANNED = {
"id": "h03-9b499eeb",
"text": "A cassette in the deck, still rewound.\n\nWhat kept you up?",
"dur": 9.0,
"audio": "/api/tarot/opening/h03-9b499eeb.wav",
}
def test_canned_opening_reveals_without_an_llm_turn(open_tarot):
"""The whole point: when the hour has a pre-generated opening, the first
turn costs no /api/tarot/chat round-trip (and no TTS synth) at all."""
called = []
def chat(route):
called.append(route.request.url)
route.fulfill(status=200, content_type="text/event-stream",
body=sse(txt("live turn")))
pg = open_tarot(chat, clip=CANNED, audio=wav_bytes())
settle(pg)
assert reader_text(pg).startswith("A cassette in the deck")
assert called == []
assert_recovered(pg)
def test_canned_opening_survives_a_missing_clip_audio(open_tarot):
"""The .wav never arrives (404). The text still reveals at the guessed pace
and the reading settles — a clip that cannot be narrated must not freeze the
page any more than a dead TTS box does."""
pg = open_tarot(fulfill_sse(sse(txt("unused"))), clip=CANNED)
settle(pg, timeout=25000)
assert reader_text(pg).startswith("A cassette in the deck")
assert_recovered(pg)
def test_canned_opening_survives_failed_narration(open_tarot):
"""The clip downloads but the voice layer refuses to play it (VOICE_FAIL's
speakClip). Same contract: reveal, settle, recover."""
pg = open_tarot(fulfill_sse(sse(txt("unused"))), voice_js=VOICE_FAIL,
clip=CANNED, audio=wav_bytes())
settle(pg, timeout=25000)
assert reader_text(pg).startswith("A cassette in the deck")
assert_recovered(pg)
def test_voice_offline_is_stated_once_the_opening_has_spoken(open_tarot):
"""Home box unreachable: the canned opening still narrates (it is a file),
and the status bar then says the rest of the reading is silent — rather
than leaving the querent to notice the voice never came back."""
pg = open_tarot(fulfill_sse(sse(txt("unused"))), clip=CANNED, audio=wav_bytes(),
home_voice=False)
settle(pg)
pg.wait_for_function(
"""() => (document.getElementById('tarot-statusline')||{}).innerText
?.includes('voice offline')""", timeout=5000)
assert reader_text(pg).startswith("A cassette in the deck")
assert_recovered(pg)
def test_voice_up_says_nothing_about_being_offline(open_tarot):
pg = open_tarot(fulfill_sse(sse(txt("unused"))), clip=CANNED, audio=wav_bytes())
settle(pg)
assert not any("voice offline" in n.lower() for n in sys_texts(pg))