ducoterra
f51d0fe9ae
**Phase 05 final verification pass — all green.** - Verified implementation in tree: staged `.cap`/`.actions` hidden at rest with container+children `pointer-events: none` (`app.css`); `revealStageControls()` with 3 s idle timer, `pointermove`/`pointerdown` + wall-keypress wiring, `unstage()` timer/attribute cleanup (`watch.js`); `tests/e2e/test_stage_autofade.py` (5 tests) present; `test_keyboard_wall.py` rest-state assertion updated per task 01. - Gates: `uv run ruff check . && uv run ruff format --check .` — clean; `uv run pyright` — 0 errors; `uv run pytest` — **274 passed, 4 skipped** (realdisplay-gated), coverage **95%** (>90% ✓); `MIRROR_REALDISPLAY=1 uv run pytest -m realdisplay` — **3 passed, 1 skipped** (documented platform skip: no audio source on host). - Dedicated phase suite `uv run pytest tests/e2e/test_stage_autofade.py` — **5/5 in isolation**. - Criteria: idle opacity 0 + bottom-edge hit probe ✓; mouse reveal → 1 → idle fade → 0 ✓; keypress reveal + Exit visible/clickable/unstages ✓; grid rules byte-for-byte unchanged (CSS diff touches only the staged block; grid guard tests green) ✓; suite green, coverage >90% ✓; phases 01–04 suites all green (no regression) ✓. - Documentation: README.md reviewed in full — all commands, env vars (incl. `DEBUGPY_*`, 25 Mbit/s default), `--allow-cidr`, kiosk/share key tables, index-page and auto-hide descriptions match the code; the only stale "12 Mbit/s" default lives in locked PLAN.md §6, README is correct — no change made. AGENTS.md: all facts (gates, test names, starlette 1.10, e2e flags, ~95% coverage) current — verified, no changes needed. - Notable: no code changes were required this pass — the phase work was already complete and correct; nothing to fix. - Next pending phase: none — phase 05 is the last in `todo/`.
908 lines
37 KiB
Python
908 lines
37 KiB
Python
"""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 overlay is hidden at rest — caption bar and control
|
|
# cluster alike — so a shared terminal's bottom line is never
|
|
# covered (phase 05). 10 s stays green under task 02 as well: an
|
|
# Enter-press reveal lasts 3 s + 200 ms fade, so the settled `0`
|
|
# is always reachable inside the window.
|
|
viewer.wait_for_function(
|
|
"() => { const tile = document.querySelector('#watch-stage .tile');"
|
|
" if (!tile) return false;"
|
|
" return getComputedStyle(tile.querySelector('.cap')).opacity === '0'"
|
|
" && getComputedStyle(tile.querySelector('.actions')).opacity === '0'; }",
|
|
timeout=10_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"
|