undercooked/tests/e2e/conftest.py

"""Playwright fixtures for the browser suites in `tests/e2e/`.

The project depends on the `playwright` package, not on `pytest-playwright`, so
the three fixtures the plugin would provide are defined here — a headless chromium,
a fresh context and a fresh page, all function-scoped (see `browser` for why the scope
is load-bearing). A fresh context per test matters too: `sessionStorage` (the anonymous
`uc.token`, `uc.lastCode`) is scoped to it, so tests cannot observe each other's
identity.

The phase-07 story suites (task 02) need servers a browser can drive end to end:
`e2e_server` is a session-scoped factory that spins up a real `uvicorn.Server`
on an ephemeral `127.0.0.1` port (never a routable interface) with a `Settings`
resolved the way `tests/integration/conftest.py` resolves one (the authoritative
`config.ENV_KEYS` contract, ambient `UNDERCOOKED_*` / `DEBUGPY*` stripped, the
`levels_dir` override pointing at the campaign the factory names). The three
session fixtures built on it:

* `round_server` — the `tests/fixtures/levels_round/` campaign (the shipped level 1,
  layout-verbatim, `target_score` 20 / `time_limit` 60 s — one served soup scores
  44-66, so the first serve wins the round; the forced-short-round the task-02
  result-screen test runs. The task's `time_limit: 4` idea is impossible: the
  validator's `TIME_LIMIT_INVALID` floor is `MIN_TIME_LIMIT_S` (30 s) and
  `create_app`'s content gate is unconditional — 60 s with a first-serve win is
  the shortest round a valid campaign can offer);
* `small_server` — `max_players=2`, for the full-room story;
* `full_server` — `max_rooms=1`, for the process-capacity "server full" story.
"""

from __future__ import annotations

import importlib
import os
import threading
import time
from collections.abc import Callable, Iterator
from pathlib import Path
from typing import TYPE_CHECKING, Any

import httpx
import pytest
from playwright.sync_api import sync_playwright

if TYPE_CHECKING:
    from playwright.sync_api import Browser, BrowserContext, Page, ViewportSize

# The default desktop viewport; story tests that care about small screens call
# `page.set_viewport_size` before navigating.
VIEWPORT: ViewportSize = {"width": 1280, "height": 720}

REPO_ROOT = Path(__file__).resolve().parents[2]
#: The playable campaign (phase 06 task 06's test-only copy of the shipped level 1).
ROUND_LEVELS_DIR = REPO_ROOT / "tests" / "fixtures" / "levels_round"

config = importlib.import_module("undercooked.config")
app_module = importlib.import_module("undercooked.server.app")
uvicorn = importlib.import_module("uvicorn")


@pytest.fixture
def browser() -> Iterator[Browser]:
    """A headless chromium, live for exactly one test.

    Function scope is load-bearing, not shyness about startup time. The
    *synchronous* Playwright API drives the browser from an asyncio event loop it
    owns in the main thread, and that loop stays flagged as running for as long as
    `sync_playwright()` is open — so a session- or module-scoped fixture makes every
    later `async def` test in the run die with
    `RuntimeError: Runner.run() cannot be called from a running event loop`
    (pytest-asyncio, `asyncio_mode = "auto"`). Closing the whole stack inside
    the test leaves the main thread clean for whatever runs next.
    """
    with sync_playwright() as playwright:
        launched = playwright.chromium.launch(headless=True)
        try:
            yield launched
        finally:
            launched.close()


@pytest.fixture
def context(browser: Browser) -> Iterator[BrowserContext]:
    """A brand-new browser context per test: isolated storage and cookies."""
    created = browser.new_context(viewport=VIEWPORT)
    try:
        yield created
    finally:
        created.close()


@pytest.fixture
def page(context: BrowserContext) -> Iterator[Page]:
    """A tab inside `context`, closed after the test."""
    opened = context.new_page()
    try:
        yield opened
    finally:
        opened.close()


# ---------------------------------------------------------------- live servers


def _as_env_value(value: object) -> str:
    """Render a Python override value the way `load_settings` would read it."""
    if isinstance(value, bool):
        return "1" if value else "0"
    if isinstance(value, str | os.PathLike):
        return str(value)
    if isinstance(value, list | tuple):
        return ",".join(str(item) for item in value)
    return str(value)


def _wait_for_health(url: str, thread: threading.Thread) -> None:
    """Poll `/healthz` until it answers 200, failing fast if the server thread died."""
    timeout = float(os.environ.get("UNDERCOOKED_TEST_SOCKET_TIMEOUT", "30"))
    deadline = time.monotonic() + timeout
    probe = httpx.Client(timeout=1.0, trust_env=False)
    try:
        while time.monotonic() < deadline:
            try:
                if probe.get(f"{url}/healthz").status_code == 200:
                    return
            except httpx.HTTPError:
                pass
            thread.join(timeout=0.1)
            if not thread.is_alive():
                raise RuntimeError(f"live server on {url} died during startup")
    finally:
        probe.close()
    raise RuntimeError(f"live server on {url} never became healthy")


class E2EServer:
    """One live uvicorn server + its app (a context manager)."""

    def __init__(self, url: str, app: Any, server: Any, thread: threading.Thread) -> None:
        self.url = url
        self.app = app
        self._server = server
        self._thread = thread
        self._closed = False

    def __enter__(self) -> tuple[str, Any]:
        return self.url, self.app

    def __exit__(self, *exc_info: object) -> None:
        self.close()

    def close(self) -> None:
        if self._closed:
            return
        self._closed = True
        self._server.should_exit = True
        self._thread.join(timeout=15)
        if self._thread.is_alive():  # pragma: no cover - only if uvicorn stops cooperating
            pytest.fail("live server did not shut down within 15 s")


def _start_e2e_server(overrides: dict[str, object], levels_dir: Path) -> E2EServer:
    """One live uvicorn server on an ephemeral loopback port.

    The integration conftest's machine, verbatim in purpose: `Settings` from
    `**overrides` (field names, validated against `config.ENV_KEYS`) over the
    named `levels_dir` campaign, on an ephemeral `127.0.0.1` port (bound
    before the serving thread starts, so the port is known exactly — no log
    scraping, no collisions between concurrent runs).
    """
    env = {
        key: value
        for key, value in os.environ.items()
        if not key.startswith(("UNDERCOOKED_", "DEBUGPY"))
    }
    env["UNDERCOOKED_LEVELS_DIR"] = str(levels_dir)
    for name, value in overrides.items():
        env[config.ENV_KEYS[name]] = _as_env_value(value)
    settings = config.load_settings(env)
    application = app_module.create_app(settings)
    uv_config = uvicorn.Config(
        application,
        host="127.0.0.1",
        port=0,
        log_level="warning",
        lifespan="on",
    )
    sock = uv_config.bind_socket()
    url = f"http://127.0.0.1:{sock.getsockname()[1]}"
    server = uvicorn.Server(uv_config)
    thread = threading.Thread(
        target=server.run, args=([sock],), name="undercooked-e2e", daemon=True
    )
    thread.start()
    _wait_for_health(url, thread)
    return E2EServer(url, application, server, thread)


@pytest.fixture(scope="session")
def e2e_server() -> Iterator[Callable[..., E2EServer]]:
    """Factory: `e2e_server(ROUND_LEVELS_DIR, max_players=2)` — one call, one
    server. Each call starts a fresh `uvicorn.Server` on an ephemeral loopback
    port with a `Settings` resolved from the overrides (field names, validated
    against `config.ENV_KEYS`) over the named campaign. Shutdown is requested
    through `should_exit` so the lifespan stops cleanly; a session-end safety
    close covers a leaked server."""
    started: list[E2EServer] = []

    def _make(levels_dir: Path, **overrides: object) -> E2EServer:
        if any(name not in config.ENV_KEYS for name in overrides):
            unknown = sorted(name for name in overrides if name not in config.ENV_KEYS)
            pytest.fail(f"unknown settings override(s): {unknown}")
        live = _start_e2e_server(dict(overrides), levels_dir)
        started.append(live)
        return live

    def _close_all() -> None:
        for live in started:
            live.close()

    _make.close_all = _close_all  # type: ignore[attr-defined]
    yield _make
    _close_all()  # type: ignore[attr-defined]


@pytest.fixture(scope="session")
def round_server(e2e_server: Callable[..., E2EServer]) -> Iterator[E2EServer]:
    """The playable campaign (`tests/fixtures/levels_round/`): the forced-short
    round for the result-screen test (first serve wins — the module docstring)."""
    live = e2e_server(ROUND_LEVELS_DIR)
    try:
        yield live
    finally:
        live.close()


@pytest.fixture(scope="session")
def small_server(e2e_server: Callable[..., E2EServer]) -> Iterator[E2EServer]:
    """`max_players=2` — the full-room story (US-02)."""
    live = e2e_server(ROUND_LEVELS_DIR, max_players=2)
    try:
        yield live
    finally:
        live.close()


@pytest.fixture(scope="session")
def full_server(e2e_server: Callable[..., E2EServer]) -> Iterator[E2EServer]:
    """`max_rooms=1` — the process-capacity "server full" story (US-01)."""
    live = e2e_server(ROUND_LEVELS_DIR, max_rooms=1)
    try:
        yield live
    finally:
        live.close()


# ---------------------------------------------------------------- page helpers


def open_two_pages(
    browser: Browser, url: str
) -> tuple[BrowserContext, Page, BrowserContext, Page]:
    """Two tabs in two fresh contexts (the multiplayer harness, US-01/US-02).

    Returns `(context_a, page_a, context_b, page_b)`; the caller closes the
    contexts in a `finally` (closing a context is the socket drop a story test
    wants when it "closes a laptop")."""
    context_a = browser.new_context(viewport=VIEWPORT)
    context_b = browser.new_context(viewport=VIEWPORT)
    page_a = context_a.new_page()
    page_b = context_b.new_page()
    return context_a, page_a, context_b, page_b


def load_app(page: Page, url: str) -> None:
    """Load the shell and wait for its module graph to settle."""
    page.goto(f"{url}/", wait_until="networkidle")


def app_screen(page: Page) -> str | None:
    """The screen the state machine is on (the `window.__undercooked` seam)."""
    return page.evaluate("window.__undercooked ? window.__undercooked.screen : null")