"""End-to-end: the sharer's per-leg encoder settings, read back from the page.

The wall is a wall of *text*: the encoder must never buy frame rate with
resolution. `degradationPreference: "maintain-resolution"` is that trade, and
the per-leg `maxBitrate` cap must survive every renegotiation — a re-offer
(a surface switch, a viewer that re-joined) resets the encodings, and the
cap is re-applied after each negotiation and when a leg reaches `connected`.

The assertions read the parameters back from the sharer's real
RTCPeerConnection senders, through the page's `window.__probe` test handle
(AGENTS.md §6) — on the full Chromium channel this is a real read, not a
stub. The media rules still govern: pixel assertions are made on the
*viewer's* `<video>`, never on the sharer's preview.
"""

from __future__ import annotations

import time

import pytest
from playwright.sync_api import expect

from .conftest import frames_matching, start_sharing

pytestmark = [pytest.mark.e2e, pytest.mark.timeout(180)]


@pytest.fixture
def device(base_url, browser, request):
    """Open a page as its own device (isolated context, fresh peer id)."""
    opened = []

    def _open(path: str = "/", errors: list | None = None):
        context = browser.new_context(viewport={"width": 1100, "height": 720})
        opened.append(context)
        request.addfinalizer(context.close)
        page = context.new_page()
        if errors is not None:
            page.on("pageerror", lambda error: errors.append(f"pageerror: {error}"))
        page.goto(f"{base_url}{path}", wait_until="domcontentloaded")
        expect(page.locator("#net")).to_have_attribute("data-state", "online", timeout=10_000)
        return page

    return _open


# Reads the video sender's committed parameters back from the sharer's live
# legs. `state` is the leg's connectionState; `maxBitrate` is the cap the
# leg is actually encoding under (0 or undefined would mean "uncapped").
SENDER_PROBE = """() => {
  const probe = window.__probe;
  if (!probe) return null;
  return [...probe.peers.values()].map((peer) => {
    const sender = peer.pc.getSenders().find(
      (s) => s.track && s.track.kind === "video"
    );
    if (!sender) return null;
    const params = sender.getParameters();
    const encodings = params.encodings || [];
    return {
      state: peer.pc.connectionState,
      degradationPreference: params.degradationPreference,
      maxBitrate: encodings.length ? encodings[0].maxBitrate : null,
    };
  });
}"""


def _wait_for(sharer, want: int, pred, timeout: float = 45.0) -> list:
    """Poll the sharer's legs until `want` of them satisfy `pred` (or give up)."""
    legs: list = []
    deadline = time.time() + timeout
    while time.time() < deadline:
        legs = sharer.evaluate(SENDER_PROBE) or []
        if len(legs) == want and all(leg and pred(leg) for leg in legs):
            return legs
        time.sleep(0.5)
    return legs


class TestSenderFidelity:
    def test_two_live_legs_report_maintain_resolution_and_the_cap(self, device, base_url):
        import httpx

        cap = httpx.get(f"{base_url}/api/config", timeout=5).json()["maxBitrateBps"]
        assert cap > 0

        errors: list = []
        sharer = device("/mirror", errors)
        start_sharing(sharer)

        viewers = [device("/watch", errors) for _ in range(2)]
        for viewer in viewers:
            expect(viewer.locator(".tile[data-state='live']")).to_have_count(1, timeout=25_000)

        legs = _wait_for(sharer, 2, lambda leg: leg["state"] == "connected")
        assert len(legs) == 2, f"the sharer has {len(legs)} legs, wanted 2: {legs}"
        for leg in legs:
            assert leg["state"] == "connected", legs
            assert leg["degradationPreference"] == "maintain-resolution", leg
            assert leg["maxBitrate"] == cap, leg

        # The parameters are not a facade: the pixels actually arrive, moving.
        for viewer in viewers:
            frames = frames_matching(viewer, ".tile video")
            assert frames["distinct"] >= 2, f"a viewer froze: {frames}"

        expect(sharer.locator("#viewer-count")).to_have_text("2", timeout=10_000)
        assert errors == [], f"uncaught browser errors: {errors}"

    def test_the_quality_survives_a_renegotiation(self, device, base_url):
        """The bug the post-negotiation re-apply exists for.

        A live leg is renegotiated in place — a second offer on the same
        RTCPeerConnection, exactly what a surface switch or a viewer that
        re-joined does. The re-offer resets the encodings; the parameters
        must be back by the time the renegotiation completes, and the wall
        must still be moving on the viewer.
        """
        import httpx

        cap = httpx.get(f"{base_url}/api/config", timeout=5).json()["maxBitrateBps"]

        errors: list = []
        sharer = device("/mirror", errors)
        start_sharing(sharer)

        viewer = device("/watch", errors)
        expect(viewer.locator(".tile[data-state='live']")).to_have_count(1, timeout=25_000)
        legs = _wait_for(sharer, 1, lambda leg: leg["state"] == "connected")
        assert legs and legs[0]["state"] == "connected", legs
        assert legs[0]["degradationPreference"] == "maintain-resolution", legs
        assert legs[0]["maxBitrate"] == cap, legs

        immediate = sharer.evaluate(
            """async () => {
              const [id, peer] = [...window.__probe.peers.entries()][0];
              await window.__probe.renegotiate(id);
              const sender = peer.pc.getSenders().find(
                (s) => s.track && s.track.kind === "video"
              );
              const params = sender.getParameters();
              const encodings = params.encodings || [];
              return {
                state: peer.pc.connectionState,
                degradationPreference: params.degradationPreference,
                maxBitrate: encodings.length ? encodings[0].maxBitrate : null,
              };
            }"""
        )
        # renegotiate() awaits setLocalDescription *and* the re-apply before
        # it resolves, so the parameters must already be back — whatever the
        # leg's connection state is at that instant.
        assert immediate["degradationPreference"] == "maintain-resolution", immediate
        assert immediate["maxBitrate"] == cap, immediate

        # ...and once the leg reports `connected` again, the connected-state
        # re-apply confirms them and the picture is still moving.
        legs = _wait_for(sharer, 1, lambda leg: leg["state"] == "connected")
        assert legs and legs[0]["state"] == "connected", legs
        assert legs[0]["degradationPreference"] == "maintain-resolution", legs
        assert legs[0]["maxBitrate"] == cap, legs

        frames = frames_matching(viewer, ".tile video")
        assert frames["distinct"] >= 2, f"the leg froze across the renegotiation: {frames}"
        assert errors == [], f"uncaught browser errors: {errors}"