JesseMarkowitz
87a40326a2
WP-D and WP-E complete the planned v1.1 implementation packages. WP-D — recovery honesty: - backups verify the completed copy with PRAGMA integrity_check - corruption missed by quick_check is detected by the full check - existing good backups remain protected - oversized exports are still delivered but declare whether this version can import them, while the 20 MB import limit remains unchanged - backup was exercised through the real browser UI on both the normal campaign database and a campaign-shaped database over 100 MB WP-E — control-boundary contrast: - interactive control boundaries meet the WCAG 1.4.11 3:1 target - the contrast audit is now a failing gate rather than an advisory - rendered browser measurements pass for the composer, controls, tabs and nav - text contrast and focus visibility remain intact - owner reviewed and approved the before/after screenshots Reports: - planning/reports/v1.1/V1.1-WP-D-REPORT.md - planning/reports/v1.1/V1.1-WP-E-REPORT.md All planned v1.1 work packages A-E are now complete. Release validation has not yet begun.
420 lines
17 KiB
Python
420 lines
17 KiB
Python
"""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 base64
|
|
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 screenshot(self, path) -> Path:
|
|
"""The viewport as a PNG, written where you ask (v1.1 WP-E).
|
|
|
|
Evidence for a change a reader judges by looking at it: a contrast ratio
|
|
says a boundary is measurable, and a picture says what it looks like.
|
|
"""
|
|
encoded = self._call("GET", self._s("/screenshot"))["value"]
|
|
target = Path(path)
|
|
target.parent.mkdir(parents=True, exist_ok=True)
|
|
target.write_bytes(base64.b64decode(encoded))
|
|
return target
|
|
|
|
def hover(self, element: str) -> None:
|
|
"""A real pointer over an element, so `:hover` actually applies.
|
|
|
|
Dispatching a mouseover event from JavaScript does not do this: CSS
|
|
`:hover` follows the browser's own pointer state, not a synthetic event,
|
|
so a measurement taken after `dispatchEvent` reads the resting style and
|
|
reports it as the hover style. This moves the pointer (v1.1 WP-E).
|
|
"""
|
|
self._call("POST", self._s("/execute/sync"), {
|
|
"script": "arguments[0].scrollIntoView({block: 'center', inline: 'nearest'})",
|
|
"args": [{ELEMENT_KEY: element}]})
|
|
self._call("POST", self._s("/actions"), {"actions": [{
|
|
"type": "pointer", "id": "mouse", "parameters": {"pointerType": "mouse"},
|
|
"actions": [{"type": "pointerMove", "duration": 60,
|
|
"origin": {ELEMENT_KEY: element}, "x": 0, "y": 0}]}]})
|
|
|
|
def unhover(self) -> None:
|
|
"""Move the pointer off whatever it was over, and forget the input state."""
|
|
self._call("POST", self._s("/actions"), {"actions": [{
|
|
"type": "pointer", "id": "mouse", "parameters": {"pointerType": "mouse"},
|
|
"actions": [{"type": "pointerMove", "duration": 30,
|
|
"origin": "viewport", "x": 0, "y": 0}]}]})
|
|
try:
|
|
self._call("DELETE", self._s("/actions"))
|
|
except WebDriverError:
|
|
pass
|
|
|
|
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()
|