"""HTTP surface + signaling socket tests. No browser required."""

from __future__ import annotations

import json
import re
import shutil
import subprocess
from pathlib import Path

import pytest
from fastapi.testclient import TestClient
from starlette.websockets import WebSocketDisconnect

from mirror.access import parse_cidrs
from mirror.app import create_app
from mirror.config import Settings

STATIC = Path(__file__).resolve().parent.parent / "src/mirror/static"


def gated(*allow, client_address="203.0.113.9", **kwargs):
    """A client speaking as ``client_address`` to a server restricted to ``allow``."""
    settings = Settings(allow_cidrs=parse_cidrs(",".join(allow)), **kwargs)
    return TestClient(create_app(settings), client=(client_address, 44444))


EXTERNAL_REF = re.compile(r"""(?:src|href)\s*=\s*["']https?://""", re.IGNORECASE)


def handshake(socket, role, name=""):
    socket.send_json({"type": "hello", "role": role, "name": name})
    message = socket.receive_json()
    assert message["type"] == "welcome", message
    return message["peerId"]


def wait_for(socket, kind):
    """Read frames until one has ``type == kind``.

    The roster legitimately pushes ``stream_updated`` frames whenever the
    audience changes, so tests must not assume a one-message-per-action cadence.
    Any ``error`` frame seen on the way is a failure worth surfacing.
    """
    seen = []
    while True:
        message = socket.receive_json()
        if message.get("type") == kind:
            return message
        assert message.get("type") != "error", (
            f"unexpected error while waiting for {kind}: {message}"
        )
        seen.append(message)
        assert len(seen) < 25, f"never saw {kind!r}; got {seen}"


def wait_for_update_with(socket, **stream_fields):
    """`wait_for` for `stream_updated` frames, matched on the stream payload.

    `stream_updated` is pushed for audience changes too, so the test must
    keep reading until the payload itself says what is being awaited (e.g.
    the audio flag actually flipped)."""
    seen = []
    while True:
        message = socket.receive_json()
        if message.get("type") == "stream_updated" and all(
            message["stream"].get(key) == value for key, value in stream_fields.items()
        ):
            return message["stream"]
        assert message.get("type") != "error", f"unexpected error: {message}"
        seen.append(message)
        assert len(seen) < 25, f"never saw stream_updated with {stream_fields}; got {seen}"


class TestPages:
    @pytest.mark.parametrize(
        ("path", "marker"),
        [
            ("/", "Watch everything"),
            ("/mirror", "Share my screen"),
            ("/share", "Share my screen"),
            ("/watch", "tile-template"),
        ],
    )
    def test_pages_render(self, client, path, marker):
        response = client.get(path)
        assert response.status_code == 200
        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"

    def test_the_two_pages_use_the_media_apis_they_claim(self):
        share = (STATIC / "share.js").read_text()
        watch = (STATIC / "watch.js").read_text()
        assert "getDisplayMedia" in share
        assert "RTCPeerConnection" in share and "RTCPeerConnection" in watch
        assert "new WebSocket" in (STATIC / "net.js").read_text()

    def test_nothing_is_loaded_from_a_cdn(self):
        """The no-CDN rule is a hard product requirement, so assert it."""
        offenders = []
        for file in sorted(STATIC.iterdir()):
            text = file.read_text()
            if EXTERNAL_REF.search(text):
                offenders.append(file.name)
            if "@import" in text or "fonts.googleapis" in text:
                offenders.append(f"{file.name} (@import)")
        assert offenders == []

    @pytest.mark.parametrize("asset", ["app.css", "net.js", "share.js", "watch.js"])
    def test_static_assets_are_served(self, client, asset):
        assert client.get(f"/static/{asset}").status_code == 200

    def test_unknown_page_is_a_404(self, client):
        assert client.get("/nope").status_code == 404


class TestClientScripts:
    """Guards for the hand-written ES modules: they must parse, and the ids they
    reach for must exist. Both of these have already bitten once."""

    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"])
    def test_scripts_parse_as_es_modules(self, script, tmp_path):
        copy = tmp_path / "subject.mjs"
        copy.write_text((STATIC / script).read_text())
        result = subprocess.run([self.NODE, "--check", str(copy)], capture_output=True, text=True)
        assert result.returncode == 0, f"{script}: {result.stderr}"

    @pytest.mark.parametrize(
        ("script", "page"),
        [("share.js", "mirror.html"), ("watch.js", "watch.html")],
    )
    def test_every_element_the_script_reaches_for_exists(self, script, page):
        html = (STATIC / page).read_text()
        wanted = set(re.findall(r"""\$\(\"#([A-Za-z0-9_-]+)\"\)""", (STATIC / script).read_text()))
        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")):
            html = (STATIC / page).read_text()
            assert 'href="/static/app.css"' in html
            assert f'src="/static/{script}"' in html
            assert 'type="module"' in html

    def test_share_never_asks_for_an_exact_capture_size(self):
        """A hard size breaks pickers: `exact` fails or hides picker entries on
        displays/DPRs the browser cannot honour. The size request must stay a
        preference the browser can decline while still capturing."""
        source = (STATIC / "share.js").read_text()
        assert not re.search(r"\bexact\s*:", source), (
            "share.js must not use an `exact` constraint key"
        )

    def test_server_and_client_agree_on_the_bitrate_default(self):
        """`/api/config` advertises the server's per-leg cap; a browser that cannot
        reach it falls back to the hardcoded literal in `net.js`. If the two drift,
        offline clients silently get a different picture quality and nothing fails.
        One number, two places, one test that notices."""
        server = int(
            re.search(
                r"DEFAULT_MAX_BITRATE_BPS\s*=\s*(\d[\d_]*)",
                (STATIC.parent / "config.py").read_text(),
            )
            .group(1)
            .replace("_", "")
        )
        client = int(re.search(r"maxBitrateBps:\s*(\d+)", (STATIC / "net.js").read_text()).group(1))
        assert server == client, (
            f"the server default ({server}) and the net.js fallback ({client}) "
            "have drifted: a client that cannot reach /api/config would use "
            "a different bitrate silently"
        )

    def test_the_quality_options_are_the_three_presets(self):
        """`#quality` (mirror.html) and `PRESETS` (share.js) are one list in two places.

        If the option values drift from the table keys, a preset the panel offers has
        no row (or an orphan row the panel can never select), and `currentPreset()`
        falls back to `sharp` instead of noticing: the sharer gets a wall they did not
        ask for and nothing fails. One list, two places, one test that notices (the
        same shape as the bitrate-default test)."""
        html = (STATIC / "mirror.html").read_text()
        select = re.search(r'<select id="quality"[^>]*>.*?</select>', html, re.S)
        assert select, "mirror.html must render a #quality select"
        options = set(re.findall(r'value="([A-Za-z0-9_-]+)"', select.group(0)))

        source = (STATIC / "share.js").read_text()
        table = re.search(r"const PRESETS = Object\.freeze\(\{(.*?)\}\);", source, re.S)
        assert table, "share.js must define the PRESETS table"
        # Top-level keys sit two spaces in; the fields inside a preset sit four
        # spaces in, so anchoring on exactly two spaces selects the keys only.
        keys = set(re.findall(r"^  ([a-z]+): \{$", table.group(1), re.M))

        expected = {"sharp", "balanced", "light"}
        assert options == expected, f"#quality offers {sorted(options)}, not {sorted(expected)}"
        assert keys == expected, f"PRESETS keys are {sorted(keys)}, not {sorted(expected)}"
        assert options == keys, "the select offers presets the table cannot resolve"

    def test_sender_quality_is_reapplied_on_negotiation_and_connection(self):
        """`setParameters` is per-negotiation: a re-offer (a surface switch,
        a viewer that re-joined) resets the encodings. The preset's cap and
        degradation preference must therefore be re-applied after
        `setLocalDescription` *and* when the leg becomes `connected`. Both
        call sites are asserted so a future edit cannot drop one and silently
        soften the wall. The preference itself stays a single, named
        decision — the PRESETS table's `degradation` field, read at call
        time — so a hardcoded literal at the call site would be a second,
        drifting decision."""
        source = (STATIC / "share.js").read_text()
        assert re.search(r'degradationPreference\s*=\s*"', source) is None, (
            "a hardcoded degradation literal at a call site is a second decision"
        )
        assert "params.degradationPreference = PRESETS[currentPreset()].degradation" in source, (
            "the preference is not read from the PRESETS table at call time"
        )
        assert "async function applySenderQuality" in source

        def section(start: str, stop: str) -> str:
            begin = source.index(start)
            return source[begin : source.index(stop, begin)]

        negotiate = section("async function negotiate(viewerId)", "async function onSignal")
        assert "applySenderQuality" in negotiate, (
            "the cap is not re-applied after setLocalDescription"
        )
        assert negotiate.index("setLocalDescription") < negotiate.index("applySenderQuality"), (
            "the cap is applied before the encodings exist"
        )

        connected = section(
            "pc.onconnectionstatechange = () => {",
            "for (const track of state.stream.getTracks())",
        )
        assert re.search(
            r'connectionState\s*===\s*"connected".*applySenderQuality', connected, re.S
        ), "the cap is not re-applied when the leg becomes connected"

    def test_pick_surface_rethrows_cancellation_and_retries_only_no_audio_errors(self):
        """The audio-off retry must be precise. A cancelled picker
        (NotAllowedError) is a refusal — in Chrome it is also what unticking
        "Share tab audio" produces — and must be *rethrown*: retrying would
        silently re-open the picker and share a surface the person meant not
        to. Only the errors that genuinely mean "this source has no audio"
        (a failed audio *constraint*, not a failed capture) may drop the
        audio requirement; a bare retry in a catch-all would hide real
        capture failures and turn refusals into silent shares."""
        source = (STATIC / "share.js").read_text()
        # pickSurface and its named-error set live between these markers; the
        # first `catch` in that span is pickSurface's own.
        begin = source.index("async function pickSurface")
        section = source[begin : source.index("function publish", begin)]
        catch = section[section.index("catch") :]
        assert "NotAllowedError" in catch, "the catch block must name the refusal error"
        assert re.search(r"NotAllowedError[\s\S]*?throw\s+err", catch), (
            "a cancelled picker must be rethrown, not retried"
        )
        for name in (
            "NotSupportedError",
            "OverconstrainedError",
            "NotFoundError",
            "TypeError",
        ):
            assert name in section, f"the audio-off retry must name {name}"
        assert "return await navigator.mediaDevices.getDisplayMedia" not in catch, (
            "the audio-off retry must be conditional on the named errors, "
            "not a bare retry in a catch-all"
        )
        assert re.search(r"audiolessError\(err\)\s*\)\s*throw\s+err", section), (
            "an error the named set does not cover must surface, not be retried"
        )

    def test_no_bare_publish_frame_wipes_the_roster(self):
        """The wire's `publish` is a full-metadata frame. A bare
        `{ type: "publish" }` frame parses with the model's defaults and
        would wipe width/height/audio off the roster on every surface switch
        (the mute/unmute republish path) — the panel and the wall would
        disagree from a single event. The full frame is the only publish
        share.js may send."""
        source = (STATIC / "share.js").read_text()
        assert source.count('type: "publish"') == 1, (
            "share.js must send exactly one publish frame (the full one); "
            "a bare { type: 'publish' } frame would reset the metadata"
        )

    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.

        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.
        """
        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")
        }
        assert ids["mirror.html"] & ids["watch.html"] == shared
        assert ids["index.html"] & ids["mirror.html"] <= shared | {"livebar"}
        for page, found in ids.items():
            assert len(found) == len(set(found)), f"{page} declares an id twice"

    def test_videos_are_muted_and_inline_so_autoplay_is_not_blocked(self):
        for page in ("mirror.html", "watch.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"


class TestStylesheet:
    def test_hidden_always_means_gone(self):
        """Every component that sets its own `display` outranks the UA's `[hidden]`.

        Without this rule the share page shows the LIVE badge and the Stop button
        while nothing is being shared, and every wall tile offers an Exit button
        it is not on stage for.
        """
        css = (STATIC / "app.css").read_text()
        assert re.search(r"\[hidden\]\s*\{[^}]*display:\s*none\s*!important", css), (
            "app.css must declare a global `[hidden] { display: none !important }`"
        )
        # Components toggled with `hidden` in JS. All but `.placeholder` set
        # their own `display`, which would outrank the UA's `[hidden]` without
        # the global rule; `.placeholder` is guarded because its block is the
        # escape hatch that keeps its content inside the stage.
        for selector in (".btn", ".livetag", ".icon-btn", ".toast", ".net", ".placeholder"):
            assert selector in css, f"{selector} is toggled by `hidden`; it lives in app.css"

    def test_no_rule_reintroduces_a_conflicting_display_for_hidden(self):
        """Every rule whose selector mentions `[hidden]` must actually hide."""
        css = re.sub(r"/\*.*?\*/", "", (STATIC / "app.css").read_text(), flags=re.S)
        offenders = [
            selector.strip()
            for selector, body in re.findall(r"([^{}]+)\{([^{}]*)\}", css)
            if "[hidden]" in selector and not re.search(r"display\s*:\s*none", body)
        ]
        assert offenders == []

    def test_a_hidden_control_is_not_revealed_only_by_hover(self):
        """Kiosk rule: a control hidden by default must be reachable without a pointer.

        The tile's controls (`.actions`) start at `opacity: 0`. If the only rule
        that raises them to full opacity were a `:hover` rule, a mouseless kiosk
        would have no mute/expand controls at all. So while any `.actions` rule
        hides them, the stylesheet must also contain a `.actions` reveal rule
        with a non-hover trigger — keyboard focus or the selection state.
        """
        css = re.sub(r"/\*.*?\*/", "", (STATIC / "app.css").read_text(), flags=re.S)
        actions_rules = [
            (selector.strip(), body)
            for selector, body in re.findall(r"([^{}]+)\{([^{}]*)\}", css)
            if ".actions" in selector
        ]
        hiding = [
            selector
            for selector, body in actions_rules
            if re.search(r"opacity\s*:\s*0(?![0-9])", body)
        ]
        non_hover_reveals = [
            selector
            for selector, body in actions_rules
            if re.search(r"opacity\s*:\s*1(?![0-9])", body)
            and re.search(r":focus-visible|:focus-within|data-selected", selector)
        ]
        if hiding:
            assert non_hover_reveals, (
                f".actions starts hidden ({sorted(hiding)}) but no rule reveals it "
                "without a pointer — a mouseless kiosk would have no tile controls"
            )

    def _rules(self) -> list[tuple[str, str]]:
        r"""app.css as ``(selector, body)`` pairs, comments stripped.

        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
        (``overflow``/``aspect-ratio``/``min-height`` together).
        """
        css = re.sub(r"/\*.*?\*/", "", (STATIC / "app.css").read_text(), flags=re.S)
        return [(sel.strip(), body) for sel, body in re.findall(r"([^{}]+)\{([^{}]*)\}", css)]

    def test_the_share_stage_is_one_cell(self):
        """The stage's children overlap on purpose.

        In plain flow an idle ``<video>`` is intrinsically 300x150, so the
        placeholder (and the Share button in it) starts 151px below the video
        at *every* width and runs out of the ``overflow: hidden`` box.
        ``grid-template: 1fr / 1fr`` makes the stage one cell; ``grid-area:
        1 / 1`` on every child is what puts them all in it. Either half
        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"
        assert any(
            re.search(r"grid-template\s*:\s*1fr\s*/\s*1fr", body)
            or (
                re.search(r"grid-template-rows\s*:\s*1fr", body)
                and re.search(r"grid-template-columns\s*:\s*1fr", body)
            )
            for body in stage_bodies
        ), (
            ".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 > *"]
        assert child_bodies and any(
            re.search(r"grid-area\s*:\s*1\s*/\s*1", body) for body in child_bodies
        ), (
            "app.css must put every .stage child in that one cell (grid-area: 1 / 1) — "
            "without it the placeholder stacks below the 300x150 video and the Share "
            "button is clipped by overflow: hidden"
        )

    def test_the_stage_has_a_floor_and_a_ceiling(self):
        """An aspect-derived box with no floor is a clip.

        ``aspect-ratio`` alone made the stage 180px tall at 320px wide (the
        idle placeholder needs 240px there) and 543px tall in a 480px-tall
        window (the Share button below the fold). The floor and the ceiling
        make the ratio the *preferred* shape instead of a hard derivation, and
        the second check is the general statement of the bug: any rule that
        keeps ``overflow: hidden`` on an aspect-ratio box must carry a
        ``min-height`` in the same block, or the hidden clip silently swallows
        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"
        assert any(re.search(r"min-height\s*:", body) for body in stage_bodies), (
            ".stage must keep its floor (min-height)"
        )
        assert any(re.search(r"max-height\s*:", body) for body in stage_bodies), (
            ".stage must keep its ceiling (max-height)"
        )
        for sel, body in rules:
            if re.search(r"overflow[^:]*:\s*[^;]*hidden", body) and re.search(
                r"aspect-ratio\s*:", body
            ):
                assert re.search(r"min-height\s*:", body), (
                    f"{sel} combines `overflow: hidden` with `aspect-ratio` but has no "
                    "`min-height` in the same block — an aspect-derived box with no "
                    "floor clips its content silently"
                )


class TestAddressGate:
    """mirror has no accounts; the address is the access control."""

    ALLOW = "10.2.0.0/22"

    def test_unset_allows_everyone_which_is_the_development_default(self):
        with gated(client_address="203.0.113.9") as client:
            assert client.get("/watch").status_code == 200

    @pytest.mark.parametrize(
        "path", ["/", "/mirror", "/watch", "/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:
            assert client.get(path).status_code == 403

    def test_the_probe_stays_answerable_from_anywhere(self):
        """The container health check comes from loopback, which the list excludes."""
        with gated(self.ALLOW) as client:
            response = client.get("/healthz")
            assert response.status_code == 200
            assert response.json()["status"] == "ok"

    def test_a_browser_gets_an_explanation_rather_than_json(self):
        with gated(self.ALLOW) as client:
            response = client.get("/mirror", headers={"accept": "text/html"})
            assert response.status_code == 403
            assert "text/html" in response.headers["content-type"]
            body = response.text
            assert self.ALLOW in body
            assert "203.0.113.9" in body  # which address was judged, or this is undiagnosable

    def test_the_refusal_page_cannot_depend_on_assets_the_client_cannot_fetch(self):
        with gated(self.ALLOW) as client:
            body = client.get("/", headers={"accept": "text/html"}).text
            assert not EXTERNAL_REF.search(body)
            assert "/static/" not in body

    def test_a_fetch_gets_the_machine_readable_refusal(self):
        with gated(self.ALLOW) as client:
            response = client.get("/api/config")
            assert response.status_code == 403
            assert response.json() == {
                "error": "forbidden",
                "message": "This mirror only accepts connections from 10.2.0.0/22.",
                "you": "203.0.113.9",
            }

    def test_a_refusal_is_never_cacheable(self):
        with gated(self.ALLOW) as client:
            for headers in ({"accept": "text/html"}, {}):
                response = client.get("/api/config", headers=headers)
                assert response.headers["cache-control"] == "no-store"

    def test_a_refused_address_never_learns_the_layout(self):
        with gated(self.ALLOW) as client:
            response = client.get("/api/kiosk")
            assert response.status_code == 403
            assert "staged" not in response.text

    def test_the_real_address_gets_the_layout_read_with_no_store(self):
        with gated(self.ALLOW, client_address="10.2.1.10") as client:
            response = client.get("/api/kiosk")
            assert response.status_code == 200
            assert response.headers["cache-control"] == "no-store"

    def test_the_socket_closes_before_it_accepts(self):
        with (
            gated(self.ALLOW) as client,
            pytest.raises(WebSocketDisconnect),
            client.websocket_connect("/ws") as socket,
        ):
            socket.send_json({"type": "hello", "role": "viewer"})

    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
            with client.websocket_connect("/ws") as socket:
                assert handshake(socket, "viewer")

    def test_a_trusted_local_proxy_carries_the_client_through(self):
        """The fleet shape: nginx on the same host, X-Forwarded-For set by nginx."""
        with gated(self.ALLOW, client_address="127.0.0.1") as client:
            assert client.get("/watch", headers={"x-forwarded-for": "10.2.1.5"}).status_code == 200
            with client.websocket_connect("/ws", headers={"x-forwarded-for": "10.2.1.5"}) as socket:
                assert handshake(socket, "viewer")

    def test_a_trusted_proxy_carrying_a_foreign_client_is_still_refused(self):
        with gated(self.ALLOW, client_address="127.0.0.1") as client:
            assert client.get("/watch", headers={"x-forwarded-for": "8.8.8.8"}).status_code == 403
            with (
                pytest.raises(WebSocketDisconnect),
                client.websocket_connect("/ws", headers={"x-forwarded-for": "8.8.8.8"}),
            ):
                pass

    def test_an_untrusted_peer_cannot_write_its_own_pass(self):
        with gated(self.ALLOW, client_address="203.0.113.9") as client:
            response = client.get("/watch", headers={"x-forwarded-for": "10.2.1.5"})
            assert response.status_code == 403
            assert response.json()["you"] == "203.0.113.9"

    def test_both_address_families_can_be_admitted_side_by_side(self):
        with gated(f"{self.ALLOW}, fd00::/64", client_address="fd00::1234") as client:
            assert client.get("/watch").status_code == 200


class TestJsonApi:
    def test_config_advertises_the_media_defaults(self, client):
        body = client.get("/api/config").json()
        assert body["iceServers"] == []
        assert body["maxBitrateBps"] == 2_000_000
        assert "version" in body

    def test_config_is_a_whitelist(self, client):
        body = client.get("/api/config").json()
        assert set(body) == {"version", "iceServers", "maxBitrateBps", "pingIntervalMs"}

    def test_streams_starts_empty(self, client):
        body = client.get("/api/streams").json()
        assert body["mirrors"] == []
        assert body["stats"]["mirrors"] == 0

    def test_healthz(self, client):
        assert client.get("/healthz").json()["status"] == "ok"

    def test_kiosk_answers_exactly_the_layout_dto(self, client):
        """Shape guard: the read endpoint is the socket's `layout` DTO, nothing more.

        It answers the same object the socket pushes (task 02's single-DTO
        invariant), so the key set — including the `type` tag — is pinned: a new
        key here is a decision (00_phase.md), not a leak.
        """
        body = client.get("/api/kiosk").json()
        assert set(body) == {"type", "staged", "fullscreen", "followers"}
        assert body == {"type": "layout", "staged": None, "fullscreen": False, "followers": 0}


class TestSignalingSocket:
    def test_hello_must_come_first(self, client):
        with client.websocket_connect("/ws") as socket:
            socket.send_json({"type": "publish", "label": "sneaky"})
            message = socket.receive_json()
            assert message["type"] == "error"
            assert message["code"] == "hello_required"
            assert message["fatal"] is True

    def test_malformed_json_is_rejected(self, client):
        with client.websocket_connect("/ws") as socket:
            socket.send_text("{not json")
            assert socket.receive_json()["code"] == "bad_json"

    def test_a_non_object_frame_is_rejected(self, client):
        with client.websocket_connect("/ws") as socket:
            socket.send_text("[1, 2, 3]")
            assert socket.receive_json()["code"] == "bad_json"

    def test_an_unknown_role_is_refused(self, client):
        with client.websocket_connect("/ws") as socket:
            socket.send_json({"type": "hello", "role": "spectator"})
            assert socket.receive_json()["code"] == "bad_role"

    def test_unknown_message_type_is_reported_but_survivable(self, client):
        with client.websocket_connect("/ws") as socket:
            handshake(socket, "viewer", "V")
            socket.send_json({"type": "telepathy"})
            assert socket.receive_json()["code"] == "bad_message"
            socket.send_json({"type": "ping"})
            assert socket.receive_json()["type"] == "pong"

    def test_ping_roundtrips(self, client):
        with client.websocket_connect("/ws") as socket:
            handshake(socket, "viewer")
            socket.send_json({"type": "ping"})
            assert socket.receive_json() == {"type": "pong"}

    def test_an_oversized_frame_is_refused(self, client):
        with client.websocket_connect("/ws") as socket:
            handshake(socket, "viewer")
            socket.send_text(json.dumps({"type": "ping", "padding": "x" * 70_000}))
            assert socket.receive_json()["code"] == "too_large"

    def test_roster_flows_from_sharer_to_viewer(self, client):
        with client.websocket_connect("/ws") as sharer, client.websocket_connect("/ws") as viewer:
            sharer_id = handshake(sharer, "sharer", "Linux Box")
            viewer_id = handshake(viewer, "viewer", "Phone")

            sharer.send_json({"type": "publish", "label": "Terminal", "width": 1280, "height": 720})
            added = wait_for(viewer, "stream_added")["stream"]
            assert added["id"] == sharer_id
            assert added["label"] == "Terminal"
            assert added["name"] == "Linux Box"
            assert added["width"] == 1280

            viewer.send_json({"type": "subscribe", "streamId": sharer_id})
            joined = wait_for(sharer, "peer_joined")
            assert joined["viewerId"] == viewer_id
            assert joined["name"] == "Phone"

            sharer.send_json({"type": "signal", "to": viewer_id, "payload": {"sdp": {"s": "o"}}})
            relayed = wait_for(viewer, "signal")
            assert relayed["from"] == sharer_id
            assert relayed["payload"] == {"sdp": {"s": "o"}}

            viewer.send_json({"type": "signal", "to": sharer_id, "payload": {"c": "ice"}})
            assert wait_for(sharer, "signal")["payload"] == {"c": "ice"}

    def test_capture_metadata_survives_the_wire(self, client):
        """Regression: `displaySurface` once arrived as "" because the model expected
        snake_case and unknown keys were being ignored."""
        with client.websocket_connect("/ws") as sharer, client.websocket_connect("/ws") as viewer:
            handshake(sharer, "sharer", "A")
            handshake(viewer, "viewer")
            sharer.send_json(
                {
                    "type": "publish",
                    "label": "proj.gl — zsh",
                    "displaySurface": "window",
                    "width": 1920,
                    "height": 1080,
                    "audio": True,
                }
            )
            stream = wait_for(viewer, "stream_added")["stream"]
            assert stream["displaySurface"] == "window"
            assert stream["audio"] is True
            assert (stream["width"], stream["height"]) == (1920, 1080)
            assert client.get("/api/streams").json()["mirrors"][0]["displaySurface"] == "window"

    def test_the_audio_flag_survives_the_wire_and_a_republish_flips_it(self, client):
        """Regression for the audio work: `audio` is the one bit that tells a
        viewer whether to expect sound. A publish with audio on must arrive
        with the flag set (and the REST roster must agree); a republish that
        drops it must flip it via `stream_updated` for **every** subscribed
        viewer; and an `unpublish` must leave the wall empty — a viewer that
        joins afterwards sees an empty roster, so the flip cannot leak into a
        stream that no longer exists. The policy half is the same one the
        capture-metadata test pins; this records it for the audio phase so it
        cannot regress."""
        with (
            client.websocket_connect("/ws") as sharer,
            client.websocket_connect("/ws") as one,
            client.websocket_connect("/ws") as two,
        ):
            sharer_id = handshake(sharer, "sharer", "A")
            handshake(one, "viewer", "one")
            handshake(two, "viewer", "two")
            sharer.send_json({"type": "publish", "label": "Screen", "audio": True})
            assert wait_for(one, "stream_added")["stream"]["audio"] is True
            assert wait_for(two, "stream_added")["stream"]["audio"] is True
            assert client.get("/api/streams").json()["mirrors"][0]["audio"] is True

            one.send_json({"type": "subscribe", "streamId": sharer_id})
            two.send_json({"type": "subscribe", "streamId": sharer_id})
            wait_for(sharer, "peer_joined")
            wait_for(sharer, "peer_joined")

            # A shape change (audio off) is a republish: each subscribed
            # viewer is re-offered and the flag arrives flipped in
            # `stream_updated`. The subscribes already pushed one (audience
            # change), so keep reading until the republish's own update
            # lands on *each* viewer.
            sharer.send_json({"type": "publish", "label": "Screen", "audio": False})
            for viewer in (one, two):
                assert wait_for_update_with(viewer, audio=False)["audio"] is False
            assert client.get("/api/streams").json()["mirrors"][0]["audio"] is False

            # Teardown: every viewer is told, and the wall is empty
            # afterwards — no stray update may keep advertising the gone
            # stream's audio.
            sharer.send_json({"type": "unpublish"})
            wait_for(one, "stream_removed")
            wait_for(two, "stream_removed")
            body = client.get("/api/streams").json()
            assert body["mirrors"] == []
            assert body["stats"]["mirrors"] == 0

        # The empty wall is unaffected: a viewer that joins after the
        # teardown sees an empty roster, with no audio bit on anything.
        with client.websocket_connect("/ws") as late:
            late.send_json({"type": "hello", "role": "viewer", "name": "late"})
            welcome = late.receive_json()
            assert welcome["type"] == "welcome", welcome
            assert welcome["streams"] == []

    def test_both_viewers_of_one_mirror_see_their_own_leg(self, client):
        with (
            client.websocket_connect("/ws") as sharer,
            client.websocket_connect("/ws") as one,
            client.websocket_connect("/ws") as two,
        ):
            sharer_id = handshake(sharer, "sharer", "A")
            first = handshake(one, "viewer", "one")
            second = handshake(two, "viewer", "two")
            sharer.send_json({"type": "publish", "label": "Screen"})
            wait_for(one, "stream_added")
            wait_for(two, "stream_added")

            one.send_json({"type": "subscribe", "streamId": sharer_id})
            two.send_json({"type": "subscribe", "streamId": sharer_id})
            assert wait_for(sharer, "peer_joined")["viewerId"] == first
            assert wait_for(sharer, "peer_joined")["viewerId"] == second

            sharer.send_json({"type": "signal", "to": first, "payload": {"n": 1}})
            sharer.send_json({"type": "signal", "to": second, "payload": {"n": 2}})
            assert wait_for(one, "signal")["payload"] == {"n": 1}
            assert wait_for(two, "signal")["payload"] == {"n": 2}

    def test_a_mirror_disappears_for_viewers_when_the_sharer_leaves(self, client):
        with client.websocket_connect("/ws") as sharer, client.websocket_connect("/ws") as viewer:
            sharer_id = handshake(sharer, "sharer", "A")
            handshake(viewer, "viewer", "V")
            sharer.send_json({"type": "publish", "label": "Screen"})
            wait_for(viewer, "stream_added")
            viewer.send_json({"type": "subscribe", "streamId": sharer_id})
            wait_for(sharer, "peer_joined")

            sharer.close()
            assert wait_for(viewer, "stream_removed")["streamId"] == sharer_id

    def test_unpublish_removes_the_mirror(self, client):
        with client.websocket_connect("/ws") as sharer, client.websocket_connect("/ws") as viewer:
            sharer_id = handshake(sharer, "sharer")
            handshake(viewer, "viewer")
            sharer.send_json({"type": "publish", "label": "Screen"})
            wait_for(viewer, "stream_added")
            sharer.send_json({"type": "unpublish"})
            assert wait_for(viewer, "stream_removed")["streamId"] == sharer_id
            assert client.get("/api/streams").json()["mirrors"] == []

    def test_signalling_a_peer_that_never_subscribed_is_refused(self, client):
        with client.websocket_connect("/ws") as sharer, client.websocket_connect("/ws") as viewer:
            sharer_id = handshake(sharer, "sharer")
            viewer_id = handshake(viewer, "viewer")
            sharer.send_json({"type": "publish", "label": "Screen"})
            assert wait_for(viewer, "stream_added")["stream"]["sharerId"] == sharer_id
            # The viewer knows the mirror exists but has not subscribed to it.
            sharer.send_json({"type": "signal", "to": viewer_id, "payload": {}})
            assert wait_for(sharer, "error")["code"] == "not_subscribed"

    def test_a_viewer_cannot_reach_a_sharer_it_never_subscribed_to(self, client):
        with client.websocket_connect("/ws") as sharer, client.websocket_connect("/ws") as viewer:
            sharer_id = handshake(sharer, "sharer", "A")
            handshake(viewer, "viewer")
            sharer.send_json({"type": "publish", "label": "Screen"})
            wait_for(viewer, "stream_added")
            viewer.send_json({"type": "signal", "to": sharer_id, "payload": {}})
            assert wait_for(viewer, "error")["code"] == "not_subscribed"

    def test_subscribing_to_a_mirror_that_is_not_there_reports_cleanly(self, client):
        with client.websocket_connect("/ws") as viewer:
            handshake(viewer, "viewer")
            viewer.send_json({"type": "subscribe", "streamId": "p404"})
            message = wait_for(viewer, "error")
            assert message["code"] == "unknown_stream"
            assert message["fatal"] is False

    def test_editing_the_display_name_reaches_viewers(self, client):
        with client.websocket_connect("/ws") as sharer, client.websocket_connect("/ws") as viewer:
            sharer_id = handshake(sharer, "sharer", "Old name")
            handshake(viewer, "viewer")
            sharer.send_json({"type": "publish", "label": "Screen"})
            wait_for(viewer, "stream_added")
            sharer.send_json({"type": "hello", "role": "sharer", "name": "Design Laptop"})
            assert wait_for(viewer, "stream_updated")["stream"]["name"] == "Design Laptop"
            assert client.get("/api/streams").json()["mirrors"][0]["sharerId"] == sharer_id

    def test_stats_track_the_fan_out(self, client):
        with client.websocket_connect("/ws") as sharer, client.websocket_connect("/ws") as viewer:
            sharer_id = handshake(sharer, "sharer", "A")
            handshake(viewer, "viewer", "V")
            sharer.send_json({"type": "publish", "label": "Screen"})
            wait_for(viewer, "stream_added")
            viewer.send_json({"type": "subscribe", "streamId": sharer_id})
            wait_for(sharer, "peer_joined")
            stats = client.get("/api/streams").json()["stats"]
            assert stats["mirrors"] == 1
            assert stats["watching"] == 1
            assert stats["connections"] == 2

    def test_one_peer_cannot_impersonate_another(self, client):
        """`to` is addressed by peer id; a viewer must not be able to spoof a sharer."""
        with client.websocket_connect("/ws") as sharer, client.websocket_connect("/ws") as viewer:
            sharer_id = handshake(sharer, "sharer", "A")
            handshake(viewer, "viewer", "V")
            sharer.send_json({"type": "publish", "label": "Screen"})
            wait_for(viewer, "stream_added")
            viewer.send_json({"type": "publish", "label": "forged"})
            assert wait_for(viewer, "error")["code"] == "not_sharer"
            assert client.get("/api/streams").json()["stats"]["mirrors"] == 1
            assert sharer_id and viewer

    def test_arrange_roundtrips_to_viewers_the_sender_and_the_read_endpoint(self, client):
        """The whole path: publish, arrange, and every observer — the sender, the
        other viewers, and `GET /api/kiosk` — says the same thing."""
        with (
            client.websocket_connect("/ws") as sharer,
            client.websocket_connect("/ws") as viewer,
            client.websocket_connect("/ws") as second,
        ):
            sharer_id = handshake(sharer, "sharer", "A")
            handshake(viewer, "viewer", "V1")
            handshake(second, "viewer", "V2")
            sharer.send_json({"type": "publish", "label": "Screen"})
            wait_for(viewer, "stream_added")
            wait_for(second, "stream_added")

            viewer.send_json({"type": "arrange", "staged": sharer_id, "fullscreen": True})
            sent = wait_for(viewer, "layout")
            assert sent["staged"] == sharer_id
            assert sent["fullscreen"] is True
            assert wait_for(second, "layout") == sent
            body = client.get("/api/kiosk").json()
            assert (body["staged"], body["fullscreen"], body["followers"]) == (sharer_id, True, 0)

            # A peer that joins after the arrange is already arranged: its
            # welcome carries the state, so a page can navigate and obey
            # without asking. (Joins while the wall still exists — closing the
            # sharer's socket would legitimately reset the stage.)
            with client.websocket_connect("/ws") as late:
                late.send_json({"type": "hello", "role": "viewer", "name": "late"})
                welcome = late.receive_json()
                assert welcome["layout"]["staged"] == sharer_id
                assert welcome["layout"]["fullscreen"] is True

    def test_kiosk_followers_are_counted_and_recounted_on_leave(self, client):
        with (
            client.websocket_connect("/ws") as plain,
            client.websocket_connect("/ws") as kiosk1,
            client.websocket_connect("/ws") as kiosk2,
        ):
            handshake(plain, "viewer", "plain")
            kiosk1.send_json({"type": "hello", "role": "viewer", "name": "K1", "kiosk": True})
            assert kiosk1.receive_json()["type"] == "welcome"
            kiosk2.send_json({"type": "hello", "role": "viewer", "name": "K2", "kiosk": True})
            # The joiner counts itself: it is connected and it declared kiosk.
            assert kiosk2.receive_json()["layout"]["followers"] == 2

            # The plain viewer was told by every kiosk join, in order.
            assert wait_for(plain, "layout")["followers"] == 1
            assert wait_for(plain, "layout")["followers"] == 2
            assert wait_for(kiosk1, "layout")["followers"] == 2

            kiosk1.send_json({"type": "arrange", "fullscreen": True})
            sent = wait_for(kiosk1, "layout")
            assert sent["fullscreen"] is True
            assert sent["followers"] == 2
            assert wait_for(plain, "layout")["followers"] == 2
            assert wait_for(kiosk2, "layout")["followers"] == 2

            kiosk1.close()
            # A follower leaving re-counts: the remaining viewers hear `1`.
            assert wait_for(kiosk2, "layout")["followers"] == 1
            assert wait_for(plain, "layout")["followers"] == 1
            assert client.get("/api/kiosk").json()["followers"] == 1

    def test_arranging_a_mirror_that_is_not_there_is_refused_without_closing(self, client):
        with client.websocket_connect("/ws") as viewer:
            handshake(viewer, "viewer")
            viewer.send_json({"type": "arrange", "staged": "p404"})
            error = wait_for(viewer, "error")
            assert error["code"] == "unknown_stream"
            assert error["fatal"] is False
            # The socket is still up, and nothing changed.
            viewer.send_json({"type": "ping"})
            assert viewer.receive_json() == {"type": "pong"}
            assert client.get("/api/kiosk").json()["staged"] is None

    def test_a_fullscreen_only_arrange_leaves_the_stage_and_null_returns_the_wall(self, client):
        """End to end: absent ≠ null. A `fullscreen`-only frame keeps the staged
        mirror; an explicit `staged: null` is the only way back to the wall."""
        with (
            client.websocket_connect("/ws") as sharer,
            client.websocket_connect("/ws") as viewer,
        ):
            sharer_id = handshake(sharer, "sharer")
            handshake(viewer, "viewer")
            sharer.send_json({"type": "publish", "label": "Screen"})
            wait_for(viewer, "stream_added")

            viewer.send_json({"type": "arrange", "staged": sharer_id})
            wait_for(viewer, "layout")

            viewer.send_json({"type": "arrange", "fullscreen": True})
            frame = wait_for(viewer, "layout")
            assert frame["staged"] == sharer_id  # absent field left alone
            assert frame["fullscreen"] is True

            viewer.send_json({"type": "arrange", "staged": None})
            frame = wait_for(viewer, "layout")
            assert frame["staged"] is None  # explicit null returns the wall
            assert frame["fullscreen"] is True  # and touches nothing else
            body = client.get("/api/kiosk").json()
            assert (body["staged"], body["fullscreen"]) == (None, True)

    def test_a_sharer_may_arrange_but_only_viewers_and_the_sender_are_told(self, client):
        with (
            client.websocket_connect("/ws") as sharer_a,
            client.websocket_connect("/ws") as sharer_b,
            client.websocket_connect("/ws") as viewer,
        ):
            a_id = handshake(sharer_a, "sharer", "A")
            b_id = handshake(sharer_b, "sharer", "B")
            handshake(viewer, "viewer")
            sharer_a.send_json({"type": "publish", "label": "A"})
            wait_for(viewer, "stream_added")
            sharer_b.send_json({"type": "publish", "label": "B"})
            wait_for(viewer, "stream_added")
            viewer.send_json({"type": "subscribe", "streamId": a_id})
            wait_for(sharer_a, "peer_joined")
            wait_for(sharer_a, "stream_updated")
            viewer.send_json({"type": "subscribe", "streamId": b_id})
            wait_for(sharer_b, "peer_joined")
            wait_for(sharer_b, "stream_updated")
            for _ in range(2):
                wait_for(viewer, "stream_updated")

            sharer_a.send_json({"type": "arrange", "staged": a_id, "fullscreen": True})
            sent = wait_for(sharer_a, "layout")  # a sharer-sender gets its ack
            assert sent["staged"] == a_id
            assert wait_for(viewer, "layout")["fullscreen"] is True

            # sharer_b is neither the sender nor a viewer: its next frame must be
            # the unsubscribe below, not a layout — a fanned-out layout would
            # have arrived first and broken this.
            viewer.send_json({"type": "unsubscribe", "streamId": b_id})
            message = sharer_b.receive_json()
            assert message["type"] == "peer_left", (
                f"a non-viewing sharer received {message!r}: layout may not fan out to sharers"
            )

    def test_the_staged_mirror_going_away_resets_the_stage_for_every_viewer(self, client):
        with (
            client.websocket_connect("/ws") as sharer,
            client.websocket_connect("/ws") as viewer1,
            client.websocket_connect("/ws") as viewer2,
        ):
            sharer_id = handshake(sharer, "sharer")
            handshake(viewer1, "viewer")
            handshake(viewer2, "viewer")
            sharer.send_json({"type": "publish", "label": "Screen"})
            wait_for(viewer1, "stream_added")
            wait_for(viewer2, "stream_added")

            viewer1.send_json({"type": "arrange", "staged": sharer_id, "fullscreen": True})
            wait_for(viewer1, "layout")
            wait_for(viewer2, "layout")

            sharer.close()
            assert wait_for(viewer1, "layout")["staged"] is None
            assert wait_for(viewer2, "layout")["staged"] is None
            assert client.get("/api/kiosk").json()["staged"] is None