mirror/tests/e2e/test_kiosk_layout.py

"""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"