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.

**Downloads (v1.1 WP-C).** The same snap Firefox saves a download without any
dialog when its profile says where to, and the folder is under `$HOME`.
`firefox_download_prefs` is that profile, `require_under_home` refuses a folder
the sandbox would not let it write, and `wait_for_download` decides when a file
has actually finished arriving — never the click that started it.
"""

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"
#: The W3C key an element reference is returned under.
ELEMENT_KEY = "element-6066-11e4-a52e-4f735466cecf"
#: What Firefox names a download while it is still arriving.
PARTIAL_SUFFIXES = (".part",)


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]


def geckodriver_version() -> str:
    try:
        out = subprocess.run([GECKODRIVER, "--version"], capture_output=True, text=True, timeout=30)
        return (out.stdout.splitlines() or ["?"])[0].strip()
    except (OSError, subprocess.SubprocessError):
        return "?"


class WebDriverError(RuntimeError):
    pass


# ------------------------------------------------------------------ downloads

def require_under_home(path: Path) -> Path:
    """`path`, resolved, if it is inside the user's home; otherwise refuse.

    The snap sandbox will not write elsewhere, and a download folder under
    `/tmp` would also put evidence where a reboot deletes it.
    """
    resolved = Path(path).expanduser().resolve()
    home = Path.home().resolve()
    if resolved != home and home not in resolved.parents:
        raise WebDriverError(f"{resolved} is not under {home}; the browser cannot write there")
    return resolved


def firefox_download_prefs(directory: Path) -> dict:
    """Profile preferences that save every download to `directory`, unasked."""
    return {
        "browser.download.folderList": 2,  # 2 = the folder named below
        "browser.download.dir": str(directory),
        "browser.download.useDownloadDir": True,
        "browser.download.start_downloads_in_tmp_dir": False,
        "browser.download.always_ask_before_handling_new_types": False,
        "browser.helperApps.neverAsk.saveToDisk": "application/json,application/octet-stream",
        "browser.download.manager.showWhenStarting": False,
        "browser.download.alwaysOpenPanel": False,
        "browser.download.panel.shown": True,
    }


def wait_for_download(directory: Path, before: set[str], *, timeout: float = 60,
                      poll: float = 0.2, stable_polls: int = 3) -> Path:
    """The file a download wrote into `directory`, once it has finished.

    Finished means all of these, at once:
    - a name that was not in `before` (the listing taken before the click);
    - no in-progress file (`*.part`) left in the folder;
    - more than zero bytes;
    - the same size for `stable_polls` consecutive polls.

    A first appearance is not a finished download, and a zero-byte or partial
    file never counts. Raises `WebDriverError` when nothing finishes in time.
    """
    deadline = time.monotonic() + timeout
    last: dict[str, int] = {}
    steady: dict[str, int] = {}
    while time.monotonic() < deadline:
        names = {p.name for p in directory.iterdir()} if directory.exists() else set()
        partial = any(n.endswith(PARTIAL_SUFFIXES) for n in names)
        fresh = sorted(n for n in names - before if not n.endswith(PARTIAL_SUFFIXES))
        for name in fresh:
            size = (directory / name).stat().st_size
            steady[name] = steady.get(name, 0) + 1 if last.get(name) == size else 1
            last[name] = size
            if not partial and size > 0 and steady[name] >= stable_polls:
                return directory / name
        time.sleep(poll)
    listing = sorted(p.name for p in directory.iterdir()) if directory.exists() else []
    raise WebDriverError(f"no finished download in {directory} within {timeout}s; saw {listing}")


# -------------------------------------------------------------------- browser

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

    def __init__(self, *, headless: bool = True, log: Path | None = None,
                 download_dir: 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 []
        options: dict = {"args": args}
        self.download_dir = None
        if download_dir is not None:
            self.download_dir = require_under_home(download_dir)
            self.download_dir.mkdir(parents=True, exist_ok=True)
            options["prefs"] = firefox_download_prefs(self.download_dir)
        answer = self._call("POST", "/session", {"capabilities": {"alwaysMatch": {
            "browserName": "firefox",
            "moz:firefoxOptions": options,
            # 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", "?")
        self.capabilities = answer["capabilities"]

    # ------------------------------------------------------------- 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})

    def reload(self) -> None:
        self._call("POST", self._s("/refresh"), {})

    @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 element_by_js(self, script: str, *args):
        """An element a script returns, as a reference `click` can use, or None."""
        value = self.js(script, *args)
        if isinstance(value, dict) and ELEMENT_KEY in value:
            return value[ELEMENT_KEY]
        return None

    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:
        """A real click, on an element first scrolled to the middle of the view.

        WebDriver scrolls a target only as far as its edge, and the play page's
        composer is fixed to the bottom of the window: a control just under it
        (a failure notice's details, a turn's Inspect button) is then covered,
        and the click is intercepted. A reader scrolls it clear first; so does
        this (v1.1 WP-C).
        """
        self._call("POST", self._s("/execute/sync"), {
            "script": "arguments[0].scrollIntoView({block: 'center', inline: 'nearest'})",
            "args": [{ELEMENT_KEY: element}]})
        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]

    # -------------------------------------------------------------- windows

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

    def new_tab(self) -> str:
        return self._call("POST", self._s("/window/new"), {"type": "tab"})["value"]["handle"]

    def switch_to(self, handle: str) -> None:
        self._call("POST", self._s("/window"), {"handle": handle})

    def close_window(self) -> None:
        self._call("DELETE", self._s("/window"))

    # ------------------------------------------------------------- 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=""):
        if self.wait_js(script, timeout=timeout):
            return True
        raise WebDriverError(f"condition never held: {what or script}")

    def wait_js(self, script: str, *, timeout=90) -> bool:
        """Whether `script` became true within `timeout`. For a check to record,
        where `wait_until` is for a precondition that must hold."""
        deadline = time.monotonic() + timeout
        while time.monotonic() < deadline:
            if self.js(f"return ({script})"):
                return True
            time.sleep(0.25)
        return False


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()