"""The Linux-native capture path: a real X11 screen, no fake media devices.

The rest of the suite drives Chromium's fake capture device, which is
deterministic but synthetic. This module is the counterweight: it starts its own
Xvfb server, runs a *headed* Chromium on it, and asks that browser to mirror the
whole monitor through Chromium's real X11 capture backend. A second Chromium
process — a different machine as far as WebRTC cares — watches it.

Skipped unless ``MIRROR_REALDISPLAY=1``, because it needs an X server binary and
a headed browser::

    MIRROR_REALDISPLAY=1 uv run pytest -m realdisplay
"""

from __future__ import annotations

import os
import shutil
import socket
import subprocess
import time

import httpx
import pytest
from playwright.sync_api import expect, sync_playwright

pytestmark = [
    pytest.mark.e2e,
    pytest.mark.realdisplay,
    pytest.mark.timeout(240),
    pytest.mark.skipif(shutil.which("Xvfb") is None, reason="Xvfb is not installed"),
    pytest.mark.skipif(
        os.environ.get("MIRROR_REALDISPLAY", "") not in {"1", "true", "yes"},
        reason="set MIRROR_REALDISPLAY=1 to exercise the real X11 capture backend",
    ),
]

# Deliberately not 16:9: Chromium's fake device reports 1280x720 no matter what,
# so an aspect ratio of exactly 1280/800 is a fingerprint of the real screen.
XVFB_GEOMETRY = "1280x800x24"
XVFB_ASPECT = 1280 / 800
FAKE_DEVICE_SIZE = (1280, 720)

# Flags learned the hard way, and worth keeping written down:
#
# * No `--use-fake-device-for-media-stream`: that is exactly what we are trying
#   not to use here.
# * No `--use-fake-ui-for-media-stream` either. It short-circuits the picker
#   *before* the X11 backend selects a source, and every capture then dies with
#   `NotReadableError: Could not start video source` in ~55ms. Choosing the
#   source with `--auto-select-desktop-capture-source` grants and selects in one
#   step, which is what makes unattended desktop capture work.
# * `--ozone-platform=x11` keeps Chromium on its native X11 capturer instead of
#   xdg-desktop-portal/PipeWire. Combined with `browser_env()` this makes a
#   permission dialog on the developer's own session impossible.
REAL_CAPTURE_ARGS = [
    "--auto-select-desktop-capture-source=Entire screen",
    "--ozone-platform=x11",
    "--autoplay-policy=no-user-gesture-required",
    "--no-sandbox",
    "--disable-dev-shm-usage",
    "--no-proxy-server",
]

# Never reachable: without a session bus and without a Wayland socket there is no
# portal to ask, so these tests cannot interrupt a human.
UNREACHABLE_SESSION_VARS = (
    "WAYLAND_DISPLAY",
    "DBUS_SESSION_BUS_ADDRESS",
    "XDG_CURRENT_DESKTOP",
    "XDG_BACKEND",
)


def browser_env(display: str | None = None) -> dict[str, str]:
    """An environment that can only reach the display this test owns."""
    env = {key: value for key, value in os.environ.items() if key not in UNREACHABLE_SESSION_VARS}
    env["XDG_SESSION_TYPE"] = "x11"
    env["MIRROR_ICE_SERVERS"] = ""
    if display is None:
        env.pop("DISPLAY", None)
    else:
        env["DISPLAY"] = display
    return env


# Starts a pixel-sampling loop without blocking, so the test can drive motion on
# the captured screen while the watcher is measuring it.
PROBE = """() => {
  const video = document.querySelector('.tile video');
  const canvas = document.createElement('canvas');
  canvas.width = 96;
  canvas.height = 60;
  const ctx = canvas.getContext('2d', {willReadFrequently: true});
  window.__probe = (async () => {
    const seen = new Set();
    for (let i = 0; i < 14; i++) {
      await new Promise((r) => setTimeout(r, 120));
      try { ctx.drawImage(video, 0, 0, 96, 60); } catch {}
      const d = ctx.getImageData(0, 0, 96, 60).data;
      let h = 0;
      for (let j = 0; j < d.length; j += 4) {
        h = ((h * 31) + d[j] + d[j + 1] * 3 + d[j + 2] * 7) >>> 0;
      }
      seen.add(h);
    }
    return {distinct: seen.size, width: video.videoWidth, height: video.videoHeight};
  })();
  return 'started';
}"""


def _free_display() -> str:
    for number in range(90, 120):
        path = f"/tmp/.X11-unix/X{number}"
        if os.path.exists(path):
            continue
        with socket.socket(socket.AF_UNIX) as probe:
            try:
                probe.connect(path)
            except OSError:
                return f":{number}"
            else:
                continue
    raise RuntimeError("no free X display in :90..:119")


def _socket_path(display: str) -> str:
    """`:90` -> `/tmp/.X11-unix/X90`, the file whose existence means "ready"."""
    return f"/tmp/.X11-unix/X{display.lstrip(':')}"


@pytest.fixture(scope="module")
def x_display():
    display = _free_display()
    ready = _socket_path(display)
    server = subprocess.Popen(
        ["Xvfb", display, "-screen", "0", XVFB_GEOMETRY, "-nolisten", "tcp"],
        stdout=subprocess.DEVNULL,
        stderr=subprocess.DEVNULL,
    )
    deadline = time.time() + 20
    while time.time() < deadline:
        if os.path.exists(ready):
            break
        if server.poll() is not None:  # pragma: no cover
            raise RuntimeError(f"Xvfb exited immediately with code {server.returncode}")
        time.sleep(0.15)
    else:  # pragma: no cover
        server.terminate()
        raise RuntimeError(f"Xvfb never created {ready}")
    yield display
    server.terminate()
    try:
        server.wait(timeout=5)
    except subprocess.TimeoutExpired:  # pragma: no cover
        server.kill()


@pytest.fixture(scope="module")
def browsers(x_display):
    """Two Chromium *processes* from one Playwright driver.

    The sharer is headed and lives on the Xvfb display, so what it captures is a
    genuine desktop. The watcher is headless: it never touches the display, which
    is exactly why it can be a separate process.
    """
    previous = os.environ.get("DISPLAY")
    os.environ["DISPLAY"] = x_display
    try:
        with sync_playwright() as playwright:
            sharer = playwright.chromium.launch(
                channel="chromium",
                headless=False,
                args=REAL_CAPTURE_ARGS,
                env=browser_env(x_display),
            )
            watcher = playwright.chromium.launch(
                channel="chromium",
                headless=True,
                args=[
                    "--no-sandbox",
                    "--disable-dev-shm-usage",
                    "--no-proxy-server",
                    "--ozone-platform=x11",
                ],
                env=browser_env(),
            )
            try:
                yield {"sharer": sharer, "watcher": watcher}
            finally:
                sharer.close()
                watcher.close()
    finally:
        if previous is not None:
            os.environ["DISPLAY"] = previous
        else:  # pragma: no cover
            os.environ.pop("DISPLAY", None)


@pytest.fixture
def share_page(browsers, base_url):
    page = browsers["sharer"].new_context(viewport={"width": 1200, "height": 760}).new_page()
    page.goto(f"{base_url}/mirror", wait_until="domcontentloaded")
    expect(page.locator("#net")).to_have_attribute("data-state", "online", timeout=20_000)
    yield page
    page.close()


@pytest.fixture
def watch_page(browsers, base_url):
    page = browsers["watcher"].new_context().new_page()
    page.goto(f"{base_url}/watch", wait_until="domcontentloaded")
    expect(page.locator("#net")).to_have_attribute("data-state", "online", timeout=20_000)
    yield page
    page.close()


class TestRealX11Capture:
    def test_the_real_desktop_is_captured_not_a_synthetic_one(self, base_url, share_page):
        errors: list[str] = []
        share_page.on("pageerror", lambda error: errors.append(str(error)))

        share_page.locator("#start").click()
        expect(share_page.locator("body")).to_have_attribute("data-live", "on", timeout=40_000)

        settings = share_page.evaluate(
            """() => {
              const track = document.querySelector('#preview').srcObject.getVideoTracks()[0];
              return {label: track.label, ...track.getSettings()};
            }"""
        )
        assert settings["displaySurface"] == "monitor", f"not a monitor capture: {settings}"
        assert settings["label"], "a real capture always names its source"
        assert not settings["label"].startswith("screen:-"), (
            f"{settings['label']} is Chromium's synthetic capture device"
        )
        size = (settings["width"], settings["height"])
        assert size != FAKE_DEVICE_SIZE, (
            f"got exactly {size}: that is Chromium's fake capture device, not the X11 screen"
        )
        assert settings["width"] >= 640, f"implausibly small capture: {settings}"
        # Chromium scales a monitor capture to the capturing window, so compare
        # the aspect ratio: 1280x800 in, 1200x750 out is the real screen.
        assert abs(settings["width"] / settings["height"] - XVFB_ASPECT) < 0.06, (
            f"{size} does not match the {XVFB_GEOMETRY} desktop aspect "
            f"({XVFB_ASPECT:.3f}); this is not our Xvfb screen"
        )

        # The roster must carry the real geometry, not a made-up one.
        time.sleep(0.6)
        mirror = httpx.get(f"{base_url}/api/streams", timeout=5).json()["mirrors"][0]
        assert (mirror["width"], mirror["height"]) == size
        assert mirror["displaySurface"] == "monitor"
        assert errors == [], f"uncaught browser errors: {errors}"

    def test_a_real_desktop_reaches_a_second_machine_and_moves(self, base_url, browsers):
        share_page = (
            browsers["sharer"].new_context(viewport={"width": 1200, "height": 760}).new_page()
        )
        share_page.goto(f"{base_url}/mirror", wait_until="domcontentloaded")
        expect(share_page.locator("#net")).to_have_attribute("data-state", "online", timeout=20_000)
        share_page.locator("#start").click()
        expect(share_page.locator("body")).to_have_attribute("data-live", "on", timeout=40_000)

        watch_page = browsers["watcher"].new_context().new_page()
        watch_page.goto(f"{base_url}/watch", wait_until="domcontentloaded")
        expect(watch_page.locator("#net")).to_have_attribute("data-state", "online", timeout=20_000)
        expect(watch_page.locator(".tile[data-state='live']")).to_have_count(1, timeout=60_000)

        # Start measuring on the watcher, then paint: cycling the captured
        # machine's background is what a real presenter's screen does constantly.
        assert watch_page.evaluate(PROBE) == "started"
        for colour in ("#c81e1e", "#0b0d10", "#1e6ec8", "#0b0d10", "#1ec85a"):
            share_page.evaluate("(colour) => { document.body.style.background = colour; }", colour)
            time.sleep(0.28)
        result = watch_page.evaluate("() => window.__probe")

        assert result["width"] >= 640 and result["height"] >= 400, (
            f"the watcher decoded no picture: {result}"
        )
        assert abs(result["width"] / result["height"] - XVFB_ASPECT) < 0.06, result
        assert result["distinct"] >= 2, (
            f"the watcher received a frozen frame: {result} — real desktop motion "
            f"never made it across the peer connection"
        )

        # And the sharer agrees that someone is watching its desktop.
        expect(share_page.locator("#viewer-count")).to_have_text("1", timeout=20_000)

        # Stopping the share tears the mirror away for the watcher as well.
        share_page.locator("#stop").click()
        expect(watch_page.locator(".tile")).to_have_count(0, timeout=20_000)

        watch_page.close()
        share_page.close()