diff --git a/docs/superpowers/plans/2026-07-07-wabot-modernization.md b/docs/superpowers/plans/2026-07-07-wabot-modernization.md new file mode 100644 index 0000000..323dd7e --- /dev/null +++ b/docs/superpowers/plans/2026-07-07-wabot-modernization.md @@ -0,0 +1,3306 @@ +# WABot Modernization Implementation Plan + +> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. + +**Goal:** Rebuild wabot on Selenium 4 + uv with pytest and Sphinx docs, replacing pickle/dill browser persistence with JSON session records + a session-ID reattach mechanism. + +**Architecture:** A `Browser` facade drives declarative `Page` objects with typed field wrappers; a `SessionStore` (JSON in platformdirs) plus two `DriverHost` implementations (`ExternalServer`, detached `ManagedService`) make browser sessions survive the Python process; a `ReattachingRemote` webdriver subclass adopts saved sessions. Stealth behavior (random delays, Gaussian clicks) is a swappable pacing policy. + +**Tech Stack:** Python ≥3.10, selenium ≥4.45 <5, platformdirs, Pillow, uv (`uv_build` backend), pytest (unit + `integration` marker), ruff, Sphinx + sphinx-rtd-theme. + +**Spec:** `docs/superpowers/specs/2026-07-07-wabot-modernization-design.md` (approved). Work on branch `feature/modernization`. + +--- + +## Verified API facts (do not re-derive; verified 2026-07-07 on this machine) + +These were verified against installed selenium 4.45.0 source and live runs with chromedriver 150 + chromium and geckodriver 0.37.0. Trust them over your training data: + +1. `webdriver.Remote(command_executor=url, options=opts)` — `options` is required in practice (raises `TypeError` if None); `desired_capabilities`/`browser_profile` were removed in 4.10.0. +2. **Reattach pattern (verified live end-to-end):** subclass `selenium.webdriver.remote.webdriver.WebDriver`, stash the session id under a private attr (base `__init__` resets `self.session_id = None` *before* calling `start_session`), and override `start_session(self, capabilities)` to set only `self.session_id` and `self.caps`. Nothing else needed for normal WebDriver commands. +3. `GET /status` works on standalone chromedriver and geckodriver: `{"value": {"ready": bool, "message": str}}`. **chromedriver reports `ready:true` even with active sessions; geckodriver (single-session) flips `ready:false` while busy** — so liveness checks must accept an HTTP 200 with valid JSON and must NOT require `ready == true`. +4. `driver.quit()` on a plain `Remote` deletes the session + browser but **never kills the standalone driver server** (that's what makes reuse viable). After a reattached driver quits, other handles to that session get `InvalidSessionIdException`. +5. `SeleniumManager().binary_paths(["--browser", "chrome"])` → `{"driver_path": ..., "browser_path": ...}` exists but is explicitly beta — guard with try/except; prefer `shutil.which` first. +6. `ActionChains.move_to_element_with_offset(el, x, y)` offsets are **from the element's in-view center** in all of Selenium 4 (was top-left in Selenium 3 JSON-Wire). +7. Firefox full-page screenshot helpers (`get_full_page_screenshot_as_file` etc.) exist **only on `webdriver.Firefox`, not on `Remote`**. Over Remote you must register the raw command: `driver.command_executor._commands["FULL_PAGE_SCREENSHOT"] = ("GET", "/session/$sessionId/moz/screenshot/full")` then `driver.execute("FULL_PAGE_SCREENSHOT")["value"]` (base64). No Chrome equivalent. +8. Headless: Chrome `options.add_argument("--headless=new")`; Firefox `options.add_argument("-headless")`. The old `options.headless` property is gone. +9. User-agent: Chrome `add_argument(f"--user-agent={ua}")`; Firefox `options.set_preference("general.useragent.override", ua)` (no FirefoxProfile needed). +10. `driver.switch_to.alert` is a property (legacy `switch_to_alert()` removed); `EC.alert_is_present()` unchanged. +11. `options.binary_location = "/usr/bin/chromium"` still correct for pointing chromedriver at Chromium. +12. uv: `[build-system] requires = ["uv_build>=0.11,<0.12"]`, `build-backend = "uv_build"`, src layout `src/wabot/__init__.py` is the default (no module-name config needed). `[dependency-groups] dev = [...]` (PEP 735) is installed by `uv sync` by default. +13. pytest: a `-m` on the CLI **overrides** `-m` in `addopts` (last-wins) — verified on pytest 9.1.1. So `addopts = '-m "not integration"'` + `uv run pytest -m integration` works. +14. Sphinx: `html_theme = "sphinx_rtd_theme"` only — do **not** add the theme to `extensions`. `sphinx-rtd-theme>=3` supports sphinx 6–9. Build: `uv run sphinx-build -M html docs docs/_build`. + +--- + +## File structure + +``` +wabot/ (repo root, branch feature/modernization) +├── pyproject.toml # NEW — uv project, replaces setup.py +├── uv.lock # NEW — generated by uv sync +├── README.md # rewritten in Task 18 +├── CLAUDE.md # updated in Task 18 +├── src/wabot/ +│ ├── __init__.py # public API: browser(), sessions(), destroy(), re-exports +│ ├── pacing.py # NoPacing / HumanPacing +│ ├── sessions.py # SessionRecord, SessionStore +│ ├── hosts.py # build_options, service_alive, ExternalServer, +│ │ # ManagedService, find_driver_binary, stop_service, +│ │ # ReattachingRemote, attach +│ ├── fields.py # PageObject, TextField, SelectField, CheckField, NullField +│ ├── page.py # Page (element maps, click, forms, alerts, waits) +│ ├── browser.py # Browser facade +│ └── screenshot.py # save_full_page (firefox raw cmd / chromium stitch) +├── tests/ +│ ├── unit/ # test_pacing.py, test_sessions.py, test_hosts.py, +│ │ # test_reattach.py, test_fields.py, test_page.py, +│ │ # test_browser.py, test_api.py +│ ├── integration/ # conftest.py, test_ephemeral.py, test_persistence.py, +│ │ # test_external_server.py +│ └── fixtures/ # login.html, welcome.html +└── docs/ # conf.py, index.rst, quickstart.rst, sessions.rst, + # page-objects.rst, api.rst +``` + +Deleted in Task 1: legacy `wabot/` package (old `api.py`, `create_browser.py`, `page.py`, `fields.py`) and `setup.py`. The old code stays reachable in git history; everything worth porting is reproduced verbatim in this plan. + +--- + +### Task 1: Scaffold — uv project, src layout, remove legacy package + +**Files:** +- Create: `pyproject.toml` +- Create: `src/wabot/__init__.py` +- Create: `tests/unit/__init__.py`, `tests/integration/__init__.py` (empty files) +- Delete: `setup.py`, `wabot/__init__.py`, `wabot/api.py`, `wabot/create_browser.py`, `wabot/page.py`, `wabot/fields.py` + +- [ ] **Step 1: Verify you are on the feature branch** + +Run: `git -C /home/mathew/dev/Zavage-Software/wabot branch --show-current` +Expected: `feature/modernization`. All subsequent commands run from the repo root `/home/mathew/dev/Zavage-Software/wabot`. + +- [ ] **Step 2: Delete the legacy package and setup.py** + +```bash +git rm -r wabot setup.py +``` + +- [ ] **Step 3: Write `pyproject.toml`** + +```toml +[project] +name = "wabot" +version = "0.2.0" +description = "Stateful Selenium browser automation with sessions that survive the Python process" +readme = "README.md" +authors = [{ name = "Mathew Guest", email = "t3h.zavage@gmail.com" }] +requires-python = ">=3.10" +dependencies = [ + "selenium>=4.45,<5", + "platformdirs>=4", + "pillow>=10", +] + +[dependency-groups] +dev = [ + "pytest>=8", + "pytest-cov>=5", + "ruff>=0.8", + "sphinx>=7", + "sphinx-rtd-theme>=3", +] + +[build-system] +requires = ["uv_build>=0.11,<0.12"] +build-backend = "uv_build" + +[tool.pytest.ini_options] +addopts = '-m "not integration"' +markers = [ + "integration: real-browser tests (run with: uv run pytest -m integration)", +] +testpaths = ["tests"] + +[tool.ruff] +line-length = 100 +src = ["src", "tests"] + +[tool.ruff.lint] +select = ["E", "F", "W", "I", "UP", "B", "SIM"] +``` + +Note: if `uv sync` reports it cannot resolve sphinx for Python 3.10, change `requires-python` to `">=3.11"` — the library has no 3.10-specific consumers; this is the only sanctioned deviation. + +- [ ] **Step 4: Create the package skeleton** + +`src/wabot/__init__.py`: + +```python +"""wabot — stateful Selenium browser automation with reattachable sessions.""" +``` + +Create empty `tests/unit/__init__.py` and `tests/integration/__init__.py`. + +- [ ] **Step 5: Sync and sanity-check** + +Run: `uv sync` +Expected: creates `.venv` and `uv.lock`; installs selenium 4.45.x, platformdirs, pillow + dev group. + +Run: `uv run pytest` +Expected: `no tests ran` (exit code 5 is fine at this stage). + +Run: `uv run python -c "import wabot; print(wabot.__doc__)"` +Expected: prints the docstring. + +- [ ] **Step 6: Commit** + +```bash +git add pyproject.toml uv.lock src tests +git commit -m "build: replace setup.py with uv project, src layout" +``` + +--- + +### Task 2: `pacing.py` — NoPacing / HumanPacing + +**Files:** +- Create: `src/wabot/pacing.py` +- Test: `tests/unit/test_pacing.py` + +- [ ] **Step 1: Write the failing tests** + +`tests/unit/test_pacing.py`: + +```python +from wabot.pacing import HumanPacing, NoPacing + + +class TestNoPacing: + def test_delay_is_zero_for_every_action(self): + p = NoPacing() + assert p.delay("text") == 0.0 + assert p.delay("select") == 0.0 + assert p.delay("unknown-action") == 0.0 + + def test_click_offset_is_none(self): + assert NoPacing().click_offset(200, 50) is None + + +class TestHumanPacing: + def test_delays_fall_in_documented_ranges(self): + p = HumanPacing() + for _ in range(50): + assert 3.0 <= p.delay("text") <= 5.0 + assert 6.0 <= p.delay("select") <= 11.0 + assert 2.0 <= p.delay("checkbox") <= 3.0 + assert 3.0 <= p.delay("navigate") <= 6.0 + + def test_unknown_action_has_no_delay(self): + assert HumanPacing().delay("unknown-action") == 0.0 + + def test_click_offset_stays_inside_element(self): + p = HumanPacing() + for _ in range(200): + offset = p.click_offset(100, 40) + assert offset is not None + x, y = offset + # offsets are measured from the element CENTER (selenium 4) + assert -49 <= x <= 49 + assert -19 <= y <= 19 + + def test_tiny_elements_get_plain_click(self): + p = HumanPacing() + assert p.click_offset(4, 40) is None + assert p.click_offset(40, 4) is None +``` + +- [ ] **Step 2: Run tests to verify they fail** + +Run: `uv run pytest tests/unit/test_pacing.py -v` +Expected: FAIL — `ModuleNotFoundError: No module named 'wabot.pacing'` + +- [ ] **Step 3: Write the implementation** + +`src/wabot/pacing.py`: + +```python +"""Pacing policies: how human-like the bot's interactions are. + +``HumanPacing`` reproduces legacy wabot stealth behavior (random delays +between actions, Gaussian click offsets). ``NoPacing`` is instant and +exact — used by the test suite and for trusted-site automation. +""" + +from __future__ import annotations + +import random + + +class NoPacing: + """Instant actions, exact clicks.""" + + def delay(self, action: str) -> float: + """Seconds to sleep before performing ``action``.""" + return 0.0 + + def click_offset(self, width: int, height: int) -> tuple[int, int] | None: + """Offset from the element center to click at, or None for a plain click.""" + return None + + +class HumanPacing(NoPacing): + """Random delays and Gaussian click offsets (anti-bot stealth).""" + + DELAYS: dict[str, tuple[float, float]] = { + "text": (3.0, 5.0), + "select": (6.0, 11.0), + "checkbox": (2.0, 3.0), + "navigate": (3.0, 6.0), + } + MIN_DIMENSION = 5 # elements thinner than this get a plain centered click + + def delay(self, action: str) -> float: + lo, hi = self.DELAYS.get(action, (0.0, 0.0)) + return random.uniform(lo, hi) + + def click_offset(self, width: int, height: int) -> tuple[int, int] | None: + if width < self.MIN_DIMENSION or height < self.MIN_DIMENSION: + return None + return (self._axis_offset(width), self._axis_offset(height)) + + @staticmethod + def _axis_offset(size: int) -> int: + # Selenium 4 measures offsets from the element's in-view center, so a + # human-looking click is a Gaussian around 0 clamped inside the element. + max_offset = size // 2 - 1 + offset = int(random.gauss(0.0, size / 7.0)) + return max(-max_offset, min(max_offset, offset)) +``` + +- [ ] **Step 4: Run tests to verify they pass** + +Run: `uv run pytest tests/unit/test_pacing.py -v` +Expected: 6 passed + +- [ ] **Step 5: Commit** + +```bash +git add src/wabot/pacing.py tests/unit/test_pacing.py +git commit -m "feat: pacing policies (HumanPacing default behavior, NoPacing for tests)" +``` + +--- +### Task 3: `sessions.py` — SessionRecord + SessionStore + +**Files:** +- Create: `src/wabot/sessions.py` +- Test: `tests/unit/test_sessions.py` + +- [ ] **Step 1: Write the failing tests** + +`tests/unit/test_sessions.py`: + +```python +from datetime import datetime, timedelta, timezone + +from wabot.sessions import SessionRecord, SessionStore + + +def make_record(name="scraper1", created_at=None): + return SessionRecord( + name=name, + executor_url="http://127.0.0.1:9515", + session_id="abc123", + browser="chromium", + created_at=(created_at or datetime.now(timezone.utc)).isoformat(), + service_pid=4242, + service_port=9515, + ) + + +class TestSessionStore: + def test_get_missing_returns_none(self, tmp_path): + store = SessionStore(path=tmp_path / "sessions.json") + assert store.get("nope") is None + + def test_save_and_get_round_trip(self, tmp_path): + store = SessionStore(path=tmp_path / "sessions.json") + store.save(make_record()) + rec = store.get("scraper1") + assert rec == make_record(created_at=datetime.fromisoformat(rec.created_at)) + assert rec.session_id == "abc123" + assert rec.service_pid == 4242 + + def test_save_is_readable_by_a_fresh_store_instance(self, tmp_path): + path = tmp_path / "sessions.json" + SessionStore(path=path).save(make_record()) + assert SessionStore(path=path).get("scraper1") is not None + + def test_names_lists_sessions_sorted(self, tmp_path): + store = SessionStore(path=tmp_path / "sessions.json") + store.save(make_record(name="zeta")) + store.save(make_record(name="alpha")) + assert store.names() == ["alpha", "zeta"] + + def test_remove(self, tmp_path): + store = SessionStore(path=tmp_path / "sessions.json") + store.save(make_record()) + store.remove("scraper1") + assert store.get("scraper1") is None + store.remove("scraper1") # removing twice must not raise + + def test_stale_records_are_evicted_on_load(self, tmp_path): + store = SessionStore(path=tmp_path / "sessions.json", max_age=timedelta(days=3)) + old = datetime.now(timezone.utc) - timedelta(days=4) + store.save(make_record(name="old", created_at=old)) + store.save(make_record(name="fresh")) + assert store.names() == ["fresh"] + + def test_corrupt_file_is_moved_aside_not_crashed_on(self, tmp_path): + path = tmp_path / "sessions.json" + path.write_text("{ this is not json") + store = SessionStore(path=path) + assert store.load() == {} + assert (tmp_path / "sessions.json.bad").exists() + store.save(make_record()) # store is usable again + assert store.get("scraper1") is not None + + def test_default_path_is_under_platformdirs(self): + from wabot.sessions import default_store_path + + assert default_store_path().name == "sessions.json" + assert "wabot" in str(default_store_path()) +``` + +- [ ] **Step 2: Run tests to verify they fail** + +Run: `uv run pytest tests/unit/test_sessions.py -v` +Expected: FAIL — `ModuleNotFoundError: No module named 'wabot.sessions'` + +- [ ] **Step 3: Write the implementation** + +`src/wabot/sessions.py`: + +```python +"""Persistence of reattachable browser sessions as plain JSON. + +Replaces the legacy pickle/dill approach: instead of serializing a live +webdriver object, we record only what is needed to reattach — the +executor URL and session id — plus bookkeeping for managed services. +""" + +from __future__ import annotations + +import dataclasses +import json +import logging +from dataclasses import dataclass +from datetime import datetime, timedelta, timezone +from pathlib import Path + +import platformdirs + +LOGGER = logging.getLogger("wabot") + + +@dataclass +class SessionRecord: + """Everything needed to find a saved browser session again.""" + + name: str + executor_url: str + session_id: str + browser: str + created_at: str # ISO-8601, UTC + service_pid: int | None = None + service_port: int | None = None + + +def default_store_path() -> Path: + return Path(platformdirs.user_data_dir("wabot")) / "sessions.json" + + +class SessionStore: + """A JSON file mapping session name -> SessionRecord.""" + + def __init__(self, path: Path | None = None, max_age: timedelta = timedelta(days=3)): + self.path = Path(path) if path is not None else default_store_path() + self.max_age = max_age + + def load(self) -> dict[str, SessionRecord]: + """Read all records, evicting stale ones and surviving a corrupt file.""" + if not self.path.exists(): + return {} + try: + raw = json.loads(self.path.read_text()) + records = {name: SessionRecord(**data) for name, data in raw.items()} + except (json.JSONDecodeError, TypeError, AttributeError): + bad = self.path.with_name(self.path.name + ".bad") + self.path.rename(bad) + LOGGER.warning("corrupt session store moved to %s; starting fresh", bad) + return {} + fresh = {} + for name, record in records.items(): + created = datetime.fromisoformat(record.created_at) + if datetime.now(timezone.utc) - created >= self.max_age: + LOGGER.info("evicting stale session %r (created %s)", name, record.created_at) + continue + fresh[name] = record + if len(fresh) != len(records): + self._write(fresh) + return fresh + + def get(self, name: str) -> SessionRecord | None: + return self.load().get(name) + + def save(self, record: SessionRecord) -> None: + records = self.load() + records[record.name] = record + self._write(records) + + def remove(self, name: str) -> None: + records = self.load() + if records.pop(name, None) is not None: + self._write(records) + + def names(self) -> list[str]: + return sorted(self.load()) + + def _write(self, records: dict[str, SessionRecord]) -> None: + self.path.parent.mkdir(parents=True, exist_ok=True) + payload = {name: dataclasses.asdict(rec) for name, rec in records.items()} + self.path.write_text(json.dumps(payload, indent=2)) +``` + +Note the corrupt-file test writes `sessions.json` + `.bad` → `sessions.json.bad`, which is why `_write` uses `with_name(self.path.name + ".bad")` (not `with_suffix`, which would replace `.json`). + +- [ ] **Step 4: Run tests to verify they pass** + +Run: `uv run pytest tests/unit/test_sessions.py -v` +Expected: 8 passed + +- [ ] **Step 5: Commit** + +```bash +git add src/wabot/sessions.py tests/unit/test_sessions.py +git commit -m "feat: JSON SessionStore with stale eviction (replaces pickle persistence)" +``` + +--- + +### Task 4: `hosts.py` (part 1) — browser normalization, options builder, ExternalServer, service_alive + +**Files:** +- Create: `src/wabot/hosts.py` +- Test: `tests/unit/test_hosts.py` + +- [ ] **Step 1: Write the failing tests** + +`tests/unit/test_hosts.py`: + +```python +import http.server +import json +import threading + +import pytest + +from wabot.hosts import ExternalServer, build_options, normalize_browser, service_alive + + +class TestNormalizeBrowser: + def test_chrome_is_an_alias_for_chromium(self): + assert normalize_browser("chrome") == "chromium" + + def test_firefox_and_chromium_pass_through(self): + assert normalize_browser("firefox") == "firefox" + assert normalize_browser("chromium") == "chromium" + + def test_unknown_browser_raises(self): + with pytest.raises(ValueError, match="unsupported browser"): + normalize_browser("safari") + + +class TestBuildOptions: + def test_chromium_headless_and_user_agent(self): + opts = build_options("chromium", headless=True, user_agent="MyBot/1.0") + assert "--headless=new" in opts.arguments + assert "--user-agent=MyBot/1.0" in opts.arguments + + def test_chromium_defaults_have_no_headless_or_ua(self): + opts = build_options("chromium") + assert not any(a.startswith(("--headless", "--user-agent")) for a in opts.arguments) + + def test_firefox_headless_and_user_agent(self): + opts = build_options("firefox", headless=True, user_agent="MyBot/1.0") + assert "-headless" in opts.arguments + assert opts.preferences.get("general.useragent.override") == "MyBot/1.0" + + def test_options_type_matches_browser(self): + assert type(build_options("firefox")).__module__.startswith("selenium.webdriver.firefox") + assert type(build_options("chrome")).__module__.startswith("selenium.webdriver.chrome") + + +class _StatusHandler(http.server.BaseHTTPRequestHandler): + ready = True + + def do_GET(self): + if self.path == "/status": + body = json.dumps({"value": {"ready": self.ready, "message": ""}}).encode() + self.send_response(200) + self.send_header("Content-Type", "application/json") + self.end_headers() + self.wfile.write(body) + else: + self.send_response(404) + self.end_headers() + + def log_message(self, *args): # keep test output clean + pass + + +@pytest.fixture +def status_server(): + server = http.server.HTTPServer(("127.0.0.1", 0), _StatusHandler) + thread = threading.Thread(target=server.serve_forever, daemon=True) + thread.start() + yield f"http://127.0.0.1:{server.server_address[1]}" + server.shutdown() + + +class TestServiceAlive: + def test_true_when_status_answers(self, status_server): + assert service_alive(status_server) is True + + def test_true_even_when_not_ready(self, status_server): + # geckodriver reports ready=false while its single session is in use; + # aliveness must not require ready==true + _StatusHandler.ready = False + try: + assert service_alive(status_server) is True + finally: + _StatusHandler.ready = True + + def test_false_when_nothing_listens(self): + assert service_alive("http://127.0.0.1:1") is False + + +class TestExternalServer: + def test_returns_url_when_alive(self, status_server): + assert ExternalServer(status_server + "/").ensure_running() == status_server + + def test_raises_when_dead(self): + with pytest.raises(ConnectionError, match="no WebDriver server"): + ExternalServer("http://127.0.0.1:1").ensure_running() +``` + +- [ ] **Step 2: Run tests to verify they fail** + +Run: `uv run pytest tests/unit/test_hosts.py -v` +Expected: FAIL — `ModuleNotFoundError: No module named 'wabot.hosts'` + +- [ ] **Step 3: Write the implementation** + +`src/wabot/hosts.py` (this file grows in Tasks 5 and 6; start with exactly this): + +```python +"""Driver hosts: the processes that keep a browser alive, and how to reach them. + +Persistence-capable browsers are always driven over the WebDriver wire +protocol (a URL). Two hosts provide that URL: ``ExternalServer`` (a +Selenium Grid or standalone driver the user runs) and ``ManagedService`` +(a chromedriver/geckodriver wabot spawns detached so it outlives Python). +""" + +from __future__ import annotations + +import json +import logging +import urllib.request + +from selenium import webdriver + +LOGGER = logging.getLogger("wabot") + +BROWSER_ALIASES = {"chrome": "chromium"} +DRIVER_BINARIES = {"chromium": "chromedriver", "firefox": "geckodriver"} + + +def normalize_browser(browser: str) -> str: + browser = BROWSER_ALIASES.get(browser, browser) + if browser not in DRIVER_BINARIES: + raise ValueError( + f"unsupported browser {browser!r}; use 'firefox' or 'chromium' (alias: 'chrome')" + ) + return browser + + +def build_options( + browser: str, *, headless: bool = False, user_agent: str | None = None +) -> webdriver.ChromeOptions | webdriver.FirefoxOptions: + """Browser-appropriate Options. UA defaults to the browser's own.""" + browser = normalize_browser(browser) + if browser == "chromium": + opts = webdriver.ChromeOptions() + binary = _chromium_binary() + if binary: + opts.binary_location = binary + if headless: + opts.add_argument("--headless=new") + if user_agent: + opts.add_argument(f"--user-agent={user_agent}") + return opts + opts = webdriver.FirefoxOptions() + if headless: + opts.add_argument("-headless") + if user_agent: + opts.set_preference("general.useragent.override", user_agent) + return opts + + +def _chromium_binary() -> str | None: + import shutil + + for name in ("chromium", "chromium-browser", "google-chrome", "chrome"): + found = shutil.which(name) + if found: + return found + return None + + +def service_alive(url: str, timeout: float = 2.0) -> bool: + """True if a WebDriver server answers GET /status at ``url``. + + Liveness only: geckodriver reports ready=false while its single session + is in use, so the ``ready`` flag is deliberately ignored. + """ + try: + with urllib.request.urlopen(f"{url.rstrip('/')}/status", timeout=timeout) as resp: + json.loads(resp.read()) + return resp.status == 200 + except (OSError, ValueError): + return False + + +class ExternalServer: + """A WebDriver server somebody else runs (Selenium Grid, bare driver).""" + + def __init__(self, url: str): + self.url = url.rstrip("/") + + def ensure_running(self) -> str: + if not service_alive(self.url): + raise ConnectionError(f"no WebDriver server answering at {self.url}/status") + return self.url +``` + +- [ ] **Step 4: Run tests to verify they pass** + +Run: `uv run pytest tests/unit/test_hosts.py -v` +Expected: 12 passed + +- [ ] **Step 5: Commit** + +```bash +git add src/wabot/hosts.py tests/unit/test_hosts.py +git commit -m "feat: options builder, /status liveness check, ExternalServer host" +``` + +--- +### Task 5: `hosts.py` (part 2) — ManagedService: spawn a detached driver + +**Files:** +- Modify: `src/wabot/hosts.py` (append; also extend the import block) +- Test: `tests/unit/test_hosts.py` (append) + +- [ ] **Step 1: Write the failing tests** + +Append to `tests/unit/test_hosts.py`. The tests use a stub "driver" executable — a tiny Python script that serves `/status` on `--port=N` — so spawning/detaching/polling is tested without any browser: + +```python +import os +import signal +import stat +import sys +import time + +from wabot.hosts import ManagedService, find_driver_binary, stop_service + +STUB_DRIVER = """\ +#!{python} +import http.server, json, sys +port = int(sys.argv[1].split("=", 1)[1]) +class H(http.server.BaseHTTPRequestHandler): + def do_GET(self): + body = json.dumps({{"value": {{"ready": True, "message": ""}}}}).encode() + self.send_response(200); self.end_headers(); self.wfile.write(body) + def log_message(self, *a): pass +http.server.HTTPServer(("127.0.0.1", port), H).serve_forever() +""" + + +@pytest.fixture +def stub_driver(tmp_path): + path = tmp_path / "chromedriver" + path.write_text(STUB_DRIVER.format(python=sys.executable)) + path.chmod(path.stat().st_mode | stat.S_IEXEC) + return path + + +class TestFindDriverBinary: + def test_env_var_override_wins(self, monkeypatch): + monkeypatch.setenv("WABOT_CHROMEDRIVER", "/custom/chromedriver") + assert find_driver_binary("chromium") == "/custom/chromedriver" + + def test_path_lookup(self, monkeypatch, stub_driver): + monkeypatch.delenv("WABOT_CHROMEDRIVER", raising=False) + monkeypatch.setenv("PATH", str(stub_driver.parent)) + assert find_driver_binary("chromium") == str(stub_driver) + + +class TestManagedService: + def test_spawns_detached_and_answers_status(self, stub_driver, tmp_path): + svc = ManagedService("chromium", log_dir=tmp_path, driver_binary=str(stub_driver)) + url = svc.ensure_running() + try: + assert service_alive(url) + assert svc.pid is not None and svc.port is not None + assert url == f"http://127.0.0.1:{svc.port}" + # detached: the spawned process is in its own session, so it + # would survive this python process exiting + assert os.getsid(svc.pid) != os.getsid(os.getpid()) + assert (tmp_path / f"chromedriver-{svc.port}.log").exists() + finally: + stop_service(svc.pid) + + def test_startup_timeout_raises(self, tmp_path): + # /bin/true exits immediately and never serves /status + svc = ManagedService( + "chromium", log_dir=tmp_path, driver_binary="/bin/true", startup_timeout=1.0 + ) + with pytest.raises(TimeoutError, match="did not answer /status"): + svc.ensure_running() + + +class TestStopService: + def test_stops_a_live_process_and_reports_dead_ones(self, stub_driver, tmp_path): + svc = ManagedService("chromium", log_dir=tmp_path, driver_binary=str(stub_driver)) + svc.ensure_running() + assert stop_service(svc.pid) is True + deadline = time.monotonic() + 5 + while time.monotonic() < deadline: + try: + os.kill(svc.pid, 0) + except ProcessLookupError: + break + time.sleep(0.05) + assert stop_service(999999) is False # no such pid +``` + +- [ ] **Step 2: Run tests to verify they fail** + +Run: `uv run pytest tests/unit/test_hosts.py -v` +Expected: new tests FAIL — `ImportError: cannot import name 'ManagedService'`; the 12 Task-4 tests still pass. + +- [ ] **Step 3: Write the implementation** + +Append to `src/wabot/hosts.py`, and extend the top import block to: + +```python +import json +import logging +import os +import shutil +import signal +import socket +import subprocess +import time +import urllib.request +from pathlib import Path + +import platformdirs +from selenium import webdriver +``` + +(then remove the now-redundant `import shutil` inside `_chromium_binary`). Appended code: + +```python +def find_driver_binary(browser: str) -> str: + """Locate chromedriver/geckodriver: env override, PATH, then Selenium Manager.""" + name = DRIVER_BINARIES[normalize_browser(browser)] + override = os.environ.get(f"WABOT_{name.upper()}") + if override: + return override + found = shutil.which(name) + if found: + return found + try: + # Selenium Manager is bundled with selenium but explicitly beta + # ("may change") — keep it as a guarded fallback, never the primary. + from selenium.webdriver.common.selenium_manager import SeleniumManager + + args = ["--browser", "chrome" if browser == "chromium" else browser] + chromium = _chromium_binary() + if browser == "chromium" and chromium: + args += ["--browser-path", chromium] + return SeleniumManager().binary_paths(args)["driver_path"] + except Exception as ex: # noqa: BLE001 — beta API, failure modes unknown + raise FileNotFoundError( + f"could not find {name!r} on PATH (set WABOT_{name.upper()} to override); " + f"Selenium Manager fallback also failed: {ex}" + ) from ex + + +def _free_port() -> int: + with socket.socket() as sock: + sock.bind(("127.0.0.1", 0)) + return sock.getsockname()[1] + + +class ManagedService: + """A driver service wabot spawns detached, so it outlives this process.""" + + def __init__( + self, + browser: str, + log_dir: Path | None = None, + driver_binary: str | None = None, + startup_timeout: float = 15.0, + ): + self.browser = normalize_browser(browser) + self.log_dir = Path(log_dir) if log_dir else Path(platformdirs.user_data_dir("wabot")) + self.driver_binary = driver_binary + self.startup_timeout = startup_timeout + self.pid: int | None = None + self.port: int | None = None + + def ensure_running(self) -> str: + binary = self.driver_binary or find_driver_binary(self.browser) + self.port = _free_port() + url = f"http://127.0.0.1:{self.port}" + self.log_dir.mkdir(parents=True, exist_ok=True) + log_path = self.log_dir / f"{DRIVER_BINARIES[self.browser]}-{self.port}.log" + with open(log_path, "ab") as log: + process = subprocess.Popen( + [binary, f"--port={self.port}"], + stdout=log, + stderr=log, + stdin=subprocess.DEVNULL, + start_new_session=True, # own session: survives this python process + ) + self.pid = process.pid + LOGGER.info("spawned %s (pid %s) on %s, log %s", binary, self.pid, url, log_path) + deadline = time.monotonic() + self.startup_timeout + while time.monotonic() < deadline: + if service_alive(url): + return url + if process.poll() is not None: + break # exited early; fall through to the error + time.sleep(0.2) + raise TimeoutError( + f"{binary} did not answer /status on {url} within " + f"{self.startup_timeout}s (see {log_path})" + ) + + +def stop_service(pid: int) -> bool: + """SIGTERM a managed driver service. False if it was already gone.""" + try: + os.kill(pid, signal.SIGTERM) + return True + except (ProcessLookupError, PermissionError): + return False +``` + +- [ ] **Step 4: Run tests to verify they pass** + +Run: `uv run pytest tests/unit/test_hosts.py -v` +Expected: 17 passed (12 from Task 4 + 5 new) + +- [ ] **Step 5: Commit** + +```bash +git add src/wabot/hosts.py tests/unit/test_hosts.py +git commit -m "feat: ManagedService spawns detached driver services that outlive python" +``` + +--- + +### Task 6: `hosts.py` (part 3) — ReattachingRemote + attach() + +**Files:** +- Modify: `src/wabot/hosts.py` (append) +- Test: `tests/unit/test_reattach.py` + +- [ ] **Step 1: Write the failing tests** + +`tests/unit/test_reattach.py`. Constructing a `Remote` performs **no HTTP until the first command** (verified), so adoption can be unit-tested without a server; the dead-session path monkeypatches the probe: + +```python +import pytest +from selenium.common.exceptions import WebDriverException + +from wabot.hosts import ReattachingRemote, attach, build_options + + +class TestReattachingRemote: + def test_adopts_session_id_without_creating_a_session(self): + driver = ReattachingRemote( + "http://127.0.0.1:1", "saved-session-id", options=build_options("chromium") + ) + # base __init__ resets session_id to None BEFORE start_session runs; + # our override must still win + assert driver.session_id == "saved-session-id" + assert isinstance(driver.caps, dict) + + def test_works_for_firefox_options_too(self): + driver = ReattachingRemote( + "http://127.0.0.1:1", "sid", options=build_options("firefox") + ) + assert driver.session_id == "sid" + + +class TestAttach: + def test_returns_driver_when_probe_succeeds(self, monkeypatch): + monkeypatch.setattr( + ReattachingRemote, "current_url", property(lambda self: "http://example.com") + ) + driver = attach("http://127.0.0.1:1", "sid", "chromium") + assert driver is not None + assert driver.session_id == "sid" + + def test_returns_none_when_session_is_dead(self, monkeypatch): + def boom(self): + raise WebDriverException("invalid session id") + + monkeypatch.setattr(ReattachingRemote, "current_url", property(boom)) + assert attach("http://127.0.0.1:1", "sid", "chromium") is None + + def test_returns_none_when_server_is_unreachable(self): + # no server on port 1: connection refused is immediate; attach must not raise + assert attach("http://127.0.0.1:1", "sid", "chromium") is None +``` + +- [ ] **Step 2: Run tests to verify they fail** + +Run: `uv run pytest tests/unit/test_reattach.py -v` +Expected: FAIL — `ImportError: cannot import name 'ReattachingRemote'` + +- [ ] **Step 3: Write the implementation** + +Append to `src/wabot/hosts.py`; add these imports to the top block: + +```python +from selenium.common.exceptions import WebDriverException +from selenium.webdriver.remote.webdriver import WebDriver as RemoteWebDriver +``` + +Appended code: + +```python +class ReattachingRemote(RemoteWebDriver): + """A Remote driver that adopts an existing session instead of creating one. + + Selenium has no supported reattach API; this relies on two internals + that have been stable across 4.x (pin selenium <5): ``execute()`` + injects ``self.session_id`` into every command, and ``start_session`` + is the only place a NEW_SESSION request happens. + """ + + def __init__(self, command_executor: str, session_id: str, options): + # NOT self.session_id: the base __init__ resets that to None before + # calling start_session, so stash the id under a private name. + self._reattach_session_id = session_id + super().__init__(command_executor=command_executor, options=options) + + def start_session(self, capabilities: dict) -> None: + # Skip Command.NEW_SESSION entirely; adopt the saved session. + self.session_id = self._reattach_session_id + self.caps = capabilities # requested caps: fine for normal commands (no CDP/BiDi) + + +def attach(executor_url: str, session_id: str, browser: str): + """Return a live driver adopting the saved session, or None if it is dead.""" + driver = ReattachingRemote(executor_url, session_id, options=build_options(browser)) + try: + _ = driver.current_url # first real HTTP call; raises if the session is gone + return driver + except WebDriverException as ex: + LOGGER.warning("saved session %s is dead: %s", session_id, ex) + return None + except OSError as ex: # server itself unreachable (connection refused, timeout) + LOGGER.warning("no server at %s: %s", executor_url, ex) + return None +``` + +Note: selenium's HTTP layer (urllib3) can surface connection failures as +`urllib3.exceptions.HTTPError` subclasses that are **not** `OSError`. If +`test_returns_none_when_server_is_unreachable` fails with an unhandled +`MaxRetryError`, add it explicitly: + +```python +from urllib3.exceptions import HTTPError as _Urllib3HTTPError +``` + +and change the last except clause to `except (OSError, _Urllib3HTTPError) as ex:`. +Whichever variant makes the test pass is the correct final code. + +- [ ] **Step 4: Run tests to verify they pass** + +Run: `uv run pytest tests/unit/test_reattach.py -v` +Expected: 5 passed + +Run: `uv run pytest` +Expected: all unit tests so far pass (pacing 6, sessions 8, hosts 17, reattach 5 = 36) + +- [ ] **Step 5: Commit** + +```bash +git add src/wabot/hosts.py tests/unit/test_reattach.py +git commit -m "feat: session reattach via ReattachingRemote (adopts saved session id)" +``` + +--- +### Task 7: `fields.py` — typed field wrappers + +**Files:** +- Create: `src/wabot/fields.py` +- Test: `tests/unit/test_fields.py` + +- [ ] **Step 1: Write the failing tests** + +`tests/unit/test_fields.py`. Fields talk to a `page` object (which owns the driver and the form helpers), so a `MagicMock` page is the natural test double. Pacing must come from the page's browser: + +```python +from unittest.mock import MagicMock + +import pytest + +from wabot.fields import CheckField, NullField, PageObject, SelectField, TextField +from wabot.pacing import NoPacing + +ACCESSORS = ("id", "username") + + +@pytest.fixture +def page(): + page = MagicMock(name="page") + page.pacing = NoPacing() + return page + + +class TestPageObject: + def test_locates_element_via_page_driver(self, page): + obj = PageObject(page, accessors=ACCESSORS, name="username") + page.driver.find_element.assert_called_once_with(by="id", value="username") + assert obj.el is page.driver.find_element.return_value + + def test_unknown_attributes_fall_through_to_element(self, page): + obj = PageObject(page, accessors=ACCESSORS, name="username") + assert obj.tag_name is obj.el.tag_name + + def test_click_delegates_to_page(self, page): + obj = PageObject(page, accessors=ACCESSORS, name="username") + assert obj.click() is page.click.return_value + page.click.assert_called_once_with(obj.el) + + +class TestTextField: + def test_set_value_uses_page_helper(self, page): + field = TextField(page, accessors=ACCESSORS, name="username") + result = field.set_value("mathew") + page.set_el_value.assert_called_once_with(field.el, "mathew") + assert result is page.set_el_value.return_value + + def test_get_value_uses_page_helper(self, page): + field = TextField(page, accessors=ACCESSORS, name="username") + assert field.get_value() is page.get_el_value.return_value + + +class TestSelectField: + def test_wraps_element_in_selenium_select(self, page, monkeypatch): + import wabot.fields as fields_mod + + fake_select_cls = MagicMock(name="Select") + monkeypatch.setattr(fields_mod, "Select", fake_select_cls) + field = SelectField(page, accessors=("id", "state"), name="state") + fake_select_cls.assert_called_once_with(field.el) + assert field.dropdown is fake_select_cls.return_value + + def test_set_value_by_value_and_by_text(self, page, monkeypatch): + import wabot.fields as fields_mod + + monkeypatch.setattr(fields_mod, "Select", MagicMock()) + field = SelectField(page, accessors=("id", "state"), name="state") + field.set_value(value="CO") + page.set_select_value.assert_called_with(field.dropdown, value="CO", text=None) + field.set_value(text="Colorado") + page.set_select_value.assert_called_with(field.dropdown, value=None, text="Colorado") + + +class TestCheckField: + def test_set_and_get_checked(self, page): + field = CheckField(page, accessors=("id", "agree"), name="agree") + field.set_checked(True) + page.set_checkbox.assert_called_once_with(field.el, True) + field.get_checked() + page.get_checkbox_value.assert_called_once_with(field.el, False) + + +class TestNullField: + def test_is_falsy(self): + assert not NullField(name="missing") + + def test_attribute_access_raises_with_element_name(self): + with pytest.raises(AttributeError, match="missing"): + NullField(name="missing").click() + + def test_pattern_if_el_guards_work(self): + el = NullField(name="missing") + if el: + pytest.fail("NullField must be falsy") +``` + +- [ ] **Step 2: Run tests to verify they fail** + +Run: `uv run pytest tests/unit/test_fields.py -v` +Expected: FAIL — `ModuleNotFoundError: No module named 'wabot.fields'` + +- [ ] **Step 3: Write the implementation** + +`src/wabot/fields.py`: + +```python +"""Typed wrappers around page elements. + +``page[key]`` returns one of these instead of a raw WebElement. Unknown +attribute access falls through to the underlying selenium element, so a +wrapper can be used anywhere an element can. ``NullField`` is the falsy +null-object returned for elements that could not be found, enabling +``if el:`` guards instead of exception handling. +""" + +from __future__ import annotations + +import logging +import time + +from selenium.webdriver.support.ui import Select + +LOGGER = logging.getLogger("wabot") + + +class PageObject: + """Base wrapper; subclasses add typed get/set behavior.""" + + def __init__(self, page, accessors=None, name=None): + self.page = page + self.accessors = accessors + self.name = name + self.el = self._locate() if accessors else None + + def _locate(self): + by, value = self.accessors + return self.page.driver.find_element(by=by, value=value) + + @property + def pacing(self): + return self.page.pacing + + def _pace(self, action: str) -> None: + delay = self.pacing.delay(action) + if delay: + time.sleep(delay) + + def __getattr__(self, name): + if name == "el": # guard: never recurse if __init__ didn't finish + raise AttributeError(name) + return getattr(self.el, name) + + def click(self): + return self.page.click(self.el) + + def click_and_go(self): + self._pace("navigate") + return self.page.click_and_go(self.el) + + +class TextField(PageObject): + def get_value(self): + return self.page.get_el_value(self.el) + + def set_value(self, value): + LOGGER.info("[%s] set_text(%r)", self.name, value) + self._pace("text") + return self.page.set_el_value(self.el, value) + + +class SelectField(PageObject): + def __init__(self, *args, **kwargs): + super().__init__(*args, **kwargs) + self.dropdown = Select(self.el) + + def get_value(self): + return self.page.get_select_value(self.dropdown) + + def set_value(self, value=None, text=None): + LOGGER.info("[%s] set_select(%r)", self.name, value if value is not None else text) + self._pace("select") + return self.page.set_select_value(self.dropdown, value=value, text=text) + + def select_by_index(self, index): + self._pace("select") + return self.dropdown.select_by_index(index) + + +class CheckField(PageObject): + def get_checked(self, ignore_disabled=False): + return self.page.get_checkbox_value(self.el, ignore_disabled) + + def set_checked(self, checked): + LOGGER.info("[%s] set_checked(%r)", self.name, checked) + self._pace("checkbox") + return self.page.set_checkbox(self.el, checked) + + +class NullField: + """Falsy placeholder for an element that was not found.""" + + def __init__(self, name=None): + # bypass __getattr__ tricks: plain attribute set + object.__setattr__(self, "name", name) + + def __bool__(self): + return False + + def __getattr__(self, attr): + raise AttributeError( + f"element {self.name!r} was not found on the page (attribute {attr!r})" + ) +``` + +- [ ] **Step 4: Run tests to verify they pass** + +Run: `uv run pytest tests/unit/test_fields.py -v` +Expected: 11 passed + +- [ ] **Step 5: Commit** + +```bash +git add src/wabot/fields.py tests/unit/test_fields.py +git commit -m "feat: typed field wrappers with pacing (port of legacy fields.py)" +``` + +--- + +### Task 8: `page.py` (part 1) — element maps and proxy resolution + +**Files:** +- Create: `src/wabot/page.py` +- Test: `tests/unit/test_page.py` + +- [ ] **Step 1: Write the failing tests** + +`tests/unit/test_page.py`: + +```python +from unittest.mock import MagicMock + +from selenium.common.exceptions import NoSuchElementException + +from wabot.fields import CheckField, NullField, SelectField, TextField +from wabot.page import Page +from wabot.pacing import NoPacing + + +class BasePage(Page): + elements = { + "username": ("el", ("id", "username")), + "state": ("select", ("id", "state")), + "agree": ("checkbox", ("id", "agree")), + "rows": ("els", ("css selector", "tr.row")), + } + + +class ChildPage(BasePage): + elements = { + "username": ("el", ("id", "user_name_v2")), # override + "extra": ("el", ("id", "extra")), + } + + +def make_browser(): + browser = MagicMock(name="browser") + browser.pacing = NoPacing() + return browser + + +class TestElementResolution: + def test_own_elements_found(self): + page = BasePage(make_browser()) + assert page.find_element_locators("username") == ("el", ("id", "username")) + + def test_mro_walk_inherits_parent_elements(self): + page = ChildPage(make_browser()) + assert page.find_element_locators("state") == ("select", ("id", "state")) + assert page.find_element_locators("extra") == ("el", ("id", "extra")) + + def test_child_overrides_parent(self): + page = ChildPage(make_browser()) + assert page.find_element_locators("username") == ("el", ("id", "user_name_v2")) + + def test_missing_key_returns_none(self): + assert BasePage(make_browser()).find_element_locators("nope") is None + + +class TestGetProxy: + def test_typed_dispatch(self): + page = BasePage(make_browser()) + assert isinstance(page["username"], TextField) + assert isinstance(page["agree"], CheckField) + + def test_select_dispatch(self, monkeypatch): + import wabot.fields as fields_mod + + monkeypatch.setattr(fields_mod, "Select", MagicMock()) + page = BasePage(make_browser()) + assert isinstance(page["state"], SelectField) + + def test_els_returns_find_elements_result(self): + browser = make_browser() + page = BasePage(browser) + result = page["rows"] + browser.driver.find_elements.assert_called_once_with( + by="css selector", value="tr.row" + ) + assert result is browser.driver.find_elements.return_value + + def test_unknown_key_returns_falsy_nullfield(self): + el = BasePage(make_browser())["nope"] + assert isinstance(el, NullField) + assert not el + + def test_element_not_on_page_returns_nullfield(self): + browser = make_browser() + browser.driver.find_element.side_effect = NoSuchElementException("gone") + el = BasePage(browser)["username"] + assert isinstance(el, NullField) + + def test_custom_field_class_dispatch(self): + class MyField(TextField): + pass + + class CustomPage(Page): + elements = {"thing": (MyField, ("id", "thing"))} + + assert isinstance(CustomPage(make_browser())["thing"], MyField) + + +class TestVerify: + def test_default_verify_is_true(self): + assert BasePage(make_browser()).verify() is True +``` + +- [ ] **Step 2: Run tests to verify they fail** + +Run: `uv run pytest tests/unit/test_page.py -v` +Expected: FAIL — `ModuleNotFoundError: No module named 'wabot.page'` + +- [ ] **Step 3: Write the implementation** + +`src/wabot/page.py` (interaction helpers arrive in Task 9; start with exactly this): + +```python +"""Page: the base class consumers subclass to model a website page. + +Pages declare a class-level ``elements`` map:: + + class Login(wabot.Page): + elements = { + "username": ("el", ("id", "username")), + "state": ("select", ("id", "state")), + "agree": ("checkbox", ("id", "agree")), + "rows": ("els", ("css selector", "tr.row")), + "special": (MyFieldClass, ("id", "special")), + } + +Lookup walks the MRO, so subclasses inherit and override parent maps. +``page[key]`` returns a typed field wrapper — never a raw WebElement — +and a falsy ``NullField`` when the element cannot be found. +""" + +from __future__ import annotations + +import logging + +from selenium.common.exceptions import NoSuchElementException + +from .fields import CheckField, NullField, SelectField, TextField + +LOGGER = logging.getLogger("wabot") + +ALERT_TIMEOUT = 3 +PAGE_LOAD_TIMEOUT = 10 + +FIELD_TYPES = {"el": TextField, "select": SelectField, "checkbox": CheckField} + + +class Page: + elements: dict = {} + + def __init__(self, browser): + self.browser = browser + self.driver = browser.driver + LOGGER.info("loaded page %s", type(self).__name__) + + @property + def pacing(self): + return self.browser.pacing + + def verify(self) -> bool: + """Override to gate ``Browser.set_page`` (return False to refuse).""" + return True + + def find_element_locators(self, key): + """First match for ``key`` walking the class hierarchy (MRO order).""" + for cls in type(self).__mro__: + locators = getattr(cls, "elements", {}).get(key) + if locators: + return locators + return None + + def get_proxy(self, key): + locators = self.find_element_locators(key) + if not locators: + LOGGER.warning("element not in page map: %s", key) + return NullField(name=key) + obj_type, accessors = locators[0], locators[1] + try: + if obj_type == "els": + by, value = accessors + return self.driver.find_elements(by=by, value=value) + field_cls = FIELD_TYPES.get(obj_type) + if field_cls is None and isinstance(obj_type, type): + field_cls = obj_type + if field_cls is not None: + return field_cls(page=self, accessors=accessors, name=key) + except NoSuchElementException: + LOGGER.warning("element %r not present on current page", key) + return NullField(name=key) + LOGGER.error("unknown element type for %r: %r", key, obj_type) + return NullField(name=key) + + def __getitem__(self, key): + return self.get_proxy(key) +``` + +- [ ] **Step 4: Run tests to verify they pass** + +Run: `uv run pytest tests/unit/test_page.py -v` +Expected: 12 passed + +- [ ] **Step 5: Commit** + +```bash +git add src/wabot/page.py tests/unit/test_page.py +git commit -m "feat: Page element maps with MRO inheritance and typed dispatch" +``` + +--- + +### Task 9: `page.py` (part 2) — clicks, forms, alerts, waits + +**Files:** +- Modify: `src/wabot/page.py` (append methods to `Page`; extend imports) +- Test: `tests/unit/test_page.py` (append) + +- [ ] **Step 1: Write the failing tests** + +Append to `tests/unit/test_page.py`: + +```python +import wabot.page as page_mod +from wabot.pacing import HumanPacing + + +class PlainPage(Page): + elements = {} + + +def make_element(width=100, height=30, displayed=True, enabled=True): + el = MagicMock(name="element") + el.size = {"width": width, "height": height} + el.is_displayed.return_value = displayed + el.is_enabled.return_value = enabled + return el + + +class TestClick: + def test_refuses_null_element(self): + assert PlainPage(make_browser()).click(None) is False + assert PlainPage(make_browser()).click(NullField(name="x")) is False + + def test_plain_click_when_pacing_gives_no_offset(self): + page = PlainPage(make_browser()) # NoPacing -> click_offset None + el = make_element() + assert page.click(el) is True + el.click.assert_called_once_with() + + def test_offset_click_uses_action_chains(self, monkeypatch): + browser = make_browser() + browser.pacing = HumanPacing() + chains = MagicMock(name="ActionChains") + monkeypatch.setattr(page_mod, "ActionChains", chains) + page = PlainPage(browser) + el = make_element(width=100, height=40) + assert page.click(el) is True + chains.assert_called_once_with(browser.driver) + args = chains.return_value.move_to_element_with_offset.call_args.args + assert args[0] is el + # offsets are center-relative and must stay inside the element + assert -49 <= args[1] <= 49 and -19 <= args[2] <= 19 + chains.return_value.move_to_element_with_offset.return_value.click.return_value.perform.assert_called_once() + + def test_never_clickable_element_fails(self): + page = PlainPage(make_browser()) + el = make_element(displayed=False) + assert page.click(el, clickable_timeout=0.3) is False + el.click.assert_not_called() + + +class TestFormHelpers: + def test_set_el_value_clears_types_and_verifies(self): + page = PlainPage(make_browser()) + el = make_element() + el.get_attribute.return_value = "mathew" + assert page.set_el_value(el, "mathew") is True + el.clear.assert_called_once_with() + el.send_keys.assert_called_once_with("mathew") + + def test_set_el_value_returns_false_on_mismatch(self): + page = PlainPage(make_browser()) + el = make_element() + el.get_attribute.return_value = "wrong" + assert page.set_el_value(el, "mathew") is False + + def test_set_el_value_none_just_clears(self): + page = PlainPage(make_browser()) + el = make_element() + assert page.set_el_value(el, None) is True + el.clear.assert_called_once_with() + el.send_keys.assert_not_called() + + def test_get_el_value_and_text_handle_null(self): + page = PlainPage(make_browser()) + assert page.get_el_value(None) is None + assert page.get_el_text(None) is None + + def test_set_checkbox_clicks_only_when_state_differs(self): + page = PlainPage(make_browser()) + el = make_element() + el.is_selected.return_value = False + page.set_checkbox(el, True) + el.click.assert_called_once() + el.click.reset_mock() + el.is_selected.return_value = True + page.set_checkbox(el, True) + el.click.assert_not_called() + + +class TestStaleness: + def test_fresh_element_is_not_stale(self): + page = PlainPage(make_browser()) + assert page.is_element_stale(make_element()) is False + + def test_stale_element_is_detected(self): + from selenium.common.exceptions import StaleElementReferenceException + + class StaleEl: # NOT a MagicMock: mutating type(mock) would poison every mock + @property + def tag_name(self): + raise StaleElementReferenceException("stale") + + page = PlainPage(make_browser()) + assert page.is_element_stale(StaleEl()) is True +``` + +- [ ] **Step 2: Run tests to verify they fail** + +Run: `uv run pytest tests/unit/test_page.py -v` +Expected: new tests FAIL with `AttributeError: 'PlainPage' object has no attribute 'click'`; Task-8 tests still pass. + +- [ ] **Step 3: Write the implementation** + +Extend the import block of `src/wabot/page.py` to: + +```python +import logging +import time + +from selenium.common.exceptions import ( + NoSuchElementException, + StaleElementReferenceException, + TimeoutException, + WebDriverException, +) +from selenium.webdriver import ActionChains +from selenium.webdriver.support import expected_conditions as EC +from selenium.webdriver.support.ui import WebDriverWait + +from .fields import CheckField, NullField, SelectField, TextField +``` + +(`screenshot.py` is deliberately NOT imported at module level — it doesn't exist until Task 10; `save_screenshot` below imports it lazily. Create the placeholder now so that lazy import resolves. `src/wabot/screenshot.py`:) + +```python +"""Full-page screenshots. Implemented in the screenshot task.""" +``` + +Append these methods to `class Page`: + +```python + # ---- clicking ----------------------------------------------------- + + def click(self, el, clickable_timeout: float = 10.0) -> bool: + """Click, at a pacing-chosen offset when the policy asks for one.""" + if not el: + LOGGER.warning("refusing to click null element") + return False + try: + size = el.size + except StaleElementReferenceException: + LOGGER.error("failed to click element: stale reference") + return False + if not self._wait_clickable(el, clickable_timeout): + return False + offset = self.pacing.click_offset(size["width"], size["height"]) + try: + if offset is None: + el.click() + else: + x, y = offset # measured from element center (selenium 4) + ActionChains(self.driver).move_to_element_with_offset(el, x, y).click().perform() + return True + except WebDriverException as ex: + LOGGER.error("click failed: %s", ex) + return False + + def _wait_clickable(self, el, timeout: float) -> bool: + deadline = time.monotonic() + timeout + while True: + try: + if el.is_displayed() and el.is_enabled(): + return True + except StaleElementReferenceException: + return False + if time.monotonic() >= deadline: + LOGGER.error("element never became clickable") + return False + time.sleep(0.2) + + def click_and_go(self, el) -> bool: + """Click, accept any alert, then wait for the page to change.""" + if not self.click(el): + return False + self.accept_alert() + return self._wait_for_element_to_go_stale(el) + + def _wait_for_element_to_go_stale(self, el) -> bool: + try: + WebDriverWait(self.driver, PAGE_LOAD_TIMEOUT).until( + lambda _driver: self.is_element_stale(el) + ) + return True + except TimeoutException: + LOGGER.error("timed out waiting for page load") + return False + + def is_element_stale(self, el) -> bool: + try: + _ = el.tag_name + return False + except StaleElementReferenceException: + return True + + # ---- alerts ------------------------------------------------------- + + def accept_alert(self, accept: bool = True, timeout: float = ALERT_TIMEOUT): + """Accept (or dismiss) a JS alert if one appears within ``timeout``. + + Returns the alert text, or False if no alert appeared. There is a + timeout penalty when no alert exists. + """ + try: + WebDriverWait(self.driver, timeout).until(EC.alert_is_present()) + except TimeoutException: + LOGGER.debug("no js alert present") + return False + alert = self.driver.switch_to.alert + text = alert.text + if accept: + alert.accept() + else: + alert.dismiss() + LOGGER.debug("handled js alert: %s", text) + return text + + # ---- form helpers ------------------------------------------------- + + def get_el_value(self, el): + if not el: + return None + return el.get_attribute("value") + + def get_el_text(self, el): + if not el: + return None + return el.text + + def set_el_value(self, el, value, slow_type: bool = False) -> bool: + if not el: + return False + el.clear() + if value is None: + return True + try: + if slow_type: + for char in str(value): + el.send_keys(char) + time.sleep(self.pacing.delay("text") / 10.0) + else: + el.send_keys(value) + except WebDriverException as ex: + LOGGER.error("failed to send keys, element in unknown state: %s", ex) + return False + actual = self.get_el_value(el) + if str(actual) != str(value): + LOGGER.error("field value mismatch: wanted %r, field has %r", value, actual) + return False + return True + + def get_select_value(self, select): + if not select: + LOGGER.error("tried to get select value of null element") + return None + try: + return select.first_selected_option.get_attribute("value") + except NoSuchElementException: + return None + + def set_select_value(self, select, value=None, text=None) -> bool: + if not select: + return False + try: + if value is not None: + select.select_by_value(str(value)) + return True + if text is not None: + select.select_by_visible_text(text) + return True + except (NoSuchElementException, WebDriverException) as ex: + LOGGER.error("failed to set select: %s", ex) + return False + + def set_checkbox(self, el, checked: bool) -> bool: + if not el: + return False + if not el.is_enabled(): + return False + if el.is_selected() != checked: + self.click(el) + return True + + def get_checkbox_value(self, el, ignore_disabled: bool = False): + """True if (enabled and) checked; None for a null element.""" + if not el: + return None + return (ignore_disabled or el.is_enabled()) and el.is_selected() + + def get_el_identifier(self, el): + """A quick human-readable identifier for logging.""" + for attr in ("id", "name", "class"): + value = el.get_attribute(attr) + if value: + return value + return el.tag_name + + # ---- screenshots ---------------------------------------------------- + + def save_screenshot(self, filename: str) -> bool: + from . import screenshot as screenshot_mod + + LOGGER.info("saving full-page screenshot: %s", filename) + return screenshot_mod.save_full_page(self.driver, filename) +``` + +- [ ] **Step 4: Run tests to verify they pass** + +Run: `uv run pytest tests/unit/test_page.py -v` +Expected: 23 passed (12 from Task 8 + 11 new) + +- [ ] **Step 5: Commit** + +```bash +git add src/wabot/page.py src/wabot/screenshot.py tests/unit/test_page.py +git commit -m "feat: Page interaction helpers (paced clicks, forms, alerts, waits)" +``` + +--- +### Task 10: `screenshot.py` — full-page screenshots + +**Files:** +- Modify: `src/wabot/screenshot.py` (replace the Task-9 placeholder) +- Test: `tests/unit/test_screenshot.py` + +- [ ] **Step 1: Write the failing tests** + +`tests/unit/test_screenshot.py`: + +```python +import base64 +from io import BytesIO +from unittest.mock import MagicMock + +from PIL import Image + +from wabot.screenshot import FULL_PAGE_SCREENSHOT, save_full_page + + +def png_bytes(width, height, color=(200, 30, 30)): + buf = BytesIO() + Image.new("RGB", (width, height), color).save(buf, format="PNG") + return buf.getvalue() + + +class TestFirefoxPath: + def test_registers_raw_command_and_decodes_base64(self, tmp_path): + driver = MagicMock(name="driver") + driver.capabilities = {"browserName": "firefox"} + driver.command_executor._commands = {} + driver.execute.return_value = {"value": base64.b64encode(png_bytes(50, 80)).decode()} + out = tmp_path / "shot.png" + assert save_full_page(driver, str(out)) is True + # helper methods only exist on webdriver.Firefox, so the raw + # geckodriver command must have been registered for Remote support + assert driver.command_executor._commands[FULL_PAGE_SCREENSHOT] == ( + "GET", + "/session/$sessionId/moz/screenshot/full", + ) + driver.execute.assert_called_once_with(FULL_PAGE_SCREENSHOT) + assert Image.open(out).size == (50, 80) + + +class TestChromiumPath: + def test_single_viewport_page_is_saved_directly(self, tmp_path): + driver = MagicMock(name="driver") + driver.capabilities = {"browserName": "chrome"} + # page fits in one viewport: total == viewport + driver.execute_script.side_effect = lambda script: { + "return document.body.parentNode.scrollWidth": 100, + "return document.body.parentNode.scrollHeight": 60, + "return document.documentElement.clientWidth": 100, + "return window.innerHeight": 60, + }.get(script, None) + driver.get_screenshot_as_png.return_value = png_bytes(100, 60) + out = tmp_path / "shot.png" + assert save_full_page(driver, str(out)) is True + assert Image.open(out).size == (100, 60) + + def test_tall_page_is_stitched_from_tiles(self, tmp_path): + driver = MagicMock(name="driver") + driver.capabilities = {"browserName": "chrome"} + driver.execute_script.side_effect = lambda script: { + "return document.body.parentNode.scrollWidth": 100, + "return document.body.parentNode.scrollHeight": 150, # 3 tiles of 60 + "return document.documentElement.clientWidth": 100, + "return window.innerHeight": 60, + }.get(script, None) + driver.get_screenshot_as_png.return_value = png_bytes(100, 60) + out = tmp_path / "shot.png" + assert save_full_page(driver, str(out)) is True + assert Image.open(out).size == (100, 150) + # scrolled at least twice beyond the initial position + scroll_calls = [ + c for c in driver.execute_script.call_args_list + if c.args and str(c.args[0]).startswith("window.scrollTo") + ] + assert len(scroll_calls) >= 2 +``` + +- [ ] **Step 2: Run tests to verify they fail** + +Run: `uv run pytest tests/unit/test_screenshot.py -v` +Expected: FAIL — `ImportError: cannot import name 'FULL_PAGE_SCREENSHOT'` + +- [ ] **Step 3: Write the implementation** + +Replace `src/wabot/screenshot.py` entirely: + +```python +"""Full-page screenshots for both browser families. + +Firefox: geckodriver has a native endpoint, but selenium only exposes it +on ``webdriver.Firefox`` — not on ``Remote`` — so the raw command is +registered by hand and works for both. + +Chromium: no full-page endpoint exists; the page is captured viewport by +viewport while scrolling, then stitched with PIL (port of the legacy +workaround, using in-memory PNGs instead of temp files). +""" + +from __future__ import annotations + +import base64 +import logging +import time +from io import BytesIO + +from PIL import Image + +LOGGER = logging.getLogger("wabot") + +FULL_PAGE_SCREENSHOT = "fullPageScreenshot" +_SCROLL_SETTLE_SECONDS = 0.2 + + +def save_full_page(driver, filename: str) -> bool: + """Save a full-page PNG of the current page to ``filename``.""" + if driver.capabilities.get("browserName") == "firefox": + return _firefox_full_page(driver, filename) + return _stitch_chromium(driver, filename) + + +def _firefox_full_page(driver, filename: str) -> bool: + driver.command_executor._commands.setdefault( + FULL_PAGE_SCREENSHOT, ("GET", "/session/$sessionId/moz/screenshot/full") + ) + b64 = driver.execute(FULL_PAGE_SCREENSHOT)["value"] + with open(filename, "wb") as fp: + fp.write(base64.b64decode(b64)) + return True + + +def _stitch_chromium(driver, filename: str) -> bool: + total_width = driver.execute_script("return document.body.parentNode.scrollWidth") + total_height = driver.execute_script("return document.body.parentNode.scrollHeight") + viewport_width = driver.execute_script("return document.documentElement.clientWidth") + viewport_height = driver.execute_script("return window.innerHeight") + + if total_width <= viewport_width and total_height <= viewport_height: + with open(filename, "wb") as fp: + fp.write(driver.get_screenshot_as_png()) + return True + + stitched = Image.new("RGB", (total_width, total_height)) + y = 0 + while y < total_height: + x = 0 + while x < total_width: + driver.execute_script(f"window.scrollTo({x}, {y})") + time.sleep(_SCROLL_SETTLE_SECONDS) + tile = Image.open(BytesIO(driver.get_screenshot_as_png())) + # the last row/column can't scroll a full viewport: paste aligned + # to the bottom/right edge instead of duplicating content + paste_x = min(x, total_width - viewport_width) + paste_y = min(y, total_height - viewport_height) + stitched.paste(tile, (max(paste_x, 0), max(paste_y, 0))) + x += viewport_width + y += viewport_height + stitched.save(filename) + LOGGER.debug("stitched %sx%s screenshot -> %s", total_width, total_height, filename) + return True +``` + +- [ ] **Step 4: Run tests to verify they pass** + +Run: `uv run pytest tests/unit/test_screenshot.py -v` +Expected: 3 passed. Also run `uv run pytest` — all unit tests still pass. + +- [ ] **Step 5: Commit** + +```bash +git add src/wabot/screenshot.py tests/unit/test_screenshot.py +git commit -m "feat: full-page screenshots (firefox raw command, chromium PIL stitch)" +``` + +--- + +### Task 11: `browser.py` — the Browser facade + +**Files:** +- Create: `src/wabot/browser.py` +- Test: `tests/unit/test_browser.py` + +Naming note: the spec says "`reset_good_status()` retained"; under the approved free-API-redesign decision it is renamed to the cleaner `reset()` (same semantics: clears the broken-state flag). This is the only intentional naming deviation from the spec. + +- [ ] **Step 1: Write the failing tests** + +`tests/unit/test_browser.py`: + +```python +from unittest.mock import MagicMock + +import pytest +from selenium.common.exceptions import WebDriverException + +from wabot.browser import Browser +from wabot.pacing import HumanPacing, NoPacing +from wabot.page import Page + + +class Login(Page): + elements = {"username": ("el", ("id", "username"))} + + def do_login(self): + return "logged-in" + + def explode(self): + raise WebDriverException("browser gone") + + +class Unverifiable(Page): + def verify(self): + return False + + +@pytest.fixture +def bot(): + return Browser(MagicMock(name="driver"), pacing=NoPacing()) + + +class TestConstruction: + def test_good_flag_starts_true(self, bot): + assert bot.good is True + + def test_default_pacing_is_human(self): + assert isinstance(Browser(MagicMock()).pacing, HumanPacing) + + +class TestSetPage: + def test_sets_and_returns_true(self, bot): + assert bot.set_page(Login) is True + assert isinstance(bot.page, Login) + + def test_failed_verify_refuses_switch(self, bot): + bot.set_page(Login) + assert bot.set_page(Unverifiable) is False + assert isinstance(bot.page, Login) # unchanged + + +class TestDelegation: + def test_getattr_delegates_to_page(self, bot): + bot.set_page(Login) + assert bot.do_login() == "logged-in" + + def test_getitem_delegates_to_page(self, bot): + bot.set_page(Login) + el = bot["username"] + assert el.name == "username" + + def test_getattr_without_page_raises(self, bot): + with pytest.raises(AttributeError): + bot.do_login + + +class TestPerform: + def test_returns_method_result(self, bot): + bot.set_page(Login) + assert bot.perform("do_login") == "logged-in" + + def test_missing_method_returns_none(self, bot): + bot.set_page(Login) + assert bot.perform("nope") is None + + def test_webdriver_exception_flips_good(self, bot): + bot.set_page(Login) + assert bot.perform("explode") is None + assert bot.good is False + + def test_broken_state_refuses_actions_until_reset(self, bot): + bot.set_page(Login) + bot.perform("explode") + assert bot.perform("do_login") is None # refused + assert bot.do_login() is None # __getattr__ path also refused + bot.reset() + assert bot.perform("do_login") == "logged-in" + + +class TestQuit: + def test_quit_quits_driver_and_removes_session(self): + driver, store = MagicMock(), MagicMock() + bot = Browser(driver, session_name="s1", store=store) + bot.quit() + driver.quit.assert_called_once_with() + store.remove.assert_called_once_with("s1") + + def test_quit_without_session_only_quits(self): + driver = MagicMock() + Browser(driver).quit() + driver.quit.assert_called_once_with() +``` + +- [ ] **Step 2: Run tests to verify they fail** + +Run: `uv run pytest tests/unit/test_browser.py -v` +Expected: FAIL — `ModuleNotFoundError: No module named 'wabot.browser'` + +- [ ] **Step 3: Write the implementation** + +`src/wabot/browser.py`: + +```python +"""Browser: the facade consumers drive. + +Holds the webdriver plus a *current page* object. Unknown attribute +access is delegated to the current page (``bot.login()`` invokes +``bot.page.login()``); ``bot[key]`` resolves elements on it. Any +``perform()`` failure flips ``good = False``, after which page actions +are refused until ``reset()`` — "log and refuse, don't crash". +""" + +from __future__ import annotations + +import logging + +from selenium.common.exceptions import WebDriverException + +from .pacing import HumanPacing + +LOGGER = logging.getLogger("wabot") + +# attributes that must never be delegated to the page +_OWN_ATTRS = frozenset( + {"driver", "pacing", "session_name", "good", "page", "_store"} +) + + +def _refused(*_args, **_kwargs): + return None + + +class Browser: + def __init__(self, driver, *, pacing=None, session_name=None, store=None): + self.driver = driver + self.pacing = pacing if pacing is not None else HumanPacing() + self.session_name = session_name + self._store = store + self.good = True + self.page = None + + def set_page(self, page_cls) -> bool: + """Instantiate ``page_cls`` and make it current if its verify() passes.""" + page = page_cls(self) + if not page.verify(): + LOGGER.error("failed to verify page: %s", page_cls.__name__) + return False + self.page = page + return True + + def __getattr__(self, name): + if name in _OWN_ATTRS or name.startswith("__"): + raise AttributeError(name) + if self.page is None: + raise AttributeError( + f"{name!r}: no current page — call set_page() first" + ) + if not self.good: + LOGGER.warning("broken state — refusing page action %r (call reset())", name) + return _refused + return getattr(self.page, name) + + def __getitem__(self, key): + return self.page[key] + + def perform(self, method: str, *args, **kwargs): + """Invoke a page method, trapping failures instead of raising.""" + if not self.good: + LOGGER.warning("broken state — refusing %r (call reset())", method) + return None + try: + fn = getattr(self.page, method) + except AttributeError: + LOGGER.error("page %s has no action %r", type(self.page).__name__, method) + return None + try: + return fn(*args, **kwargs) + except WebDriverException: + LOGGER.exception("page action %r failed; flipping good=False", method) + self.good = False + return None + + def reset(self) -> None: + """Clear the broken-state flag after an exception.""" + self.good = True + + def quit(self) -> None: + """Quit the browser; forget the saved session if there is one.""" + try: + self.driver.quit() + finally: + if self._store is not None and self.session_name: + self._store.remove(self.session_name) +``` + +- [ ] **Step 4: Run tests to verify they pass** + +Run: `uv run pytest tests/unit/test_browser.py -v` +Expected: 13 passed + +- [ ] **Step 5: Commit** + +```bash +git add src/wabot/browser.py tests/unit/test_browser.py +git commit -m "feat: Browser facade with working refuse-after-exception guard" +``` + +--- + +### Task 12: `__init__.py` — public API: browser(), sessions(), destroy() + +**Files:** +- Modify: `src/wabot/__init__.py` +- Test: `tests/unit/test_api.py` + +- [ ] **Step 1: Write the failing tests** + +`tests/unit/test_api.py`. All webdriver construction is monkeypatched — these tests exercise the wiring, not selenium: + +```python +from unittest.mock import MagicMock + +import pytest + +import wabot +from wabot.sessions import SessionRecord, SessionStore + + +@pytest.fixture +def store(tmp_path): + return SessionStore(path=tmp_path / "sessions.json") + + +@pytest.fixture +def fake_selenium(monkeypatch): + """Patch every constructor that would talk to a real browser.""" + fakes = MagicMock(name="fakes") + fakes.Remote.return_value.session_id = "new-session-id" + monkeypatch.setattr(wabot, "_new_remote", fakes.Remote) + monkeypatch.setattr(wabot, "_new_local", fakes.Local) + monkeypatch.setattr(wabot, "attach", fakes.attach) + monkeypatch.setattr(wabot, "service_alive", fakes.service_alive) + fakes.managed = MagicMock(name="ManagedService_instance") + fakes.managed.ensure_running.return_value = "http://127.0.0.1:7777" + fakes.managed.pid, fakes.managed.port = 4321, 7777 + monkeypatch.setattr(wabot, "ManagedService", MagicMock(return_value=fakes.managed)) + return fakes + + +class TestEphemeral: + def test_no_session_no_host_uses_local_driver(self, fake_selenium, store): + bot = wabot.browser(store=store) + fake_selenium.Local.assert_called_once() + assert bot.driver is fake_selenium.Local.return_value + assert store.names() == [] # nothing persisted + + def test_no_session_with_host_uses_remote_without_saving(self, fake_selenium, store): + wabot.browser(host="http://grid:4444", store=store) + fake_selenium.Remote.assert_called_once() + assert fake_selenium.Remote.call_args.args[0] == "http://grid:4444" + assert store.names() == [] + + +class TestPersistentCreate: + def test_managed_service_created_and_recorded(self, fake_selenium, store): + bot = wabot.browser(session="s1", browser="chromium", store=store) + record = store.get("s1") + assert record == SessionRecord( + name="s1", + executor_url="http://127.0.0.1:7777", + session_id="new-session-id", + browser="chromium", + created_at=record.created_at, + service_pid=4321, + service_port=7777, + ) + assert bot.session_name == "s1" + + def test_external_host_records_no_pid(self, fake_selenium, store, monkeypatch): + external = MagicMock() + external.ensure_running.return_value = "http://grid:4444" + del external.pid # ExternalServer has no pid/port attributes + del external.port + monkeypatch.setattr(wabot, "ExternalServer", MagicMock(return_value=external)) + wabot.browser(session="s1", host="http://grid:4444", store=store) + record = store.get("s1") + assert record.executor_url == "http://grid:4444" + assert record.service_pid is None and record.service_port is None + + +class TestReattach: + def make_record(self, store): + from datetime import datetime, timezone + + store.save( + SessionRecord( + name="s1", + executor_url="http://127.0.0.1:7777", + session_id="old-session", + browser="chromium", + created_at=datetime.now(timezone.utc).isoformat(), + ) + ) + + def test_live_record_is_reattached(self, fake_selenium, store): + self.make_record(store) + fake_selenium.service_alive.return_value = True + bot = wabot.browser(session="s1", store=store) + fake_selenium.attach.assert_called_once_with( + "http://127.0.0.1:7777", "old-session", "chromium" + ) + assert bot.driver is fake_selenium.attach.return_value + fake_selenium.Remote.assert_not_called() + + def test_dead_record_falls_through_to_fresh_creation(self, fake_selenium, store): + self.make_record(store) + fake_selenium.service_alive.return_value = True + fake_selenium.attach.return_value = None # session gone + wabot.browser(session="s1", store=store) + fake_selenium.Remote.assert_called_once() + assert store.get("s1").session_id == "new-session-id" + + def test_dead_server_skips_attach_entirely(self, fake_selenium, store): + self.make_record(store) + fake_selenium.service_alive.return_value = False + wabot.browser(session="s1", store=store) + fake_selenium.attach.assert_not_called() + fake_selenium.Remote.assert_called_once() + + +class TestHousekeeping: + def test_sessions_lists_names(self, store): + assert wabot.sessions(store=store) == [] + + def test_destroy_quits_kills_and_removes(self, fake_selenium, store, monkeypatch): + from datetime import datetime, timezone + + stop = MagicMock() + monkeypatch.setattr(wabot, "stop_service", stop) + store.save( + SessionRecord( + name="s1", + executor_url="http://127.0.0.1:7777", + session_id="old-session", + browser="chromium", + created_at=datetime.now(timezone.utc).isoformat(), + service_pid=4321, + ) + ) + fake_selenium.service_alive.return_value = True + assert wabot.destroy("s1", store=store) is True + fake_selenium.attach.return_value.quit.assert_called_once_with() + stop.assert_called_once_with(4321) + assert store.get("s1") is None + + def test_destroy_missing_returns_false(self, store): + assert wabot.destroy("nope", store=store) is False + + +class TestPublicSurface: + def test_all_exports_exist(self): + for name in wabot.__all__: + assert hasattr(wabot, name), name +``` + +- [ ] **Step 2: Run tests to verify they fail** + +Run: `uv run pytest tests/unit/test_api.py -v` +Expected: FAIL — `AttributeError: module 'wabot' has no attribute 'browser'` + +- [ ] **Step 3: Write the implementation** + +Replace `src/wabot/__init__.py` entirely: + +```python +"""wabot — stateful Selenium browser automation with reattachable sessions. + +Quickstart:: + + import wabot + + bot = wabot.browser(session="scraper1", browser="firefox") + # ... process exits; the browser stays open ... + bot = wabot.browser(session="scraper1", browser="firefox") # reattaches +""" + +from __future__ import annotations + +import logging +from datetime import datetime, timezone + +from selenium import webdriver +from selenium.common.exceptions import WebDriverException + +from .browser import Browser +from .fields import CheckField, NullField, PageObject, SelectField, TextField +from .hosts import ( + ExternalServer, + ManagedService, + attach, + build_options, + normalize_browser, + service_alive, + stop_service, +) +from .page import Page +from .pacing import HumanPacing, NoPacing +from .sessions import SessionRecord, SessionStore + +__all__ = [ + "Browser", + "CheckField", + "HumanPacing", + "NoPacing", + "NullField", + "Page", + "PageObject", + "SelectField", + "SessionRecord", + "SessionStore", + "TextField", + "browser", + "destroy", + "sessions", +] + +LOGGER = logging.getLogger("wabot") + + +def _new_remote(url: str, options): + return webdriver.Remote(command_executor=url, options=options) + + +def _new_local(browser_name: str, options): + if browser_name == "chromium": + return webdriver.Chrome(options=options) + return webdriver.Firefox(options=options) + + +def browser( + session: str | None = None, + browser: str = "firefox", + host: str | None = None, + *, + headless: bool = False, + user_agent: str | None = None, + pacing=None, + store: SessionStore | None = None, +) -> Browser: + """Create a Browser, reattaching to a saved session when one exists. + + Args: + session: Persistence name. None (default) = ephemeral: the browser + is not saved and (without ``host``) dies with this process. + browser: "firefox" or "chromium" ("chrome" is an alias). + host: URL of an external WebDriver server (Selenium Grid or a bare + driver). Without it, persistent sessions get a wabot-managed + detached driver service. + headless: Run the browser headless. + user_agent: Override the browser's user agent (default: browser's own). + pacing: A pacing policy; defaults to HumanPacing (stealth delays). + store: SessionStore override, mainly for tests. + """ + browser_name = normalize_browser(browser) + options = build_options(browser_name, headless=headless, user_agent=user_agent) + + if session is None: + if host: + return Browser(_new_remote(host, options), pacing=pacing) + return Browser(_new_local(browser_name, options), pacing=pacing) + + store = store if store is not None else SessionStore() + record = store.get(session) + if record is not None: + if service_alive(record.executor_url): + driver = attach(record.executor_url, record.session_id, record.browser) + if driver is not None: + LOGGER.info("reattached to session %r", session) + return Browser(driver, pacing=pacing, session_name=session, store=store) + LOGGER.warning("saved session %r is dead; creating a fresh browser", session) + store.remove(session) + + host_obj = ExternalServer(host) if host else ManagedService(browser_name) + url = host_obj.ensure_running() + driver = _new_remote(url, options) + store.save( + SessionRecord( + name=session, + executor_url=url, + session_id=driver.session_id, + browser=browser_name, + created_at=datetime.now(timezone.utc).isoformat(), + service_pid=getattr(host_obj, "pid", None), + service_port=getattr(host_obj, "port", None), + ) + ) + LOGGER.info("created persistent session %r on %s", session, url) + return Browser(driver, pacing=pacing, session_name=session, store=store) + + +def sessions(store: SessionStore | None = None) -> list[str]: + """Names of saved (non-stale) sessions.""" + return (store if store is not None else SessionStore()).names() + + +def destroy(name: str, store: SessionStore | None = None) -> bool: + """Quit a saved session's browser, stop its managed service, forget it.""" + store = store if store is not None else SessionStore() + record = store.get(name) + if record is None: + return False + if service_alive(record.executor_url): + driver = attach(record.executor_url, record.session_id, record.browser) + if driver is not None: + try: + driver.quit() + except WebDriverException as ex: + LOGGER.warning("quit failed while destroying %r: %s", name, ex) + if record.service_pid: + stop_service(record.service_pid) + store.remove(name) + return True +``` + +- [ ] **Step 4: Run tests to verify they pass** + +Run: `uv run pytest tests/unit/test_api.py -v` +Expected: 11 passed + +Run: `uv run pytest` +Expected: full unit suite passes (97 tests: pacing 6, sessions 8, hosts 17, reattach 5, fields 11, page 23, screenshot 3, browser 13, api 11) + +- [ ] **Step 5: Commit** + +```bash +git add src/wabot/__init__.py tests/unit/test_api.py +git commit -m "feat: public API — wabot.browser()/sessions()/destroy() with reattach flow" +``` + +--- +### Task 13: Integration fixtures — static HTML + local server + conftest + +**Files:** +- Create: `tests/fixtures/login.html`, `tests/fixtures/welcome.html` +- Create: `tests/integration/conftest.py` + +- [ ] **Step 1: Create the HTML fixtures** + +`tests/fixtures/login.html`: + +```html + + +Login + +

Test Login

+
+ + + + +
+ + +``` + +`tests/fixtures/welcome.html`: + +```html + + +Welcome + +

Welcome!

+

You made it.

+ + +``` + +- [ ] **Step 2: Write the conftest** + +`tests/integration/conftest.py`: + +```python +"""Shared fixtures for real-browser integration tests. + +Every test in this directory is marked ``integration`` automatically and +runs with NoPacing (human delays would make the suite glacial). Browsers +run headless against static HTML served from tests/fixtures. +""" + +import functools +import http.server +import threading +from pathlib import Path + +import pytest + +FIXTURES = Path(__file__).parent.parent / "fixtures" +BROWSERS = ["firefox", "chromium"] + + +def pytest_collection_modifyitems(items): + for item in items: + item.add_marker(pytest.mark.integration) + + +@pytest.fixture(scope="session") +def site_url(): + """Serve tests/fixtures over HTTP on an ephemeral port.""" + handler = functools.partial( + http.server.SimpleHTTPRequestHandler, directory=str(FIXTURES) + ) + server = http.server.ThreadingHTTPServer(("127.0.0.1", 0), handler) + thread = threading.Thread(target=server.serve_forever, daemon=True) + thread.start() + yield f"http://127.0.0.1:{server.server_address[1]}" + server.shutdown() + + +@pytest.fixture +def store(tmp_path): + from wabot.sessions import SessionStore + + return SessionStore(path=tmp_path / "sessions.json") +``` + +- [ ] **Step 3: Verify collection behaves** + +Run: `uv run pytest --collect-only tests/integration` +Expected: collects 0 tests so far, no errors. + +Run: `uv run pytest` +Expected: unit suite unchanged; integration dir contributes nothing yet. + +- [ ] **Step 4: Commit** + +```bash +git add tests/fixtures tests/integration/conftest.py +git commit -m "test: integration fixtures — static site server, auto-marking conftest" +``` + +--- + +### Task 14: Integration — ephemeral browsing through the page-object API + +**Files:** +- Create: `tests/integration/test_ephemeral.py` + +- [ ] **Step 1: Write the test** + +`tests/integration/test_ephemeral.py`: + +```python +"""Drive the full page-object stack against a real headless browser.""" + +import pytest + +import wabot + +from .conftest import BROWSERS + + +class LoginPage(wabot.Page): + elements = { + "title": ("el", ("id", "page-title")), + "username": ("el", ("id", "username")), + "state": ("select", ("id", "state")), + "agree": ("checkbox", ("id", "agree")), + "submit": ("el", ("id", "submit-btn")), + } + + def verify(self): + return self.driver.title == "Login" + + def log_in(self, username): + self["username"].set_value(username) + self["state"].set_value(value="CO") + self["agree"].set_checked(True) + return self["submit"].click_and_go() + + +class WelcomePage(wabot.Page): + elements = {"greeting": ("el", ("id", "greeting"))} + + def verify(self): + return self.driver.title == "Welcome" + + +@pytest.mark.parametrize("browser_name", BROWSERS) +def test_login_flow_end_to_end(browser_name, site_url): + bot = wabot.browser(browser=browser_name, headless=True, pacing=wabot.NoPacing()) + try: + bot.driver.get(f"{site_url}/login.html") + assert bot.set_page(LoginPage) is True + assert bot.log_in("mathew") is True + assert bot.set_page(WelcomePage) is True + assert bot["greeting"].text == "You made it." + assert bot.good is True + finally: + bot.driver.quit() + + +@pytest.mark.parametrize("browser_name", BROWSERS) +def test_full_page_screenshot(browser_name, site_url, tmp_path): + bot = wabot.browser(browser=browser_name, headless=True, pacing=wabot.NoPacing()) + try: + bot.driver.get(f"{site_url}/login.html") + bot.set_page(LoginPage) + out = tmp_path / "shot.png" + assert bot.save_screenshot(str(out)) is True + assert out.stat().st_size > 0 + finally: + bot.driver.quit() +``` + +- [ ] **Step 2: Run the integration tests** + +Run: `uv run pytest -m integration tests/integration/test_ephemeral.py -v` +Expected: 4 passed (firefox + chromium × 2 tests). First firefox run may download geckodriver via Selenium Manager (network). If a browser is missing on the machine, that parametrization may error — investigate before proceeding; do not mark xfail. + +Run: `uv run pytest` +Expected: unit suite only (integration deselected by default). + +- [ ] **Step 3: Commit** + +```bash +git add tests/integration/test_ephemeral.py +git commit -m "test: end-to-end page-object flow on real headless browsers" +``` + +--- + +### Task 15: Integration — the flagship: cross-process session pickup + +**Files:** +- Create: `tests/integration/test_persistence.py` + +- [ ] **Step 1: Write the test** + +`tests/integration/test_persistence.py`: + +```python +"""The flagship feature: a second Python PROCESS picks up a saved browser.""" + +import json +import subprocess +import sys +import textwrap + +import pytest + +import wabot + +from .conftest import BROWSERS + +REATTACH_SCRIPT = textwrap.dedent( + """ + import json, sys + import wabot + from wabot.sessions import SessionStore + + store = SessionStore(path=sys.argv[1]) + bot = wabot.browser(session=sys.argv[2], pacing=wabot.NoPacing(), store=store) + value = bot.driver.find_element("id", "username").get_attribute("value") + print(json.dumps({"url": bot.driver.current_url, "username_value": value})) + """ +) + + +@pytest.mark.parametrize("browser_name", BROWSERS) +def test_second_process_picks_up_saved_browser(browser_name, site_url, store, tmp_path): + bot = wabot.browser( + session="flagship", browser=browser_name, headless=True, + pacing=wabot.NoPacing(), store=store, + ) + try: + bot.driver.get(f"{site_url}/login.html") + bot.driver.find_element("id", "username").send_keys("persisted-value") + + # simulate "the process exits": drop our handle without quitting + record = store.get("flagship") + assert record is not None and record.service_pid is not None + + result = subprocess.run( + [sys.executable, "-c", REATTACH_SCRIPT, str(store.path), "flagship"], + capture_output=True, text=True, timeout=120, check=True, + ) + payload = json.loads(result.stdout.strip().splitlines()[-1]) + assert payload["url"] == f"{site_url}/login.html" + assert payload["username_value"] == "persisted-value" + finally: + assert wabot.destroy("flagship", store=store) is True + + +def test_destroyed_session_is_really_gone(site_url, store): + from wabot.hosts import service_alive + + bot = wabot.browser( + session="doomed", browser="chromium", headless=True, + pacing=wabot.NoPacing(), store=store, + ) + bot.driver.get(f"{site_url}/login.html") + record = store.get("doomed") + assert wabot.destroy("doomed", store=store) is True + assert store.get("doomed") is None + assert wabot.destroy("doomed", store=store) is False + # the managed driver service was SIGTERMed + import time + + deadline = time.monotonic() + 10 + while time.monotonic() < deadline and service_alive(record.executor_url): + time.sleep(0.2) + assert not service_alive(record.executor_url) +``` + +- [ ] **Step 2: Run the tests** + +Run: `uv run pytest -m integration tests/integration/test_persistence.py -v` +Expected: 3 passed. This proves: managed service spawned detached → session recorded as JSON → a *fresh Python process* reattached by name and saw the same page **and the typed-in field value** → destroy() cleaned everything up. + +- [ ] **Step 3: Commit** + +```bash +git add tests/integration/test_persistence.py +git commit -m "test: flagship cross-process browser pickup + destroy cleanup" +``` + +--- + +### Task 16: Integration — external-server mode + +**Files:** +- Create: `tests/integration/test_external_server.py` + +- [ ] **Step 1: Write the test** + +A manually spawned chromedriver plays the "external Selenium server the user runs" — same wire protocol, no Java Grid needed: + +`tests/integration/test_external_server.py`: + +```python +"""host= mode: wabot connects to a WebDriver server it does not manage.""" + +import shutil +import subprocess +import time + +import pytest + +import wabot +from wabot.hosts import service_alive + + +@pytest.fixture +def external_chromedriver(): + binary = shutil.which("chromedriver") + if binary is None: + pytest.skip("chromedriver not on PATH") + proc = subprocess.Popen( + [binary, "--port=19515"], + stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, + ) + url = "http://127.0.0.1:19515" + deadline = time.monotonic() + 15 + while time.monotonic() < deadline and not service_alive(url): + time.sleep(0.2) + assert service_alive(url), "chromedriver never came up" + yield url + proc.terminate() + proc.wait(timeout=10) + + +def test_persistent_session_on_external_host(external_chromedriver, site_url, store): + bot = wabot.browser( + session="ext", browser="chromium", host=external_chromedriver, + headless=True, pacing=wabot.NoPacing(), store=store, + ) + bot.driver.get(f"{site_url}/welcome.html") + record = store.get("ext") + assert record.executor_url == external_chromedriver + assert record.service_pid is None # not ours to kill + + # same process, second browser() call: must reattach, not create + bot2 = wabot.browser(session="ext", store=store, pacing=wabot.NoPacing()) + assert bot2.driver.session_id == record.session_id + assert bot2.driver.current_url == f"{site_url}/welcome.html" + + bot2.driver.quit() + store.remove("ext") + + +def test_ephemeral_on_external_host(external_chromedriver, site_url, store): + bot = wabot.browser( + host=external_chromedriver, browser="chromium", + headless=True, pacing=wabot.NoPacing(), store=store, + ) + bot.driver.get(f"{site_url}/welcome.html") + assert store.names() == [] # session=None never persists + bot.driver.quit() +``` + +- [ ] **Step 2: Run the tests** + +Run: `uv run pytest -m integration tests/integration/test_external_server.py -v` +Expected: 2 passed + +Run: `uv run pytest -m integration` +Expected: entire integration suite passes (9 tests) + +- [ ] **Step 3: Commit** + +```bash +git add tests/integration/test_external_server.py +git commit -m "test: external WebDriver server mode (persist + ephemeral)" +``` + +--- + +### Task 17: Sphinx documentation + +**Files:** +- Create: `docs/conf.py`, `docs/index.rst`, `docs/quickstart.rst`, `docs/sessions.rst`, `docs/page-objects.rst`, `docs/api.rst` + +- [ ] **Step 1: Write the Sphinx config** + +`docs/conf.py`: + +```python +project = "wabot" +author = "Mathew Guest" +release = "0.2.0" + +extensions = [ + "sphinx.ext.autodoc", + "sphinx.ext.napoleon", +] + +# sphinx_rtd_theme registers itself via entry point — do NOT add to extensions +html_theme = "sphinx_rtd_theme" + +autodoc_member_order = "bysource" +napoleon_google_docstring = True +``` + +(No `sys.path` hack needed: `uv run sphinx-build` runs inside the synced venv where wabot is installed.) + +- [ ] **Step 2: Write the doc pages** + +`docs/index.rst`: + +```rst +wabot +===== + +Stateful Selenium browser automation with sessions that survive the +Python process. + +.. toctree:: + :maxdepth: 2 + + quickstart + sessions + page-objects + api +``` + +`docs/quickstart.rst`: + +```rst +Quickstart +========== + +Install +------- + +.. code-block:: bash + + uv add wabot # or: pip install . + +You also need Firefox or Chromium. Driver binaries (geckodriver / +chromedriver) are found on PATH or downloaded automatically by Selenium +Manager. + +First scrape +------------ + +.. code-block:: python + + import wabot + + class Login(wabot.Page): + elements = { + "username": ("el", ("id", "username")), + "password": ("el", ("id", "password")), + "submit": ("el", ("id", "submit-btn")), + } + + def verify(self): + return "Login" in self.driver.title + + def log_in(self, user, password): + self["username"].set_value(user) + self["password"].set_value(password) + return self["submit"].click_and_go() + + bot = wabot.browser(browser="firefox", headless=True) + bot.driver.get("https://example.com/login") + bot.set_page(Login) + bot.log_in("me", "secret") + +By default wabot paces itself like a human (multi-second random delays, +randomized click positions). Pass ``pacing=wabot.NoPacing()`` for full +speed on sites you trust. +``` + +`docs/sessions.rst`: + +```rst +Persistent sessions +=================== + +The flagship feature: a browser created with a ``session`` name survives +the Python process, and any later process can pick it up by name. + +.. code-block:: python + + import wabot + + bot = wabot.browser(session="scraper1", browser="firefox") + bot.driver.get("https://example.com/login") + # ... log in, then the process exits; the browser stays open ... + + # hours later, a different process: + bot = wabot.browser(session="scraper1", browser="firefox") + print(bot.driver.current_url) # still logged in + +How it works +------------ + +A session is just ``(executor_url, session_id)`` stored as JSON in the +platform data dir. Nothing is pickled. The browser is hosted by a driver +service that outlives Python: + +* **Managed (default):** wabot spawns chromedriver/geckodriver as a + detached process and records its port and pid. +* **External:** pass ``host="http://localhost:4444"`` to use a Selenium + Grid or standalone server you run yourself. wabot then never kills the + server — only the session. + +Reattaching constructs a driver that *adopts* the saved session id +instead of creating a new session. Dead sessions (browser closed, server +gone, record older than 3 days) are pruned automatically and a fresh +browser is created instead. + +Housekeeping +------------ + +.. code-block:: python + + wabot.sessions() # ["scraper1", ...] + wabot.destroy("scraper1") # quit browser, stop managed service, forget + +Selenium Grid caveat +-------------------- + +Grid 4 kills idle sessions after **5 minutes** by default. If your +process sleeps longer between reattaches, raise it:: + + java -jar selenium-server.jar standalone --session-timeout 86400 +``` + +`docs/page-objects.rst`: + +```rst +Page objects +============ + +Model each website page as a :class:`wabot.Page` subclass with a +class-level ``elements`` map; drive them through one +:class:`wabot.Browser`. + +.. code-block:: python + + class Inventory(wabot.Page): + elements = { + "search": ("el", ("id", "search-box")), + "region": ("select", ("id", "region")), + "instock": ("checkbox", ("id", "in-stock-only")), + "rows": ("els", ("css selector", "table#items tr")), + } + +Element types +------------- + +=========== ========================================= ===================== +Type key Returned wrapper Main methods +=========== ========================================= ===================== +``el`` :class:`wabot.TextField` get_value / set_value +``select`` :class:`wabot.SelectField` set_value(value|text) +``checkbox`` :class:`wabot.CheckField` get/set_checked +``els`` list of raw WebElements (selenium API) +a class your :class:`wabot.PageObject` subclass whatever you define +=========== ========================================= ===================== + +Missing elements come back as a falsy :class:`wabot.NullField`, so guard +with ``if el:`` instead of try/except. + +Inheritance +----------- + +Element lookup walks the MRO: subclasses inherit the parent map and can +override individual keys — useful for sites with shared page chrome. + +Verification and delegation +--------------------------- + +``Browser.set_page(PageClass)`` refuses the switch if the page's +``verify()`` returns False. Once set, unknown attributes on the Browser +delegate to the current page: ``bot.log_in(...)`` calls +``bot.page.log_in(...)``. After any trapped failure the Browser refuses +further actions until ``bot.reset()``. +``` + +`docs/api.rst`: + +```rst +API reference +============= + +.. automodule:: wabot + :members: browser, sessions, destroy + +.. autoclass:: wabot.Browser + :members: + +.. autoclass:: wabot.Page + :members: + +.. automodule:: wabot.fields + :members: PageObject, TextField, SelectField, CheckField, NullField + +.. automodule:: wabot.pacing + :members: + +.. automodule:: wabot.sessions + :members: + +.. automodule:: wabot.hosts + :members: ExternalServer, ManagedService, attach, service_alive +``` + +- [ ] **Step 3: Build and verify** + +Run: `uv run sphinx-build -M html docs docs/_build -W` +Expected: `build succeeded` with zero warnings (`-W` turns warnings into errors). Open `docs/_build/html/index.html` if in doubt. + +- [ ] **Step 4: Commit** + +```bash +git add docs/conf.py docs/index.rst docs/quickstart.rst docs/sessions.rst docs/page-objects.rst docs/api.rst +git commit -m "docs: sphinx site — quickstart, sessions guide, page objects, API ref" +``` + +--- + +### Task 18: Finish — README, CLAUDE.md, lint, full verification + +**Files:** +- Modify: `README.md` (replace entirely) +- Modify: `CLAUDE.md` (update commands + architecture sections) + +- [ ] **Step 1: Rewrite README.md** + +```markdown +# WABot — Web Automator Bot + +Stateful Selenium browser automation for Python. Model pages as classes, +drive them through one facade, and — the flagship — keep the browser +alive after your process exits so another Python process can pick it up +by name. + +```python +import wabot + +bot = wabot.browser(session="scraper1", browser="firefox") +bot.driver.get("https://example.com/login") +# ... log in; process exits; browser stays open ... + +# a different process, later: +bot = wabot.browser(session="scraper1", browser="firefox") # reattaches +``` + +Human-like pacing (random delays, randomized click positions) is on by +default; pass `pacing=wabot.NoPacing()` for full speed. + +## Development + +```bash +uv sync # install everything +uv run pytest # unit tests (fast, no browsers) +uv run pytest -m integration # real headless-browser tests +uv run ruff check . && uv run ruff format --check . +uv run sphinx-build -M html docs docs/_build # docs +``` + +Docs: quickstart, persistent-sessions guide, and API reference under +`docs/` (build locally with the command above). +``` + +- [ ] **Step 2: Update CLAUDE.md** + +Replace the "What this is" third paragraph and the whole "Architecture" and "Gotchas" sections to match the new reality. Replace the sentence about no test suite with: + +```markdown +Tooling: uv (`uv sync`), pytest (`uv run pytest`; real-browser suite via +`uv run pytest -m integration`), ruff, Sphinx (`uv run sphinx-build -M html docs docs/_build`). +Selenium 4 (pinned <5: session reattach relies on stable 4.x internals). +``` + +In the Architecture section: `BrowserProxy` → `Browser` (`src/wabot/browser.py`), add `pacing.py` (HumanPacing/NoPacing policies), and replace the dill/pickle description with: sessions persist as JSON records `(executor_url, session_id)` in `platformdirs.user_data_dir("wabot")/sessions.json`; `hosts.py` spawns detached driver services (`ManagedService`) or connects to external ones (`ExternalServer`); `ReattachingRemote.start_session` adopts the saved session id. Delete the Gotchas bullets about `LOGGER.trace`, undeclared deps, and star exports (all fixed); keep the Gotcha that external hosts must outlive Python and add: geckodriver reports `ready:false` while its single session is active, so liveness checks ignore the `ready` flag. + +- [ ] **Step 3: Full verification** + +Run: `uv run ruff check . && uv run ruff format --check .` +Expected: clean (fix any complaints — they will be import order or line length). + +Run: `uv run pytest` +Expected: all unit tests pass. + +Run: `uv run pytest -m integration` +Expected: all integration tests pass. + +Run: `uv run sphinx-build -M html docs docs/_build -W` +Expected: build succeeded. + +Run: `uv build` +Expected: produces `dist/wabot-0.2.0.tar.gz` and a wheel — proves the uv_build backend + src layout are correct. + +- [ ] **Step 4: Commit** + +```bash +git add README.md CLAUDE.md +git commit -m "docs: rewrite README and CLAUDE.md for the modernized library" +``` + +--- + +## Execution notes + +- Tasks 1–12 are pure TDD with no browser required; Tasks 13–16 need the real Firefox/Chromium on this machine; Task 17–18 close out docs and hygiene. +- If an integration test fails, use superpowers:systematic-debugging — do not weaken assertions or mark xfail to get green. +- After Task 18, use superpowers:finishing-a-development-branch (the user pushes to Gitea manually and opens the PR).