"""End-to-end: a hand on another machine drives the /kiosk display.

Phase 09, task 03: the whole loop the TODO item claims — a mirror on the wall,
a box across the room with no keyboard, and a person on a laptop or phone who
can put a mirror on the stage, send it fullscreen, or put the wall back. The
controller is asserted *on the display's DOM and on the server's numbers*,
never on the controller's optimism (a page that paints a result and hopes is
the regression this suite exists to catch), and the media is asserted as
decoded, changing pixels on the display's staged ``<video>`` (AGENTS.md §5).

Unlike the kiosk suite, this one **clicks**: a person with a mouse driving a
machine that has none is the entire feature. Isolated contexts are the unit
of "a device": each is its own socket and its own peer id, which is exactly
what "another client" means to the server. Uncaught page errors and console
errors are captured from the first line of every test and surfaced through
the ``errors == []`` assertion.

Two invariants are asserted here because they are the product's, not the
browser's:

* **A controller costs a sharer nothing** — with only ``/remote`` open, the
  stream reports ``viewers: 0`` and the server's ``watching`` count is 0: the
  controller never subscribes and never opens a media leg (PLAN.md §6's mesh
  cost). Opening a real ``/watch`` is what moves those numbers.
* **``followers`` counts displays, never controllers** — a second ``/remote``
  does not raise the count, so "no display following" keeps meaning what it
  says.

Fullscreen is asserted on the page's own knobs — ``body[data-fullscreen]``
and the server's 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``). Asserting on the
browser's concession would turn this suite into a flake.
"""

from __future__ import annotations

import time

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 streams_api(base_url) -> dict:
    return httpx.get(f"{base_url}/api/streams", timeout=5).json()


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


def wait_for_streams(base_url, predicate, what: str, timeout: float = 15.0) -> dict:
    deadline = time.time() + timeout
    last: dict = {}
    while time.time() < deadline:
        last = streams_api(base_url)
        if predicate(last):
            return last
        time.sleep(0.2)
    raise AssertionError(f"{what}; last /api/streams was {last}")


def wait_row(remote, timeout: int = 15_000) -> None:
    """The controller's list holds one row: `welcome` has landed."""
    expect(remote.locator("#remote-list .mirror-row")).to_have_count(1, timeout=timeout)


class TestTheLoop:
    def test_show_on_wall_and_back_to_the_wall(self, device, base_url):
        """The whole loop: a click on the controller arranges the display, and
        "back to the wall" undoes it — asserted on the display, not the click."""
        errors: list = []
        sharer = device("/mirror", errors)
        start_sharing(sharer)
        (stream_id,) = mirror_ids(base_url, 1)

        display = device("/kiosk", errors)
        expect(display.locator("body")).to_have_attribute("data-layout", "wall")
        controller = device("/remote", errors)
        wait_row(controller)
        # The row the click lands on names the live mirror.
        assert controller.locator(".mirror-row").get_attribute("data-stream") == stream_id

        # One click: "Show on wall". The display must obey, on the server's
        # numbers as well as its own DOM.
        controller.locator(".mirror-row .show").click()
        expect(display.locator("body")).to_have_attribute("data-layout", "stage", timeout=15_000)
        expect(display.locator(f"#kiosk-stage .tile[data-stream='{stream_id}']")).to_have_count(
            1, timeout=15_000
        )
        state = wait_for_kiosk(base_url, lambda s: s["staged"] == stream_id, "the display staged")
        assert state["staged"] == stream_id, state

        # The controller reflects the server's state, not its own click: the
        # row says the mirror is on the wall.
        expect(controller.locator(".mirror-row[data-on]")).to_have_count(1, timeout=15_000)

        controller.locator("#remote-back").click()
        expect(display.locator("body")).to_have_attribute("data-layout", "wall", timeout=15_000)
        expect(display.locator("#kiosk-stage .tile")).to_have_count(0)
        wait_for_kiosk(base_url, lambda s: s["staged"] is None, "the display unstaged")

        assert errors == [], f"uncaught browser errors: {errors}"

    def test_the_controller_moves_real_media_to_the_display(self, device, base_url):
        """Pixels, not classes: the assertion is decoded, *changing* frames on
        the display's staged <video> (AGENTS.md §5) — a CSS class would prove
        nothing crossed the wire."""
        errors: list = []
        sharer = device("/mirror", errors)
        start_sharing(sharer)
        (stream_id,) = mirror_ids(base_url, 1)

        display = device("/kiosk", errors)
        controller = device("/remote", errors)
        wait_row(controller)
        controller.locator(".mirror-row .show").click()
        expect(display.locator("body")).to_have_attribute("data-layout", "stage", timeout=15_000)
        expect(
            display.locator(f"#kiosk-stage .tile[data-stream='{stream_id}'][data-state='live']")
        ).to_have_count(1, timeout=25_000)

        frames = frames_matching(display, "#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}"

        assert errors == [], f"uncaught browser errors: {errors}"


class TestFullscreen:
    def test_fullscreen_is_reported_not_promised(self, device, base_url):
        """The controller requests fullscreen; the display obeys in its state.
        Never assert ``document.fullscreenElement`` here: headless Chromium
        may refuse an element ``requestFullscreen()`` without user activation,
        and the real box's window manager forces the window fullscreen anyway
        (``docs/kiosk-silverblue/window-config.ini``). The page's own knob and
        the server's state are what the app controls; the browser's
        concession is a bonus, not the guarantee."""
        errors: list = []
        sharer = device("/mirror", errors)
        start_sharing(sharer)
        mirror_ids(base_url, 1)

        display = device("/kiosk", errors)
        controller = device("/remote", errors)
        wait_row(controller)
        # Normalize the shared server state: the suites share one session
        # server, and the kiosk suite's first test leaves `fullscreen: true`
        # behind (a teardown resets `staged`, never `fullscreen`). The button
        # toggles from the server's value, so a leaked true would make this
        # click flip it to false — reset it through the product, the way a
        # controller would.
        if kiosk_state(base_url)["fullscreen"]:
            controller.locator("#remote-fullscreen").click()
            wait_for_kiosk(base_url, lambda s: s["fullscreen"] is False, "the state normalized")
        controller.locator(".mirror-row .show").click()
        expect(display.locator("body")).to_have_attribute("data-layout", "stage", timeout=15_000)

        controller.locator("#remote-fullscreen").click()
        state = wait_for_kiosk(
            base_url, lambda s: s["fullscreen"] is True, "the server's state went fullscreen"
        )
        assert state["fullscreen"] is True, state
        expect(display.locator("body")).to_have_attribute("data-fullscreen", "true", timeout=15_000)
        # The controller's button reads the server's value, not its own click.
        expect(controller.locator("#remote-fullscreen")).to_have_attribute("aria-pressed", "true")

        assert errors == [], f"uncaught browser errors: {errors}"


class TestConvergence:
    def test_two_controllers_converge_without_touching_the_second(self, device, base_url):
        """One acts, the other was never touched: its readout and the
        ``data-on`` row update from the server's frame, so two open
        controllers can never disagree (task 02's no-optimistic-paint rule,
        proven from the outside)."""
        errors: list = []
        sharer = device("/mirror", errors)
        start_sharing(sharer)
        (stream_id,) = mirror_ids(base_url, 1)

        display = device("/kiosk", errors)
        first = device("/remote", errors)
        second = device("/remote", errors)  # nothing is ever clicked here
        wait_row(first)
        wait_row(second)

        first.locator(".mirror-row .show").click()
        expect(display.locator("body")).to_have_attribute("data-layout", "stage", timeout=15_000)

        # The untouched controller: its row carries the marker and the button
        # reads the server's state.
        row = second.locator(".mirror-row[data-on]")
        expect(row).to_have_count(1, timeout=15_000)
        assert row.get_attribute("data-stream") == stream_id
        expect(row.locator(".show")).to_have_text("On the wall")
        # Its readout agrees too: the staged mirror's title is the row's title.
        title = row.locator(".title").inner_text()
        expect(second.locator("#remote-staged")).to_have_text(title, timeout=15_000)

        # And the server, which the display obeys, names the same mirror.
        state = kiosk_state(base_url)
        assert state["staged"] == stream_id, state

        assert errors == [], f"uncaught browser errors: {errors}"


class TestACostsNothing:
    def test_a_controller_costs_the_sharer_nothing(self, device, base_url):
        """The guard that keeps /remote from becoming a stealth watcher:
        the controller never subscribes, so the sharer uploads nothing for
        it — the per-stream ``viewers`` and the server's ``watching`` count
        stay at zero until a real ``/watch`` joins (PLAN.md §6)."""
        errors: list = []
        sharer = device("/mirror", errors)
        start_sharing(sharer)
        (stream_id,) = mirror_ids(base_url, 1)

        def only(stream_id):
            def check(payload):
                mirror = next(m for m in payload["mirrors"] if m["id"] == stream_id)
                return mirror["viewers"] == 0 and payload["stats"]["watching"] == 0

            return check

        # The baseline: nothing is watching before anyone opens a page.
        wait_for_streams(base_url, only(stream_id), "the idle baseline")

        controller = device("/remote", errors)
        wait_row(controller)  # the page is connected and rendered…
        # …and still consumes nothing: no subscription, no leg.
        wait_for_streams(base_url, only(stream_id), "the controller not counted")

        device("/watch", errors)
        wait_for_streams(
            base_url,
            lambda payload: (
                next(m for m in payload["mirrors"] if m["id"] == stream_id)["viewers"] == 1
                and payload["stats"]["watching"] == 1
            ),
            "the real watcher counted",
        )

        assert errors == [], f"uncaught browser errors: {errors}"


class TestFollowers:
    def test_followers_and_the_readout_agree_and_a_controller_is_not_a_display(
        self, device, base_url
    ):
        """`/api/kiosk` and the controller's `#remote-followers` read the same
        number at 0, at 1 with a display open, and still at 1 with a second
        /remote added — a controller is not a display, so "1 display
        following" keeps meaning what it says."""
        errors: list = []
        sharer = device("/mirror", errors)
        start_sharing(sharer)

        assert kiosk_state(base_url)["followers"] == 0, "the suite must start with no displays"

        controller = device("/remote", errors)
        wait_row(controller)  # welcome has landed: the readout is the server's
        assert kiosk_state(base_url)["followers"] == 0
        expect(controller.locator("#remote-followers")).to_contain_text("no display following")

        device("/kiosk", errors)
        wait_for_kiosk(base_url, lambda s: s["followers"] == 1, "the display counted")
        expect(controller.locator("#remote-followers")).to_contain_text(
            "1 display following", timeout=15_000
        )

        second = device("/remote", errors)
        wait_row(second)
        assert kiosk_state(base_url)["followers"] == 1, "a controller is not a display"
        expect(second.locator("#remote-followers")).to_contain_text(
            "1 display following", timeout=15_000
        )

        assert errors == [], f"uncaught browser errors: {errors}"


class TestTheDisplayIsAControllerToo:
    def test_a_keypress_on_the_display_moves_the_remote_readout(self, device, base_url):
        """The TODO's "any client can manipulate", in the direction that would
        otherwise be forgotten: with a keyboard plugged in to the box, `n`
        stages a mirror *and* publishes the intent, so the controller — which
        receives no keypress — follows it."""
        errors: list = []
        sharer = device("/mirror", errors)
        start_sharing(sharer)
        (stream_id,) = mirror_ids(base_url, 1)

        display = device("/kiosk", errors)
        controller = device("/remote", errors)
        wait_row(controller)
        expect(controller.locator("#remote-staged")).to_have_text("no mirror")

        # The one input in this test: a keypress on the display.
        display.keyboard.press("n")
        expect(display.locator("body")).to_have_attribute("data-layout", "stage", timeout=15_000)

        # The untouched controller follows: its readout names the staged
        # mirror and its row carries the marker.
        row = controller.locator(".mirror-row")
        title = row.locator(".title").inner_text()
        expect(controller.locator("#remote-staged")).to_have_text(title, timeout=15_000)
        expect(row).to_have_count(1)
        expect(controller.locator(".mirror-row[data-on]")).to_have_count(1, timeout=15_000)
        assert row.get_attribute("data-stream") == stream_id

        assert errors == [], f"uncaught browser errors: {errors}"


class TestGoneIsAState:
    def test_an_ended_mirror_unstages_the_display_and_clears_the_controller(self, device, base_url):
        """Gone is a state: when the mirror ends, the display falls back to
        the wall and the controller's row and readout clear — no stale tile,
        no stale readout, no stale server state."""
        errors: list = []
        sharer = device("/mirror", errors)
        start_sharing(sharer)
        (stream_id,) = mirror_ids(base_url, 1)

        display = device("/kiosk", errors)
        controller = device("/remote", errors)
        wait_row(controller)
        controller.locator(".mirror-row .show").click()
        expect(display.locator("body")).to_have_attribute("data-layout", "stage", timeout=15_000)
        expect(display.locator(f"#kiosk-stage .tile[data-stream='{stream_id}']")).to_have_count(1)

        stop_sharing(sharer)

        # No display sits on a dead tile.
        expect(display.locator("body")).to_have_attribute("data-layout", "wall", timeout=10_000)
        expect(display.locator("#kiosk-stage .tile")).to_have_count(0)
        # The controller: the row is gone and the readout says no mirror.
        expect(controller.locator("#remote-list .mirror-row")).to_have_count(0, timeout=10_000)
        expect(controller.locator("#remote-staged")).to_have_text("no mirror", timeout=10_000)
        # And the server cleared the stage.
        wait_for_kiosk(base_url, lambda s: s["staged"] is None, "the layout cleared")

        assert errors == [], f"uncaught browser errors: {errors}"