tests/test_voice_input_browser.py
Rama master · réplica de lectura del repositorio original.
"""Hands-free chat input (voice-input.js), on the Exec panel's binding
(exec-mic.js): the `$` prompt becomes the control, a final result sends the way
Enter would, and anything heard while the page is TALKING is dropped rather
than sent, or the surface transcribes its own narration and answers itself.
(/cc's binding went with the page, 2026-09-29.)
WebKit headless has no SpeechRecognition, so a fake one is installed before any
page script runs; it is the same shape the engine drives (continuous session,
results accumulating across utterances, `isFinal` on the last one).
Marked `browser` so the fast smoke step skips it.
.venv/bin/pytest tests/test_voice_input_browser.py -q
"""
import json
import pytest
pytest.importorskip("playwright.sync_api")
pytestmark = pytest.mark.browser
# A continuous recognizer: results ACCUMULATE, which is what the engine's `base`
# floor exists to survive — without it the second utterance resends the first.
_FAKE_REC = """
window.__recStarts = 0;
class FakeRec {
constructor() { this.results = []; window.__rec = this; }
start() { window.__recStarts += 1; }
stop() { if (this.onend) this.onend(); }
abort() { this.aborted = true; }
say(text, final) {
this.results.push(Object.assign([{ transcript: text }], { isFinal: !!final }));
if (this.onresult) this.onresult({ results: this.results });
}
}
window.webkitSpeechRecognition = FakeRec;
"""
_HISTORY = {"messages": [], "monitorTotal": 0}
def _json_route(payload):
def handler(route):
route.fulfill(status=200, content_type="application/json",
body=json.dumps(payload))
return handler
@pytest.fixture
def phone(browser, base_url, admin_headers):
"""iPhone-sized /rd with the panel's audio stubbed out (markdown is the
real vendored marked + DOMPurify, served by the app itself)."""
ctx = browser.new_context(extra_http_headers=admin_headers,
viewport={"width": 430, "height": 932},
has_touch=True, is_mobile=True)
pg = ctx.new_page()
pg.add_init_script(_FAKE_REC)
pg.route("**/hosaka-audio.js*",
lambda r: r.fulfill(status=200,
content_type="application/javascript",
body="window.HosakaAudio={createPlayer:()=>({"
"unlock(){},speak(){return Promise.resolve()},"
"flush(){},setVolume(){},elapsed:()=>0,"
"audioDuration:()=>0,isUnlocked:()=>true,"
"gestureUnlocked:()=>true})};"))
def _chat(route):
if route.request.method == "POST":
pg.evaluate("b => window.__sent.push(JSON.parse(b))",
route.request.post_data)
route.fulfill(status=200, content_type="text/event-stream",
body='data: {"type":"done"}\n\n')
else:
route.fulfill(status=200, content_type="application/json",
body=json.dumps(_HISTORY))
pg.route("**/api/cc/exec-history", _chat)
pg.route("**/api/cc/query", _chat)
pg.goto(f"{base_url}/rd", wait_until="domcontentloaded")
pg.evaluate("() => { window.__sent = []; }")
pg.wait_for_selector("#exec-bubble", timeout=5000)
yield pg
ctx.close()
def _open(pg):
pg.tap("#exec-bubble")
pg.wait_for_selector("#exec-panel.open", timeout=4000)
pg.wait_for_selector("#exec-prompt.mic", timeout=4000)
def _listen(pg):
"""Open the panel and tap the prompt into a listening session."""
_open(pg)
pg.tap("#exec-prompt")
pg.wait_for_function(
"() => document.getElementById('exec-prompt').dataset.live === 'true'",
timeout=4000)
def test_the_prompt_is_the_control(phone):
"""`$` carries the mic; a separate button would sit where the bubble rests."""
_open(phone)
assert phone.eval_on_selector("#exec-prompt", "e => e.textContent") == "$"
assert phone.eval_on_selector("#exec-prompt",
"e => e.getAttribute('aria-pressed')") == "false"
def test_tapping_the_prompt_opens_a_session(phone):
"""One tap starts the recognizer and lights the prompt."""
_listen(phone)
assert phone.evaluate("() => window.__recStarts") == 1
assert phone.eval_on_selector("#exec-prompt", "e => e.textContent") == "●"
def test_a_final_result_sends_it(phone):
"""What she said reaches /api/cc/query as the message she would have typed."""
_listen(phone)
phone.evaluate("() => window.__rec.say('archive the climbing card', true)")
phone.wait_for_function("() => (window.__sent || []).length >= 1", timeout=4000)
sent = phone.evaluate("() => window.__sent")[0]["prompt"]
assert sent.endswith("archive the climbing card")
def test_exec_talking_is_never_sent_back(phone):
"""Heard while Exec speaks = binned: composer empty, nothing posted.
The mic cannot be paused (start() needs a gesture, so a pause would be
one-way), so the drop is the whole defence against the panel answering its
own narration.
"""
_listen(phone)
phone.evaluate("() => { window.execVoice.isSpeaking = () => true; }")
phone.evaluate("() => window.__rec.say('that was rhetorical', true)")
assert phone.eval_on_selector("#exec-minput", "e => e.textContent") == ""
assert phone.evaluate("() => window.__sent.length") == 0
assert phone.eval_on_selector("#exec-prompt", "e => e.dataset.drop") == "true"
def test_closing_the_panel_ends_the_session(phone):
"""A mic left open behind a hidden panel keeps sending with nothing on screen."""
_listen(phone)
phone.tap("#exec-ph-close")
phone.wait_for_function(
"() => document.getElementById('exec-prompt').dataset.live === 'false'",
timeout=4000)
assert phone.evaluate("() => window.__rec.aborted === true")
# ── /tarot: the reading table ──────────────────────────────────────────────
# A player that records instead of streaming PCM, and hands the TEST the
# utterance's terminal frame (__endLast) instead of sending it itself.
#
# The real upstream always ends an utterance -- and two layers synthesize one if
# it dies mid-sentence (routes_tts.died_mid_utterance, hosaka-audio's onclose) --
# so a fake that never ends would leave `speaking` true forever, which is not a
# state the app can actually reach. Driving it by hand is what lets one test
# clear the flag and another hold it true on purpose.
_FAKE_HOSAKA = """
window.__spoken = [];
window.HosakaAudio = { createPlayer: function () {
var unlocked = false;
function arm(r) {
window.__endLast = function () { if (r && r.onStatus) r.onStatus({ type: "end" }); };
}
return {
unlock: function () { unlocked = true; },
speak: function (r) { window.__spoken.push(r); arm(r); return Promise.resolve(); },
speakBuffer: function (r) { arm(r); return Promise.resolve(); },
flush: function () {}, setVolume: function () {},
elapsed: function () { return 0; }, audioDuration: function () { return 0; },
isUnlocked: function () { return unlocked; },
gestureUnlocked: function () { return unlocked; },
};
}};
"""
@pytest.fixture
def tarot(browser, base_url, admin_headers):
# Wai's phone, and touch: the prompt is tapped, not clicked.
ctx = browser.new_context(extra_http_headers=admin_headers,
viewport={"width": 430, "height": 932},
has_touch=True, is_mobile=True)
pg = ctx.new_page()
# __sent must exist before ANY page script: the opening reader turn fires
# /api/tarot/chat during load, long before the fixture's own evaluate could
# create it.
pg.add_init_script(_FAKE_REC + "\nwindow.__sent = [];")
pg.route("**/hosaka-audio.js*",
lambda r: r.fulfill(status=200,
content_type="application/javascript",
body=_FAKE_HOSAKA))
# No canned opening in the test: the LIVE path is the one with a mic under it.
pg.route("**/api/tarot/opening*", _json_route({"clip": None}))
pg.route("**/api/tarot/warm", _json_route({"ok": True, "skipped": "test"}))
pg.route("**/api/hosaka/health", _json_route({"ok": True, "home": True, "piper": True}))
def _chat(route):
pg.evaluate("b => window.__sent.push(b)", route.request.post_data)
route.fulfill(status=200, content_type="text/event-stream",
body='data: {"type":"text","delta":"The cards wait."}\n\n')
pg.route("**/api/tarot/chat", _chat)
pg.goto(f"{base_url}/tarot", wait_until="domcontentloaded")
# The opening turn is held for the first gesture when the voice is on; tap
# to start it, then let it settle so the mic is not "busy" on a live turn.
pg.mouse.click(200, 400)
pg.wait_for_function("() => typeof streaming !== 'undefined' && streaming === false",
timeout=15000)
# End the opening's narration the way the upstream would, so the reader is
# not still "speaking" (which the mic reads as a moment to drop what it
# hears) for the rest of the fixture.
pg.evaluate("() => { if (window.__endLast) window.__endLast(); }")
pg.wait_for_function("() => tarotVoice.isSpeaking() === false", timeout=4000)
pg.evaluate("() => { window.__sent = []; }")
pg.wait_for_selector("#input-prompt.mic", timeout=5000)
yield pg
ctx.close()
def _tarot_listen(pg):
pg.tap("#input-prompt")
pg.wait_for_function(
"() => document.getElementById('input-prompt').dataset.live === 'true'",
timeout=4000)
def test_tarot_prompt_is_the_control(tarot):
"""The reading table's `$` carries the mic too — same engine, same glyph."""
assert tarot.eval_on_selector("#input-prompt", "e => e.textContent") == "$"
_tarot_listen(tarot)
assert tarot.eval_on_selector("#input-prompt", "e => e.textContent") == "●"
def test_tarot_final_result_sends_it(tarot):
"""A finished utterance reaches /api/tarot/chat as the querent's answer."""
_tarot_listen(tarot)
tarot.evaluate("() => window.__rec.say('my work has been eating me', true)")
tarot.wait_for_function("() => (window.__sent || []).length >= 1", timeout=4000)
body = json.loads(tarot.evaluate("() => window.__sent")[0])
assert body["messages"][-1]["content"] == "my work has been eating me"
def test_the_reader_talking_is_never_sent_back(tarot):
"""The reader speaks over the same speaker the mic hears. Without the
isSpeaking() guard the page transcribes the reading and answers itself."""
_tarot_listen(tarot)
tarot.evaluate("() => tarotVoice.speak('The Tower falls in the third position.')")
tarot.wait_for_function("() => tarotVoice.isSpeaking() === true", timeout=4000)
tarot.evaluate("() => window.__rec.say('the tower falls in the third position', true)")
tarot.wait_for_timeout(700)
assert tarot.evaluate("() => window.__sent") == []