phase: 08_kiosk_page
Build and Push Containers / test (push) Successful in 42s
Build and Push Containers / build-and-push-app (push) Successful in 9s

**Phase 08 final verification pass — complete, all green.**

**Verified (all 5 tasks already implemented):**
- `static/wall.js` (`createWall` + `installWallKeys`), `kiosk.html`/`kiosk.js`, `GET /kiosk` route — guards: no `$("#…")` in wall.js, one keydown listener, kiosk ids no-collision across 3 pages, no audio field in `arrange`, no header/share link in markup, address gate covers `/kiosk`
- Cross-page e2e: arrival-arrangement, keypress-on-one-display moves the other (asserted on the second display's decoded moving pixels), dead mirror → `data-layout="wall"` on both, `followers` count, pointer-free meta-test

**Gates (exact outcomes):**
- `uv run ruff check . && uv run ruff format --check .` — passed
- `uv run pyright` — 0 errors
- `uv run pytest` — 341 passed, 5 skipped (realdisplay-gated), 96s
- `MIRROR_REALDISPLAY=1 uv run pytest -m realdisplay` — 4 passed, 1 skipped (documented: host has no audio source for real capture)
- Coverage 96% (>90%); only warning is a starlette/httpx third-party deprecation

**Completion criteria:** all 9 ✅ — e2e green incl. `test_kiosk_layout.py` (the one e2e diff is the phase-mandated address-gate extension to `/kiosk`; assertion logic untouched), full suite 100% passing, README documents `/kiosk`, PLAN.md untouched.

**Documentation:**
- README: fixed the Kiosk-client section that implied the display box is permanently pointed at `/watch` — now states the display page is `/kiosk` with the runbook's one-line re-point (installer default still `/watch` this phase, per docs). Rest of README verified against code (routes, config table, CLI flag, presets, key table) — consistent.
- AGENTS.md: fixed 2 stale statements — CDN test moved to `TestPages` (was `TestClientScripts`), suite runtime ~50s → ~95s. No content added.

**Notable:** PLAN.md §6 says "per-leg `maxBitrate` cap (default 12 Mbit/s)" but code/README default is 25 Mbit/s — pre-existing, left untouched (locked file; flagged for a user decision).

**Next pending phase:** `09_wall_remote_page`.
This commit is contained in:
2026-11-03 22:38:22 -05:00
parent 736da1c1ef
commit cd29156a7d
45 changed files with 2389 additions and 750 deletions
+360
View File
@@ -0,0 +1,360 @@
"""End-to-end: the /kiosk display arranges itself, and obeys one client over another.
Phase 08, task 05: the two properties the TODO item claims — **it arranges
itself on arrival** (point a display at /kiosk and it does what the shared
layout already says, with nobody standing at the box) and **one client's
action moves another client's display** (a keypress on one kiosk re-arranges
the other, asserted on the second display's decoded pixels) — plus the
``followers`` count (displays only, not plain viewers) and the teardown
fallback (a dead mirror returns every display to the wall in the same frame
set; no display may sit on a dead tile).
Isolated contexts are the unit of "a client": each is its own socket and its
own peer id, which is exactly what "another client" means to the server. The
media path itself is proven by ``test_live_mirror.py``; this suite proves the
*state* path and still asserts on real decoded frames, so nothing is taken on
trust.
The suite is pointer-free: a test that reaches for the mouse proves nothing
about a kiosk. The only input a display receives here is a keypress; the
server state is driven by raw ``WebSocket`` frames in ``page.evaluate`` — a
wire-level client, not a pointer. That is what the meta-test at the bottom
permits: it bans pointer calls, not the driver's socket.
Fullscreen is asserted on the page's own knobs — ``body[data-fullscreen]``
and the staged state — and **never on ``document.fullscreenElement``**:
headless Chromium may refuse an element ``requestFullscreen()`` without user
activation, and the real kiosk's window manager forces the window fullscreen
anyway (``docs/kiosk-silverblue/window-config.ini``). The knobs are what the
app controls (the same stance ``kiosk.js`` documents for its best-effort
request); a "strengthened" assertion on ``fullscreenElement`` turns this
suite into a flake.
"""
from __future__ import annotations
import time
from pathlib import Path
import httpx
import pytest
from playwright.sync_api import expect
from .conftest import frames_matching, start_sharing, stop_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). Contexts are closed by finalizers, so a leaked peer here
cannot surface as a tile count in another suite's test (they share one
session-scoped server).
"""
opened = []
def _open(path: str = "/", errors: list | None = None):
context, page = open_device(base_url, browser, request, path, errors)
opened.append(context)
return page
return _open
def open_device(base_url, browser, request, path: str, errors: list | None = None):
"""Open one device; return ``(context, page)``.
The context is closed exactly once — by a finalizer at test end, or by
the test itself (the guard flag is what makes a finalizer after an
explicit close a no-op).
"""
context = browser.new_context(viewport={"width": 1100, "height": 720})
closed = [False]
def _finalize() -> None:
if not closed[0]:
closed[0] = True
context.close()
request.addfinalizer(_finalize)
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 context, page
def mirror_ids(base_url, count: int, timeout: float = 15.0) -> list[str]:
"""The server's truth about what is live (AGENTS.md §8, rung 1)."""
deadline = time.time() + timeout
last: list[str] = []
while time.time() < deadline:
payload = httpx.get(f"{base_url}/api/streams", timeout=5).json()
last = [mirror["id"] for mirror in payload["mirrors"]]
if len(last) == count:
return last
time.sleep(0.2)
raise AssertionError(f"expected {count} mirror(s) on /api/streams, saw {last}")
def kiosk_state(base_url) -> dict:
return httpx.get(f"{base_url}/api/kiosk", timeout=5).json()
def wait_for_kiosk(base_url, predicate, what: str, timeout: float = 15.0) -> dict:
deadline = time.time() + timeout
last: dict = {}
while time.time() < deadline:
last = kiosk_state(base_url)
if predicate(last):
return last
time.sleep(0.2)
raise AssertionError(f"{what}; last /api/kiosk was {last}")
# The driver: a plain WebSocket client, opened from inside a page so the
# handshake runs in a real browser. It is wire-level state, not a pointer —
# the meta-test below allows it on exactly that ground. It closes the socket
# once the server has acknowledged the change: the layout is process state,
# not the driver's, so the kiosk's `welcome` (which carries it) cannot race
# the driver's exit.
DRIVER = """
async ([url, staged, fullscreen]) => {
const ws = new WebSocket(url);
await new Promise((resolve, reject) => {
ws.onopen = () => resolve();
ws.onerror = () => reject(new Error('driver socket did not open'));
});
const nextFrame = (type) => new Promise((resolve, reject) => {
const timer = setTimeout(
() => { ws.close(); reject(new Error(`driver timed out waiting for ${type}`)); },
10_000);
ws.onmessage = (event) => {
const message = JSON.parse(event.data);
if (message.type === type) { clearTimeout(timer); resolve(message); }
};
});
ws.send(JSON.stringify({ type: 'hello', role: 'viewer', name: '', device: 'e2e layout driver' }));
const welcome = await nextFrame('welcome');
ws.send(JSON.stringify({ type: 'arrange', staged, fullscreen }));
const layout = await nextFrame('layout');
ws.close();
return { peerId: welcome.peerId, layout };
}
"""
def drive_layout(page, base_url, staged: str | None, fullscreen: bool) -> dict:
"""Send one `arrange` frame over a raw socket; return the server's ack."""
ws_url = base_url.replace("http://", "ws://") + "/ws"
return page.evaluate(DRIVER, [ws_url, staged, fullscreen])
def expect_staged(page, stream_id: str, timeout: int = 15_000) -> None:
"""The display's stage holds exactly the tile the layout names."""
expect(page.locator("#kiosk-stage .tile")).to_have_count(1, timeout=timeout)
assert page.locator("#kiosk-stage .tile").get_attribute("data-stream") == stream_id
class TestArrivalIsArrangement:
def test_a_display_opened_after_an_arrange_shows_it_without_input(self, device, base_url):
"""Arrival is arrangement: the state is applied before the first tile
paints, so the person pointing a kiosk at /kiosk touches nothing."""
errors: list = []
sharer = device("/mirror", errors)
start_sharing(sharer)
(stream_id,) = mirror_ids(base_url, 1)
# The controller acts first, over a raw socket: stage + fullscreen, and
# the server's own ack says it took.
driver = device("/watch", errors)
ack = drive_layout(driver, base_url, stream_id, True)
assert ack["layout"]["staged"] == stream_id
assert ack["layout"]["fullscreen"] is True
# The display opens *after* the arrange. No key, no pointer: it must
# settle on the stage.
kiosk = device("/kiosk", errors)
expect(kiosk.locator("body")).to_have_attribute("data-layout", "stage", timeout=15_000)
# The page's own knob, not document.fullscreenElement: headless may
# refuse the request and the window manager's window-fullscreen is the
# guarantee, so the knob (what the app controls) is the assertion.
expect(kiosk.locator("body")).to_have_attribute("data-fullscreen", "true", timeout=15_000)
expect_staged(kiosk, stream_id)
# The frame assertion is what proves media crossed to the display, not
# just a CSS class (AGENTS.md §5): decoded, *changing* pixels on the
# staged <video>.
expect(kiosk.locator("#kiosk-stage .tile[data-state='live']")).to_have_count(
1, timeout=25_000
)
frames = frames_matching(kiosk, "#kiosk-stage video")
assert frames["width"] > 0, f"no decoded frame size on the display: {frames}"
assert frames["height"] > 0, f"no decoded frame size on the display: {frames}"
assert frames["distinct"] >= 2, f"the staged video is frozen: {frames}"
# One source of truth: the server says what the pixels show.
state = kiosk_state(base_url)
assert state["staged"] == stream_id
assert state["fullscreen"] is True
# The display is counted; the driver (a plain viewer) is not.
assert state["followers"] == 1, state
assert errors == [], f"uncaught browser errors: {errors}"
class TestOneClientMovesAnother:
def test_a_keypress_on_one_display_rearranges_the_other(self, device, base_url):
"""One action, two displays, one source of truth — and the second
display never receives a pointer call (or any input at all)."""
errors: list = []
sharer = device("/mirror", errors)
start_sharing(sharer)
(stream_id,) = mirror_ids(base_url, 1)
first = device("/kiosk", errors)
second = device("/kiosk", errors) # nothing is ever typed or pointed here
expect(first.locator("body")).to_have_attribute("data-layout", "wall")
expect(second.locator("body")).to_have_attribute("data-layout", "wall")
# The only input in this test: one keypress on the first display.
first.keyboard.press("Enter")
expect(first.locator("body")).to_have_attribute("data-layout", "stage", timeout=5_000)
expect(second.locator("body")).to_have_attribute("data-layout", "stage", timeout=15_000)
expect_staged(second, stream_id)
# The display is showing it, not merely labelled: a live, moving leg.
expect(second.locator("#kiosk-stage .tile[data-state='live']")).to_have_count(
1, timeout=25_000
)
frames = frames_matching(second, "#kiosk-stage video")
assert frames["width"] > 0, f"no decoded frame on the second display: {frames}"
assert frames["distinct"] >= 2, f"the second display's video is frozen: {frames}"
# The server agrees with both displays.
state = kiosk_state(base_url)
assert state["staged"] == stream_id, state
assert errors == [], f"uncaught browser errors: {errors}"
def test_n_and_p_on_one_display_walk_the_other_through_the_mirrors(self, device, base_url):
errors: list = []
for _ in range(2):
start_sharing(device("/mirror", errors))
first_id, second_id = mirror_ids(base_url, 2)
first = device("/kiosk", errors)
second = device("/kiosk", errors)
expect(second.locator("body")).to_have_attribute("data-layout", "wall")
expect(second.locator(".tile")).to_have_count(2, timeout=15_000)
# n: no cursor yet, so it lands on the first mirror…
first.keyboard.press("n")
expect_staged(second, first_id)
# …p walks to the second…
first.keyboard.press("p")
expect_staged(second, second_id)
# …and wraps back to the first.
first.keyboard.press("p")
expect_staged(second, first_id)
# The server walked the same way (checked at the wrap, where a stale
# broadcast would still show the second mirror).
state = kiosk_state(base_url)
assert state["staged"] == first_id, state
# Both mirrors survived the walking: the display's wall holds both.
expect(second.locator(".tile")).to_have_count(2)
expect(second.locator(".tile[data-state='live']")).to_have_count(2, timeout=20_000)
assert errors == [], f"uncaught browser errors: {errors}"
def test_a_dead_mirror_returns_every_display_to_the_wall(self, device, base_url):
errors: list = []
sharer = device("/mirror", errors)
start_sharing(sharer)
(stream_id,) = mirror_ids(base_url, 1)
first = device("/kiosk", errors)
second = device("/kiosk", errors)
first.keyboard.press("Enter")
expect(first.locator("body")).to_have_attribute("data-layout", "stage", timeout=5_000)
expect_staged(second, stream_id)
stop_sharing(sharer)
# No display may sit on a dead tile: both back on the wall, in time.
expect(first.locator("body")).to_have_attribute("data-layout", "wall", timeout=10_000)
expect(second.locator("body")).to_have_attribute("data-layout", "wall", timeout=10_000)
expect(first.locator("#kiosk-stage .tile")).to_have_count(0)
expect(second.locator("#kiosk-stage .tile")).to_have_count(0)
# And the wall is actually empty, not just unstaged: the dead tile is
# gone, so there is no stale <video> to clear.
expect(first.locator(".tile")).to_have_count(0, timeout=10_000)
expect(second.locator(".tile")).to_have_count(0, timeout=10_000)
state = wait_for_kiosk(base_url, lambda s: s["staged"] is None, "the layout cleared")
assert state["followers"] == 2, state
assert errors == [], f"uncaught browser errors: {errors}"
class TestFollowers:
def test_followers_counts_displays_only_over_a_real_socket_lifecycle(
self, base_url, browser, request
):
"""`followers` is how many *displays* are obeying the wall — a plain
viewer (a /watch tab) is not one, and a display leaving is noticed
without any state to poll: the count is derived from the roster."""
errors: list = []
assert kiosk_state(base_url)["followers"] == 0, "the suite must start with no displays"
context, _kiosk = open_device(base_url, browser, request, "/kiosk", errors)
wait_for_kiosk(base_url, lambda s: s["followers"] == 1, "the display counted")
# A plain viewer joins: the wall follows the same state, but the count
# is about displays, not about who is reading the state.
open_device(base_url, browser, request, "/watch", errors)
wait_for_kiosk(base_url, lambda s: s["followers"] == 1, "the viewer not counted")
# The display leaves (its socket closes through a real context close)
# and the count falls back with it.
context.close()
wait_for_kiosk(base_url, lambda s: s["followers"] == 0, "the display counted out")
assert errors == [], f"uncaught browser errors: {errors}"
class TestSuiteIsPointerFree:
"""The kiosk guarantee, enforced on this file itself.
A test that reaches for the mouse proves nothing about a kiosk. The
guarantee lives in the code, not in reviewer attention, so this meta-test
bans pointer calls on this file and fails the suite the moment one slips
back in. The driver's raw ``WebSocket`` frames (``page.evaluate``) are
wire-level, not a pointer, and are allowed *explicitly* on that ground —
this guard bans the two pointer calls (click and hover) anywhere in this
file, nothing else, and does not loosen to skip the driver's block: a
pointer call anywhere in this file is a failure.
"""
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 kiosk suite"
assert ("hover" + "(") not in source, "a pointer call slipped into the kiosk suite"
+4 -2
View File
@@ -544,7 +544,7 @@ class TestAddressGate:
like a bug in the tool.
"""
@pytest.mark.parametrize("path", ["/mirror", "/watch"])
@pytest.mark.parametrize("path", ["/mirror", "/watch", "/kiosk"])
def test_a_refused_address_sees_why_and_none_of_the_app(self, browser, restricted_url, path):
context = browser.new_context(viewport={"width": 1100, "height": 720})
page = context.new_page()
@@ -557,7 +557,9 @@ class TestAddressGate:
# front of a proxy is undiagnosable.
expect(page.locator("body")).to_contain_text("127.0.0.1")
# None of the app shell leaked past the gate.
expect(page.locator("#start, #grid, #notice")).to_have_count(0)
expect(page.locator("#start, #grid, #notice, #kiosk-grid, #kiosk-stage")).to_have_count(
0
)
finally:
context.close()
+158 -20
View File
@@ -80,6 +80,7 @@ class TestPages:
("/mirror", "Share my screen"),
("/share", "Share my screen"),
("/watch", "tile-template"),
("/kiosk", "kiosk-tile-template"),
],
)
def test_pages_render(self, client, path, marker):
@@ -88,14 +89,17 @@ class TestPages:
assert response.headers["content-type"].startswith("text/html")
assert marker in response.text
def test_pages_are_not_cacheable(self, client):
assert client.get("/watch").headers["cache-control"] == "no-store"
@pytest.mark.parametrize("path", ["/", "/mirror", "/share", "/watch", "/kiosk"])
def test_pages_are_not_cacheable(self, client, path):
assert client.get(path).headers["cache-control"] == "no-store"
def test_the_two_pages_use_the_media_apis_they_claim(self):
# The watch page's media path lives in wall.js (phase 08): the page
# builds the wall, and the wall owns the RTCPeerConnection.
share = (STATIC / "share.js").read_text()
watch = (STATIC / "watch.js").read_text()
wall = (STATIC / "wall.js").read_text()
assert "getDisplayMedia" in share
assert "RTCPeerConnection" in share and "RTCPeerConnection" in watch
assert "RTCPeerConnection" in share and "RTCPeerConnection" in wall
assert "new WebSocket" in (STATIC / "net.js").read_text()
def test_nothing_is_loaded_from_a_cdn(self):
@@ -109,7 +113,10 @@ class TestPages:
offenders.append(f"{file.name} (@import)")
assert offenders == []
@pytest.mark.parametrize("asset", ["app.css", "net.js", "share.js", "watch.js"])
@pytest.mark.parametrize(
"asset",
["app.css", "net.js", "share.js", "watch.js", "wall.js", "kiosk.js"],
)
def test_static_assets_are_served(self, client, asset):
assert client.get(f"/static/{asset}").status_code == 200
@@ -124,7 +131,7 @@ class TestClientScripts:
NODE = shutil.which("node")
@pytest.mark.skipif(NODE is None, reason="node is not installed")
@pytest.mark.parametrize("script", ["net.js", "share.js", "watch.js"])
@pytest.mark.parametrize("script", ["net.js", "share.js", "watch.js", "wall.js", "kiosk.js"])
def test_scripts_parse_as_es_modules(self, script, tmp_path):
copy = tmp_path / "subject.mjs"
copy.write_text((STATIC / script).read_text())
@@ -133,7 +140,11 @@ class TestClientScripts:
@pytest.mark.parametrize(
("script", "page"),
[("share.js", "mirror.html"), ("watch.js", "watch.html")],
[
("share.js", "mirror.html"),
("watch.js", "watch.html"),
("kiosk.js", "kiosk.html"),
],
)
def test_every_element_the_script_reaches_for_exists(self, script, page):
html = (STATIC / page).read_text()
@@ -141,8 +152,100 @@ class TestClientScripts:
missing = {name for name in wanted if f'id="{name}"' not in html}
assert not missing, f"{script} looks up {sorted(missing)} which {page} does not render"
def test_both_pages_load_the_stylesheet_and_their_module(self):
for page, script in (("mirror.html", "share.js"), ("watch.html", "watch.js")):
def test_wall_never_looks_up_elements_in_the_document(self):
"""wall.js serves /watch and /kiosk (and phase 09's /remote) from one
module; an element lookup would resolve in whichever page happens to
load it. The elements come from the page instead, which is why the
element ids must not collide across pages (see
test_the_two_pages_share_no_element_ids_but_the_shared_chrome) — so
the module is guarded at the source level, here, next to the other
client-script guards."""
source = (STATIC / "wall.js").read_text()
offenders = re.findall(r"""\$\(['\"]#([A-Za-z0-9_-]+)['\"]\)""", source)
assert not offenders, f"wall.js must not query the document; found {offenders}"
def test_the_wall_key_router_is_one_listener_in_wall_js(self):
"""One listener, one switch, per wall page — phase 01's whole point.
`installWallKeys` (wall.js) is the single keydown listener a wall page
gets (/watch, /kiosk, phase 09's /remote): a second listener on the
same document would run the same key twice (j would move the cursor
twice) or race the first. The share page is not a wall page — its
s/x/Esc listener lives in share.js and is page policy — but a wall
page module must never add a keydown listener of its own, so a second
`addEventListener("keydown"` in a wall page module fails the build.
"""
wall = (STATIC / "wall.js").read_text()
assert "export function installWallKeys" in wall, (
"installWallKeys must be defined in wall.js (the one keydown listener)"
)
assert len(re.findall(r'addEventListener\(\s*["\']keydown["\']', wall)) == 1, (
"wall.js must hold exactly one keydown listener (the router)"
)
offenders = [
file.name
for file in sorted(STATIC.glob("*.js"))
if file.name not in ("share.js", "wall.js")
and re.search(r'addEventListener\(\s*["\']keydown["\']', file.read_text())
]
assert offenders == [], (
"a wall page module added its own keydown listener; the one listener "
f"per wall page lives in wall.js: {offenders}"
)
@staticmethod
def _object_literal_around(source, pos):
"""The object literal containing ``source[pos:``.
``pos`` points at a ``type: "arrange"`` key. Walking back, the nearest
``{`` that is not inside a deeper ``{...}`` span is the literal's start
— the literal itself when ``type`` is its first member, the enclosing
object when it is not. Forward brace matching then closes it.
"""
depth = 0
start = None
for i in range(pos - 1, -1, -1):
if source[i] == "}":
depth += 1
elif source[i] == "{":
if depth == 0:
start = i
break
depth -= 1
assert start is not None, "no object literal before the arrange key"
depth = 0
for i in range(start, len(source)):
if source[i] == "{":
depth += 1
elif source[i] == "}":
depth -= 1
if depth == 0:
return source[start : i + 1]
raise AssertionError("unbalanced braces around the arrange frame")
def test_kiosk_never_publishes_an_audio_field(self):
"""The layout has no audio field (SECURITY.md §4): a display must not
become audible because another machine asked, so ``kiosk.js`` may never
put a ``muted``/``audio`` field in an ``arrange`` frame. The rule is a
security rule, so it fails the build here instead of living in a
comment (phase 08 task 04). At least one ``arrange`` frame must exist:
the display is a first-class controller, and this guard must not pass
on a file that simply stopped publishing."""
source = (STATIC / "kiosk.js").read_text()
matches = list(re.finditer(r"type\s*:\s*[\"\']arrange[\"\']", source))
assert matches, "kiosk.js publishes no arrange frames at all"
for match in matches:
literal = self._object_literal_around(source, match.start())
assert not re.search(r"\b(muted|audio)\s*:", literal), (
f"an arrange frame carries an audio field: {literal}"
)
def test_the_pages_load_the_stylesheet_and_their_module(self):
for page, script in (
("mirror.html", "share.js"),
("watch.html", "watch.js"),
("kiosk.html", "kiosk.js"),
):
html = (STATIC / page).read_text()
assert 'href="/static/app.css"' in html
assert f'src="/static/{script}"' in html
@@ -289,23 +392,55 @@ class TestClientScripts:
)
def test_the_two_pages_share_no_element_ids_but_the_shared_chrome(self):
"""Both pages consume one stylesheet, so a duplicated id means a colliding rule.
"""The pages consume one stylesheet, so a duplicated id means a colliding rule.
This is not hypothetical: a `#stage` on both pages once made the share
page's preview inherit `display: none` from the watch page's overlay.
The kiosk page's ids carry the `kiosk-` prefix for exactly this reason;
the only ids it may share with another page are the shared chrome
(net, notice, toast).
"""
shared = {"net", "notice", "toast"}
ids = {
page: set(re.findall(r'id="([A-Za-z0-9_-]+)"', (STATIC / page).read_text()))
for page in ("mirror.html", "watch.html", "index.html")
for page in ("mirror.html", "watch.html", "index.html", "kiosk.html")
}
assert ids["mirror.html"] & ids["watch.html"] == shared
assert ids["index.html"] & ids["mirror.html"] <= shared | {"livebar"}
assert ids["kiosk.html"] & ids["watch.html"] == shared
assert ids["kiosk.html"] & ids["mirror.html"] == shared
# index.html has no notice/toast (a signpost, not an app page), so its
# intersection with the kiosk is a subset of the shared chrome — the
# same shape as the index/mirror assertion above.
assert ids["kiosk.html"] & ids["index.html"] <= shared
for page, found in ids.items():
assert len(found) == len(set(found)), f"{page} declares an id twice"
def test_the_wall_pages_render_their_stage_classed(self):
"""The stage rules are class-keyed (`.stage`, phase 08 task 01): a page
whose stage lost the class renders a full-size unstyled video, which is
exactly the class of bug the stylesheet guards exist for. Both wall
pages keep it, and the kiosk's structural elements (grid, stage, hud)
exist for the layout applier to take them. (There was no
elements-a-page-must-have check before; this is the one.)
"""
kiosk = (STATIC / "kiosk.html").read_text()
for element_id in ("kiosk-grid", "kiosk-stage", "kiosk-hud"):
assert f'id="{element_id}"' in kiosk, f"kiosk.html is missing #{element_id}"
for page, stage_id in (("watch.html", "watch-stage"), ("kiosk.html", "kiosk-stage")):
stage_tags = [
tag
for tag in re.findall(r"<div[^>]*>", (STATIC / page).read_text())
if f'id="{stage_id}"' in tag
]
assert stage_tags, f'{page} is missing <div id="{stage_id}">'
assert 'class="stage"' in stage_tags[0], (
f'{page}#{stage_id} lost class="stage": the stage rules are '
"class-keyed, so an unclassed stage renders a full-size unstyled video"
)
def test_videos_are_muted_and_inline_so_autoplay_is_not_blocked(self):
for page in ("mirror.html", "watch.html"):
for page in ("mirror.html", "watch.html", "kiosk.html"):
for tag in re.findall(r"<video[^>]*>", (STATIC / page).read_text()):
assert "muted" in tag, f"{page}: {tag} would be blocked by autoplay policy"
assert "playsinline" in tag, f"{page}: {tag} would hijack iOS into fullscreen"
@@ -378,8 +513,9 @@ class TestStylesheet:
The flat ``[^{}]+\{[^{}]*\}`` pass treats an ``@media`` header as a
selector and the first child's header as its body; the child rules
that follow still come back as their own pairs, which is all these
guards need — every assertion is about a top-level block (``.stage``,
``.stage > *``) or about properties that must sit in the *same* block
guards need — every assertion is about a top-level block
(``.preview-stage``, ``.preview-stage > *``) or about properties that
must sit in the *same* block
(``overflow``/``aspect-ratio``/``min-height`` together).
"""
css = re.sub(r"/\*.*?\*/", "", (STATIC / "app.css").read_text(), flags=re.S)
@@ -396,8 +532,8 @@ class TestStylesheet:
without the other stacks the rows back.
"""
rules = self._rules()
stage_bodies = [body for sel, body in rules if sel == ".stage"]
assert stage_bodies, ".stage rule missing from app.css"
stage_bodies = [body for sel, body in rules if sel == ".preview-stage"]
assert stage_bodies, ".preview-stage rule missing from app.css"
assert any(
re.search(r"grid-template\s*:\s*1fr\s*/\s*1fr", body)
or (
@@ -409,7 +545,7 @@ class TestStylesheet:
".stage must declare an explicit single cell (grid-template: 1fr / 1fr, "
"or grid-template-rows/columns of 1fr) — otherwise its children stack into rows"
)
child_bodies = [body for sel, body in rules if sel == ".stage > *"]
child_bodies = [body for sel, body in rules if sel == ".preview-stage > *"]
assert child_bodies and any(
re.search(r"grid-area\s*:\s*1\s*/\s*1", body) for body in child_bodies
), (
@@ -431,8 +567,8 @@ class TestStylesheet:
content again.
"""
rules = self._rules()
stage_bodies = [body for sel, body in rules if sel == ".stage"]
assert stage_bodies, ".stage rule missing from app.css"
stage_bodies = [body for sel, body in rules if sel == ".preview-stage"]
assert stage_bodies, ".preview-stage rule missing from app.css"
assert any(re.search(r"min-height\s*:", body) for body in stage_bodies), (
".stage must keep its floor (min-height)"
)
@@ -460,7 +596,8 @@ class TestAddressGate:
assert client.get("/watch").status_code == 200
@pytest.mark.parametrize(
"path", ["/", "/mirror", "/watch", "/api/streams", "/api/kiosk", "/static/app.css"]
"path",
["/", "/mirror", "/watch", "/kiosk", "/api/streams", "/api/kiosk", "/static/app.css"],
)
def test_a_refused_address_gets_no_page_no_roster_no_assets(self, path):
with gated(self.ALLOW) as client:
@@ -527,6 +664,7 @@ class TestAddressGate:
def test_the_real_address_is_in_and_connects(self):
with gated(self.ALLOW, client_address="10.2.1.10") as client:
assert client.get("/watch").status_code == 200
assert client.get("/kiosk").status_code == 200
with client.websocket_connect("/ws") as socket:
assert handshake(socket, "viewer")