mirror/tests/e2e/conftest.py

"""Fixtures for the browser end-to-end suite.

The capture flags matter: with a fake UI and a fake device, Chromium answers
`getDisplayMedia()` with a real, *moving* 1280x720 monitor stream. That means
these tests exercise the genuine capture -> encode -> SRTP -> decode -> paint
path rather than a stubbed-out media source.
"""

from __future__ import annotations

import os
import socket
import subprocess
import sys
import time
from collections.abc import Iterator, Sequence
from pathlib import Path

import pytest
from playwright.sync_api import expect, sync_playwright

pytestmark = pytest.mark.e2e

REPO_ROOT = Path(__file__).resolve().parents[2]

CHROMIUM_ARGS = [
    "--use-fake-ui-for-media-stream",
    "--use-fake-device-for-media-stream",
    "--use-fake-device-for-audio-capture",
    # Deliberate — do not "fix" this away: with this flag the suite cannot
    # prove the *browser's* autoplay rule. The audio tests therefore assert
    # mirror's own `muted` state (the application's policy), and never claim
    # to exercise a real browser's autoplay behaviour. See test_audio_share.py.
    "--autoplay-policy=no-user-gesture-required",
    # Two Chromium instances must each get their own UDP ports; never proxy.
    "--no-sandbox",
    "--disable-dev-shm-usage",
    "--no-proxy-server",
]


def _free_port() -> int:
    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as probe:
        probe.bind(("127.0.0.1", 0))
        return probe.getsockname()[1]


def _start(extra_args: Sequence[str] = ()) -> tuple[str, subprocess.Popen[str]]:
    """Spawn a real `mirror serve` on a free port and wait for it to answer."""
    import httpx

    port = _free_port()
    process = subprocess.Popen(
        [
            sys.executable,
            "-m",
            "mirror",
            "serve",
            "--host",
            "127.0.0.1",
            "--port",
            str(port),
            "--log-level",
            "warning",
            *extra_args,
        ],
        cwd=REPO_ROOT,
        stdout=subprocess.PIPE,
        stderr=subprocess.STDOUT,
        text=True,
        env={
            **os.environ,
            "MIRROR_ICE_SERVERS": "",
            "PYTHONUNBUFFERED": "1",
            # A stray DEBUGPY in the developer's shell would stall the server.
            "DEBUGPY": "0",
        },
    )
    url = f"http://127.0.0.1:{port}"
    deadline = time.time() + 25
    last_error: Exception | None = None
    while time.time() < deadline:
        if process.poll() is not None:
            output = process.stdout.read() if process.stdout else ""
            raise RuntimeError(f"mirror serve exited early ({process.returncode}):\n{output}")
        try:
            # /healthz is the readiness probe *and* the one route an address gate
            # never refuses, so it works for a restricted server too.
            if httpx.get(f"{url}/healthz", timeout=1.0).status_code == 200:
                return url, process
        except Exception as exc:  # still booting
            last_error = exc
        time.sleep(0.15)
    process.terminate()
    raise RuntimeError(f"mirror serve never came up: {last_error}")


def _stop(process: subprocess.Popen[str]) -> None:
    process.terminate()
    try:
        process.wait(timeout=5)
    except subprocess.TimeoutExpired:  # pragma: no cover
        process.kill()


@pytest.fixture(scope="session")
def base_url() -> Iterator[str]:
    """A real `mirror serve` process on a real port — not a test double."""
    url, process = _start()
    yield url
    _stop(process)


@pytest.fixture(scope="session")
def restricted_url() -> Iterator[str]:
    """A server that admits only 203.0.113.0/24 — so it refuses this machine.

    That range is reserved for documentation by RFC 5737, which is why it is safe
    to use as "everybody else": nothing real is ever locked out by it.
    """
    url, process = _start(["--allow-cidr", "203.0.113.0/24"])
    yield url
    _stop(process)


@pytest.fixture(scope="session")
def browser():
    with sync_playwright() as playwright:
        # channel="chromium" = the full Chromium binary in new headless mode.
        # The trimmed "headless shell" has no media stack at all and hides
        # navigator.mediaDevices entirely.
        instance = playwright.chromium.launch(channel="chromium", headless=True, args=CHROMIUM_ARGS)
        yield instance
        instance.close()


@pytest.fixture
def context(browser):
    """One browser context == one device. Closed after every test."""
    created = browser.new_context(viewport={"width": 1100, "height": 720})
    yield created
    created.close()


@pytest.fixture
def page(context):
    return context.new_page()


def start_sharing(page) -> None:
    page.locator("#start").click()
    expect(page.locator("body")).to_have_attribute("data-live", "on", timeout=15_000)


def stop_sharing(page) -> None:
    page.locator("#stop").click()
    expect(page.locator("body")).to_have_attribute("data-live", "off", timeout=10_000)


# Returns {width, height, currentTime, distinct} for the first <video> matching.
FRAME_PROBE = """
async (selector) => {
  const video = document.querySelector(selector);
  if (!video) return {missing: true};
  const canvas = document.createElement('canvas');
  canvas.width = 48;
  canvas.height = 32;
  const ctx = canvas.getContext('2d', { willReadFrequently: true });
  const seen = new Set();
  for (let i = 0; i < 8; i++) {
    await new Promise((resolve) => setTimeout(resolve, 120));
    try { ctx.drawImage(video, 0, 0, 48, 32); } catch { /* not ready yet */ }
    const data = ctx.getImageData(0, 0, 48, 32).data;
    let hash = 0;
    for (let j = 0; j < data.length; j += 4) {
      hash = ((hash * 31) + data[j] + data[j + 1] * 3 + data[j + 2] * 7) >>> 0;
    }
    seen.add(hash);
  }
  return {
    width: video.videoWidth,
    height: video.videoHeight,
    currentTime: video.currentTime,
    paused: video.paused,
    distinct: seen.size,
  };
}
"""


def frames_matching(page, selector: str, attempts: int = 10) -> dict:
    """Poll until real, changing pixels show up (or give up and return last)."""
    result: dict = {}
    for _ in range(attempts):
        result = page.evaluate(FRAME_PROBE, selector)
        if result.get("distinct", 0) >= 2 and result.get("width", 0) > 0:
            return result
        time.sleep(0.4)
    return result