interactive-story/backend/tools/m11_webdriver.py

"""A W3C WebDriver client in one file, so browser evidence needs no dependency.

M8 and M9 drove Firefox from a harness that lived outside the repository, which
made their browser evidence unrepeatable by anyone else. This is the same thing
kept inside it, and deliberately dependency-free: WebDriver is an HTTP protocol,
`urllib` speaks HTTP, and adding Selenium to the release candidate to press
buttons would put a package in the audit surface (§23) for no capability.

Only what the release scenarios need is implemented. Anything missing is missing
because nothing used it, not because it was hard.

**One environment note, established by measurement.** Firefox here is a snap, and
its sandbox refuses a file the browser was told to open from `/tmp` — which is
what M9 recorded as "this machine cannot drive a file into the browser". The
narrower and more useful statement is that it refuses `/tmp`: a path under the
user's home works. `stage()` exists to put evidence files there, so knowledge
import can be exercised through the real file input rather than in two halves.
"""

from __future__ import annotations

import json
import os
import shutil
import socket
import subprocess
import time
import urllib.error
import urllib.request
from pathlib import Path

GECKODRIVER = shutil.which("geckodriver") or "/snap/bin/geckodriver"
#: Where files the browser must open are staged. Under $HOME because the snap
#: sandbox denies /tmp; see the module docstring.
STAGE = Path.home() / "m11-evidence"


def stage(name: str, body: str | bytes) -> str:
    STAGE.mkdir(parents=True, exist_ok=True)
    path = STAGE / name
    if isinstance(body, bytes):
        path.write_bytes(body)
    else:
        path.write_text(body)
    return str(path)


def free_port() -> int:
    with socket.socket() as s:
        s.bind(("127.0.0.1", 0))
        return s.getsockname()[1]


class WebDriverError(RuntimeError):
    pass


class Browser:
    """One headless Firefox, driven over the wire protocol."""

    def __init__(self, *, headless: bool = True, log: Path | None = None):
        self.port = free_port()
        handle = open(log, "ab") if log else subprocess.DEVNULL
        self.proc = subprocess.Popen(
            [GECKODRIVER, "--port", str(self.port)],
            stdout=handle, stderr=subprocess.STDOUT,
        )
        self.base = f"http://127.0.0.1:{self.port}"
        self._wait_for_driver()
        args = ["-headless"] if headless else []
        answer = self._call("POST", "/session", {"capabilities": {"alwaysMatch": {
            "browserName": "firefox",
            "moz:firefoxOptions": {"args": args},
            # Never silently accept a bad certificate: the endpoint policy and
            # the TLS trust union are release claims (H12, A06), and a browser
            # that ignored certificates would hide a failure of either.
            "acceptInsecureCerts": False,
        }}})["value"]
        self.session = answer["sessionId"]
        self.version = answer["capabilities"].get("browserVersion", "?")

    # ------------------------------------------------------------- plumbing

    def _wait_for_driver(self) -> None:
        deadline = time.monotonic() + 30
        while time.monotonic() < deadline:
            try:
                urllib.request.urlopen(self.base + "/status", timeout=2)
                return
            except Exception:
                time.sleep(0.2)
        raise WebDriverError("geckodriver never became ready")

    def _call(self, method: str, path: str, payload=None, timeout=120):
        data = json.dumps(payload).encode() if payload is not None else None
        request = urllib.request.Request(
            self.base + path, data=data, method=method,
            headers={"Content-Type": "application/json"},
        )
        try:
            with urllib.request.urlopen(request, timeout=timeout) as response:
                return json.loads(response.read().decode() or "{}")
        except urllib.error.HTTPError as exc:
            body = exc.read().decode()[:400]
            raise WebDriverError(f"{method} {path} -> {exc.code}: {body}") from None

    def _s(self, path: str) -> str:
        return f"/session/{self.session}{path}"

    def quit(self) -> None:
        try:
            self._call("DELETE", self._s(""))
        except Exception:
            pass
        self.proc.terminate()
        try:
            self.proc.wait(timeout=10)
        except subprocess.TimeoutExpired:
            self.proc.kill()

    # ------------------------------------------------------------ commands

    def go(self, url: str) -> None:
        self._call("POST", self._s("/url"), {"url": url})

    @property
    def url(self) -> str:
        return self._call("GET", self._s("/url"))["value"]

    @property
    def title(self) -> str:
        return self._call("GET", self._s("/title"))["value"]

    def source(self) -> str:
        return self._call("GET", self._s("/source"))["value"]

    def js(self, script: str, *args):
        return self._call("POST", self._s("/execute/sync"),
                          {"script": script, "args": list(args)})["value"]

    def find(self, css: str, *, required=True):
        try:
            answer = self._call("POST", self._s("/element"),
                                {"using": "css selector", "value": css})
        except WebDriverError:
            if required:
                raise
            return None
        return list(answer["value"].values())[0]

    def find_all(self, css: str) -> list[str]:
        answer = self._call("POST", self._s("/elements"),
                            {"using": "css selector", "value": css})
        return [list(v.values())[0] for v in answer["value"]]

    def text(self, element: str) -> str:
        return self._call("GET", self._s(f"/element/{element}/text"))["value"]

    def attr(self, element: str, name: str):
        return self._call("GET", self._s(f"/element/{element}/attribute/{name}"))["value"]

    def prop(self, element: str, name: str):
        return self._call("GET", self._s(f"/element/{element}/property/{name}"))["value"]

    def click(self, element: str) -> None:
        self._call("POST", self._s(f"/element/{element}/click"), {})

    def clear(self, element: str) -> None:
        self._call("POST", self._s(f"/element/{element}/clear"), {})

    def type(self, element: str, text: str) -> None:
        self._call("POST", self._s(f"/element/{element}/value"), {"text": text})

    def keys(self, text: str) -> None:
        """Sends keys to whatever has focus — the only way to test tab order."""
        self._call("POST", self._s("/actions"), {"actions": [{
            "type": "key", "id": "keyboard",
            "actions": [a for ch in text for a in (
                {"type": "keyDown", "value": ch}, {"type": "keyUp", "value": ch})],
        }]})

    def active(self):
        answer = self._call("GET", self._s("/element/active"))
        return list(answer["value"].values())[0]

    # ------------------------------------------------------------- waiting

    def wait_for(self, css: str, *, timeout=90, gone=False):
        deadline = time.monotonic() + timeout
        while time.monotonic() < deadline:
            found = self.find(css, required=False)
            if (found is None) if gone else (found is not None):
                return found
            time.sleep(0.25)
        raise WebDriverError(
            f"{'still present' if gone else 'never appeared'}: {css}")

    def wait_until(self, script: str, *, timeout=90, what=""):
        deadline = time.monotonic() + timeout
        while time.monotonic() < deadline:
            if self.js(f"return ({script})"):
                return True
            time.sleep(0.25)
        raise WebDriverError(f"condition never held: {what or script}")


class Site:
    """The application, served the production-shaped way, for the browser."""

    def __init__(self, backend: Path, db_path: Path, log: Path, env=None):
        self.port = free_port()
        handle = open(log, "ab")
        self.proc = subprocess.Popen(
            [str(backend / ".venv/bin/uvicorn"), "app.main:app",
             "--host", "127.0.0.1", "--port", str(self.port)],
            cwd=str(backend), stdout=handle, stderr=subprocess.STDOUT,
            env={**os.environ, "AIDND_DB_PATH": str(db_path),
                 "AIDND_DATABASE_URL": "", "DATABASE_URL": "", **(env or {})},
        )
        self.url = f"http://127.0.0.1:{self.port}"
        deadline = time.monotonic() + 90
        while time.monotonic() < deadline:
            if self.proc.poll() is not None:
                raise WebDriverError(f"server exited early; see {log}")
            try:
                urllib.request.urlopen(self.url + "/api/settings", timeout=2)
                return
            except Exception:
                time.sleep(0.15)
        raise WebDriverError(f"server never became ready; see {log}")

    def api(self, method: str, path: str, payload=None, timeout=600):
        data = json.dumps(payload).encode() if payload is not None else None
        request = urllib.request.Request(
            f"{self.url}/api{path}", data=data, method=method,
            headers={"Content-Type": "application/json"} if data else {})
        with urllib.request.urlopen(request, timeout=timeout) as response:
            body = response.read().decode()
        return json.loads(body) if body else None

    def stop(self) -> None:
        if self.proc.poll() is None:
            self.proc.terminate()
            try:
                self.proc.wait(timeout=20)
            except subprocess.TimeoutExpired:
                self.proc.kill()