Merge pull request 'Feature/modernization' (#1) from feature/modernization into master

Reviewed-on: https://git.zavage.net/Zavage-Software/wabot/pulls/1
This commit is contained in:
zavage 2026-07-12 17:48:29 -06:00
commit f6c92dfe37
44 changed files with 8316 additions and 1269 deletions

5
.gitignore vendored

@ -39,6 +39,11 @@ htmlcov/
.cache
nosetests.xml
coverage.xml
.pytest_cache/
# uv / ruff
.venv/
.ruff_cache/
# Translations
*.mo

39
CLAUDE.md Normal file

@ -0,0 +1,39 @@
# CLAUDE.md
This file provides guidance to Claude Code (claude.ai/code) when working with code in this repository.
## What this is
WABot is a Python library wrapping Selenium for stateful browser automation (login flows, multi-page form filling, scraping). Its two distinguishing features:
1. **Page-object automation with state carried between pages** — consumers model each website page as a `Page` subclass and drive them through a single `Browser` facade.
2. **Browser sessions that survive the Python process** — session identity (executor URL + session id) is persisted to disk so a *different* Python process can reattach to the same still-open browser, whether it's running under an external WebDriver server or a wabot-managed detached driver service.
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).
## Architecture
Consumer code talks to one object, `Browser` (`src/wabot/_browser.py` — underscored so the module doesn't shadow the `wabot.browser()` factory function), which composes everything else:
- **`Browser`** holds the webdriver plus a *current page* object and acts as a facade. `__getattr__` delegates unknown attributes to the current page, so `bot.login()` invokes `bot.page.login()`; `bot[key]` resolves page elements. `set_page(PageClass)` swaps the current page and calls its `verify()` gate. `perform(name, ...)` invokes a page method with exception trapping — any failure flips `self.good = False`, after which all further delegated actions are refused, forcing an explicit `reset()`. `quit()` tears down the driver and, for a wabot-managed service, stops the detached process too (a plain `driver.quit()` would leave it running); external servers are left alone.
- **`Page`** (`src/wabot/page.py`) is the base class consumers subclass. Pages declare a class-level `elements = {key: (type, accessors)}` map; lookup walks the MRO, so subclasses inherit and override parent element maps. `page[key]` returns a typed field wrapper, never a raw WebElement.
- **Field wrappers** (`src/wabot/fields.py`): `TextField`, `SelectField`, `CheckField` wrap elements with `get_value`/`set_value`; unknown attribute access falls through to the underlying selenium element. `NullField` is a falsy null-object returned for missing elements so callers can `if el:` instead of catching exceptions.
- **Pacing policies** (`src/wabot/pacing.py`): `HumanPacing` (default) and `NoPacing`. Field wrappers and `Page` ask the active policy for a per-action delay and a click offset instead of hardcoding stealth behavior, so speed/stealth is swappable per `Browser` instance.
- **Sessions** (`src/wabot/sessions.py`): `SessionRecord`/`SessionStore` persist only what's needed to reattach — `executor_url` and `session_id`, plus `service_pid`/`service_port` bookkeeping for managed services — as JSON at `platformdirs.user_data_dir("wabot")/sessions.json`, keyed by session name. Records older than `max_age` (default 3 days) are evicted on load; a corrupt store is quarantined to `sessions.json.bad` rather than crashing the caller.
- **Driver hosts** (`src/wabot/hosts.py`): two ways to obtain a WebDriver server URL. `ExternalServer` points at a Selenium Grid or bare driver the user already runs (validated via `GET /status`). `ManagedService` spawns `chromedriver`/`geckodriver` itself, detached (`start_new_session=True`) so it survives this Python process exiting; `stop_service(pid)` is how `Browser.quit()`/`destroy()` shut it back down. `ReattachingRemote.start_session` skips `NEW_SESSION` entirely and adopts a saved `session_id`; `attach()` builds one and probes `driver.current_url` to confirm the session is still live before handing it back.
- **`wabot.browser()`** (`src/wabot/__init__.py`) is the public entry point tying it together: no `session=` gives an ephemeral local-or-remote browser; a named session looks up its saved record, reattaches if the host answers, and otherwise spins up a host (external or managed) and saves a fresh record. `wabot.sessions()`/`wabot.destroy()` round out the public API.
- **Screenshots** (`src/wabot/screenshot.py`): Firefox exposes a raw `moz/screenshot/full` command over Remote for full-page capture; Chromium has no Remote equivalent, so viewport tiles are captured and stitched with Pillow instead.
**Human-like behavior is intentional, not accidental slowness**: with the default `HumanPacing`, clicks land at Gaussian-random offsets from the element's in-view center (Selenium 4 measures click offsets from the center, not top-left, unlike Selenium 3) instead of a plain click, and field setters sleep a random interval between actions. Pass `pacing=wabot.NoPacing()` for instant, exact behavior (what the test suite uses). Don't "optimize" these away.
## Gotchas
- External hosts (`ExternalServer`) must outlive the Python process — wabot never starts or stops them. Only `ManagedService`-spawned drivers are wabot's to stop.
- geckodriver reports `ready: false` on `GET /status` whenever its single session is in use; chromedriver reports `ready: true` even with an active session. Liveness checks (`service_alive`) therefore accept any HTTP 200 with valid JSON and deliberately ignore the `ready` flag — requiring it would work for chromedriver and break for geckodriver.
- Selenium is pinned `<5`: `ReattachingRemote` relies on internals that have been stable across the 4.x line but aren't a public API contract (that `start_session` is the only place a new session gets created, and that `session_id` is injected into every subsequent command). `ReattachingRemote.__init__` raises `RuntimeError` on adoption mismatch as a tripwire if a future selenium release changes this.
## Git
Remote is Gitea (`gitea@git.zavage.net:Zavage-Software/wabot.git`); default branch is `master`. Do feature work on branches — pushing is done manually by the user (no SSH auth from inside the Claude shell).

@ -1,9 +1,33 @@
# WABot - Web Automator Bot (python module)
# WABot — Web Automator Bot
Module provides a class which builds on selenium automated-browser control. What it adds
is the ability to carry and maintain-state between web pages. So for e.g. you can automate
a web browser to log-in, navigate through the menus, and fill out a form-- or scrape data.
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.
Probably pretty awful code, I'd write it a lot better if I had more time on it. Though I've
used and continue to use to scrape 100,000s of data points with zero human time, performing
complex interactions with websites.k
```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).

26
docs/api.rst Normal file

@ -0,0 +1,26 @@
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:
.. autoclass:: wabot.SessionRecord
:members:
.. autoclass:: wabot.SessionStore
:members:
.. automodule:: wabot.hosts
:members: ExternalServer, ManagedService, attach, service_alive

14
docs/conf.py Normal file

@ -0,0 +1,14 @@
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

13
docs/index.rst Normal file

@ -0,0 +1,13 @@
wabot
=====
Stateful Selenium browser automation with sessions that survive the
Python process.
.. toctree::
:maxdepth: 2
quickstart
sessions
page-objects
api

47
docs/page-objects.rst Normal file

@ -0,0 +1,47 @@
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()``.

44
docs/quickstart.rst Normal file

@ -0,0 +1,44 @@
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.

51
docs/sessions.rst Normal file

@ -0,0 +1,51 @@
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

File diff suppressed because it is too large Load Diff

@ -0,0 +1,156 @@
# WABot Modernization — Design
**Date:** 2026-07-07
**Status:** Approved (brainstorming session with Mathew)
**Approach:** Rebuild the core on the existing design (approach B) — port proven logic into cleanly separated modules; redesign the session-persistence layer; free API redesign was explicitly approved (no downstream consumers to protect).
## Context & goals
WABot is a Selenium-wrapping browser-automation library (page-object pattern, stateful multi-page flows, anti-bot stealth behavior). It is stuck on Selenium 3-era APIs because its flagship feature — saving a browser so a *different* Python process can pick it up — was implemented by pickling live webdriver objects with `dill`, which broke on Selenium 4.
Goals, as requested:
1. Modern packaging with **uv** (`pyproject.toml`, lockfile).
2. **Sphinx + sphinx-rtd-theme** documentation.
3. **pytest** suite (unit + real-browser integration).
4. Migrate to **latest Selenium 4**, keeping core functionality — especially cross-process browser pickup, rebuilt on a sound mechanism.
5. General code quality: fix latent bugs, drop dead code, keep the behavioral subtleties that make the library effective.
### Decisions from the brainstorming Q&A
| Question | Decision |
|---|---|
| API compatibility | Free to redesign; keep the concepts, not the exact signatures |
| Browser host for persistence | **Both**: external Selenium server URL *and* wabot-spawned detached driver service |
| Test depth | Unit (mocked) + integration (real headless browsers, local HTML fixtures, cross-process reattach e2e) |
| Stealth pacing | Configurable policy; human-like behavior stays the **default** |
### Key insight (why persistence broke, and the fix)
A `Remote` webdriver is a thin HTTP client around `(command_executor_url, session_id)`. Pickling the whole object was always fragile and is unworkable on Selenium 4. The robust mechanism is to persist just the URL + session ID as JSON and **reattach** by constructing a `Remote` subclass whose `start_session` adopts the saved session ID instead of creating a new session. This works against any WebDriver-protocol server: a Selenium Grid *or* a bare chromedriver/geckodriver. `dill` is removed entirely. The browser must be hosted by a process that outlives Python — same constraint as the old design, where only `remote_*` drivers were persistable.
## Tooling & layout
- `pyproject.toml` managed by uv, `uv_build` backend, `requires-python = ">=3.10"`.
- Runtime deps: `selenium>=4` (latest), `platformdirs` (replaces dead `appdirs`), `Pillow`. **Removed:** `dill`, `appdirs`.
- Dev group: `pytest`, `pytest-cov`, `ruff` (lint + format), `sphinx`, `sphinx-rtd-theme`.
- Commands: `uv sync`, `uv run pytest`, `uv run pytest -m integration`, `uv run ruff check`, `uv run sphinx-build docs docs/_build/html`.
```
wabot/
├── pyproject.toml
├── uv.lock
├── src/wabot/
│ ├── __init__.py # explicit public API (no star exports); wabot.browser() entry point
│ ├── browser.py # Browser (was BrowserProxy) — facade over driver + current page
│ ├── page.py # Page base class + element resolution
│ ├── fields.py # TextField / SelectField / CheckField / NullField
│ ├── pacing.py # HumanPacing / NoPacing policies
│ ├── sessions.py # SessionStore — JSON persistence + stale eviction
│ ├── hosts.py # ExternalServer / ManagedService driver hosts
│ └── screenshot.py # Chromium full-page stitching (PIL)
├── tests/
│ ├── unit/
│ ├── integration/
│ └── fixtures/ # static HTML pages (login form, multi-page flow)
└── docs/ # Sphinx
```
The legacy `setup.py` and top-level `wabot/` package are removed once the `src/` layout lands.
## Session persistence layer
Three cooperating pieces (this replaces all pickle logic in `create_browser.py`):
**`SessionStore`** (`sessions.py`)
- JSON file at `platformdirs.user_data_dir('wabot')/sessions.json`.
- Schema: `{name: {executor_url, session_id, browser, created_at, service_pid?, service_port?}}`.
- Entries older than a configurable max age (default 3 days, matching current behavior) are evicted on load.
- A corrupt file is renamed aside (`sessions.json.bad`) and recreated — never a crash.
**`DriverHost`** (`hosts.py`) — small interface: `ensure_running() -> executor_url`.
- `ExternalServer(url)`: a Selenium server/Grid the user runs; wabot just connects. Docs must note Grid 4 reaps idle sessions after 5 minutes unless `--session-timeout` is raised.
- `ManagedService(browser)`: wabot spawns `chromedriver`/`geckodriver` as a **detached** process (new process session, output redirected to a log file in the wabot data dir) so it outlives Python; records port + pid in the store. Driver binaries are found on `PATH` first, falling back to Selenium Manager's downloaded copies. On reattach the service's `/status` endpoint is health-checked before use.
**Reattach**
- A private `Remote` subclass overrides `start_session` to adopt the saved `session_id` (no new session created).
- Liveness validation: touch `driver.current_url`. On failure: log a warning, prune the store entry, fall through to fresh creation.
**Public flow**
```python
import wabot
bot = wabot.browser(session="scraper1", browser="firefox") # create + persist
# process exits; browser stays open
bot = wabot.browser(session="scraper1", browser="firefox") # second process reattaches
```
- `session=None` (default): ephemeral — the browser is not persisted and the store is untouched. With no `host`, this is a plain local Selenium driver that dies with the process; with a `host`, it connects to that server but the session is still not saved.
- `session="name"` + `host="http://..."` → persist via `ExternalServer`; `session="name"` without `host` → persist via `ManagedService`.
- Housekeeping: `wabot.sessions()` lists saved sessions; `wabot.destroy(name)` quits the browser, kills a managed service, removes the entry.
## Core API
**`Browser`** (`browser.py`, rename of `BrowserProxy`)
- Facade: holds driver + *current page*; `set_page(PageClass)` instantiates and runs the page's `verify()` gate; `__getattr__` delegates unknown attributes to the current page (`bot.login()` → `bot.page.login()`); `bot[key]` resolves elements on the current page.
- `good` flag properly initialized to `True` (today it is never set, so the refuse-after-exception guard is broken); `perform()` keeps trap-and-flip behavior; `reset_good_status()` retained.
- Dropped: unused `switch_after` decorator, NHSN-specific `RC` enum values, hardcoded `/home/mathew/firefox_prof` profile path, PhantomJS remnants, `firefox1/firefox2/chromium1/chromium2` naming (now just `firefox` / `chromium` — with `chrome` accepted as an alias, since chromedriver drives both — plus the optional `host`).
- User-agent becomes an optional parameter defaulting to the browser's own UA (the hardcoded 2006 MSIE 7 string is an anti-stealth beacon today).
- Logging: stdlib only — `LOGGER.trace``debug`, `LOGGER.warn``warning` (the `trace` calls crash under stdlib logging unless the host app registers a custom level).
**`Page` + fields** — declarative design preserved
- Class-level `elements = {key: (type, accessors)}`; lookup walks the MRO so subclasses inherit/override element maps; `page[key]` returns typed wrappers; `NullField` stays the falsy null-object for missing elements.
- Consolidate the duplicated find/wait logic; replace bare `except:` with specific Selenium exception types.
- Selenium 4 port subtlety: `move_to_element_with_offset` now measures from the element **center** (was top-left in Selenium 3), so the Gaussian click math is recomputed relative to center — otherwise human-like clicks land outside small elements.
**`pacing.py`** — stealth extracted into a policy
- `HumanPacing` (default): current behavior — random 311 s delays per field type, Gaussian click offsets.
- `NoPacing`: instant actions, exact clicks (used by tests and trusted-site automation).
- `Browser` holds the policy; fields and `Page.click` consult it instead of calling `time.sleep` directly.
**Screenshots** (`screenshot.py`)
- Firefox: Selenium 4 native `get_full_page_screenshot_as_file`.
- Chromium: keep the PIL viewport-stitching workaround, isolated in this module.
## Error handling
Philosophy stays *log and refuse, don't crash the scraper*:
- Missing elements → `NullField`; failed actions → `False`; exceptions in `perform()` → logged with traceback, `good = False`.
- Dead reattach → warn, prune, create fresh (never crash on a stale session).
- Corrupt store → rename aside, rebuild.
- Dead managed service → detected by `/status` health check before driving.
- No bare `except:` — catch specific Selenium exceptions so real bugs surface in logs.
## Testing
Two pytest tiers split by the `integration` marker:
**Unit** (default, fast, no browsers): element resolution through the MRO; field wrappers against a mocked driver; pacing policies; `SessionStore` round-trip, eviction, and corruption handling on `tmp_path`; reattach logic against a mocked executor.
**Integration** (`-m integration`): real headless Firefox + Chromium driving static HTML fixtures (login form, multi-page flow) served locally — no external network. Flagship test: create a session via `ManagedService`, then a **second Python process** (via `subprocess`) reattaches by name and asserts browser state (URL, filled field value) survived — proving cross-process pickup end to end. `ExternalServer` mode is tested against a locally spawned chromedriver used as the "external" URL (no Java Grid required). Integration tests always run with `NoPacing`.
## Documentation
Sphinx in `docs/`, `sphinx-rtd-theme`, autodoc + napoleon (Google-style docstrings on the public API). Pages:
1. Quickstart — install, first scrape.
2. Persistent sessions guide — the flagship feature; managed services vs external servers; the Grid `--session-timeout` caveat.
3. Page-objects guide — modeling a site with `Page` + `elements`.
4. API reference (autodoc).
Local build only (`uv run sphinx-build docs docs/_build/html`); no hosting setup for now.
## Out of scope
- CI pipeline (can be added later on Gitea).
- Read the Docs hosting.
- New automation features beyond parity (e.g., async API, non-Selenium backends).
- PhantomJS and the legacy `firefox1`-style driver taxonomy (deliberately removed).
## Known latent bugs fixed by this work
- `LOGGER.trace()` calls crash under stdlib logging.
- `_clean_old_pickles()` references undefined `final_name`.
- `BrowserProxy.get_driver()` calls `self._create_driver_chromium()` which doesn't exist on that class.
- `self.good` never initialized, breaking the refuse-after-exception guard.
- `appdirs` and `Pillow` used but undeclared in `setup.py`.

39
pyproject.toml Normal file

@ -0,0 +1,39 @@
[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"]

@ -1,44 +0,0 @@
#!/usr/bin/env python
#
# Usage:
#
# First, enable the python environment you want to install to, or if installing
# system-wide then ensure you're logged in with sufficient permissions
# (admin or root to install to system directories)
#
# installation:
#
# $ ./setup.py install
#
# de-installation:
#
# $ pip uninstall <app>
from setuptools import setup
__project__ = 'WaBoT'
__version__ = '0.1.0'
setup(
name = __project__,
version = __version__,
description = '',
author = 'Mathew Guest',
author_email = 't3h.zavage@gmail.com',
# Third-party dependencies; will be automatically installed
install_requires = (
'dill',
'selenium',
),
# Local packages to be installed (our packages)
packages = (
'wabot',
),
# Binaries/Executables to be installed to system
scripts=()
)

167
src/wabot/__init__.py Normal file

@ -0,0 +1,167 @@
"""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 contextlib
import logging
from datetime import datetime, timezone
from selenium import webdriver
from selenium.common.exceptions import WebDriverException
from ._browser import Browser
from ._sessions import SessionRecord, SessionStore
from .fields import CheckField, NullField, PageObject, SelectField, TextField
from .hosts import (
ExternalServer,
ManagedService,
attach,
build_options,
normalize_browser,
service_alive,
stop_service,
)
from .pacing import HumanPacing, NoPacing
from .page import Page
__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.
When reattaching to an existing session, ``headless`` and ``user_agent``
are ignored (the browser already 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)
if record.service_pid:
stop_service(record.service_pid) # don't leak the old managed service
store.remove(session)
host_obj = ExternalServer(host) if host else ManagedService(browser_name)
url = host_obj.ensure_running()
driver = None
try:
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),
)
)
except BaseException:
# startup or persistence failed: don't leak the driver session or the
# detached service we just spawned
if driver is not None:
with contextlib.suppress(Exception):
driver.quit()
pid = getattr(host_obj, "pid", None)
if pid:
stop_service(pid)
raise
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):
try:
driver = attach(record.executor_url, record.session_id, record.browser)
if driver is not None:
driver.quit()
except WebDriverException as ex:
LOGGER.warning("quit failed while destroying %r: %s", name, ex)
except RuntimeError as ex: # attach adoption guard can raise
LOGGER.warning("could not reattach to destroy %r: %s", name, ex)
if record.service_pid:
stop_service(record.service_pid)
store.remove(name)
return True

122
src/wabot/_browser.py Normal file

@ -0,0 +1,122 @@
"""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".
Two ways to drive the current page:
* Direct delegation ``bot.login()`` calls ``bot.page.login()``. This is
the raw path: it raises on error and does NOT touch ``good``.
* ``bot.perform("login")`` traps driver failures, flips ``good`` to
False on error, and refuses further page actions until ``reset()``.
Route anything that can fail through ``perform()`` for the guard. While
``good`` is False, refused delegation returns a no-op callable, so
``bot.x()`` yields None; reading a page *attribute* in that state is not
meaningful.
"""
from __future__ import annotations
import logging
from selenium.common.exceptions import WebDriverException
from .hosts import stop_service
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 __repr__(self) -> str:
page = type(self.page).__name__ if self.page is not None else None
return f"<Browser page={page} good={self.good}>"
def set_page(self, page_cls) -> bool:
"""Instantiate ``page_cls`` and make it current if its verify() passes.
A consumer ``verify()`` that raises propagates (it is not trapped).
"""
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):
if self.page is None:
raise RuntimeError(
f"cannot resolve element {key!r}: no current page — call set_page() first"
)
return self.page[key]
def perform(self, method: str, *args, **kwargs):
"""Invoke a page method, trapping driver 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 and forget the saved session.
For a wabot-managed driver service, its detached process is also
stopped a plain ``driver.quit()`` leaves that service running.
External servers (records without a ``service_pid``) are left alone.
Use this for full teardown; to keep a session alive for another
process, let this process exit WITHOUT calling ``quit()``.
"""
record = None
if self._store is not None and self.session_name:
record = self._store.get(self.session_name)
try:
self.driver.quit()
finally:
if record is not None and record.service_pid:
stop_service(record.service_pid)
if self._store is not None and self.session_name:
self._store.remove(self.session_name)

122
src/wabot/_sessions.py Normal file

@ -0,0 +1,122 @@
"""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
import os
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
_RECORD_FIELDS = {field.name for field in dataclasses.fields(SessionRecord)}
def _parse_created_at(value) -> datetime | None:
"""Parse a record timestamp; None means unusable (treat as stale)."""
try:
created = datetime.fromisoformat(value)
except (ValueError, TypeError):
return None
if created.tzinfo is None:
created = created.replace(tzinfo=timezone.utc)
return created
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(encoding="utf-8"))
records = {
name: SessionRecord(**{k: v for k, v in data.items() if k in _RECORD_FIELDS})
for name, data in raw.items()
}
except (json.JSONDecodeError, TypeError, AttributeError, OSError):
self._quarantine()
return {}
fresh = {}
for name, record in records.items():
created = _parse_created_at(record.created_at)
if created is None:
LOGGER.warning(
"evicting session %r with unusable created_at %r", name, record.created_at
)
continue
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 _quarantine(self) -> None:
bad = self.path.with_name(self.path.name + ".bad")
try:
self.path.replace(bad) # replace, not rename: works when .bad exists
LOGGER.warning("corrupt session store moved to %s; starting fresh", bad)
except OSError as ex:
LOGGER.warning("could not quarantine corrupt store %s: %s", self.path, ex)
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:
# NOTE: save() is read-modify-write with no cross-process lock; at this
# library's scale (a handful of sessions on human timescales) a lost
# update is acceptable. A torn/partial file is NOT — hence atomic replace.
self.path.parent.mkdir(parents=True, exist_ok=True)
payload = {name: dataclasses.asdict(rec) for name, rec in records.items()}
tmp = self.path.with_name(self.path.name + ".tmp")
tmp.write_text(json.dumps(payload, indent=2), encoding="utf-8")
os.replace(tmp, self.path)

105
src/wabot/fields.py Normal file

@ -0,0 +1,105 @@
"""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):
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})"
)

264
src/wabot/hosts.py Normal file

@ -0,0 +1,264 @@
"""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 http.client
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
from selenium.common.exceptions import WebDriverException
from selenium.webdriver.remote.webdriver import WebDriver as RemoteWebDriver
from urllib3.exceptions import HTTPError as _Urllib3HTTPError
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:
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.
The ``timeout`` bounds each socket operation, not total wall-clock time.
"""
try:
with urllib.request.urlopen(f"{url.rstrip('/')}/status", timeout=timeout) as resp:
json.loads(resp.read())
return resp.status == 200
except (OSError, ValueError, http.client.HTTPException):
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
def find_driver_binary(browser: str) -> str:
"""Locate chromedriver/geckodriver: env override, PATH, then Selenium Manager.
The Selenium Manager fallback may download a driver over the network.
"""
browser = normalize_browser(browser)
name = DRIVER_BINARIES[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"]
# Selenium Manager is beta; any failure folds into a clean FileNotFoundError
except Exception as ex:
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.
ensure_running() is not idempotent: each call spawns a new service.
"""
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
returncode = process.poll()
if returncode is not None:
raise RuntimeError(
f"{binary} exited early with code {returncode} before "
f"answering /status on {url} (see {log_path})"
)
time.sleep(0.2)
# deadline expired but the driver is still running and unresponsive:
# terminate and reap it so we never leak a process that would otherwise
# outlive the interpreter.
process.terminate()
try:
process.wait(timeout=5)
except subprocess.TimeoutExpired:
process.kill()
process.wait()
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.
If the service is a child of THIS process (the same-process spawnstop
lifecycle), it is reaped to avoid a zombie. In the cross-process reattach
case the driver is not our child and init reaps it. A driver that ignores
SIGTERM is left running (no SIGKILL escalation).
"""
try:
os.kill(pid, signal.SIGTERM)
except (ProcessLookupError, PermissionError):
return False
deadline = time.monotonic() + 5.0
while time.monotonic() < deadline:
try:
reaped, _ = os.waitpid(pid, os.WNOHANG)
except ChildProcessError:
break # not our child (cross-process) — init reaps it
if reaped:
break
time.sleep(0.05)
return True
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)
if self.session_id != session_id:
raise RuntimeError(
"session adoption failed: selenium's start_session contract "
"changed; pin selenium <5 or update ReattachingRemote"
)
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)
driver.command_executor.close()
return None
except (OSError, _Urllib3HTTPError) as ex: # server unreachable (connection refused, timeout)
LOGGER.warning("no server at %s: %s", executor_url, ex)
driver.command_executor.close()
return None

56
src/wabot/pacing.py Normal file

@ -0,0 +1,56 @@
"""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.
Also the base class for custom pacing policies; do not use
``isinstance(x, NoPacing)`` to test whether pacing is disabled
HumanPacing subclasses this.
"""
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))

280
src/wabot/page.py Normal file

@ -0,0 +1,280 @@
"""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 (except the ``els`` type, which
returns a raw list of WebElements) and a falsy ``NullField`` when the element
cannot be found.
"""
from __future__ import annotations
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
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)
if not isinstance(locators, (tuple, list)) or len(locators) < 2:
LOGGER.error("malformed element %r: expected (type, accessors), got %r", key, locators)
return NullField(name=key)
obj_type, accessors = locators[0], locators[1]
# built-in types need a (by, value) accessors pair; a custom field class
# manages its own accessors, so only the built-ins are validated here.
if (obj_type in FIELD_TYPES or obj_type == "els") and (
not isinstance(accessors, (tuple, list)) or len(accessors) != 2
):
LOGGER.error(
"malformed accessors for element %r: expected (by, value), got %r",
key,
accessors,
)
return NullField(name=key)
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.debug("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)
# ---- 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
if not self._wait_clickable(el, clickable_timeout):
return False
try:
size = el.size
except StaleElementReferenceException:
LOGGER.error("failed to click element: stale reference")
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(str(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)
# verify the field holds exactly what we typed (strict: fields that
# normalize input will read back as a mismatch)
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:
return self.click(el) # propagate: a failed click is a failed set
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)

74
src/wabot/screenshot.py Normal file

@ -0,0 +1,74 @@
"""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)
# Tiles are device pixels; the canvas/scroll math is CSS pixels. Assumes
# devicePixelRatio == 1 (true for headless chromium in the test target).
with Image.open(BytesIO(driver.get_screenshot_as_png())) as tile:
# 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

17
tests/fixtures/login.html vendored Normal file

@ -0,0 +1,17 @@
<!DOCTYPE html>
<html>
<head><title>Login</title></head>
<body>
<h1 id="page-title">Test Login</h1>
<form action="/welcome.html" method="get">
<input type="text" id="username" name="username">
<select id="state" name="state">
<option value="">--</option>
<option value="CO">Colorado</option>
<option value="NM">New Mexico</option>
</select>
<input type="checkbox" id="agree" name="agree" value="yes">
<button type="submit" id="submit-btn">Sign in</button>
</form>
</body>
</html>

8
tests/fixtures/welcome.html vendored Normal file

@ -0,0 +1,8 @@
<!DOCTYPE html>
<html>
<head><title>Welcome</title></head>
<body>
<h1 id="page-title">Welcome!</h1>
<p id="greeting">You made it.</p>
</body>
</html>

@ -0,0 +1,41 @@
"""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):
integration_dir = Path(__file__).parent
for item in items:
if integration_dir in item.path.parents:
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 import SessionStore
return SessionStore(path=tmp_path / "sessions.json")

@ -0,0 +1,60 @@
"""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()

@ -0,0 +1,66 @@
"""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()

@ -0,0 +1,80 @@
"""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 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)

0
tests/unit/__init__.py Normal file

238
tests/unit/test_api.py Normal file

@ -0,0 +1,238 @@
from unittest.mock import MagicMock
import pytest
import wabot
from wabot 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 TestResourceCleanup:
def test_managed_service_stopped_when_driver_creation_fails(
self, fake_selenium, store, monkeypatch
):
from selenium.common.exceptions import WebDriverException
stopped = []
monkeypatch.setattr(wabot, "stop_service", lambda pid: stopped.append(pid))
fake_selenium.Remote.side_effect = WebDriverException("startup failed")
with pytest.raises(WebDriverException):
wabot.browser(session="s1", browser="chromium", store=store)
assert stopped == [4321] # spawned managed service was stopped
assert store.get("s1") is None # nothing persisted
def test_dead_managed_session_stops_old_service_before_recreating(
self, fake_selenium, store, monkeypatch
):
from datetime import datetime, timezone
stopped = []
monkeypatch.setattr(wabot, "stop_service", lambda pid: stopped.append(pid))
store.save(
wabot.SessionRecord(
name="s1",
executor_url="http://127.0.0.1:7777",
session_id="old",
browser="chromium",
created_at=datetime.now(timezone.utc).isoformat(),
service_pid=9999,
service_port=7777,
)
)
fake_selenium.service_alive.return_value = True
fake_selenium.attach.return_value = None # session dead on a live server
wabot.browser(session="s1", store=store)
assert 9999 in stopped # old managed service stopped, not leaked
fake_selenium.Remote.assert_called_once() # fresh browser created
assert store.get("s1").session_id == "new-session-id"
class TestDestroyEdges:
def _save(self, store, **kw):
from datetime import datetime, timezone
defaults = dict(
name="s1",
executor_url="http://127.0.0.1:7777",
session_id="old",
browser="chromium",
created_at=datetime.now(timezone.utc).isoformat(),
)
defaults.update(kw)
store.save(wabot.SessionRecord(**defaults))
def test_destroy_external_session_does_not_stop_service(
self, fake_selenium, store, monkeypatch
):
stopped = []
monkeypatch.setattr(wabot, "stop_service", lambda pid: stopped.append(pid))
self._save(store, name="ext", executor_url="http://grid:4444", service_pid=None)
fake_selenium.service_alive.return_value = True
assert wabot.destroy("ext", store=store) is True
assert stopped == [] # external server: not ours to stop
assert store.get("ext") is None
def test_destroy_dead_session_still_stops_and_removes(self, fake_selenium, store, monkeypatch):
stopped = []
monkeypatch.setattr(wabot, "stop_service", lambda pid: stopped.append(pid))
self._save(store, service_pid=4321)
fake_selenium.service_alive.return_value = True
fake_selenium.attach.return_value = None # dead session
assert wabot.destroy("s1", store=store) is True
assert stopped == [4321] # service stopped despite dead session
assert store.get("s1") is None
class TestPlumbing:
def test_headless_and_user_agent_reach_build_options(self, fake_selenium, store, monkeypatch):
captured = {}
real_build = wabot.build_options
def spy(browser, **kwargs):
captured.update(kwargs)
return real_build(browser, **kwargs)
monkeypatch.setattr(wabot, "build_options", spy)
wabot.browser(browser="chromium", headless=True, user_agent="Bot/1.0", store=store)
assert captured == {"headless": True, "user_agent": "Bot/1.0"}
def test_pacing_reaches_browser(self, fake_selenium, store):
pacing = wabot.NoPacing()
bot = wabot.browser(browser="chromium", pacing=pacing, store=store)
assert bot.pacing is pacing
class TestPublicSurface:
def test_all_exports_exist(self):
for name in wabot.__all__:
assert hasattr(wabot, name), name

148
tests/unit/test_browser.py Normal file

@ -0,0 +1,148 @@
from types import SimpleNamespace
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
def test_direct_delegation_does_not_flip_good(self, bot):
# only perform() traps+flips; direct bot.x() is the raw path
bot.set_page(Login)
with pytest.raises(WebDriverException):
bot.explode()
assert bot.good is True
def test_getitem_without_page_raises_friendly(self, bot):
with pytest.raises(RuntimeError, match="no current page"):
bot["username"]
def test_repr_shows_page_and_good(self, bot):
assert "page=None" in repr(bot) and "good=True" in repr(bot)
bot.set_page(Login)
assert "page=Login" in repr(bot)
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"
def test_broken_state_attribute_read_returns_noop_callable(self, bot):
# documents the footgun: while broken, an attribute read yields the
# no-op refusal callable, not the underlying value
bot.set_page(Login)
bot.perform("explode") # flips good=False
assert callable(bot.anything)
class TestQuit:
def test_quit_quits_driver_and_removes_session(self):
driver, store = MagicMock(), MagicMock()
store.get.return_value = SimpleNamespace(service_pid=None) # external: nothing to stop
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()
def test_quit_stops_managed_service(self, monkeypatch):
import wabot._browser as browser_mod
stopped = []
monkeypatch.setattr(browser_mod, "stop_service", lambda pid: stopped.append(pid))
driver, store = MagicMock(), MagicMock()
store.get.return_value = SimpleNamespace(service_pid=4321) # managed session
bot = Browser(driver, session_name="s1", store=store)
bot.quit()
assert stopped == [4321]
store.remove.assert_called_once_with("s1")
def test_quit_removes_record_even_if_driver_quit_raises(self, monkeypatch):
import wabot._browser as browser_mod
monkeypatch.setattr(browser_mod, "stop_service", lambda pid: None)
driver, store = MagicMock(), MagicMock()
driver.quit.side_effect = WebDriverException("boom")
store.get.return_value = SimpleNamespace(service_pid=None)
bot = Browser(driver, session_name="s1", store=store)
with pytest.raises(WebDriverException):
bot.quit()
store.remove.assert_called_once_with("s1") # finally block still ran

131
tests/unit/test_fields.py Normal file

@ -0,0 +1,131 @@
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")
class TestPacingWiring:
def _spy_page(self, page):
page.pacing = MagicMock()
page.pacing.delay.return_value = 0.0
return page
def test_text_set_value_paces_text(self, page):
self._spy_page(page)
TextField(page, accessors=ACCESSORS, name="username").set_value("x")
page.pacing.delay.assert_called_once_with("text")
def test_select_set_value_paces_select(self, page, monkeypatch):
import wabot.fields as fields_mod
monkeypatch.setattr(fields_mod, "Select", MagicMock())
self._spy_page(page)
SelectField(page, accessors=("id", "state"), name="state").set_value(value="CO")
page.pacing.delay.assert_called_once_with("select")
def test_select_by_index_paces_select(self, page, monkeypatch):
import wabot.fields as fields_mod
monkeypatch.setattr(fields_mod, "Select", MagicMock())
self._spy_page(page)
SelectField(page, accessors=("id", "state"), name="state").select_by_index(2)
page.pacing.delay.assert_called_once_with("select")
def test_set_checked_paces_checkbox(self, page):
self._spy_page(page)
CheckField(page, accessors=("id", "agree"), name="agree").set_checked(True)
page.pacing.delay.assert_called_once_with("checkbox")
def test_click_and_go_paces_navigate(self, page):
self._spy_page(page)
PageObject(page, accessors=ACCESSORS, name="username").click_and_go()
page.pacing.delay.assert_called_once_with("navigate")
class TestTruthiness:
def test_real_field_is_truthy(self, page):
# locks the other end of the `if el:` guard contract (NullField is falsy)
assert bool(TextField(page, accessors=ACCESSORS, name="username")) is True

236
tests/unit/test_hosts.py Normal file

@ -0,0 +1,236 @@
import http.server
import json
import os
import stat
import sys
import threading
import time
import pytest
from wabot.hosts import (
ExternalServer,
ManagedService,
build_options,
find_driver_binary,
normalize_browser,
service_alive,
stop_service,
)
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()
server.server_close()
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()
class TestServiceAliveNonHttpResponder:
def test_non_http_service_on_port_is_false_and_clean_error(self):
# wrong-port misconfiguration: some non-HTTP service answers the socket
import socket
server = socket.create_server(("127.0.0.1", 0))
port = server.getsockname()[1]
def answer(count=2):
for _ in range(count):
try:
conn, _ = server.accept()
except OSError:
return
conn.sendall(b"SSH-2.0-OpenSSH_9.7\r\n")
conn.close()
thread = threading.Thread(target=answer, daemon=True)
thread.start()
try:
assert service_alive(f"http://127.0.0.1:{port}", timeout=2.0) is False
with pytest.raises(ConnectionError, match="no WebDriver server"):
ExternalServer(f"http://127.0.0.1:{port}").ensure_running()
finally:
server.close()
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
SILENT_DRIVER = """\
#!{python}
import sys, time
time.sleep(300) # accepts --port=N like a real driver but never serves /status
"""
@pytest.fixture
def silent_driver(tmp_path):
path = tmp_path / "chromedriver"
path.write_text(SILENT_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_early_exit_raises_with_returncode(self, tmp_path):
# /bin/true exits immediately: distinct from a slow timeout
svc = ManagedService(
"chromium", log_dir=tmp_path, driver_binary="/bin/true", startup_timeout=5.0
)
with pytest.raises(RuntimeError, match="exited early"):
svc.ensure_running()
def test_startup_timeout_terminates_and_does_not_leak(self, silent_driver, tmp_path):
svc = ManagedService(
"chromium", log_dir=tmp_path, driver_binary=str(silent_driver), startup_timeout=1.0
)
with pytest.raises(TimeoutError, match="did not answer /status"):
svc.ensure_running()
# the unresponsive driver must have been terminated + reaped, not leaked
deadline = time.monotonic() + 5
while time.monotonic() < deadline:
try:
os.kill(svc.pid, 0)
except ProcessLookupError:
break
time.sleep(0.05)
with pytest.raises(ProcessLookupError):
os.kill(svc.pid, 0)
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

46
tests/unit/test_pacing.py Normal file

@ -0,0 +1,46 @@
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_delays_actually_vary(self):
assert len({HumanPacing().delay("text") for _ in range(50)}) > 1
def test_click_offset_stays_inside_element(self):
p = HumanPacing()
offsets = []
for _ in range(200):
offset = p.click_offset(100, 40)
assert offset is not None
offsets.append(offset)
x, y = offset
# offsets are measured from the element CENTER (selenium 4)
assert -49 <= x <= 49
assert -19 <= y <= 19
assert any(o != (0, 0) for o in offsets)
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

296
tests/unit/test_page.py Normal file

@ -0,0 +1,296 @@
from unittest.mock import MagicMock
from selenium.common.exceptions import NoSuchElementException
import wabot.page as page_mod
from wabot.fields import CheckField, NullField, SelectField, TextField
from wabot.pacing import HumanPacing, NoPacing
from wabot.page import Page
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
def test_empty_override_still_inherits_ancestors(self):
class Grandchild(ChildPage):
elements = {} # declares nothing; must still resolve ancestor keys
page = Grandchild(make_browser())
assert page.find_element_locators("state") == ("select", ("id", "state"))
assert page.find_element_locators("extra") == ("el", ("id", "extra"))
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)
def test_malformed_locator_missing_accessors_returns_nullfield(self):
class BadPage(Page):
elements = {"x": ("el",)} # missing the accessors half
el = BadPage(make_browser())["x"]
assert isinstance(el, NullField)
assert not el
def test_malformed_accessors_arity_returns_nullfield(self):
class BadPage(Page):
elements = {"x": ("els", ("id",))} # accessors not a (by, value) pair
assert isinstance(BadPage(make_browser())["x"], NullField)
def test_unknown_type_returns_nullfield(self):
class BadPage(Page):
elements = {"x": ("selct", ("id", "x"))} # typo'd type string
assert isinstance(BadPage(make_browser())["x"], NullField)
class TestVerify:
def test_default_verify_is_true(self):
assert BasePage(make_browser()).verify() is True
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
class TestAlert:
def test_accept_alert_accepts_and_returns_text(self, monkeypatch):
browser = make_browser()
page = PlainPage(browser)
monkeypatch.setattr(page_mod, "WebDriverWait", MagicMock()) # until() won't raise
alert = browser.driver.switch_to.alert
alert.text = "confirm?"
assert page.accept_alert() == "confirm?"
alert.accept.assert_called_once_with()
def test_accept_alert_can_dismiss(self, monkeypatch):
browser = make_browser()
page = PlainPage(browser)
monkeypatch.setattr(page_mod, "WebDriverWait", MagicMock())
alert = browser.driver.switch_to.alert
alert.text = "confirm?"
page.accept_alert(accept=False)
alert.dismiss.assert_called_once_with()
def test_accept_alert_returns_false_when_no_alert(self, monkeypatch):
from selenium.common.exceptions import TimeoutException
page = PlainPage(make_browser())
wait = MagicMock()
wait.return_value.until.side_effect = TimeoutException()
monkeypatch.setattr(page_mod, "WebDriverWait", wait)
assert page.accept_alert(timeout=0) is False
class TestSelectHelpers:
def test_get_select_value(self):
select = MagicMock()
select.first_selected_option.get_attribute.return_value = "CO"
assert PlainPage(make_browser()).get_select_value(select) == "CO"
def test_get_select_value_null_returns_none(self):
assert PlainPage(make_browser()).get_select_value(None) is None
def test_set_select_value_by_value_including_zero(self):
# legacy `if value:` skipped a valid 0/"" — verify the fix
select = MagicMock()
assert PlainPage(make_browser()).set_select_value(select, value=0) is True
select.select_by_value.assert_called_once_with("0")
def test_set_select_value_by_text(self):
select = MagicMock()
assert PlainPage(make_browser()).set_select_value(select, text="Colorado") is True
select.select_by_visible_text.assert_called_once_with("Colorado")
class TestCheckboxAndCoercion:
def test_set_checkbox_returns_false_when_click_fails(self, monkeypatch):
page = PlainPage(make_browser())
el = make_element()
el.is_enabled.return_value = True
el.is_selected.return_value = False # needs toggling
monkeypatch.setattr(page, "click", lambda e: False) # click fails
assert page.set_checkbox(el, True) is False
def test_set_checkbox_disabled_returns_false(self):
page = PlainPage(make_browser())
el = make_element()
el.is_enabled.return_value = False
assert page.set_checkbox(el, True) is False
def test_set_el_value_coerces_non_str(self):
page = PlainPage(make_browser())
el = make_element()
el.get_attribute.return_value = "42"
assert page.set_el_value(el, 42) is True
el.send_keys.assert_called_once_with("42")

@ -0,0 +1,54 @@
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):
# constructing against a closed port (127.0.0.1:1) proves no HTTP happens
# until the first command
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
def test_dead_session_closes_executor(self, monkeypatch):
from selenium.webdriver.remote.remote_connection import RemoteConnection
closed = []
monkeypatch.setattr(RemoteConnection, "close", lambda self: closed.append(1))
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
assert closed == [1] # executor was closed, not leaked

@ -0,0 +1,111 @@
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)
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)
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
def test_tall_page_stitches_tiles_bottom_aligned(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) # window.scrollTo(...) -> None
driver.get_screenshot_as_png.side_effect = [
png_bytes(100, 60, (255, 0, 0)), # tile @ y=0
png_bytes(100, 60, (0, 255, 0)), # tile @ y=60
png_bytes(100, 60, (0, 0, 255)), # tile @ y=120, clamped to paste_y=90
]
out = tmp_path / "shot.png"
assert save_full_page(driver, str(out)) is True
img = Image.open(out)
assert img.size == (100, 150)
assert img.getpixel((50, 30)) == (255, 0, 0) # first tile
assert img.getpixel((50, 75)) == (0, 255, 0) # second tile (rows 60-89)
assert img.getpixel((50, 140)) == (0, 0, 255) # third tile, bottom-aligned
def test_wide_page_clamps_last_column(self, tmp_path):
driver = MagicMock(name="driver")
driver.capabilities = {"browserName": "chrome"}
driver.execute_script.side_effect = lambda script: {
"return document.body.parentNode.scrollWidth": 150, # 2 columns of 100
"return document.body.parentNode.scrollHeight": 60,
"return document.documentElement.clientWidth": 100,
"return window.innerHeight": 60,
}.get(script)
driver.get_screenshot_as_png.side_effect = [
png_bytes(100, 60, (255, 0, 0)), # column @ x=0
png_bytes(100, 60, (0, 0, 255)), # column @ x=100, clamped to paste_x=50
]
out = tmp_path / "shot.png"
assert save_full_page(driver, str(out)) is True
img = Image.open(out)
assert img.size == (150, 60)
assert img.getpixel((25, 30)) == (255, 0, 0) # left column (cols 0-49)
assert img.getpixel((140, 30)) == (0, 0, 255) # right column, clamped (cols 50-149)

125
tests/unit/test_sessions.py Normal file

@ -0,0 +1,125 @@
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())
def test_naive_created_at_is_treated_as_utc_not_crash(self, tmp_path):
# legacy/hand-written records may carry naive timestamps
import json
path = tmp_path / "sessions.json"
record = make_record()
data = {"scraper1": {**record.__dict__, "created_at": "2999-01-01T00:00:00"}}
path.write_text(json.dumps(data))
store = SessionStore(path=path)
assert store.get("scraper1") is not None # fresh: kept, assumed UTC
def test_garbage_created_at_evicts_only_that_record(self, tmp_path):
import json
path = tmp_path / "sessions.json"
good, bad = make_record(name="good"), make_record(name="bad")
data = {
"good": good.__dict__,
"bad": {**bad.__dict__, "created_at": "not-a-date"},
}
path.write_text(json.dumps(data))
store = SessionStore(path=path)
assert store.names() == ["good"] # no crash, bad evicted, file rewritten
assert "not-a-date" not in path.read_text()
def test_null_created_at_evicts_only_that_record(self, tmp_path):
import json
path = tmp_path / "sessions.json"
data = {"broken": {**make_record(name="broken").__dict__, "created_at": None}}
path.write_text(json.dumps(data))
assert SessionStore(path=path).names() == []
def test_unknown_record_fields_are_ignored_not_fatal(self, tmp_path):
# a newer wabot may add fields; older versions must not quarantine the store
import json
path = tmp_path / "sessions.json"
data = {"scraper1": {**make_record().__dict__, "future_field": 42}}
path.write_text(json.dumps(data))
assert SessionStore(path=path).get("scraper1") is not None
def test_repeat_corruption_with_existing_bad_file_does_not_crash(self, tmp_path):
path = tmp_path / "sessions.json"
(tmp_path / "sessions.json.bad").write_text("old corruption")
path.write_text("{ corrupt again")
store = SessionStore(path=path)
assert store.load() == {}
store.save(make_record())
assert store.get("scraper1") is not None
def test_write_is_atomic_no_tmp_leftover(self, tmp_path):
store = SessionStore(path=tmp_path / "sessions.json")
store.save(make_record())
assert not (tmp_path / "sessions.json.tmp").exists()

1084
uv.lock Normal file

File diff suppressed because it is too large Load Diff

@ -1,4 +0,0 @@
from .api import *
from .page import *
from .fields import *

@ -1,251 +0,0 @@
from .create_browser import *
import appdirs
import logging
# import pickle
import dill as pickle
import selenium.common.exceptions
import selenium.webdriver
import sys
import time
import traceback
import os
USER_AGENT = "Mozilla/4.0 (compatible; MSIE 7.0; Windows NT 6.0; Trident/5.0)"
REFUSE_AFTER_EXCEPTION = True
EXECUTOR_PORT = 4444
REMOTE_EXECUTOR = 'http://127.0.0.1:%s/wd/hub'
# PICKLE_FILENAME = '/tmp/nhsnwebdriverdump'
# PICKLE_FILENAME = os.path.join(
# appdirs.user_data_dir('wabot'),
# 'saved_browser_instances.pickle'
# )
LOGGER = logging.getLogger('wabot')
DEFAULT_WEBDRIVER_TYPE = 'firefox1'
class BrowserProxy:
def __init__(
self,
session_name='webdriver',
pickle_filename=None,
phantom=False,
webdriver_type=None # remote_chromium2
):
"""
BrowserProxy wraps a selenium webdriver instance and provides utility
functions for automation webpages.
"""
LOGGER.info('requesting selenium browser instance (%s): instance_name = %s', webdriver_type, session_name)
# if pickle_filename is None:
# pickle_filename = PICKLE_FILENAME
# self._pickle_filename = pickle_filename
if webdriver_type is None:
webdriver_type = 'firefox1'
assert webdriver_type in (
'firefox1',
'firefox2',
'chromium2',
'remote_chromium',
'remote_chrome',
'phantomjs',
'remote_firefox'
), 'webdriver_type must be firefox1, firefox2, chromium2, remote_chromium2, or phantomjs'
try:
self.driver_type = webdriver_type
if phantom:
pass
# driver_type = "phantomjs"
self.driver = self.get_driver(webdriver_type, session_name)
if not self.driver:
LOGGER.error('failed to get selenium webdriver')
self.good = False
return
except Exception as ex:
print('caught exception at BrowserProxy().__init__')
print(type(ex), ex)
raise
# self.page = nhsn_lo.pages.Login(self)
# self.good = True
def get_page(self):
return type(self.page)
def set_page(self, page):
LOGGER.trace('switching page to %s' % page)
newpage = page(self)
if hasattr(newpage, 'verify'):
rc = newpage.verify()
if not rc:
LOGGER.error('failed to verify page: %s' % page)
# input('failure')
return False
else:
LOGGER.trace('verified page: %s' % page)
self.page = newpage
return True
def switch_after(self, fn=None, page=None):
if not fn:
return lambda fn: self.switch_after(fn, page = page)
def wrapper(*args, **kwargs):
self.set_page(page)
return fn(*args, **kwargs)
return wrapper
def __getattr__(self, name):
"""
Access method on current page.
When object.method is referenced, python invokes
object.__getattribute__ which checks the instance
for a matching property, If none is found, python
invokes object.__getattr__, which is overridden here.
If the attribute is not found in the main class, this
will search the page object for the method. This
allows invoking page methods through the proxy class.
nhsn.foo() invokes nhsn.page.foo()
"""
# todo(mg) would this be a good spot to inject page verification?
# idea: maybe we could decorate method with error checking and
# error collection? method can return value, rc and this can
# queue actions and error check them as it goes? At the end,
# the user would be able to pull out the errors and check them
if REFUSE_AFTER_EXCEPTION and not self.good:
LOGGER.warn("broken state - refusing to invoke page action: %s" % (name))
return
try:
method = getattr(self.page, name)
return method
except AttributeError as ex:
raise
def __getitem__(self, key):
"""
Access element on the current page.
"""
return self.page[key]
def perform(self, meth, *args, **kwargs):
"""
Calls self.page.method and catches exceptions.
"""
# input('<breakpoint> at perform')
if REFUSE_AFTER_EXCEPTION and not self.good:
LOGGER.warn("broken state - refusing to invoke page action: %s" % (meth))
return
try:
x = getattr(self.page, meth)
except AttributeError as ex:
LOGGER.error("Failed to invoke page action. "\
"page = %s, method = %s" %\
(self.page.__class__, meth))
return None
# Hooks:
# x = self.page.detect_logged_out()
# y = self.page.detected_logged_in()
# print('logged out/in: ', x, y)
LOGGER.trace("invoking %s.%s" % (self.page.__class__.__name__, meth))
try:
return x(*args, **kwargs)
except selenium.common.exceptions.NoSuchWindowException:
self.good = False
except Exception as ex:
print("caught exception: %s" % (type(ex)))
exc_type, exc_value, exc_traceback = sys.exc_info()
lines = traceback.format_exception(exc_type, exc_value,
exc_traceback)
for line in lines:
print(line, end = "")
self.good = False
def reset_good_status(self):
self.good = True
def get_driver(self, browser, session_name='webdriver'):
"""
Returns selenium objects of different types
The combinations of selenium{1,2} {chrome,firefox} are supported.
In the case of selenium 1 (selenium server), the host is a constant
defined at the top of this file.
Args:
browser: The selenium webdriver requested.
If a chromium2 webdriver is requested, selenium
will try to use chromium as the browser. If
firefox2 is requested, selenium will try to use
firefox. Chromium is very fast but it is designed
to be fast for an interactive user which means it
is fast at the cost of processing power and
ram. Firefox is about as fast but uses less
resources.
Available browser drivers are:
chromium2, chromium1, firefox2, firefox1
The associated webdriver has to be installed
and runnable on the system.
Returns:
The selenium webdriver handle.
"""
LOGGER.debug('requesting selenium browser instance: type = %s' % (browser))
driver = None
browser_factory = CreateBrowser()
if browser == 'chromium2': # Selenium 2 - Chrome
driver = self._create_driver_chromium()
elif (
browser == 'remote_chromium'
or browser == 'remote_chrome'
):
driver = browser_factory._create_driver_remote_chromium(session_name)
elif browser == 'remote_firefox': # Selenium 1 - Firefox
driver = browser_factory._create_driver_remote_firefox(session_name)
# TODO(MG) need to redo/rename below. selenium2/ not remote, uses native browser
# webdrviers instead of javascript
elif browser == 'chromium1': # Selenium 1 - Chrome without working user agent switch
driver = self._create_driver_chromium1()
elif browser == 'firefox2': # Selenium 2 - Firefox
driver = self._create_driver_firefox2()
elif browser == 'phantomjs':
driver = self._create_driver_phantomjs()
else:
LOGGER.error(
'an attempt was made to request an '\
'unsupported (by this product) selenium '\
'webdriver; refusing. requested = %s'\
% (browser)
)
driver.implicitly_wait(10)
return driver

@ -1,316 +0,0 @@
import appdirs
import datetime
import logging
import os
import selenium
import pickle
LOGGER = logging.getLogger('wabot')
USER_AGENT = "Mozilla/4.0 (compatible; MSIE 7.0; Windows NT 6.0; Trident/5.0)"
REFUSE_AFTER_EXCEPTION = True
EXECUTOR_PORT = 4444
REMOTE_EXECUTOR = 'http://127.0.0.1:%s/wd/hub'
class PickledBrowserReference:
"""
Structure for saving a webdriver instance. Also includes the timestamp
to help invalidate old references.
"""
def __init__(self):
self.browser_ref = None
self.timestamp = None
# Factory Creator - instantiate Selenium1 and Selenium2 webdriver instances.
class CreateBrowser:
"""
Creates and instantiates selenium webbrowser instances.
Two strategies: remote via selenium server or local.
The advantage of using remote is the browser instance can stay
open after the process closes. This allows you to disconnect
and re-connect to the same browser instance with different
processes.
"""
def __init__(self, pickle_filename=None):
self._pickle_filename = None
if pickle_filename is not None:
self.pickle_filename = pickle_filename
self._has_cleaned = False
@property
def pickle_filename(self):
"""
Reference to the user-data-dir pickle target filename, for
saving browser references.
"""
if self._pickle_filename is None:
self._pickle_filename = os.path.join(
appdirs.user_data_dir('wabot'),
'saved_browser_instances.pickle'
)
return self._pickle_filename
else:
return self._pickle_filename
@pickle_filename.setter
def pickle_filename(self, value):
self._pickle_filename = value
def _clean_old_pickles(self):
if self._has_cleaned is True:
LOGGER.debug('not cleaning pickles twice')
return
p = self.pickle_filename
try:
fp = open(p, 'rb')
pickles = pickle.load(fp)
if not pickles:
raise Exception
LOGGER.info('cleaning any old pickle refs')
for pickle_ref in pickles:
driver_ref = pickles[pickle_ref]
ts = driver_ref.timestamp
if datetime.datetime.now() - driver_ref.timestamp >= datetime.timedelta(days=3):
LOGGER.info('deleting old pickled driver ref: %s at %s', pickle_ref, ts)
pickles.pop(final_name)
self._has_cleaned = True
except (FileNotFoundError, IOError) as ex:
LOGGER.error('unable to load pickles: no pickled drivers found')
except Exception as ex:
print(ex)
pass
def _load_from_pickle(self, final_name):
p = self.pickle_filename
# Definitely no browser instance already, we must instantiate
if not os.path.exists(p):
LOGGER.debug('no pickled file for saved browser instances (nothing saved yet)')
return
# There MAY be an open browser or an invalidated reference to a once-open browser
if os.path.exists(p):
LOGGER.debug('found pickled file for saved browser instances: %s', p)
# First, see if existing session_name browser instance exists
fp = None
self._clean_old_pickles()
try:
fp = open(p, 'rb')
# drivers = pickle.load(fp)
pickles = pickle.load(fp)
if not pickles:
raise Exception
for idx, saved_instance in enumerate(pickles):
LOGGER.debug('found saved browser instances (%s): %s', idx, saved_instance)
driver_ref = pickles.get(final_name)
if not driver_ref:
raise Exception
# if datetime.datetime.now() - driver_ref.timestamp >= datetime.timedelta(days=3):
# pickles.pop(final_name)
driver = driver_ref.browser_ref
LOGGER.debug('connected to pickled webdriver instance: %s', final_name)
url = driver.current_url # throw error if driver isn't reliable anymore
LOGGER.info('webdriver instance is ready')
# self.driver = driver
return driver
except (FileNotFoundError, IOError) as ex:
LOGGER.error('unable to connect to existing webdriver: no pickled drivers found')
except Exception as ex:
print(ex)
self.driver = None
def _save_to_pickle(self, final_name, driver_ref):
p = self.pickle_filename
pickles = {}
# Definitely no browser instance already, we must instantiate
if not os.path.exists(p):
LOGGER.debug('no pickled file for saved browser instances (nothing saved yet)')
# Create usr-app-dir if doesn't exist
dirname = os.path.dirname(p)
if not os.path.isdir(dirname):
os.mkdir(dirname)
# There MAY be an open browser or an invalidated reference to a once-open browser
if os.path.exists(p):
LOGGER.debug('found existing pickle file for saved browser instances: %s', p)
# First, see if existing session_name browser instance exists
self._clean_old_pickles()
fp = open(p, 'rb')
pickles = pickle.load(fp)
if not pickles:
raise Exception
LOGGER.debug('found saved browser instances while saving: %s', list(pickles.keys()))
fp.close()
# Save to pickle - creating file if necessary
pkl = PickledBrowserReference()
pkl.browser_ref = driver_ref
pkl.timestamp = datetime.datetime.now()
pickles[final_name] = pkl
import pprint;
print('pickles:')
pprint.pprint(pickles)
# pickle file must be created
fp = open(p, 'wb')
# drivers[final_name] = driver
LOGGER.info('saving browser instance to pickle: %s', final_name)
pickle.dump(pickles, fp)
fp.close()
def _create_driver_remote_chromium(self, session_name):
"""
Creates and returns Selenium1 chromium instance.
"""
p = self.pickle_filename
# e.g.: remote-chromium-webdriver
final_name = '{}-{}'.format('remote-chromium', session_name)
driver = None
driver = self._load_from_pickle(final_name)
if driver is not None:
LOGGER.info('webdriver instance is ready (from pickle)')
return driver
# At this point, need to instantiate a new browser instance
sel_host = REMOTE_EXECUTOR % (EXECUTOR_PORT)
LOGGER.info('instantianting new browser instance (remote_chromium)')
LOGGER.info('remote selenium: %s', sel_host)
opt = selenium.webdriver.chrome.options.Options()
opt.add_argument("--user-agent=" + USER_AGENT)
opt.add_argument("--kiosk-printing")
opt.add_argument("--focus-existing-tab-on-open=false")
driver = selenium.webdriver.Remote(
command_executor=sel_host,
desired_capabilities = opt.to_capabilities()
)
self._save_to_pickle(final_name, driver)
# self.driver = driver
return driver
# Pickle impl. is duped here
def _create_driver_remote_firefox(self, session_name):
"""
Creates and returns Selenium1 firefox instance.
"""
final_name = '{}-{}'.format('remote-firefox', session_name)
driver = None
driver = self._load_from_pickle(final_name)
if driver is not None:
LOGGER.info('webdriver instance is ready (from pickle)')
return driver
# At this point, need to instantiate a new browser instance
sel_host = REMOTE_EXECUTOR % (EXECUTOR_PORT)
LOGGER.info('instantianting new browser instance (remote_firefox)')
LOGGER.info('remote selenium: %s', sel_host)
profile = selenium.webdriver.FirefoxProfile()
profile.set_preference("general.useragent.override", USER_AGENT)
driver = selenium.webdriver.Remote(
# SELENIUM1_SERVER_PATH,
sel_host,
selenium.webdriver.DesiredCapabilities.FIREFOX.copy(),
browser_profile = profile
)
self._save_to_pickle(final_name, driver)
return driver
def _create_driver_chromium2(self):
opt = selenium.webdriver.chrome.options.Options()
opt.add_argument("--user-agent=" + USER_AGENT)
opt.add_argument("--kiosk-printing")
driver = selenium.webdriver.Chrome(chrome_options = opt)
self.driver = driver
return driver
def _create_driver_chromium1(self):
# Selenium 1 - Chrome without working user agent switch
# These two methods of creation ChromeOptions are equivalent objects
options = selenium.webdriver.ChromeOptions()
options.add_argument("--user-agent=" + USER_AGENT)
driver = selenium.webdriver.Remote(desired_capabilities = options.to_capabilities())
driver = selenium.webdriver.Remote(SELENIUM1_SERVER_PATH,
selenium.webdriver.DesiredCapabilities.CHROME.copy())
return driver
def _create_driver_firefox2(self):
# tmp = selenium.webdriver.FirefoxProfile()
# tmp = None
profile = None
# filename = "/tmp/firefox_profile"
# try:
# fp = open(filename, "rb")
# profile = pickle.load(fp)
# except:
# pass
if not profile:
profile = selenium.webdriver.FirefoxProfile(profile_directory = "/home/mathew/firefox_prof")
profile.set_preference("general.useragent.override", USER_AGENT )
profile.set_preference("browser.helperApps.neverAsk.saveToDisk", "text/csv");
profile.set_preference("network.http.redirection-limit", "0" )
# profile.set_preference("javascript.enabled", False )
# profile.set_preference("print.always_print_silent", True)
profile.set_preference("print.print_to_file", True)
profile.set_preference("print.print_to_filename", "/tmp/print.pdf")
profile.update_preferences()
profile.set_preference("network.http.redirection-limit", "0" )
# with open("/tmp/firefox_profile", "wb") as fp:
# pickle.dump(profile, fp, pickle.HIGHEST_PROTOCOL)
# driver = selenium.webdriver.Firefox()
driver = selenium.webdriver.Firefox(profile)
return driver
def _create_driver_phantomjs(self):
# Note(MG): Selenium support for PhantomJS has been deprecated, please use headless
# driver = selenium.webdriver.PhantomJS()
# return driver
opt = selenium.webdriver.chrome.options.Options()
opt.add_argument("--user-agent=" + USER_AGENT)
opt.add_argument("--kiosk-printing")
opt.add_argument("--headless")
driver = selenium.webdriver.Chrome(chrome_options = opt)
driver.set_window_size(838, 907)
self.driver = driver
return driver

@ -1,110 +0,0 @@
import logging
import random
import selenium
import time
LOGGER = logging.getLogger('wabot')
class PageObject:
"""
Wrapper around page element that provides an object orientated interface
to elements. Can be sublcassed to integrate more complicated instruction.
"""
def __init__(self, page, accessors=None, name=None):
self.page = page
self.driver = page.driver
self.accessors = accessors
self.name = name
# self.el = None
def __getattr__(self, name):
# if not self.el:
# raise AttributeError
try:
return getattr(self.el, name)
except AttributeError as ex:
raise
def selenium_element(self, accessors=None):
"""
Creates and returns a selenium webelement for
interfacing with the page.
"""
if not accessors:
accessors = self.accessors
accessor = accessors
by = accessor[0]
value = accessor[1]
el = self.page.driver.find_element(by=by, value=value)
return el
def click(self):
return self.page.click(self.el)
def click_and_go(self):
time.sleep(random.uniform(3,6))
return self.page.click_and_go(self.el)
# def get_value(self):
# def set_value(self, value):
# def custom, e.g. zoom(self, x, y, dx, dy)
class TextField(PageObject):
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
el = self.selenium_element(kwargs['accessors'])
self.el = el
def get_value(self):
return self.page.get_el_value(self.el)
def set_value(self, value):
LOGGER.info('[%s]->set_text (%s)' % (self.name, value))
time.sleep(random.uniform(3,5))
return self.page.set_el_value(self.el, value)
class SelectField(PageObject):
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
el = self.selenium_element(kwargs['accessors'])
dropdown = selenium.webdriver.support.ui.Select(el)
self.el = el
self.dropdown = dropdown
def get_value(self):
return self.page.get_select_value(self.dropdown)
def set_value(self, value):
LOGGER.info('[%s]->set_select (%s)' % (self.name, value))
time.sleep(random.uniform(6,11))
return self.page.set_select_value(self.dropdown, value)
def select_by_index(self, idx):
return self.dropdown.select_by_index(idx)
class CheckField(PageObject):
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
el = self.selenium_element(kwargs['accessors'])
self.el = el
def get_checked(self, ignore=False):
return self.page.get_checkbox_value(self.el, ignore)
def set_checked(self, checked):
LOGGER.info('[%s]->set_checked (%s)' % (self.name, checked))
time.sleep(random.uniform(2,3))
return self.page.set_checkbox(self.el, checked)
class NullField(PageObject):
def __init__(self, el, *args, **kwargs):
self.el = el
def __getattr__(self, attr):
raise Exception
def __bool__(self):
return False

@ -1,537 +0,0 @@
import enum
import random
import selenium.webdriver
import time
from selenium.webdriver.support.ui import WebDriverWait
from selenium.webdriver.support import expected_conditions as EC
import selenium.common.exceptions
from .fields import *
import selenium.webdriver.support.ui
from selenium.webdriver.common.by import By
from PIL import Image
import os
import inspect
import logging
LOGGER = logging.getLogger('wabot')
ENABLE_GOTO_CLINIC_SELECT_OPTIMIZATION = True
ALERT_TIMEOUT = 3
class OBJ_T(enum.Enum):
ELEMENT = 1
ELEMENT_ARRAY = 2
SELECT = 3
CUSTOM = 4
class RC(enum.Enum):
FAILURE = 0
SUCCESS = 1
EVENT_WITHIN_21_DAYS = 2
DUPLICATE_EVENT = 3
NO_FACILITY = 4
MISSING_REPORTING_PLAN = 5
class Page:
"""
Provides ancillary utility methods such as finding
elements, interacting with forms, and taking screenshots.
"""
def __init__(self, parent):
self.parent = parent
self.driver = parent.driver
LOGGER.info("Loaded Page() '%s'", self.__class__.__name__)
def find_element_locators(self, key):
"""
Returns the generator data for an element, i.e. what
class to use and the arguments to construct it with.
This is found/stored in the static class definition,
page.elements = {key: (accessors)}
"""
class_hierarchy = inspect.getmro(type(self))
locators = None
for cls in class_hierarchy:
if not hasattr(cls, 'elements'): # no elements defined here...
continue
locators = cls.elements.get(key)
if not locators: # our element isn't defined here
continue
return locators # return first found match
def get_proxy(self, key):
locators = self.find_element_locators(key)
if not locators:
LOGGER.warn('element not found: %s' % (key))
return NullField(None, page=self)
obj_type = locators[0]
accessors = locators[1]
# if not el:
# LOGGER.warn('failed to make proxy object: %s' % key)
# return NullField(None, page=self)
if obj_type == 'el':
obj = TextField(page=self, accessors=accessors, name=key)
return obj
elif obj_type == 'select':
obj = SelectField(page=self, accessors=accessors, name=key)
return obj
elif obj_type == 'checkbox':
obj = CheckField(page=self, accessors=accessors, name=key)
return obj
elif obj_type == 'els':
el = self.find_element(key)
return self.find_element(key)
elif isinstance(obj_type, type):
# by = accessors[0]
# value = accessors[1]
# el = self.driver.find_element(by=by, value=value)
obj = obj_type(idelement=accessors[1], page=self, accessors=accessors,
name=key)
return obj
else:
LOGGER.error('failed to create page element type: %s' % (obj_type))
LOGGER.error('requested unknown page element: %s' % key)
return
def __getitem__(self, key):
# input('__getitem__ %s' % key)
# return self.find_element(key) # DEPRECATED
return self.get_proxy(key)
def find_element(self, key):
"""
Accesses element on page
"""
locators = self.find_element_locators(key)
if not locators:
LOGGER.error('failed to find page element: %s' % (key))
return False
try:
it = iter(locators)
type_ = next(it)
if type_ == 'custom':
cls = next(it)
locators = []
for locator in it:
locators.append(locator)
obj = cls(self, OBJ_T.CUSTOM, locators)
# print(obj)
return obj
for locator in it:
# print('locator =', locator)
if type_ == 'el' or type_ == 'checkbox':
try:
by = locator[0]
value = locator[1]
el = self.driver.find_element(by=by, value=value)
LOGGER.trace('found single element (%s) by %s = %s', key, by, value)
return el
except Exception as ex:
LOGGER.warn('failed to find single element (%s) by %s = %s', key, by, value)
continue
elif type_ == 'els':
try:
by = locator[0]
value = locator[1]
els = self.driver.find_elements(by=by, value=value)
LOGGER.trace('found elements (%s) by %s = %s', key, by, value)
return els
except Exception as ex:
LOGGER.warn('failed to find any elements (%s) by %s = %s', key, by, value)
continue
elif type_ == 'select':
try:
by = locator[0]
value = locator[1]
el = self.driver.find_element(by=by, value=value)
LOGGER.trace('found dropdown element (%s) by %s = %s', key, by, value)
dropdown = selenium.webdriver.support.ui.Select(el)
return dropdown
except Exception as ex:
LOGGER.warn('failed to find dropdown (%s) by %s = %s', key, by, value)
continue
else:
LOGGER.error('unable to find element (%s): unknown type = %s' % (key, type_))
return
except Exception as ex:
print(ex)
def verify(self):
# The page proxy api calls PageObject.verify() if it exists, which
# should call super().verify(). It is necessary for every class in
# the hierarchy to have a callable verify(). Every class in the
# hierarchy should implement verify method or subclass from here.
return True
def click_and_go(self, el):
rc = self.click(el)
if not rc:
return rc
self.accept_alert()
LOGGER.debug('waiting for page to load...')
rc = self._wait_for_element_to_go_stale(el)
if not rc:
LOGGER.error('failed: timed out waiting for page load')
return False
return rc
def _wait_for_element_to_go_stale(self, el):
try:
wait = selenium.webdriver.support.ui.WebDriverWait(self.driver, 10)
wait.until(lambda driver: self.is_element_stale(el))
return True
except selenium.common.exceptions.TimeoutException as ex:
LOGGER.error('failed: timed out waiting for page load')
return False
def click(self, el):
"""
Clicks element at a random coordinate based on the size of the
element.
By default, selenium clicks elements at pos(0, 0). NHSN records
where things are clicked at.
"""
if not el:
LOGGER.warn("refusing to click null element")
return False
try:
size = el.size
except selenium.common.exceptions.StaleElementReferenceException as ex:
LOGGER.error('failed to click element: stale reference')
return False
# Use a guassian distribution to click more often towards the center
width = size["width"]
if width > 4:
x = random.gauss((width/2), (width/7))
if x < 0: x = 1
elif x > width: x = width - 1
else:
i = 0
while i < 10:
try:
if el.is_displayed() and el.is_enabled():
el.click()
return True
except selenium.common.exceptions.ElementNotVisibleException as ex:
LOGGER.error('failed to click element: not visible')
return False
except selenium.common.exceptions.StaleElementReferenceException as ex:
LOGGER.error('failed to click element: stale reference')
return False
time.sleep(.2)
i = i+1
return False
height = size["height"]
if height > 4:
y = random.gauss((height/2), (height/7))
if y < 0: y = 1
elif y > height: y = height -1
else:
el.click()
return
LOGGER.debug( "clicking %s (dim: x = %s, y = %s) at %d, %d" %
(self.get_el_identifier(el), size["width"], size["height"], x,y) )
i = 0
n = 20
while i < n:
if el.is_displayed() and el.is_enabled():
break
if i == n:
raise Exception("unable to click element")
time.sleep(.5)
i = i + 1
# return el.click()
x = int(x)
y = int(y)
# print('size is', width, height)
# print('clicking through selenium actions at', x, y)
actions = selenium.webdriver.ActionChains(self.driver)
actions.move_to_element_with_offset(el, x, y)
actions.click()
try:
actions.perform()
except Exception as ex: # type is selenium timeout... not sure
print(ex)
LOGGER.error("%s" % (ex))
return False
return True
def save_screenshot(self, filename):
LOGGER.info("saving page screenshot: %s" % (filename))
# Chromium2 screenshot only captures viewable area,
# selenium is waiting on WebDriver which is waiting
# on chromium. Doesn't look like it will be fixed soon.
# more info:
# https://code.google.com/p/chromedriver/issues/detail?id=294
#
# For now, the workaround is stitch it together for chromium.
# Firefox2 works fine but has it's own problems, hence the
# chromium stitch.
if self.parent.driver_type == "chromium2":
self.chrome_take_full_page_screenshot(filename)
else:
self.driver.get_screenshot_as_file(filename)
def chrome_take_full_page_screenshot(self, file):
self.driver.maximize_window()
# Global.scroll_to_zero()
time.sleep(0.2)
# Log.Debug("Starting chrome full page screenshot workaround ...")
total_width = self.driver.execute_script("return document.body.offsetWidth")
total_height = self.driver.execute_script("return document.body.parentNode.scrollHeight")
viewport_width = self.driver.execute_script("return document.body.clientWidth")
viewport_height = self.driver.execute_script("return window.innerHeight")
# Log.Debug("Total: ({0}, {1}), Viewport: ({2},{3})".format(total_width, total_height,viewport_width,viewport_height))
rectangles = []
i = 0
while i < total_height:
ii = 0
top_height = i + viewport_height
if top_height > total_height:
top_height = total_height
while ii < total_width:
top_width = ii + viewport_width
if top_width > total_width:
top_width = total_width
# Log.Debug("Appending rectangle ({0},{1},{2},{3})".format(ii,i,top_width,top_height))
rectangles.append((ii,i,top_width,top_height))
ii = ii + viewport_width
i = i + viewport_height
stitched_image = Image.new('RGB', (total_width, total_height))
previous = None
part = 0
for rectangle in rectangles:
if not previous is None:
self.driver.execute_script("window.scrollTo({0}, {1})".format(rectangle[0], rectangle[1]))
# Log.Debug("Scrolled To ({0},{1})".format(rectangle[0], rectangle[1]))
time.sleep(0.2)
tmp_path = "/tmp/"
file_name = "{0}scroll_{1}_part_{2}.png".format(tmp_path, 1, part)
# file_name = "/tmp/screen.png"
# Log.Debug("Capturing {0} ...".format(file_name))
self.driver.get_screenshot_as_file(file_name)
screenshot = Image.open(file_name)
# offset = (rectangle[0], rectangle[1])
if rectangle[1] + viewport_height > total_height:
offset = (rectangle[0], total_height - viewport_height)
else:
offset = (rectangle[0], rectangle[1])
# Log.Debug("Adding to stitched image with offset ({0}, {1})".format(offset[0],offset[1]))
stitched_image.paste(screenshot, offset)
del screenshot
os.remove(file_name)
part = part + 1
previous = rectangle
stitched_image.save(file)
# Log.Debug("Finishing chrome full page screenshot workaround ...")
return True
def get_el_value(self, el):
if not el:
return None
val = el.get_attribute("value")
# LOGGER.debug('peeked at field <%s>, value = %s'
# % (self.get_el_identifier(el), val))
return val
def get_el_text(self, el):
if not el:
return None
return el.text
def set_el_value(self, el, value, js=False, slow_type=False):
if not el:
return False
prev_val = self.get_el_value(el)
if value is None:
el.clear()
return
el.clear()
if js:
# doesn't work
rc = self.driver.execute_script("arguments[0].setAttribute('value', '%s');arguments[0].onchange();" % (value), el)
else:
if slow_type:
try:
for ch in value:
el.send_keys(ch)
time.sleep(1)
except Exception as ex:
LOGGER.error("failed to send keys, element in unknown state!!: %s" % (ex))
return False
else:
try:
el.send_keys(value)
except Exception as ex:
LOGGER.error("failed to send keys, element in unknown state!!: %s" % (ex))
return False
el_value = self.get_el_value(el)
if str(el_value) != str(value):
print(type(el_value), type(value))
print("values didn't match.", value, el_value)
return False
# LOGGER.info('set field <%s> -> %s; previous = %s'
# % (self.get_el_identifier(el), el_value, prev_val))
return True
def get_select_value(self, select):
if not select:
LOGGER.error("tried to get select value of NULL element")
return None
try:
value = select.first_selected_option.get_attribute("value")
except selenium.common.exceptions.NoSuchElementException as ex:
value = None
return value
def set_select_value(self, select, value=None, text=None):
if not select:
return False
# LOGGER.trace("setting select value (%s) for (%s)"\
# % (value, self.get_el_identifier(select._el)))
if value:
try:
select.select_by_value(str(value))
return True
except Exception as ex:
return False
elif text:
select.select_by_visible_text(text)
return False
def set_checkbox(self, el, checked):
if not el:
return False
is_enabled = el.is_enabled()
is_selected = el.is_selected()
# print("enabled, selected, check:", is_enabled, is_selected, checked)
if not is_enabled:
return False
if checked:
if not is_selected:
# LOGGER.trace("checking unchecked box (%s)" % (self.get_el_identifier(el)))
self.click(el)
else:
if is_selected:
# LOGGER.trace("unchecking checked box (%s)" % (self.get_el_identifier(el)))
self.click(el)
val = self.get_checkbox_value(el)
if val != checked:
pass
# False != None
# print('checkbox not checked properly?', val, checked)
# input('checkbox not checked properly?')
# todo(mathew guest) assert checked
return True
def get_checkbox_value(self, el, ignore_disabled=False):
"""
Returns the checked status of a checkbox element. True if enabled
and checked, False if disabled or unchecked.
"""
if not el:
return None
return (ignore_disabled or el.is_enabled()) and el.is_selected()
def get_el_identifier(self, el):
"""
Returns a quick identifier to refer an element by.
"""
id_ = el.get_attribute("id")
name = el.get_attribute("name")
class_ = el.get_attribute("style")
if id_:
return id_
if name:
return name
if class_:
return class_
def accept_alert(self, accept=True, timeout=ALERT_TIMEOUT):
"""
Looks for and tries to accept a javascript alert if it exists.
returns bool: whether or not an alert was accepted.
There is a timeout penalty when calling this method if there is no alert.
"""
try:
selenium.webdriver.support.ui.WebDriverWait(self.driver, timeout).\
until(selenium.webdriver.support.expected_conditions.alert_is_present(),
'Timed out waiting alert' )
alert = self.driver.switch_to_alert()
text = alert.text
if accept:
alert.accept()
else:
alert.dismiss()
LOGGER.trace('caught js alert: %s' % text)
return text
except selenium.common.exceptions.TimeoutException:
LOGGER.trace('no js alert present')
return False
def is_element_stale(self, webelement):
"""
Checks if a webelement is stale.
@param webelement: A selenium webdriver webelement
"""
try:
webelement.tag_name
except selenium.common.exceptions.StaleElementReferenceException:
return True
except NameError:
pass
except:
pass
return False