phase: 01_keyboard_kiosk_control
Build and Push Containers / test (push) Successful in 35s
Build and Push Containers / build-and-push-app (push) Successful in 10s

**Phase 01 final verification pass — complete, 100% green**

- Verified all 4 tasks done (files in `complete/`); implementation review: key router in `watch.js`, share-page router, CSS focus/selection affordances, `?` overlay, pointer-free E2E suite — all sound.
- Gates: `ruff check` + `ruff format --check` ✅ · `pyright` 0 errors ✅ · `uv run pytest` → 223 passed, 2 skipped (realdisplay), 1 pre-existing FastAPI deprecation warning · `MIRROR_REALDISPLAY=1 pytest -m realdisplay` → 2 passed ✅ · `pytest -m e2e` → 51 passed (incl. `test_keyboard_wall.py`, which a meta-test keeps pointer-free) · `node --check` on all static JS ✅.
- Completion criteria: all 6 ✅ (j/k+Enter, f/g/n/p/1–9/m/?/Esc keyboard-only E2E; no hover-only controls via CSS guard + focus-ring E2E; `#name` text-safe; e2e/realdisplay green; key map documented in README "Kiosk client" and `docs/kiosk-silverblue.md`).
- README fixed: added missing env vars `MIRROR_MAX_BITRATE`, `MIRROR_PING_INTERVAL`, `DEBUGPY_WAIT` to the config table; added the one-time `rpm-ostree` layering note before the kiosk install snippet.
- AGENTS.md fixed: removed stale "roadmap is intentionally empty" parenthetical (3 phases pending); coverage ~94% → ~95%. No content added.
- **Unresolved, needs your call:** README/SECURITY.md say `mirror.reeseapps.com`; `docs/kiosk-silverblue.md` (reference-client naming table with real IPs) says `mirror.reeselink.com`. I couldn't verify which is the live deployment domain, so I changed neither — say the word and I'll unify.
- Next pending phase: `02_sharp_capture`.
This commit is contained in:
2026-11-01 20:53:28 -05:00
parent f3de5b70ff
commit 1e816da5a8
39 changed files with 1726 additions and 31 deletions
+902
View File
@@ -0,0 +1,902 @@
"""End-to-end: the keyboard is a complete control surface for mirror.
Phase 01, tasks 01-04: one focus-aware keydown router plus a keyboard cursor
over the tiles (j/k/Up/Down, Esc), the wall commands (Enter/f, g, n/p, 1-9,
m), the ``?`` help overlay, the focus-visible affordances — no control is
reachable *only* by hover — and the /mirror keys (s, x, Esc). Every command
here is driven with ``page.keyboard.press()`` and asserted on the DOM state
the key changes — ``data-selected``, focus, the ``staged`` class, the
``muted`` flag, computed styles — never on whether a handler "ran".
The suite is pointer-free: a test that reaches for the pointer proves nothing
about a kiosk. The meta-test at the bottom enforces that on this file
itself, so the guarantee cannot rot.
"""
from __future__ import annotations
import os
import re
import socket
import subprocess
import sys
import time
from pathlib import Path
import pytest
from playwright.sync_api import expect
from .conftest import start_sharing
pytestmark = [pytest.mark.e2e, pytest.mark.timeout(180)]
@pytest.fixture
def device(base_url, browser, request):
"""Open a page as its own device (isolated context, fresh peer id).
Uncaught page errors and console errors are captured from the first line of
the test and surfaced through the ``errors == []`` assertion
(AGENTS.md §8: print pageerror/console from the first line of the test).
"""
opened = []
def _open(path: str = "/", errors: list | None = None):
context = browser.new_context(viewport={"width": 1100, "height": 720})
opened.append(context)
request.addfinalizer(context.close)
page = context.new_page()
if errors is not None:
page.on("pageerror", lambda error: errors.append(f"pageerror: {error}"))
page.on(
"console",
lambda message: (
errors.append(f"console: {message.text}") if message.type == "error" else None
),
)
page.goto(f"{base_url}{path}", wait_until="domcontentloaded")
expect(page.locator("#net")).to_have_attribute("data-state", "online", timeout=10_000)
return page
return _open
def _non_loopback_ipv4() -> str | None:
"""The address a *neighbour* would use to reach this machine, or None.
A UDP connect sends no packet; it only resolves the default route, which
is exactly the interface a LAN device would see.
"""
with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as probe:
try:
probe.connect(("8.8.8.8", 80))
except OSError:
return None
address = probe.getsockname()[0]
return address if address != "127.0.0.1" else None
@pytest.fixture
def insecure_base_url():
"""A `mirror serve` on a plain-HTTP non-localhost address.
That is a non-secure context: the browser hides `navigator.mediaDevices`
entirely — the exact trap /mirror documents for a bare `mirror serve` on a
LAN IP (README, docs/kiosk-silverblue.md "One trap worth knowing").
"""
host = _non_loopback_ipv4()
if host is None:
pytest.skip("no non-loopback interface: the non-secure-context trap is untestable here")
import httpx
from .conftest import _free_port
port = _free_port()
process = subprocess.Popen(
[
sys.executable,
"-m",
"mirror",
"serve",
"--host",
"0.0.0.0",
"--port",
str(port),
"--log-level",
"warning",
],
cwd=Path(__file__).resolve().parents[2],
stdout=subprocess.PIPE,
stderr=subprocess.STDOUT,
text=True,
env={
**os.environ,
"MIRROR_ICE_SERVERS": "",
"PYTHONUNBUFFERED": "1",
"DEBUGPY": "0",
},
)
url = f"http://{host}:{port}"
last_error: Exception | None = None
deadline = time.time() + 25
while time.time() < deadline:
if process.poll() is not None:
output = process.stdout.read() if process.stdout else ""
raise RuntimeError(f"mirror serve exited early ({process.returncode}):\n{output}")
try:
if httpx.get(f"{url}/healthz", timeout=1.0).status_code == 200:
break
except Exception as exc: # still booting
last_error = exc
time.sleep(0.15)
else:
process.terminate()
raise RuntimeError(f"mirror serve never came up: {last_error}")
yield url
process.terminate()
try:
process.wait(timeout=5)
except subprocess.TimeoutExpired: # pragma: no cover
process.kill()
def tile_ids(page) -> list[str]:
return page.locator(".tile").evaluate_all("els => els.map(e => e.dataset.stream)")
def selected_id(page) -> str | None:
return page.evaluate(
"() => { const el = document.querySelector('.tile[data-selected]');"
" return el ? el.dataset.stream : null; }"
)
def focus_follows_selection(page) -> bool:
return page.evaluate(
"() => { const el = document.querySelector('.tile[data-selected]');"
" return el !== null && document.activeElement === el; }"
)
class TestKeyboardCursor:
def _two_mirror_wall(self, device, errors: list) -> tuple:
for _ in range(2):
start_sharing(device("/mirror", errors))
viewer = device("/watch", errors)
expect(viewer.locator(".tile")).to_have_count(2, timeout=15_000)
return viewer, tile_ids(viewer)
def test_j_selects_the_first_tile_and_focus_follows(self, device):
errors: list = []
viewer, ids = self._two_mirror_wall(device, errors)
viewer.keyboard.press("j")
assert selected_id(viewer) == ids[0]
assert focus_follows_selection(viewer), "focus must follow the selection"
assert errors == [], f"uncaught browser errors: {errors}"
def test_j_and_k_wrap_at_the_ends(self, device):
errors: list = []
viewer, ids = self._two_mirror_wall(device, errors)
viewer.keyboard.press("j")
assert selected_id(viewer) == ids[0]
# k from the first wraps to the last.
viewer.keyboard.press("k")
assert selected_id(viewer) == ids[1]
# j from the last wraps back to the first.
viewer.keyboard.press("j")
assert selected_id(viewer) == ids[0]
assert errors == [], f"uncaught browser errors: {errors}"
def test_arrow_keys_drive_the_same_cursor(self, device):
errors: list = []
viewer, ids = self._two_mirror_wall(device, errors)
viewer.keyboard.press("ArrowDown")
assert selected_id(viewer) == ids[0]
viewer.keyboard.press("ArrowDown")
assert selected_id(viewer) == ids[1]
viewer.keyboard.press("ArrowUp")
assert selected_id(viewer) == ids[0]
# The cursor must not fight real scrolling: the router suppresses the
# arrow's default while it handles the key.
viewer.keyboard.press("ArrowUp")
assert selected_id(viewer) == ids[1]
assert errors == [], f"uncaught browser errors: {errors}"
def test_ctrl_chords_never_run_a_command(self, device):
errors: list = []
viewer, ids = self._two_mirror_wall(device, errors)
viewer.keyboard.press("j")
assert selected_id(viewer) == ids[0]
# Synthesised ctrlKey: true — the browser's chords are never ours.
viewer.keyboard.press("Control+j")
assert selected_id(viewer) == ids[0]
viewer.keyboard.press("Control+k")
assert selected_id(viewer) == ids[0]
viewer.keyboard.press("Control+ArrowDown")
assert selected_id(viewer) == ids[0]
assert errors == [], f"uncaught browser errors: {errors}"
def test_shifted_letters_drive_the_same_commands(self, device):
errors: list = []
viewer, ids = self._two_mirror_wall(device, errors)
# Keys are matched case-insensitively: a caps-locked kiosk keyboard, or
# a remote hand holding Shift, must work exactly the same.
viewer.keyboard.press("Shift+J")
assert selected_id(viewer) == ids[0]
# Shift+F stages through the same command as f.
viewer.keyboard.press("Shift+F")
expect(viewer.locator("body")).to_have_class(re.compile("staged"), timeout=5_000)
assert viewer.locator("#watch-stage .tile").get_attribute("data-stream") == ids[0]
assert errors == [], f"uncaught browser errors: {errors}"
class TestEscape:
def test_escape_leaves_the_stage(self, device):
errors: list = []
start_sharing(device("/mirror", errors))
viewer = device("/watch", errors)
expect(viewer.locator(".tile")).to_have_count(1, timeout=15_000)
# Staged the way the kiosk stages — select, then Enter — because this
# suite is pointer-free; Esc is what the kiosk uses to leave the stage.
viewer.keyboard.press("j")
viewer.keyboard.press("Enter")
expect(viewer.locator("body")).to_have_class(re.compile("staged"), timeout=5_000)
viewer.keyboard.press("Escape")
expect(viewer.locator("body")).not_to_have_class(re.compile("staged"), timeout=5_000)
# Leaving the stage must not have disturbed the stream.
expect(viewer.locator(".tile[data-state='live']")).to_have_count(1, timeout=20_000)
assert errors == [], f"uncaught browser errors: {errors}"
def test_escape_without_a_stage_is_a_noop(self, device):
errors: list = []
start_sharing(device("/mirror", errors))
viewer = device("/watch", errors)
expect(viewer.locator(".tile")).to_have_count(1, timeout=15_000)
viewer.keyboard.press("Escape")
expect(viewer.locator("body")).not_to_have_class(re.compile("staged"))
expect(viewer.locator(".tile")).to_have_count(1)
expect(viewer.locator(".tile[data-state='live']")).to_have_count(1, timeout=20_000)
assert errors == [], f"uncaught browser errors: {errors}"
class TestWallCommands:
"""Task 02: Enter/f, g, n/p, 1-9, m on /watch.
Every command is driven with ``page.keyboard.press()`` and asserted on the
DOM state the key changes — the staged tile id, the ``staged`` class, the
video's ``muted`` flag, the ``#toast`` — never on whether a handler "ran".
"""
def _two_mirror_wall(self, device, errors: list) -> tuple:
for _ in range(2):
start_sharing(device("/mirror", errors))
viewer = device("/watch", errors)
expect(viewer.locator(".tile")).to_have_count(2, timeout=15_000)
return viewer, tile_ids(viewer)
def test_enter_stages_the_selected_mirror(self, device):
errors: list = []
viewer, ids = self._two_mirror_wall(device, errors)
viewer.keyboard.press("j")
viewer.keyboard.press("Enter")
expect(viewer.locator("body")).to_have_class(re.compile("staged"), timeout=5_000)
expect(viewer.locator("#watch-stage .tile")).to_have_count(1)
assert viewer.locator("#watch-stage .tile").get_attribute("data-stream") == ids[0]
assert errors == [], f"uncaught browser errors: {errors}"
def test_enter_with_nothing_selected_stages_the_first_tile(self, device):
errors: list = []
viewer, ids = self._two_mirror_wall(device, errors)
viewer.keyboard.press("Enter")
expect(viewer.locator("body")).to_have_class(re.compile("staged"), timeout=5_000)
expect(viewer.locator("#watch-stage .tile")).to_have_count(1)
assert viewer.locator("#watch-stage .tile").get_attribute("data-stream") == ids[0]
assert errors == [], f"uncaught browser errors: {errors}"
def test_f_stages_when_unstaged_and_holds_the_stage_when_staged(self, device):
errors: list = []
viewer, ids = self._two_mirror_wall(device, errors)
# Unstaged: `f` stages like Enter (first tile, nothing selected).
viewer.keyboard.press("f")
expect(viewer.locator("body")).to_have_class(re.compile("staged"), timeout=5_000)
assert viewer.locator("#watch-stage .tile").get_attribute("data-stream") == ids[0]
# Staged: `f` re-requests fullscreen. A kiosk's window manager can drop
# the window out of fullscreen without the page noticing, so the
# command must never tear the stage down — headless or not.
viewer.keyboard.press("f")
expect(viewer.locator("body")).to_have_class(re.compile("staged"), timeout=5_000)
expect(viewer.locator("#watch-stage .tile")).to_have_count(1)
assert viewer.locator("#watch-stage .tile").get_attribute("data-stream") == ids[0]
assert errors == [], f"uncaught browser errors: {errors}"
def test_g_returns_to_the_wall(self, device):
errors: list = []
viewer, _ = self._two_mirror_wall(device, errors)
viewer.keyboard.press("j")
viewer.keyboard.press("Enter")
expect(viewer.locator("body")).to_have_class(re.compile("staged"), timeout=5_000)
viewer.keyboard.press("g")
expect(viewer.locator("body")).not_to_have_class(re.compile("staged"), timeout=5_000)
expect(viewer.locator("#watch-stage .tile")).to_have_count(0)
expect(viewer.locator(".tile")).to_have_count(2)
# g on an already-unstaged wall is a silent no-op: the wall *is* the
# visible result "back to the wall".
viewer.keyboard.press("g")
expect(viewer.locator(".tile")).to_have_count(2)
expect(viewer.locator("#toast")).not_to_be_visible()
# Leaving the stage must not have disturbed the streams.
expect(viewer.locator(".tile[data-state='live']")).to_have_count(2, timeout=20_000)
assert errors == [], f"uncaught browser errors: {errors}"
def test_n_and_p_step_between_mirrors_without_killing_the_peer(self, device):
errors: list = []
viewer, ids = self._two_mirror_wall(device, errors)
a, b = ids[0], ids[1]
expect(viewer.locator(".tile[data-state='live']")).to_have_count(2, timeout=30_000)
# Stage A by keyboard.
viewer.keyboard.press("j")
viewer.keyboard.press("Enter")
expect(viewer.locator("#watch-stage .tile")).to_have_count(1, timeout=5_000)
assert viewer.locator("#watch-stage .tile").get_attribute("data-stream") == a
# Mark A's <video> node: staging B must *move* it back to the grid,
# never recreate it — the element moves, srcObject is never reassigned.
assert viewer.evaluate(
"""(id) => {
const el = document.querySelector('#watch-stage video');
el.dataset.e2e = "same-node";
return el.srcObject !== null;
}""",
a,
), "A's video must hold a live stream before the switch"
viewer.keyboard.press("n")
expect(viewer.locator("#watch-stage .tile")).to_have_count(1, timeout=5_000)
assert viewer.locator("#watch-stage .tile").get_attribute("data-stream") == b
assert viewer.evaluate(
"""(id) => {
const el = document.querySelector(`.tile[data-stream="${id}"] video`);
return el !== null && el.dataset.e2e === "same-node" && el.srcObject !== null;
}""",
a,
), "A's <video> must be the same node, still holding its stream"
# p wraps back to A — and the stepping must not have killed either mirror.
viewer.keyboard.press("p")
expect(viewer.locator("#watch-stage .tile")).to_have_count(1, timeout=5_000)
assert viewer.locator("#watch-stage .tile").get_attribute("data-stream") == a
expect(viewer.locator(".tile[data-state='live']")).to_have_count(2, timeout=20_000)
assert errors == [], f"uncaught browser errors: {errors}"
def test_digit_jump_stages_the_nth_mirror(self, device):
errors: list = []
viewer, ids = self._two_mirror_wall(device, errors)
viewer.keyboard.press("2")
expect(viewer.locator("body")).to_have_class(re.compile("staged"), timeout=5_000)
expect(viewer.locator("#watch-stage .tile")).to_have_count(1)
assert viewer.locator("#watch-stage .tile").get_attribute("data-stream") == ids[1]
# 1 takes the first mirror back on stage; stage() unstages the other.
viewer.keyboard.press("1")
expect(viewer.locator("#watch-stage .tile")).to_have_count(1, timeout=5_000)
assert viewer.locator("#watch-stage .tile").get_attribute("data-stream") == ids[0]
# The wall survives the jump: both mirrors still live.
expect(viewer.locator(".tile[data-state='live']")).to_have_count(2, timeout=20_000)
assert errors == [], f"uncaught browser errors: {errors}"
def test_m_flips_the_staged_videos_muted_flag(self, device):
errors: list = []
viewer, _ = self._two_mirror_wall(device, errors)
viewer.keyboard.press("j")
viewer.keyboard.press("Enter")
expect(viewer.locator("#watch-stage video")).to_have_count(1, timeout=5_000)
before = viewer.evaluate("() => document.querySelector('#watch-stage video').muted")
viewer.keyboard.press("m")
after = viewer.evaluate("() => document.querySelector('#watch-stage video').muted")
assert after == (not before), f"m must flip muted ({before} -> {after})"
# The tile's audio button tracks the same state through setMuted.
assert (
viewer.locator("#watch-stage .audio").get_attribute("aria-pressed")
== str(not after).lower()
), "aria-pressed must follow the muted flag"
# And back again.
viewer.keyboard.press("m")
again = viewer.evaluate("() => document.querySelector('#watch-stage video').muted")
assert again == before, f"m must flip back ({after} -> {again})"
assert errors == [], f"uncaught browser errors: {errors}"
def test_missing_mirror_number_toasts_instead_of_silence(self, device):
errors: list = []
sharer = device("/mirror", errors)
start_sharing(sharer)
viewer = device("/watch", errors)
expect(viewer.locator(".tile")).to_have_count(1, timeout=15_000)
viewer.keyboard.press("5")
# The wall is unchanged…
expect(viewer.locator("body")).not_to_have_class(re.compile("staged"))
expect(viewer.locator(".tile")).to_have_count(1)
# …and the no-op says why — a silent no-op is undiagnosable from a kiosk.
expect(viewer.locator("#toast")).to_be_visible()
expect(viewer.locator("#toast")).to_contain_text("no mirror 5")
assert errors == [], f"uncaught browser errors: {errors}"
def test_commands_on_an_empty_wall_toast_the_reason(self, device):
errors: list = []
viewer = device("/watch", errors)
expect(viewer.locator("#empty")).to_be_visible()
viewer.keyboard.press("Enter")
expect(viewer.locator("body")).not_to_have_class(re.compile("staged"))
expect(viewer.locator("#toast")).to_be_visible()
expect(viewer.locator("#toast")).to_contain_text("no mirrors to stage")
viewer.keyboard.press("n")
expect(viewer.locator("body")).not_to_have_class(re.compile("staged"))
expect(viewer.locator("#toast")).to_contain_text("no mirrors to switch to")
assert errors == [], f"uncaught browser errors: {errors}"
class TestFocusVisibleControls:
"""Task 03: no control is reachable *only* by hover.
Everything here is keyboard-only: ``Tab`` moves focus, and the assertions
are on what the kiosk *sees* — computed opacity, the focus ring,
``data-selected`` — never on whether a handler "ran". No test in this class
drives the pointer.
"""
def _one_mirror_wall(self, device, errors: list):
start_sharing(device("/mirror", errors))
viewer = device("/watch", errors)
expect(viewer.locator(".tile")).to_have_count(1, timeout=15_000)
return viewer
def _two_mirror_wall(self, device, errors: list) -> tuple:
for _ in range(2):
start_sharing(device("/mirror", errors))
viewer = device("/watch", errors)
expect(viewer.locator(".tile")).to_have_count(2, timeout=15_000)
return viewer, tile_ids(viewer)
@staticmethod
def _tab_to_a_tile(viewer, attempts: int = 12) -> str | None:
"""Tab from the top of the document until focus lands on a tile."""
for _ in range(attempts):
viewer.keyboard.press("Tab")
tile = viewer.evaluate(
"() => { const el = document.activeElement;"
" return el && el.classList.contains('tile') ? el.dataset.stream : null; }"
)
if tile:
return tile
return None
def test_tab_focus_reveals_the_tile_controls(self, device):
errors: list = []
viewer = self._one_mirror_wall(device, errors)
assert self._tab_to_a_tile(viewer), "Tab must reach the tile without a pointer"
# The opacity transition is 150 ms; wait for the settled value.
viewer.wait_for_function(
"() => { const el = document.activeElement;"
" return el && el.classList.contains('tile')"
" && getComputedStyle(el.querySelector('.actions')).opacity === '1'; }",
timeout=5_000,
)
assert errors == [], f"uncaught browser errors: {errors}"
def test_keyboard_focus_draws_a_visible_ring(self, device, tmp_path):
errors: list = []
viewer = self._one_mirror_wall(device, errors)
assert self._tab_to_a_tile(viewer)
ring = viewer.evaluate(
"() => { const s = getComputedStyle(document.activeElement);"
" return { width: s.outlineWidth, style: s.outlineStyle, color: s.outlineColor }; }"
)
assert ring["style"] == "solid", f"the focus ring must be visible: {ring}"
assert ring["width"] == "2px", ring
assert ring["color"] != "rgba(0, 0, 0, 0)", f"the ring must not be transparent: {ring}"
# Screenshot artifact: a kiosk is debugged from another machine, so a
# focused tile must be legible in a still frame.
shot = tmp_path / "focused_tile.png"
viewer.screenshot(path=str(shot))
assert shot.stat().st_size > 0
assert errors == [], f"uncaught browser errors: {errors}"
def test_tab_focus_moves_the_selection_with_it(self, device):
errors: list = []
viewer, ids = self._two_mirror_wall(device, errors)
seen: list[str] = []
for _ in range(16):
viewer.keyboard.press("Tab")
on_tile = viewer.evaluate(
"() => document.activeElement !== null"
" && document.activeElement.classList.contains('tile')"
)
if on_tile:
seen.append(selected_id(viewer))
# While a tile itself holds focus, data-selected must be that tile.
assert viewer.evaluate(
"() => document.activeElement === document.querySelector"
"('.tile[data-selected]')"
), "the selected tile must be the focused tile"
if len(seen) == 2:
break
assert seen == ids, f"Tab must walk the grid in order {ids}; walked {seen}"
assert errors == [], f"uncaught browser errors: {errors}"
def test_tab_to_a_tile_button_is_not_stolen_back_to_the_tile(self, device):
errors: list = []
viewer = self._one_mirror_wall(device, errors)
assert self._tab_to_a_tile(viewer)
# The selected tile carries its controls, so the next Tab reaches a
# control *inside* the tile — and must stay there: the focusin
# handler must not pull focus back off the control the operator is on.
viewer.keyboard.press("Tab")
assert viewer.evaluate(
"() => document.activeElement !== null"
" && document.activeElement.closest('.actions') !== null"
), "Tab must be able to land on a control inside the tile"
assert errors == [], f"uncaught browser errors: {errors}"
def test_the_wall_stays_quiet_until_something_is_focused(self, device):
errors: list = []
viewer = self._one_mirror_wall(device, errors)
# Nothing hovered, focused or selected: the controls stay hidden, so a
# shared display keeps its quiet look.
opacity = viewer.evaluate(
"() => getComputedStyle(document.querySelector('.tile .actions')).opacity"
)
assert opacity == "0", (
f"the wall must stay quiet until a tile is focused or selected (got {opacity})"
)
assert errors == [], f"uncaught browser errors: {errors}"
def test_focus_moves_to_the_stage_and_back_with_g(self, device):
errors: list = []
viewer = self._one_mirror_wall(device, errors)
viewer.keyboard.press("Enter")
expect(viewer.locator("body")).to_have_class(re.compile("staged"), timeout=5_000)
assert viewer.evaluate(
"() => document.activeElement !== null"
" && document.activeElement.parentElement.id === 'watch-stage'"
), "staging must hand focus to the staged tile"
# The staged caption is legible while the tile holds focus (the
# opacity transition is 200 ms; wait for the settled value).
viewer.wait_for_function(
"() => document.activeElement !== null"
" && getComputedStyle(document.activeElement.querySelector('.cap')).opacity === '1'",
timeout=5_000,
)
# m still works with focus sitting on the stage.
before = viewer.evaluate("() => document.querySelector('#watch-stage video').muted")
viewer.keyboard.press("m")
after = viewer.evaluate("() => document.querySelector('#watch-stage video').muted")
assert after == (not before), f"m must flip muted on the staged tile ({before} -> {after})"
# g returns to the wall and hands focus back to the tile's grid slot.
# The hand-back is deferred until the fullscreen exit has settled
# (the exit would otherwise drop the focus back to <body>), so wait
# for it rather than asserting on the keypress alone.
viewer.keyboard.press("g")
expect(viewer.locator("body")).not_to_have_class(re.compile("staged"), timeout=5_000)
viewer.wait_for_function(
"() => { const el = document.activeElement;"
" return el !== null && el.classList.contains('tile')"
" && el.parentElement.id === 'grid'"
" && el === document.querySelector('.tile[data-selected]'); }",
timeout=5_000,
)
assert errors == [], f"uncaught browser errors: {errors}"
def test_escape_after_staging_returns_focus_to_the_grid(self, device):
errors: list = []
viewer = self._one_mirror_wall(device, errors)
viewer.keyboard.press("Enter")
expect(viewer.locator("body")).to_have_class(re.compile("staged"), timeout=5_000)
viewer.keyboard.press("Escape")
expect(viewer.locator("body")).not_to_have_class(re.compile("staged"), timeout=5_000)
viewer.wait_for_function(
"() => { const el = document.activeElement;"
" return el !== null && el.classList.contains('tile')"
" && el.parentElement.id === 'grid'; }",
timeout=5_000,
)
assert errors == [], f"uncaught browser errors: {errors}"
class TestHelpOverlay:
"""Task 04: `?` shows the key list; `?`/`Esc` close it; focus never moves.
The overlay is the element ``watch.js`` reaches for (``#wall-help``) and it
exists in ``watch.html`` (the element-existence guard enforces the pairing)
— asserted here on the DOM state, as always, never on whether a handler
"ran".
"""
def _one_mirror_wall(self, device, errors: list):
start_sharing(device("/mirror", errors))
viewer = device("/watch", errors)
expect(viewer.locator(".tile")).to_have_count(1, timeout=15_000)
return viewer
def test_question_mark_opens_the_key_list(self, device):
errors: list = []
viewer = self._one_mirror_wall(device, errors)
expect(viewer.locator("#wall-help")).to_be_hidden()
viewer.keyboard.press("?")
expect(viewer.locator("#wall-help")).to_be_visible()
# The list is the locked key map from 00_phase.md — a kiosk operator
# can read back every command the wall accepts.
for text in (
"next mirror",
"previous mirror",
"Stage the selected mirror fullscreen",
"Back to the wall",
"Stage the next / previous mirror",
"Stage the Nth mirror",
"Mute / unmute the staged mirror",
"Close this list, then leave the stage",
):
expect(viewer.locator("#wall-help")).to_contain_text(text)
assert errors == [], f"uncaught browser errors: {errors}"
def test_question_mark_closes_the_key_list(self, device):
errors: list = []
viewer = self._one_mirror_wall(device, errors)
viewer.keyboard.press("?")
expect(viewer.locator("#wall-help")).to_be_visible()
viewer.keyboard.press("?")
expect(viewer.locator("#wall-help")).to_be_hidden()
assert errors == [], f"uncaught browser errors: {errors}"
def test_the_list_says_how_to_close_it(self, device):
errors: list = []
viewer = self._one_mirror_wall(device, errors)
viewer.keyboard.press("?")
expect(viewer.locator("#wall-help .hint")).to_contain_text("to close this list")
assert errors == [], f"uncaught browser errors: {errors}"
def test_escape_closes_the_overlay_before_it_unstages(self, device):
errors: list = []
viewer = self._one_mirror_wall(device, errors)
viewer.keyboard.press("j")
viewer.keyboard.press("Enter")
expect(viewer.locator("body")).to_have_class(re.compile("staged"), timeout=5_000)
# ? over the stage: the overlay sits above the stage (z-index 70 over
# 50), so it is readable with a mirror on stage.
viewer.keyboard.press("?")
expect(viewer.locator("#wall-help")).to_be_visible()
# First Esc: the overlay closes, the stage stays.
viewer.keyboard.press("Escape")
expect(viewer.locator("#wall-help")).to_be_hidden()
expect(viewer.locator("body")).to_have_class(re.compile("staged"))
# Second Esc: the stage goes.
viewer.keyboard.press("Escape")
expect(viewer.locator("body")).not_to_have_class(re.compile("staged"), timeout=5_000)
expect(viewer.locator("#watch-stage .tile")).to_have_count(0)
assert errors == [], f"uncaught browser errors: {errors}"
def test_focus_stays_on_the_selected_tile_through_the_overlay(self, device):
errors: list = []
viewer = self._one_mirror_wall(device, errors)
viewer.keyboard.press("j")
assert viewer.evaluate(
"() => document.activeElement !== null"
" && document.activeElement.classList.contains('tile')"
), "the selected tile must hold focus before the overlay"
for key in ("?", "?", "Escape"):
viewer.keyboard.press(key)
assert viewer.evaluate(
"() => document.activeElement !== null"
" && document.activeElement.classList.contains('tile')"
), f"focus must stay on the tile after {key!r}"
assert errors == [], f"uncaught browser errors: {errors}"
class TestSharePageKeys:
"""Task 04: `s`/`x`/`Esc` on /mirror, and the no-silent-failure rule.
The sharer here is driven by the keyboard exactly as a kiosk operator would
drive it; ``start_sharing`` (a pointer helper on conftest) is used only to
stand up *wall* mirrors, never to start a share under test.
"""
def test_s_starts_a_share_and_x_stops_it(self, device):
errors: list = []
page = device("/mirror", errors)
expect(page.locator("body")).to_have_attribute("data-live", "off")
# A keydown is a user activation, so the fake UI approves the OS picker
# and the real capture -> preview path runs.
page.keyboard.press("s")
expect(page.locator("body")).to_have_attribute("data-live", "on", timeout=15_000)
page.keyboard.press("x")
expect(page.locator("body")).to_have_attribute("data-live", "off", timeout=10_000)
assert errors == [], f"uncaught browser errors: {errors}"
def test_s_while_sharing_toasts_instead_of_opening_a_second_picker(self, device):
errors: list = []
page = device("/mirror", errors)
page.keyboard.press("s")
expect(page.locator("body")).to_have_attribute("data-live", "on", timeout=15_000)
page.keyboard.press("s")
expect(page.locator("#toast")).to_be_visible()
expect(page.locator("#toast")).to_contain_text("already sharing")
assert errors == [], f"uncaught browser errors: {errors}"
def test_x_while_idle_toasts_instead_of_silence(self, device):
errors: list = []
page = device("/mirror", errors)
page.keyboard.press("x")
expect(page.locator("#toast")).to_be_visible()
expect(page.locator("#toast")).to_contain_text("not sharing")
expect(page.locator("body")).to_have_attribute("data-live", "off")
assert errors == [], f"uncaught browser errors: {errors}"
def test_s_and_x_type_text_in_the_name_field_instead(self, device):
errors: list = []
page = device("/mirror", errors)
page.locator("#name").focus()
page.keyboard.press("s")
expect(page.locator("#name")).to_have_value("s")
page.keyboard.press("x")
expect(page.locator("#name")).to_have_value("sx")
# A key in the display-name field is text, not a command: no share
# started, no share stopped, no picker.
expect(page.locator("body")).to_have_attribute("data-live", "off")
assert errors == [], f"uncaught browser errors: {errors}"
def test_ctrl_chords_never_share(self, device):
errors: list = []
page = device("/mirror", errors)
page.keyboard.press("Control+s")
expect(page.locator("body")).to_have_attribute("data-live", "off")
expect(page.locator("#toast")).not_to_be_visible()
assert errors == [], f"uncaught browser errors: {errors}"
def test_s_says_why_when_capturing_is_unavailable(self, browser, insecure_base_url):
"""The no-silent-failure path: `#start` disabled must not make `s` die quietly.
A plain-HTTP non-localhost origin is a non-secure context, so the
browser hides mediaDevices entirely and the page disables `#start` on
load. Pressing `s` routes through the button's own path (start() re-
checks captureSupport() and re-shows the notice) instead of failing
silently.
"""
errors: list = []
context = browser.new_context(viewport={"width": 1100, "height": 720})
page = context.new_page()
page.on("pageerror", lambda error: errors.append(f"pageerror: {error}"))
page.on(
"console",
lambda message: (
errors.append(f"console: {message.text}") if message.type == "error" else None
),
)
page.goto(f"{insecure_base_url}/mirror", wait_until="domcontentloaded")
# The premise: this really is the trap — the browser has hidden the
# media API, exactly as it does on a bare-LAN-IP deployment.
assert page.evaluate("() => window.isSecureContext") is False
assert page.evaluate("() => navigator.mediaDevices === undefined")
expect(page.locator("#net")).to_have_attribute("data-state", "online", timeout=10_000)
# The page says so and disables the button on load…
expect(page.locator("#notice")).to_be_visible()
expect(page.locator("#notice")).to_contain_text("secure context")
expect(page.locator("#start")).to_be_disabled()
# …and `s` does not fail silently: the notice stays up, nothing starts.
page.keyboard.press("s")
expect(page.locator("#notice")).to_be_visible()
expect(page.locator("#notice")).to_contain_text("secure context")
expect(page.locator("body")).to_have_attribute("data-live", "off")
assert errors == [], f"uncaught browser errors: {errors}"
context.close()
class TestSuiteIsPointerFree:
"""The kiosk guarantee, enforced on the file itself.
A test that reaches for the pointer proves nothing about a kiosk. The
guarantee lives in the code, not in reviewer attention, so this meta-test
bans the pointer calls on this file and fails the suite the moment one
slips back in.
"""
def test_the_suite_never_reaches_for_the_pointer(self):
source = Path(__file__).read_text()
# Built from parts so this very assertion does not contain the banned
# literals (a plain string here would trip the check on itself).
assert ("cli" + "ck(") not in source, "a pointer call slipped into the keyboard suite"
assert ("hover" + "(") not in source, "a pointer call slipped into the keyboard suite"