diff --git a/.github/workflows/python-release.yml b/.github/workflows/python-release.yml new file mode 100644 index 00000000..d2c434fd --- /dev/null +++ b/.github/workflows/python-release.yml @@ -0,0 +1,52 @@ +name: Manual PyPI Publish + +# Mirrors release.yml's manual, button-triggered shape for the Python adapter. +# Unlike npm (NPM_TOKEN), PyPI uses trusted publishing (OIDC) β€” no secret. +# Bump the version in packages/selenium-py-devtools/pyproject.toml before running. + +on: + workflow_dispatch: + inputs: + target: + description: 'Publish target' + required: true + type: choice + default: pypi + options: + - pypi + - testpypi + +defaults: + run: + working-directory: packages/selenium-py-devtools + +jobs: + release: + runs-on: ubuntu-latest + environment: ${{ inputs.target }} + permissions: + id-token: write # PyPI trusted publishing (OIDC) β€” no token/secret needed + steps: + - uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd # v6.0.2 + with: + ref: 'main' + - uses: actions/setup-python@a26af69be951a213d495a4c3e4e4022e16d87065 # v5 + with: + python-version: '3.12' + - name: πŸ§ͺ Unit tests (release guard) + run: PYTHONPATH=src python -m unittest discover -s tests + - name: πŸ“¦ Build sdist + wheel + run: | + python -m pip install --upgrade build + python -m build + - name: πŸš€ Publish to PyPI + if: ${{ inputs.target == 'pypi' }} + uses: pypa/gh-action-pypi-publish@cef221092ed1bacb1cc03d23a2d87d1d172e277b # release/v1 + with: + packages-dir: packages/selenium-py-devtools/dist + - name: πŸš€ Publish to TestPyPI + if: ${{ inputs.target == 'testpypi' }} + uses: pypa/gh-action-pypi-publish@cef221092ed1bacb1cc03d23a2d87d1d172e277b # release/v1 + with: + packages-dir: packages/selenium-py-devtools/dist + repository-url: https://test.pypi.org/legacy/ diff --git a/.github/workflows/python.yml b/.github/workflows/python.yml new file mode 100644 index 00000000..7a56aca4 --- /dev/null +++ b/.github/workflows/python.yml @@ -0,0 +1,37 @@ +name: Python Adapter + +on: + push: + branches: + - main + paths: + - packages/selenium-py-devtools/** + - packages/shared/** + - .github/workflows/python.yml + pull_request: + paths: + - packages/selenium-py-devtools/** + - packages/shared/** + - .github/workflows/python.yml + +defaults: + run: + working-directory: packages/selenium-py-devtools + +jobs: + test: + runs-on: ubuntu-latest + strategy: + matrix: + python-version: ['3.9', '3.12'] + steps: + - uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd # v6.0.2 + - uses: actions/setup-python@a26af69be951a213d495a4c3e4e4022e16d87065 # v5 + with: + python-version: ${{ matrix.python-version }} + - name: Contract is in sync with shared + run: | + python scripts/gen_contract.py + git diff --exit-code src/wdio_selenium_devtools/_contract.py + - name: πŸ§ͺ Unit tests + run: PYTHONPATH=src python -m unittest discover -s tests diff --git a/examples/selenium/python-test/.gitignore b/examples/selenium/python-test/.gitignore new file mode 100644 index 00000000..24de4231 --- /dev/null +++ b/examples/selenium/python-test/.gitignore @@ -0,0 +1,2 @@ +# Screencast videos are generated next to the test file. +*.webm diff --git a/examples/selenium/python-test/web_form.py b/examples/selenium/python-test/web_form.py new file mode 100644 index 00000000..3cfc4810 --- /dev/null +++ b/examples/selenium/python-test/web_form.py @@ -0,0 +1,49 @@ +"""A normal Selenium (Python) test β€” with WebdriverIO DevTools added. + +Only two lines differ from a plain Selenium script (marked ← devtools): +``import wdio_selenium_devtools`` and ``devtools.enable()``. Everything the +driver does is then captured and shown live in the DevTools dashboard. + +Run it: + + pip install wdio-selenium-devtools selenium + python examples/selenium/python-test/web_form.py + +The dashboard opens in a dedicated window and captures every command. It stays +open after the test so you can inspect it β€” close the window (or Ctrl-C) to +finish. The screencast .webm is written next to this file. Requires a +ChromeDriver matching your Chrome (Selenium 4.6+ auto-manages one if none is on +PATH). +""" + +import wdio_selenium_devtools as devtools # ← devtools +from selenium import webdriver +from selenium.webdriver.chrome.options import Options +from selenium.webdriver.common.by import By + +devtools.enable() # ← devtools: open the dashboard + capture every command + +options = Options() +options.add_argument("--headless=new") # remove this line to watch the browser +options.add_argument("--window-size=1280,1024") # bigger viewport β†’ fuller screencast +driver = webdriver.Chrome(options=options) +try: + driver.get("https://www.selenium.dev/selenium/web/web-form.html") + + title = driver.title + + driver.implicitly_wait(0.5) + + text_box = driver.find_element(by=By.NAME, value="my-text") + submit_button = driver.find_element(by=By.CSS_SELECTOR, value="button") + + text_box.send_keys("Selenium") + submit_button.click() + + message = driver.find_element(by=By.ID, value="message") + text = message.text + print("form submitted, received message:", text) # shows in the dashboard's Console +finally: + driver.quit() + devtools.wait_for_dashboard_close() # ← devtools: keep the UI open to inspect + devtools.disable() # ← devtools diff --git a/packages/selenium-py-devtools/.gitignore b/packages/selenium-py-devtools/.gitignore new file mode 100644 index 00000000..69b17f32 --- /dev/null +++ b/packages/selenium-py-devtools/.gitignore @@ -0,0 +1,8 @@ +__pycache__/ +*.py[cod] +.pytest_cache/ +.venv/ +venv/ +build/ +dist/ +*.egg-info/ diff --git a/packages/selenium-py-devtools/README.md b/packages/selenium-py-devtools/README.md new file mode 100644 index 00000000..b269d904 --- /dev/null +++ b/packages/selenium-py-devtools/README.md @@ -0,0 +1,193 @@ +# wdio-selenium-devtools (Python) + +Python Selenium adapter for the WebdriverIO DevTools dashboard β€” the fourth +adapter alongside the JS WebdriverIO / Nightwatch / Selenium-JS ones. It feeds +the **same backend and UI**, unchanged, over the language-neutral +`{scope, data}` WebSocket contract. + +**Status: Phase 1 + 2.** Live command capture + test tree; browser console & +network via BiDi; screencast video; and a dashboard window that auto-opens and +tears down with the run. Verified against real headless Chrome. See +[Roadmap](#roadmap) for what's deferred. + +## Install (dev) + +```bash +pip install -e packages/selenium-py-devtools # or: pip install wdio-selenium-devtools (when published) +``` + +The transport is **dependency-free** (stdlib WebSocket client). The only thing +on top of your own `selenium` install is this package; `pytest` is optional. + +## Use + +**With pytest (recommended) β€” no code changes to your tests:** + +```bash +WDIO_DEVTOOLS=1 pytest tests/ # DEVTOOLS_PORT= also opts in (and attaches) +``` + +The bundled plugin auto-captures the run, opens the dashboard in a dedicated +window, and β€” after the run β€” **keeps it open so you can inspect it**; close the +window (or Ctrl-C) to finish. Nothing devtools-specific goes in your test files. + +**Without pytest** (any script / unittest) β€” add two lines to a normal Selenium +script (`devtools.enable()` + `devtools.wait_for_dashboard_close()`): + +```python +import wdio_selenium_devtools as devtools + +devtools.enable() # open dashboard + capture every command +# ... your normal selenium code, ending with driver.quit() ... +devtools.wait_for_dashboard_close() # keep the UI open to inspect (no-op if headless) +devtools.disable() +``` + +Runnable example: [`examples/selenium/python-test/web_form.py`](../../examples/selenium/python-test/web_form.py). +If the backend can't be launched or reached, `enable()` warns and returns +`None` β€” capture is skipped, your tests still run. + +**ChromeDriver:** you need one matching your Chrome (a mismatch breaks all +Selenium, not just this). Selenium 4.6+ auto-manages it when no `chromedriver` +is on `PATH`; otherwise keep it current (`brew upgrade chromedriver`). + +## What it captures + +| Data | How | Scope | Phase | +|---|---|---|---| +| Commands (driver + element) | wrap `WebDriver.execute()` β€” the single chokepoint all commands flow through | `commands` | 1 | +| Session metadata | read `session_id` + `caps` on the first ready command | `metadata` | 1 | +| Test / suite tree | pytest plugin (`pytest_runtest_logreport` / `sessionfinish`) | `suites` | 1 | +| Browser console + JS errors | Selenium **BiDi** (`driver.script` handlers) | `consoleLogs` | 2 | +| Network requests | Selenium **BiDi** (low-level subscribe, no interception) | `networkRequests` | 2 | +| DOM snapshot (preview iframe) | inject `packages/script`, re-inject per navigation, drain mutations | `mutations` | 2 | +| Screencast video | screenshot polling β†’ ffmpeg-encoded `.webm` | `screencast` | 2 | + +Element actions (`click`, `send_keys`, `text`, …) are captured for free: they +delegate to `self._parent.execute`, so the one wrapper sees them as +`clickElement`, `getElementText`, etc. + +**BiDi is auto-enabled** β€” the adapter injects the `webSocketUrl` capability +into the `newSession` request so console/network work out-of-box (opt out with +`WDIO_DEVTOOLS_BIDI=0`). **Screencast** needs `ffmpeg` on PATH to encode the +`.webm`; without it, recording is skipped (one warning, no error). + +## Dashboard window lifecycle + +Like the JS adapters, `enable()` opens the dashboard in a dedicated, closable +Chrome window; closing that window (backend `clientDisconnected`) shuts the run +down, and ending the process (exit / Ctrl-C) closes the window. Auto-open is on +when stdout is a TTY; force it with `WDIO_DEVTOOLS_OPEN=1` or disable with `=0`. + +## Layout + +``` +src/wdio_selenium_devtools/ + __init__.py public API β€” enable() / disable() / get_capturer() + constants.py defaults, env-var names, skip sets, pinned backend version + types.py TypedDicts for the wire payloads (mirror packages/shared) + _contract.py GENERATED from packages/shared β€” scope names + CONTRACT_VERSION + utils.py framework-agnostic helpers (now_ms, iso, to_jsonable, call_source) + frames.py pure builders for each {scope,data} payload + transport.py stdlib WebSocket client (handshake, masked frames, ping/pong, control reader) + capturer.py SessionCapturer: command IDs, normalizeβ†’send, metadata-once + instrumentation.py execute() wrap + BiDi auto-enable + session-setup hook + bidi.py BiDi console/JS-error + network capture (pure mapping + wiring) + screencast.py screenshot-polling recorder + ffmpeg webm encode + backend.py launch-or-attach the Node backend + port discovery + lifecycle.py dashboard window open/close + shutdown-on-disconnect + pytest_plugin.py suite/test tree feeder (opt-in) +scripts/gen_contract.py regenerate _contract.py from shared (dev-time; also a drift-guard) +tests/ stdlib-unittest unit tests (no selenium/pytest needed) +e2e_check.py real-Chrome smoke (plain script) +e2e/test_smoke.py real-Chrome smoke (pytest + plugin) +(examples live at repo root: examples/selenium/python-test/web_form.py) +``` + +## Backend & publishing + +Two artifacts, two registries β€” pip can't resolve the Node backend, so each +coupling is handled explicitly rather than via a `workspace:^`-style resolver: + +| | Local (monorepo) | Published | +|---|---|---| +| **Adapter** (this package) | `pip install -e` | PyPI: `pip install wdio-selenium-devtools` | +| **Backend + UI** (Node) | `node packages/backend/dist/index.js` | npm: `npx @wdio/devtools-backend@` | +| **Wire contract** (`shared`) | regenerated into `_contract.py` | the generated `_contract.py` ships in the wheel | + +`enable()` obtains the backend in this order (local vs published falls out of it): + +1. `DEVTOOLS_PORT` set β†’ attach to an already-running backend (CI, manual). +2. `DEVTOOLS_BACKEND_CMD` set β†’ spawn that explicit command. +3. monorepo `packages/backend/dist/index.js` present β†’ spawn it (**local dev**). +4. else β†’ `npx @wdio/devtools-backend@` (**published**). + +The pinned `BACKEND_NPM_VERSION` in `backend.py` is the version link β€” there is +no auto-resolution, so it's bumped deliberately alongside a contract change. + +Regenerate the contract after any change to `packages/shared`: + +```bash +python3 packages/selenium-py-devtools/scripts/gen_contract.py +``` + +It fails loudly if a scope the adapter needs disappeared from `shared` β€” a +build-time drift alarm. + +## Test + +```bash +# unit (no deps): +PYTHONPATH=src python3 -m unittest discover -s tests -v + +# e2e (needs selenium + a running backend; Selenium Manager fetches the driver): +DEVTOOLS_PORT=3000 PYTHONPATH=src python3 e2e_check.py +DEVTOOLS_PORT=3000 PYTHONPATH=src pytest e2e/test_smoke.py -p wdio_selenium_devtools.pytest_plugin -q +``` + +## Release (approach A) + +Two workflows, mirroring the JS split (`ci.yml` tests / `release.yml` publish): + +- **`python.yml`** β€” runs on PRs + pushes touching this package or `shared`: + unit tests on Python 3.9 + 3.12, and a contract-drift check (regenerate + `_contract.py`, fail on any diff). Zero repo config needed. +- **`python-release.yml`** β€” **manual** (`workflow_dispatch`, like the JS + "Manual NPM Publish"), target `pypi` or `testpypi`. Builds the sdist + wheel + and publishes via **trusted publishing (OIDC)** β€” no token/secret. + +The wheel does **not** bundle the backend β€” approach A fetches a pinned +`@wdio/devtools-backend` via `npx` at runtime (Node 18+ required). Bundling it +(approach B/C) is a GA-time change. + +**One-time setup before the first publish** (this is what claims the PyPI name): + +1. On PyPI, add a **pending trusted publisher** for project + `wdio-selenium-devtools` β†’ owner `webdriverio`, repo `devtools`, workflow + `python-release.yml`, environment `pypi` (repeat on TestPyPI with env + `testpypi` if you want a dry run first). +2. Create matching GitHub **Environments** `pypi` (and `testpypi`). +3. Run the workflow β€” the first successful publish creates and claims the name. + +Each release: bump `version` in `pyproject.toml`, then run the workflow (PyPI +rejects re-uploading an existing version). + +## Roadmap + +- **Phase 2 (done)** β€” BiDi console/network + screenshot-polling screencast. + Not yet: a CDP `Page.startScreencast` push-mode fast-path, per-command + screenshots, and performance capture. +- **Phase 3** β€” trace export, preserve-and-rerun, action snapshots. Per the + architecture, the heavy post-processing is a candidate to live server-side in + the backend (written once) rather than re-implemented here. + +## Design notes + +- **Backend/UI unchanged.** This adapter only produces the wire frames; the + server routes and renders them exactly as for the JS adapters. +- **Capture never breaks tests.** Commands are recorded around the real call; + errors are captured *and re-raised* unchanged; a missing dashboard is a no-op. +- **Contract drift** is the main long-term risk (see the integration artifact). + Mitigated two ways: `_contract.py` is generated from `packages/shared` (scope + names + `CONTRACT_VERSION`), and the generator fails if a required scope + vanishes. Full field-level type generation is a future step. diff --git a/packages/selenium-py-devtools/e2e/test_smoke.py b/packages/selenium-py-devtools/e2e/test_smoke.py new file mode 100644 index 00000000..c6ca3f4c --- /dev/null +++ b/packages/selenium-py-devtools/e2e/test_smoke.py @@ -0,0 +1,35 @@ +"""pytest smoke that exercises the plugin (suites) + instrumentation (commands). + +Run against a running backend: + DEVTOOLS_PORT=63763 PYTHONPATH=src \ + pytest e2e/test_smoke.py -p wdio_selenium_devtools.pytest_plugin -q + +Uses a data: URL so it needs no network. +""" + +import pytest +from selenium import webdriver +from selenium.webdriver.chrome.options import Options +from selenium.webdriver.common.by import By + +PAGE = "data:text/html,

Hello DevTools

link" + + +@pytest.fixture +def driver(): + opts = Options() + opts.add_argument("--headless=new") + opts.add_argument("--no-sandbox") + drv = webdriver.Chrome(options=opts) + yield drv + drv.quit() + + +def test_homepage_loads(driver): + driver.get(PAGE) + assert "Hello DevTools" in driver.find_element(By.CSS_SELECTOR, "h1").text + + +def test_link_is_clickable(driver): + driver.get(PAGE) + driver.find_element(By.CSS_SELECTOR, "a").click() diff --git a/packages/selenium-py-devtools/e2e_check.py b/packages/selenium-py-devtools/e2e_check.py new file mode 100644 index 00000000..bc9e6225 --- /dev/null +++ b/packages/selenium-py-devtools/e2e_check.py @@ -0,0 +1,47 @@ +#!/usr/bin/env python3 +"""End-to-end smoke: real headless Chrome β†’ adapter β†’ backend. + +Run against a running backend (set DEVTOOLS_PORT if not 3000): + DEVTOOLS_PORT=63763 PYTHONPATH=src python3 e2e_check.py + +Uses a data: URL so it needs no network. +""" + +import sys +import time + +import wdio_selenium_devtools as devtools + +PAGE = "data:text/html,

Hello DevTools

link" + + +def main() -> int: + capturer = devtools.enable() + if capturer is None: + print("backend not reachable β€” start it first") + return 1 + + from selenium import webdriver + from selenium.webdriver.chrome.options import Options + from selenium.webdriver.common.by import By + + opts = Options() + opts.add_argument("--headless=new") + opts.add_argument("--no-sandbox") + driver = webdriver.Chrome(options=opts) + try: + driver.get(PAGE) + heading = driver.find_element(By.CSS_SELECTOR, "h1") + print("h1 text:", heading.text) + driver.find_element(By.CSS_SELECTOR, "a").click() + print("title:", driver.execute_script("return document.title")) + finally: + driver.quit() + time.sleep(1) # let frames flush + devtools.disable() + print("done") + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/packages/selenium-py-devtools/pyproject.toml b/packages/selenium-py-devtools/pyproject.toml new file mode 100644 index 00000000..abce70a2 --- /dev/null +++ b/packages/selenium-py-devtools/pyproject.toml @@ -0,0 +1,28 @@ +[build-system] +requires = ["hatchling"] +build-backend = "hatchling.build" + +[project] +name = "wdio-selenium-devtools" +version = "0.1.0" +description = "Python Selenium adapter for the WebdriverIO DevTools dashboard" +readme = "README.md" +requires-python = ">=3.9" +license = { text = "MIT" } +authors = [{ name = "WebdriverIO" }] +keywords = ["selenium", "webdriver", "devtools", "pytest", "debugging"] + +# Transport is dependency-free (stdlib). selenium is the user's own dep; we +# only patch it when present, so it's not a hard requirement to import. +dependencies = [] + +[project.optional-dependencies] +selenium = ["selenium>=4.6"] +test = ["pytest>=7", "selenium>=4.6"] + +# Auto-discovered by pytest; inert unless WDIO_DEVTOOLS / DEVTOOLS_PORT is set. +[project.entry-points.pytest11] +wdio_selenium_devtools = "wdio_selenium_devtools.pytest_plugin" + +[tool.hatch.build.targets.wheel] +packages = ["src/wdio_selenium_devtools"] diff --git a/packages/selenium-py-devtools/scripts/gen_contract.py b/packages/selenium-py-devtools/scripts/gen_contract.py new file mode 100644 index 00000000..32a5c7b4 --- /dev/null +++ b/packages/selenium-py-devtools/scripts/gen_contract.py @@ -0,0 +1,98 @@ +#!/usr/bin/env python3 +"""Generate ``_contract.py`` from ``packages/shared`` β€” the Python side of the +wire contract, derived from the single TS source of truth. + +Runs only in the monorepo (dev time); the generated file is committed and ships +inside the wheel, so published installs need neither this script nor `shared`. + +It doubles as a drift-guard: if a scope the adapter relies on disappears from +shared's ``TraceLog`` / ``WS_SCOPE``, generation fails loudly rather than +letting the Python side silently send a name the UI no longer understands. + +Run: python3 scripts/gen_contract.py +""" + +from __future__ import annotations + +import json +import re +import sys +from pathlib import Path + +# Data scopes the Python adapter emits β€” each must exist as a TraceLog key. +REQUIRED_DATA_SCOPES = { + "SCOPE_METADATA": "metadata", + "SCOPE_COMMANDS": "commands", + "SCOPE_CONSOLE_LOGS": "consoleLogs", + "SCOPE_NETWORK_REQUESTS": "networkRequests", + "SCOPE_SUITES": "suites", + "SCOPE_SCREENCAST": "screencast", + "SCOPE_SOURCES": "sources", + "SCOPE_MUTATIONS": "mutations", +} + + +def _repo_root() -> Path: + for parent in Path(__file__).resolve().parents: + if (parent / "packages" / "shared" / "package.json").exists(): + return parent + raise SystemExit("could not locate the monorepo root (packages/shared)") + + +def _shared_version(shared: Path) -> str: + pkg = json.loads((shared / "package.json").read_text()) + return pkg["version"] + + +def _trace_log_keys(types_ts: str) -> list[str]: + m = re.search(r"export interface TraceLog \{(.*?)\n\}", types_ts, re.DOTALL) + if not m: + raise SystemExit("could not find `interface TraceLog` in shared/types.ts") + return re.findall(r"^\s*(\w+)\??:", m.group(1), re.MULTILINE) + + +def _ws_scopes(routes_ts: str) -> dict[str, str]: + m = re.search(r"export const WS_SCOPE = \{(.*?)\n\} as const", routes_ts, re.DOTALL) + if not m: + raise SystemExit("could not find `WS_SCOPE` in shared/routes.ts") + return dict(re.findall(r"(\w+):\s*'([^']+)'", m.group(1))) + + +def main() -> int: + root = _repo_root() + shared = root / "packages" / "shared" + version = _shared_version(shared) + data_keys = _trace_log_keys((shared / "src" / "types.ts").read_text()) + control = _ws_scopes((shared / "src" / "routes.ts").read_text()) + + # Drift-guard. + missing = [v for v in REQUIRED_DATA_SCOPES.values() if v not in data_keys] + if missing: + raise SystemExit( + f"contract drift: scope(s) {missing} no longer in shared TraceLog " + f"(present: {data_keys}). Update the adapter or shared." + ) + + lines = [ + "# GENERATED by scripts/gen_contract.py from packages/shared.", + "# Do not edit by hand β€” run the script to regenerate.", + f'CONTRACT_VERSION = "{version}"', + "", + ] + for const, value in REQUIRED_DATA_SCOPES.items(): + lines.append(f'{const} = "{value}"') + lines += [ + "", + f"DATA_SCOPES = frozenset({sorted(data_keys)!r})", + f"CONTROL_SCOPES = frozenset({sorted(control.values())!r})", + "", + ] + out = shared.parent / "selenium-py-devtools" / "src" / "wdio_selenium_devtools" / "_contract.py" + out.write_text("\n".join(lines)) + print(f"wrote {out.relative_to(root)} (contract v{version}, " + f"{len(data_keys)} data scopes, {len(control)} control scopes)") + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/packages/selenium-py-devtools/src/wdio_selenium_devtools/__init__.py b/packages/selenium-py-devtools/src/wdio_selenium_devtools/__init__.py new file mode 100644 index 00000000..39f43a98 --- /dev/null +++ b/packages/selenium-py-devtools/src/wdio_selenium_devtools/__init__.py @@ -0,0 +1,148 @@ +"""wdio-selenium-devtools β€” Python Selenium adapter for the DevTools dashboard. + +Public API: + + import wdio_selenium_devtools as devtools + devtools.enable() # connect + instrument; reads DEVTOOLS_HOST/PORT + ... run selenium ... + devtools.disable() + +Under pytest, the bundled plugin calls these for you (gated on the +``WDIO_DEVTOOLS`` / ``DEVTOOLS_PORT`` env vars). The transport has no +third-party dependency; the only requirement on top is selenium itself. +""" + +from __future__ import annotations + +import os +import subprocess +import sys +from typing import Optional + +from . import backend, instrumentation, lifecycle +from ._contract import CONTRACT_VERSION +from .capturer import SessionCapturer +from .constants import DEFAULT_HOST, DEFAULT_PORT, ENV_HOST, ENV_PORT +from .logcapture import LogCapturer +from .terminal import TerminalCapturer +from .transport import WSClient + +__version__ = "0.1.0" +__all__ = [ + "enable", "disable", "get_capturer", "dashboard_url", + "wait_for_dashboard_close", "CONTRACT_VERSION", +] + +_active: dict = { + "capturer": None, "transport": None, "process": None, "url": None, + "handle": None, "terminal": None, "logs": None, +} + + +def enable( + host: Optional[str] = None, + port: Optional[int] = None, + *, + webdriver_cls: Optional[type] = None, +) -> Optional[SessionCapturer]: + """Connect to the backend and instrument Selenium. Idempotent. + + With no host/port and no ``DEVTOOLS_PORT``, the backend is launched + automatically (see :mod:`.backend`). Returns the SessionCapturer, or None if + the dashboard can't be reached/launched β€” a missing dashboard must never + break the user's test run. + """ + if _active["capturer"] is not None: + return _active["capturer"] + + process = None + try: + if host is not None or port is not None: + host = host or os.environ.get(ENV_HOST, DEFAULT_HOST) + port = int(port or os.environ.get(ENV_PORT, DEFAULT_PORT)) + else: + host, port, process = backend.launch_or_attach() + except (OSError, RuntimeError, TimeoutError) as exc: + print(f"[wdio-devtools] could not start dashboard ({exc}); " + f"continuing without capture", file=sys.stderr) + return None + + transport = WSClient(host, port, on_control=lifecycle.on_control) + try: + transport.connect() + except OSError as exc: + print( + f"[wdio-devtools] dashboard not reachable at {host}:{port} " + f"({exc}); continuing without capture", + file=sys.stderr, + ) + if process is not None: + process.terminate() + return None + + capturer = SessionCapturer(transport) + instrumentation.install(capturer, webdriver_cls) + # Surface the runner's output in the dashboard Console: Python logging + # (selenium + the adapter's own events) and the test's stdout. + logs = LogCapturer(capturer) + logs.start() + term = TerminalCapturer(capturer) + term.start() + url = f"http://{host}:{port}" + _active.update( + capturer=capturer, transport=transport, process=process, url=url, + terminal=term, logs=logs, + ) + + # Open the dashboard window and wire exit/signal + control-frame teardown so + # closing the window (clientDisconnected) or ending the process both tidy up. + handle = lifecycle.open_dashboard(url) if lifecycle.auto_open_enabled() else None + _active["handle"] = handle + lifecycle.register_exit_handlers(disable, handle) + return capturer + + +def disable() -> None: + # Close the dashboard window + unregister exit/signal handlers first, so a + # re-enable() starts clean. Idempotent and defensive β€” never raises. + lifecycle.unregister_exit_handlers() + instrumentation.uninstall() + term = _active["terminal"] + if term is not None: # restore stdout/stderr before tearing the transport down + term.stop() + logs = _active["logs"] + if logs is not None: # detach the logging handler + restore logger levels + logs.stop() + transport = _active["transport"] + if transport is not None: + transport.close() + process = _active["process"] + if process is not None: # only set when we launched it ourselves + process.terminate() + try: + process.wait(timeout=5) + except subprocess.TimeoutExpired: + process.kill() # backend ignored SIGTERM β€” force it + _active.update( + capturer=None, transport=None, process=None, url=None, handle=None, + terminal=None, logs=None, + ) + + +def get_capturer() -> Optional[SessionCapturer]: + return _active["capturer"] + + +def dashboard_url() -> Optional[str]: + """URL of the connected dashboard, or None if capture isn't active.""" + return _active["url"] + + +def wait_for_dashboard_close() -> None: + """Block until the user closes the dashboard window, so you can inspect the + run after your test finishes. Returns immediately if no dashboard window is + open (headless/CI) β€” safe to always call before ``disable()``.""" + if lifecycle.dashboard_window_open(): + print(f"[wdio-devtools] dashboard live at {dashboard_url()} β€” " + "close the window to finish.", file=sys.stderr) + lifecycle.wait_for_shutdown() diff --git a/packages/selenium-py-devtools/src/wdio_selenium_devtools/_contract.py b/packages/selenium-py-devtools/src/wdio_selenium_devtools/_contract.py new file mode 100644 index 00000000..01818ab2 --- /dev/null +++ b/packages/selenium-py-devtools/src/wdio_selenium_devtools/_contract.py @@ -0,0 +1,15 @@ +# GENERATED by scripts/gen_contract.py from packages/shared. +# Do not edit by hand β€” run the script to regenerate. +CONTRACT_VERSION = "1.0.0" + +SCOPE_METADATA = "metadata" +SCOPE_COMMANDS = "commands" +SCOPE_CONSOLE_LOGS = "consoleLogs" +SCOPE_NETWORK_REQUESTS = "networkRequests" +SCOPE_SUITES = "suites" +SCOPE_SCREENCAST = "screencast" +SCOPE_SOURCES = "sources" +SCOPE_MUTATIONS = "mutations" + +DATA_SCOPES = frozenset(['actionSnapshots', 'commands', 'config', 'consoleLogs', 'logs', 'metadata', 'mutations', 'networkRequests', 'screencast', 'sources', 'suites']) +CONTROL_SCOPES = frozenset(['clearCommands', 'clearExecutionData', 'clientConnected', 'clientDisconnected', 'config', 'replaceCommand', 'testStopped']) diff --git a/packages/selenium-py-devtools/src/wdio_selenium_devtools/backend.py b/packages/selenium-py-devtools/src/wdio_selenium_devtools/backend.py new file mode 100644 index 00000000..0693cb99 --- /dev/null +++ b/packages/selenium-py-devtools/src/wdio_selenium_devtools/backend.py @@ -0,0 +1,119 @@ +"""Locate or launch the dashboard backend. + +Python can't declare a dependency on the Node ``@wdio/devtools-backend`` the way +the JS adapters do (no cross-ecosystem resolution). So the backend is obtained +at runtime, and the resolution order encodes the local-vs-published split: + + 1. DEVTOOLS_PORT set β†’ attach to an already-running backend (CI, manual) + 2. DEVTOOLS_BACKEND_CMD set β†’ spawn that explicit command + 3. monorepo dist present β†’ node packages/backend/dist/index.js (LOCAL dev) + 4. else β†’ npx @wdio/devtools-backend@ (PUBLISHED) + +The pinned version below is bumped deliberately alongside a contract change β€” +there is no auto-resolution, so this constant *is* the version link. +""" + +from __future__ import annotations + +import os +import re +import shlex +import shutil +import subprocess +import threading +import time +from pathlib import Path +from typing import List, Optional, Tuple + +from .constants import ( + BACKEND_NPM_PACKAGE, + BACKEND_NPM_VERSION, + BACKEND_SPAWN_TIMEOUT_S, + DEFAULT_HOST, + ENV_BACKEND_CMD, + ENV_HOST, + ENV_PORT, +) + +# Match the ACTUAL bound port from Fastify's "Server listening at http://…:PORT" +# line β€” NOT the earlier "Starting … on port 3000" line, which is only the +# *preferred* port. When 3000 is busy the backend negotiates a different port, +# so keying off the preferred port connects to the wrong (or a dead) socket. +# Greedy `.*` so the IPv6 form (http://[::1]:PORT) resolves to the final :PORT. +_PORT_RE = re.compile(r"listening at .*:(\d+)") + + +def _find_monorepo_backend(start: Optional[Path] = None) -> Optional[Path]: + """Walk up from ``start`` (default: this module) for a built backend. Present + only in a monorepo checkout; None from an installed wheel.""" + base = start or Path(__file__).resolve() + for parent in base.parents: + candidate = parent / "packages" / "backend" / "dist" / "index.js" + if candidate.exists(): + return candidate + return None + + +def _drain(proc: subprocess.Popen) -> None: + """Keep reading the backend's stdout so its pipe never fills and blocks it.""" + + def pump() -> None: + assert proc.stdout is not None + for _ in proc.stdout: + pass + + threading.Thread(target=pump, daemon=True).start() + + +def _spawn_and_wait_for_port( + cmd: List[str], timeout: float = BACKEND_SPAWN_TIMEOUT_S +) -> Tuple[subprocess.Popen, int]: + proc = subprocess.Popen( + cmd, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True, bufsize=1 + ) + assert proc.stdout is not None + deadline = time.time() + timeout + while time.time() < deadline: + line = proc.stdout.readline() + if not line: + if proc.poll() is not None: + raise RuntimeError( + f"backend exited (code {proc.returncode}) before reporting a port" + ) + continue + match = _PORT_RE.search(line) + if match: + _drain(proc) + return proc, int(match.group(1)) + proc.terminate() + raise TimeoutError("backend did not report a port within the timeout") + + +def launch_or_attach() -> Tuple[str, int, Optional[subprocess.Popen]]: + """Return ``(host, port, process)``. ``process`` is None when we attached to + a backend we don't own (caller must not terminate it).""" + host = os.environ.get(ENV_HOST, DEFAULT_HOST) + + if os.environ.get(ENV_PORT): + return host, int(os.environ[ENV_PORT]), None + + explicit = os.environ.get(ENV_BACKEND_CMD) + if explicit: + proc, port = _spawn_and_wait_for_port(shlex.split(explicit)) + return host, port, proc + + local = _find_monorepo_backend() + if local is not None: + proc, port = _spawn_and_wait_for_port(["node", str(local)]) + return host, port, proc + + npx = shutil.which("npx") + if npx is None: + raise RuntimeError( + "Node.js not found β€” install Node 18+ (the dashboard backend is a Node " + "app), or set DEVTOOLS_PORT to an already-running dashboard." + ) + proc, port = _spawn_and_wait_for_port( + [npx, "-y", f"{BACKEND_NPM_PACKAGE}@{BACKEND_NPM_VERSION}"] + ) + return host, port, proc diff --git a/packages/selenium-py-devtools/src/wdio_selenium_devtools/bidi.py b/packages/selenium-py-devtools/src/wdio_selenium_devtools/bidi.py new file mode 100644 index 00000000..b731a0a5 --- /dev/null +++ b/packages/selenium-py-devtools/src/wdio_selenium_devtools/bidi.py @@ -0,0 +1,412 @@ +"""Selenium BiDi capture β€” browser console, JS exceptions, and network. + +Mirrors the JS ``core/bidi.ts`` + ``selenium-devtools/bidi.ts`` split: the pure +eventβ†’frame mapping helpers (``console_kwargs`` / ``request_sent_kwargs`` / +``response_completed_kwargs``) take plain dicts/objects and are unit-testable +without selenium; ``attach`` does the selenium wiring and is defensive β€” a BiDi +failure is a logged no-op, never a raised error into the user's test. + +Two selenium-version realities shape this module (selenium 4.36): + +* BiDi only opens when the session was created with ``webSocketUrl`` truthy + (``options.web_socket_url = True`` at build). We can't set that from inside + the ``execute`` wrapper β€” the session already exists β€” so attach() checks the + capability and degrades if it's missing. +* selenium's high-level ``network.add_request_handler`` *intercepts* (pauses) + requests. We deliberately avoid it: we subscribe to the network events via + the low-level connection so requests are observed but never stalled. +""" + +from __future__ import annotations + +import sys +from typing import Any, Dict, List, Optional, Tuple + +from .capturer import SessionCapturer +from .constants import ( + BIDI_CAPABILITY, + BIDI_LEVEL_MAP, + BIDI_NET_BEFORE_REQUEST, + BIDI_NET_RESPONSE_COMPLETED, +) +from .utils import now_ms + + +def _warn(message: str) -> None: + print(f"[wdio-devtools] BiDi: {message}", file=sys.stderr) + + +# ── pure mapping helpers (no selenium) ─────────────────────────────────────── + + +def normalize_level(level: Any) -> str: + """Map a BiDi log level onto the shared LogLevel union (fallback: log).""" + return BIDI_LEVEL_MAP.get(str(level or "").lower(), "log") + + +def remote_value_to_py(value: Any) -> Any: + """Deserialize one BiDi RemoteValue into a JSON-friendly Python value. + + The reverse of selenium's ``Script.__convert_to_local_value``: console args + arrive as ``{"type": ..., "value": ...}`` RemoteValues, not raw values, so + ``console.log('a', {b:1}, 42)`` yields dicts we unwrap into ``'a'``, + ``{'b': 1}``, ``42``. Anything unrecognized degrades to its string form. + """ + if not isinstance(value, dict) or "type" not in value: + return value + kind = value.get("type") + inner = value.get("value") + if kind in ("null", "undefined"): + return None + if kind in ("string", "boolean", "number"): + # BiDi encodes NaN/Infinity/-0 as the strings "NaN"/"Infinity"/"-0" β€” + # passed through as-is since JSON can't represent the float specials. + return inner + if kind == "bigint": + try: + return int(inner) + except (TypeError, ValueError): + return str(inner) + if kind in ("array", "set") and isinstance(inner, list): + return [remote_value_to_py(item) for item in inner] + if kind in ("object", "map") and isinstance(inner, list): + out: Dict[str, Any] = {} + for pair in inner: + if isinstance(pair, (list, tuple)) and len(pair) == 2: + key = remote_value_to_py(pair[0]) + out[str(key)] = remote_value_to_py(pair[1]) + return out + if kind in ("date", "regexp"): + return inner + # error/function/node/window/symbol/promise/… β€” no serializable value. + return value.get("value", kind) + + +def _args_from_entry(entry: Any) -> Optional[List[Any]]: + """Deserialized console args if the entry carries any, else None. + + Returns None (not []) when ``args`` is absent so the caller can fall back to + ``.text`` β€” an empty list is a real console call with no arguments. + """ + raw = _attr(entry, "args", None) + if not isinstance(raw, list): + return None + return [remote_value_to_py(v) for v in raw] + + +def console_kwargs(entry: Any) -> Tuple[str, List[Any]]: + """(level, args) for capturer.capture_console from a BiDi console entry. + + Accepts selenium's ConsoleLogEntry dataclass (``.level`` / ``.method`` / + ``.args`` / ``.text``) or a plain dict β€” so tests pass dicts, no selenium + needed. Prefers ``method`` (the actual console.X call β€” log/info/warn/error/ + debug) over ``level`` (coarser), and maps every RemoteValue arg, falling + back to ``.text`` only when no ``args`` are present. + """ + level = _attr(entry, "method", None) or _attr(entry, "level", "info") + args = _args_from_entry(entry) + if args is None: + text = _attr(entry, "text", None) + if text is None: + text = _attr(entry, "message", "") + args = [text] + return normalize_level(level), args + + +def js_error_kwargs(entry: Any) -> Tuple[str, List[Any]]: + """(level, args) for a BiDi JavaScript exception β€” always ``error`` level. + + JavaScriptLogEntry carries ``text`` (the message) and a ``stacktrace`` dict + rather than ``args``. We render message + formatted stack as a single arg so + the Console panel shows the full error, never an empty/duplicate entry. + """ + text = _attr(entry, "text", None) + if text is None: + text = _attr(entry, "message", "") + message = str(text or "") + stack = _format_stacktrace(_attr(entry, "stacktrace", None)) + combined = f"{message}\n{stack}" if stack else message + return "error", [combined] + + +def _format_stacktrace(stacktrace: Any) -> str: + """Render a BiDi ``StackTrace`` ({callFrames:[{functionName,url,lineNumber, + columnNumber}]}) into ``at fn (url:line:col)`` lines β€” empty string if none.""" + if not isinstance(stacktrace, dict): + return "" + frames = stacktrace.get("callFrames") + if not isinstance(frames, list): + return "" + lines: List[str] = [] + for frame in frames: + if not isinstance(frame, dict): + continue + fn = frame.get("functionName") or "" + url = frame.get("url") or "" + line = frame.get("lineNumber") + col = frame.get("columnNumber") + location = url + if line is not None: + location = f"{url}:{line}" + if col is not None: + location = f"{url}:{line}:{col}" + lines.append(f" at {fn} ({location})" if location else f" at {fn}") + return "\n".join(lines) + + +def request_sent_kwargs(params: Dict[str, Any]) -> Optional[Dict[str, Any]]: + """kwargs for the initial (pending) network frame, or None if unidentifiable. + + ``params`` is the BiDi ``network.beforeRequestSent`` event params β€” the + ``.params`` dict on selenium's NetworkEvent. + """ + request = params.get("request") or {} + request_id = str(request.get("request") or params.get("id") or "") + if not request_id: + return None + start_time = int(params.get("timestamp") or now_ms()) + return { + "request_id": request_id, + "url": request.get("url") or "", + "method": request.get("method") or "GET", + "status": None, + "timestamp": now_ms(), + "start_time": start_time, + "request_type": request_type_for(request.get("url") or ""), + "request_headers": headers_to_object(request.get("headers")), + } + + +def response_completed_kwargs( + params: Dict[str, Any], pending: Dict[str, Dict[str, Any]] +) -> Optional[Dict[str, Any]]: + """kwargs for the finalized network frame, merged over the pending request. + + Returns None when the matching request wasn't seen (out-of-order events) β€” + the caller skips rather than inventing a half-populated entry. + """ + request = params.get("request") or {} + request_id = str(request.get("request") or params.get("id") or "") + prev = pending.get(request_id) + if prev is None: + return None + response = params.get("response") or {} + start_time = int(prev.get("start_time") or now_ms()) + end_time, time = _response_timing( + request.get("timings"), start_time, params.get("timestamp") + ) + merged = dict(prev) + merged.update( + status=_int_or(response.get("status"), prev.get("status")), + status_text=response.get("statusText"), + timestamp=now_ms(), + end_time=end_time, + time=time, + size=_int_or(response.get("bytesReceived"), None), + request_type=request_type_for( + prev.get("url") or "", response.get("mimeType") + ), + response_headers=headers_to_object(response.get("headers")), + ) + return merged + + +def request_type_for(url: str, mime_type: Optional[str] = None) -> str: + """Classify a request into the dashboard's Network-tab categories. + + Prefers the response mime type; falls back to URL-extension heuristics. + Ported from core/net.ts getRequestType so the wire shape matches the JS + adapters exactly. + """ + ct = (mime_type or "").lower() + u = url.lower() + if "text/html" in ct: + return "document" + if "text/css" in ct: + return "stylesheet" + if "javascript" in ct or "ecmascript" in ct: + return "script" + if "image/" in ct: + return "image" + if "font/" in ct or "woff" in ct: + return "font" + if "application/json" in ct: + return "fetch" + if u.endswith(".html") or u.endswith(".htm"): + return "document" + if u.endswith(".css"): + return "stylesheet" + if u.endswith(".js") or u.endswith(".mjs"): + return "script" + if any(u.endswith(ext) for ext in (".png", ".jpg", ".jpeg", ".gif", ".svg", + ".webp", ".ico")): + return "image" + if any(u.endswith(ext) for ext in (".woff", ".woff2", ".ttf", ".eot", + ".otf")): + return "font" + return "xhr" + + +def headers_to_object(headers: Any) -> Optional[Dict[str, str]]: + """Flatten BiDi's ``[{name, value:{value}}]`` header list to a lowercased + ``{name: value}`` dict. Returns None for a non-list (absent) input.""" + if not isinstance(headers, list): + return None + out: Dict[str, str] = {} + for h in headers: + if not isinstance(h, dict): + continue + name = str(h.get("name") or "").lower() + if not name: + continue + value = h.get("value") + if isinstance(value, str): + out[name] = value + elif isinstance(value, dict) and isinstance(value.get("value"), str): + out[name] = value["value"] + else: + out[name] = str(value) + return out + + +def _attr(obj: Any, name: str, default: Any) -> Any: + if isinstance(obj, dict): + return obj.get(name, default) + return getattr(obj, name, default) + + +def _int_or(value: Any, fallback: Any) -> Any: + try: + return int(value) + except (TypeError, ValueError): + return fallback + + +def _response_timing( + timings: Any, start_time: int, timestamp: Any +) -> Tuple[int, int]: + """(end_time, duration_ms) preferring the browser's FetchTimingInfo β€” it's + immune to BiDi events arriving batched in one tick (which collapses the + event timestamps and yields 0-duration requests). Falls back to the event + timestamp delta when timings are unavailable.""" + if isinstance(timings, dict): + req = timings.get("requestTime") + end = timings.get("responseEnd") + if isinstance(req, (int, float)) and isinstance(end, (int, float)) and end > req: + time = round(end - req) + return start_time + time, time + end_time = _int_or(timestamp, None) + if end_time is None: + end_time = now_ms() + return end_time, max(0, end_time - start_time) + + +# ── selenium wiring (defensive) ─────────────────────────────────────────────── + + +def _bidi_enabled(driver: Any) -> bool: + caps = getattr(driver, "caps", None) + return bool(isinstance(caps, dict) and caps.get(BIDI_CAPABILITY)) + + +def _attach_console(driver: Any, capturer: SessionCapturer) -> bool: + try: + script = driver.script + except Exception as exc: # noqa: BLE001 β€” any selenium/BiDi failure is a no-op + _warn(f"script channel unavailable: {exc}") + return False + + def on_console_entry(entry: Any) -> None: + try: + level, args = console_kwargs(entry) + capturer.capture_console(level, args, source="browser") + except Exception as exc: # noqa: BLE001 + _warn(f"console handler threw: {exc}") + + def on_js_error(entry: Any) -> None: + try: + level, args = js_error_kwargs(entry) + capturer.capture_console(level, args, source="browser") + except Exception as exc: # noqa: BLE001 + _warn(f"JS error handler threw: {exc}") + + try: + script.add_console_message_handler(on_console_entry) + script.add_javascript_error_handler(on_js_error) + return True + except Exception as exc: # noqa: BLE001 + _warn(f"console/JS handlers failed to attach: {exc}") + return False + + +def _attach_network(driver: Any, capturer: SessionCapturer) -> bool: + """Subscribe to network events WITHOUT interception (see module docstring). + + Uses the low-level connection so requests are only observed. Returns False + (and logs) on any failure β€” network BiDi is best-effort. + """ + try: + conn = driver.network.conn + from selenium.webdriver.common.bidi.network import NetworkEvent # lazy + from selenium.webdriver.common.bidi.session import Session # lazy + except Exception as exc: # noqa: BLE001 + _warn(f"network channel unavailable: {exc}") + return False + + pending: Dict[str, Dict[str, Any]] = {} + + def on_request_sent(event: Any) -> None: + try: + kwargs = request_sent_kwargs(getattr(event, "params", {}) or {}) + if kwargs is not None: + pending[kwargs["request_id"]] = kwargs + capturer.capture_network(**kwargs) + except Exception as exc: # noqa: BLE001 + _warn(f"beforeRequestSent handler threw: {exc}") + + def on_response_completed(event: Any) -> None: + try: + kwargs = response_completed_kwargs( + getattr(event, "params", {}) or {}, pending + ) + if kwargs is not None: + pending.pop(kwargs["request_id"], None) + capturer.capture_network(**kwargs) + except Exception as exc: # noqa: BLE001 + _warn(f"responseCompleted handler threw: {exc}") + + try: + conn.execute( + Session(conn).subscribe( + BIDI_NET_BEFORE_REQUEST, BIDI_NET_RESPONSE_COMPLETED + ) + ) + conn.add_callback(NetworkEvent(BIDI_NET_BEFORE_REQUEST), on_request_sent) + conn.add_callback( + NetworkEvent(BIDI_NET_RESPONSE_COMPLETED), on_response_completed + ) + return True + except Exception as exc: # noqa: BLE001 + _warn(f"network subscribe failed: {exc}") + return False + + +def attach(driver: Any, capturer: SessionCapturer) -> bool: + """Wire BiDi console + network capture onto ``driver``. + + Returns True if at least one channel attached. A driver without the + ``webSocketUrl`` capability (BiDi not enabled at build time) is skipped with + a one-line warning β€” capture continues via the command stream only. + """ + if not _bidi_enabled(driver): + _warn( + f"{BIDI_CAPABILITY} not set on the session β€” enable BiDi with " + "options.web_socket_url = True to capture console/network" + ) + return False + attached = 0 + if _attach_console(driver, capturer): + attached += 1 + if _attach_network(driver, capturer): + attached += 1 + return attached > 0 diff --git a/packages/selenium-py-devtools/src/wdio_selenium_devtools/capturer.py b/packages/selenium-py-devtools/src/wdio_selenium_devtools/capturer.py new file mode 100644 index 00000000..e98f5987 --- /dev/null +++ b/packages/selenium-py-devtools/src/wdio_selenium_devtools/capturer.py @@ -0,0 +1,152 @@ +"""Session capturer β€” the Python analogue of core's ``SessionCapturerBase``. + +Owns the transport, a per-session command counter, and the normalizeβ†’send +path. Deliberately thin: it turns already-captured data into wire frames and +pushes them. No driver knowledge lives here (that's instrumentation.py); no +post-processing lives here (that's the backend's job, per the architecture). +""" + +from __future__ import annotations + +import threading +from typing import Any, List, Optional, Protocol + +from . import frames +from ._contract import ( + SCOPE_COMMANDS, + SCOPE_CONSOLE_LOGS, + SCOPE_METADATA, + SCOPE_MUTATIONS, + SCOPE_NETWORK_REQUESTS, + SCOPE_SCREENCAST, + SCOPE_SOURCES, + SCOPE_SUITES, +) +from .types import SuiteStats +from .utils import now_ms, to_jsonable + + +class Transport(Protocol): + connected: bool + + def send_json(self, scope: str, data: Any) -> bool: ... + def close(self) -> None: ... + + +class SessionCapturer: + def __init__(self, transport: Transport) -> None: + self._tx = transport + self._command_counter = 0 + self._lock = threading.Lock() + self._metadata_sent = False + self.session_id: Optional[str] = None + + # ── metadata ─────────────────────────────────────────────────────────────── + + def ensure_metadata( + self, session_id: str, capabilities: Optional[dict], url: Optional[str] + ) -> None: + if self._metadata_sent or not session_id: + return + self._metadata_sent = True + self.session_id = session_id + self._tx.send_json( + SCOPE_METADATA, + frames.metadata(session_id, to_jsonable(capabilities or {}), url), + ) + + # ── commands ─────────────────────────────────────────────────────────────── + + def capture_command( + self, + *, + command: str, + args: Any, + result: Any = None, + error: Optional[BaseException] = None, + start_time: int, + call_source: Optional[str], + screenshot: Optional[str] = None, + ) -> None: + with self._lock: + self._command_counter += 1 + command_id = self._command_counter + norm_args = args if isinstance(args, list) else ([] if args is None else [args]) + entry = frames.command_log( + command=command, + args=to_jsonable(norm_args), + result=to_jsonable(result), + error=error, + timestamp=now_ms(), + start_time=start_time, + call_source=call_source, + command_id=command_id, + screenshot=screenshot, + ) + self._tx.send_json(SCOPE_COMMANDS, [entry]) + + # ── console / network ──────────────────────────────────────────────────────── + + def capture_console(self, level: str, args: List[Any], source: str = "browser") -> None: + self._tx.send_json( + SCOPE_CONSOLE_LOGS, + [frames.console_log(level=level, args=to_jsonable(args), + timestamp=now_ms(), source=source)], + ) + + def capture_network(self, **kwargs: Any) -> None: + self._tx.send_json(SCOPE_NETWORK_REQUESTS, [frames.network_request(**kwargs)]) + + # ── screencast ───────────────────────────────────────────────────────────── + + def send_screencast( + self, + *, + video_path: str, + video_file: str, + frame_count: int, + duration: int, + start_time: Optional[int], + ) -> None: + # Screencast is a single post-run frame keyed to the session; skip if + # metadata never resolved a session id (nothing for the UI to attach to). + if not self.session_id: + return + self._tx.send_json( + SCOPE_SCREENCAST, + frames.screencast( + session_id=self.session_id, + video_path=video_path, + video_file=video_file, + frame_count=frame_count, + duration=duration, + start_time=start_time, + ), + ) + + # ── sources / mutations ────────────────────────────────────────────────────── + + def send_sources(self, sources: dict) -> None: + """Send a ``{absolute_path: source_text}`` map β€” the Source tab shows the + file a command's ``callSource`` points at. Empty map is a no-op.""" + if sources: + self._tx.send_json(SCOPE_SOURCES, sources) + + def send_mutations(self, mutations: List[Any]) -> None: + """Forward DOM mutations captured by the injected browser script + (packages/script) β€” the UI renders them into the snapshot iframe.""" + if mutations: + self._tx.send_json(SCOPE_MUTATIONS, mutations) + + # ── suites ─────────────────────────────────────────────────────────────────── + + def send_suites(self, suites: List[SuiteStats]) -> None: + # The UI expects Record[] β€” one single-key record per + # suite β€” mirroring core's TestReporterBase.sendUpstream. A plain array + # of suites won't render. Empty-payload guard matches the JS behavior. + payload = [{s["uid"]: s} for s in suites if s.get("uid")] + if payload: + self._tx.send_json(SCOPE_SUITES, payload) + + def close(self) -> None: + self._tx.close() diff --git a/packages/selenium-py-devtools/src/wdio_selenium_devtools/constants.py b/packages/selenium-py-devtools/src/wdio_selenium_devtools/constants.py new file mode 100644 index 00000000..3bf53917 --- /dev/null +++ b/packages/selenium-py-devtools/src/wdio_selenium_devtools/constants.py @@ -0,0 +1,80 @@ +"""Module-level constants β€” the single home for connection defaults, env-var +names, the skip sets, and the pinned backend version. No internal imports.""" + +from __future__ import annotations + +import os + +# ── Connection defaults ────────────────────────────────────────────────────── +DEFAULT_HOST = "localhost" +DEFAULT_PORT = 3000 +WORKER_PATH = "/worker" +CONNECT_TIMEOUT_S = 5.0 + +# ── Environment variables that configure the adapter ───────────────────────── +ENV_HOST = "DEVTOOLS_HOST" +ENV_PORT = "DEVTOOLS_PORT" +ENV_BACKEND_CMD = "DEVTOOLS_BACKEND_CMD" +ENV_OPT_IN = "WDIO_DEVTOOLS" +ENV_BIDI = "WDIO_DEVTOOLS_BIDI" # "0"/"false"/"no"/"off" disables BiDi auto-enable +ENV_OPEN = "WDIO_DEVTOOLS_OPEN" # "0"/"false"/"no"/"off" disables dashboard auto-open + +# ── Backend launch ─────────────────────────────────────────────────────────── +# Pinned backend version fetched via npx from a published install. There is no +# cross-ecosystem resolver, so this constant *is* the version link β€” bump it in +# the same change that regenerates _contract.py. +BACKEND_NPM_VERSION = "1.7.0" +BACKEND_NPM_PACKAGE = "@wdio/devtools-backend" +BACKEND_SPAWN_TIMEOUT_S = 40.0 + +# ── Instrumentation ────────────────────────────────────────────────────────── +# Selenium commands that are bookkeeping/noise rather than user-meaningful. +# `screenshot`/`elementScreenshot` are skipped so the screencast recorder's +# per-command frame capture (get_screenshot_as_base64) doesn't flood the +# Actions timeline. +SKIP_COMMANDS = frozenset( + {"newSession", "quit", "status", "getLog", "getAllSessions", "getSessions", + "screenshot", "elementScreenshot"} +) + +# Stack-frame path fragment to skip when resolving a command's call source β€” +# the adapter's own package. The selenium library dir is added at runtime by +# instrumentation (resolved from selenium.__file__), NOT matched by the +# substring "/selenium/" β€” that would wrongly skip a user's own test file living +# under a path like examples/selenium/... . +_PACKAGE_DIR = os.path.dirname(__file__) +SKIP_STACK_FRAMES = (_PACKAGE_DIR,) + +# ── BiDi ────────────────────────────────────────────────────────────────────── +# The capability the driver must advertise for selenium's BiDi channel to open +# (set via ``options.web_socket_url = True`` at build time). Without it, +# accessing ``driver.script`` / ``driver.network`` raises β€” attach() degrades. +BIDI_CAPABILITY = "webSocketUrl" +# BiDi network event names we subscribe to WITHOUT interception β€” a plain +# session.subscribe, so requests are observed but never paused (interception +# would stall the user's page loads if a callback failed to continue them). +BIDI_NET_BEFORE_REQUEST = "network.beforeRequestSent" +BIDI_NET_RESPONSE_COMPLETED = "network.responseCompleted" +# selenium's BiDi log entries already carry lowercase levels; this normalizes +# the stragglers to the shared LogLevel union. Unmapped levels fall back to log. +BIDI_LEVEL_MAP = { + "debug": "debug", + "info": "info", + "warn": "warn", + "warning": "warn", + "error": "error", + "severe": "error", + "log": "log", + "trace": "trace", +} + +# ── Screencast ──────────────────────────────────────────────────────────────── +# Frames are captured synchronously (one per command) on the main thread β€” see +# screencast.py for why a background poll thread is avoided. Screenshots via +# WebDriver are always PNG. +SCREENCAST_IMAGE_FORMAT = "png" +# Skip encoding below this many frames β€” a single still isn't a video. +SCREENCAST_MIN_FRAMES = 2 +# Output filename stem; the session id + .webm suffix are appended. +SCREENCAST_FILENAME_PREFIX = "selenium-py-video" +# The `screencast` wire scope is generated into _contract.py (SCOPE_SCREENCAST). diff --git a/packages/selenium-py-devtools/src/wdio_selenium_devtools/frames.py b/packages/selenium-py-devtools/src/wdio_selenium_devtools/frames.py new file mode 100644 index 00000000..b5deeb48 --- /dev/null +++ b/packages/selenium-py-devtools/src/wdio_selenium_devtools/frames.py @@ -0,0 +1,193 @@ +"""Builders for the wire frames the dashboard renders. + +Each function returns the ``data`` payload for a ``{scope, data}`` frame. +Shapes mirror ``packages/shared/src/types.ts``. Keeping them here β€” pure and +side-effect free β€” makes them unit-testable and the single place the contract +lives on the Python side. +""" + +from __future__ import annotations + +from typing import Any, List, Optional + +from .types import ( + CommandLog, + ConsoleLog, + Metadata, + NetworkRequest, + ScreencastInfo, + SuiteStats, + TestStats, +) +from .utils import iso + + +def metadata( + session_id: str, + capabilities: Optional[dict] = None, + url: Optional[str] = None, +) -> Metadata: + caps = capabilities or {} + return { + "type": "testrunner", # TraceType.Testrunner + "sessionId": session_id, + "url": url, + "capabilities": caps, + "desiredCapabilities": caps, + "testEnv": "python-selenium", + } + + +def command_log( + *, + command: str, + args: List[Any], + result: Any = None, + error: Optional[BaseException] = None, + timestamp: int, + start_time: int, + call_source: Optional[str], + command_id: int, + screenshot: Optional[str] = None, +) -> CommandLog: + entry: CommandLog = { + "command": command, + "args": args, + "result": result, + "timestamp": timestamp, + "startTime": start_time, + "callSource": call_source, + "id": command_id, + } + if error is not None: + entry["error"] = { + "name": type(error).__name__, + "message": str(error), + } + if screenshot: + entry["screenshot"] = screenshot + return entry + + +def console_log( + *, level: str, args: List[Any], timestamp: int, source: str = "browser" +) -> ConsoleLog: + return {"type": level, "args": args, "timestamp": timestamp, "source": source} + + +def network_request( + *, + request_id: str, + url: str, + method: str, + timestamp: int, + start_time: int, + status: Optional[int] = None, + request_type: str = "fetch", + end_time: Optional[int] = None, + status_text: Optional[str] = None, + time: Optional[int] = None, + size: Optional[int] = None, + request_headers: Optional[dict] = None, + response_headers: Optional[dict] = None, +) -> NetworkRequest: + entry: NetworkRequest = { + "id": request_id, + "url": url, + "method": method, + "status": status, + "timestamp": timestamp, + "startTime": start_time, + "endTime": end_time, + "type": request_type, + } + # Response-phase fields are absent on the initial request frame; omit rather + # than send nulls so the dashboard's "pending" state renders correctly. + if status_text is not None: + entry["statusText"] = status_text + if time is not None: + entry["time"] = time + if size is not None: + entry["size"] = size + if request_headers is not None: + entry["requestHeaders"] = request_headers + if response_headers is not None: + entry["responseHeaders"] = response_headers + return entry + + +def screencast( + *, + session_id: str, + video_path: str, + video_file: str, + frame_count: int, + duration: int, + start_time: Optional[int], +) -> ScreencastInfo: + info: ScreencastInfo = { + "sessionId": session_id, + "videoPath": video_path, + "videoFile": video_file, + "frameCount": frame_count, + "duration": duration, + } + if start_time is not None: + info["startTime"] = start_time + return info + + +def test_stats( + *, + uid: str, + title: str, + full_title: str, + parent: str, + state: str, + file: str, + start_ms: int, + end_ms: int, + call_source: Optional[str] = None, +) -> TestStats: + return { + "uid": uid, + "cid": "0-0", + "title": title, + "fullTitle": full_title, + "parent": parent, + "state": state, + "start": iso(start_ms), + "end": iso(end_ms), + "type": "test", + "file": file, + "retries": 0, + "_duration": max(0, end_ms - start_ms), + "callSource": call_source, + } + + +def suite_stats( + *, + uid: str, + title: str, + file: str, + start_ms: int, + tests: List[TestStats], + end_ms: Optional[int] = None, + state: Optional[str] = None, +) -> SuiteStats: + return { + "uid": uid, + "cid": "0-0", + "title": title, + "fullTitle": title, + "type": "suite", + "file": file, + "start": iso(start_ms), + "end": iso(end_ms) if end_ms is not None else None, + "state": state, + "tests": tests, + "suites": [], + "hooks": [], + "_duration": max(0, (end_ms - start_ms)) if end_ms is not None else 0, + } diff --git a/packages/selenium-py-devtools/src/wdio_selenium_devtools/instrumentation.py b/packages/selenium-py-devtools/src/wdio_selenium_devtools/instrumentation.py new file mode 100644 index 00000000..ba41b14e --- /dev/null +++ b/packages/selenium-py-devtools/src/wdio_selenium_devtools/instrumentation.py @@ -0,0 +1,386 @@ +"""Driver instrumentation β€” the one genuinely per-language piece. + +Every Selenium command (driver- *and* element-level, since element methods +delegate to ``self._parent.execute``) funnels through +``WebDriver.execute(driver_command, params)``. Wrapping that single chokepoint +captures the whole command stream from one place β€” cleaner than the JS +adapter's prototype patching. + +The patch target is injected so the module imports and unit-tests without +selenium present; ``install`` defaults to the real selenium class. +""" + +from __future__ import annotations + +import logging +import os +import sys +import threading +from typing import Any, Optional + +from . import bidi, frames +from .capturer import SessionCapturer +from .constants import BIDI_CAPABILITY, ENV_BIDI, SKIP_COMMANDS, SKIP_STACK_FRAMES +from .screencast import ScreencastRecorder +from .snapshot import SnapshotCapturer, start_snapshot_capture +from .sources import read_source +from .utils import call_source, now_ms + +# Operational logging β€” surfaced in the dashboard Console (the 'runner' stream). +_log = logging.getLogger("wdio_selenium_devtools") + +# Marks the adapter's OWN execute_script calls (snapshot inject/readback) so +# patched_execute skips capturing them as user commands. +_internal = threading.local() + +# User-file paths whose source we've already sent (once per file per run). +_sources_sent: set = set() + + +_skip_frames_cache: Optional[tuple] = None + + +def _skip_frames() -> tuple: + """Call-source skip fragments: the adapter package + the REAL selenium + library dir (resolved from selenium.__file__), cached. Resolving the actual + package dir avoids skipping a user test file whose path merely contains + 'selenium' (e.g. examples/selenium/...).""" + global _skip_frames_cache + if _skip_frames_cache is None: + try: + import selenium + + extra = (os.path.dirname(os.path.abspath(selenium.__file__)) + os.sep,) + except Exception: # noqa: BLE001 β€” narrow fallback if selenium isn't importable + extra = (f"{os.sep}selenium{os.sep}webdriver{os.sep}",) + _skip_frames_cache = tuple(SKIP_STACK_FRAMES) + extra + return _skip_frames_cache + + +def _internal_active() -> bool: + return getattr(_internal, "active", False) + + +# When a test framework (the pytest plugin) reports the suite tree, the adapter +# must NOT also synthesize a default one. The plugin flips this in its configure. +_external_suites = False + + +def set_external_suites(value: bool = True) -> None: + """Tell the adapter a test framework owns the suite tree (suppresses the + default single-session suite used for plain scripts).""" + global _external_suites + _external_suites = value + + +def _send_default_suite(capturer: SessionCapturer, state: str) -> None: + """Report the run as a single suite/test named after the entry script, so a + plain-script run (no test framework) still shows in the TESTS tree. No-op if + a framework is reporting suites.""" + if _external_suites: + return + entry = os.path.abspath(sys.argv[0]) if sys.argv and sys.argv[0] else "" + title = os.path.basename(entry) or "Selenium session" + ds = _state.get("default_suite") + start = ds["start"] if ds else now_ms() + _state["default_suite"] = {"start": start} + end = start if state == "running" else now_ms() + test = frames.test_stats( + uid=f"{entry or title}::session", title=title, full_title=title, + parent=title, state=state, file=entry, start_ms=start, end_ms=end, + ) + suite = frames.suite_stats( + uid=entry or title, title=title, file=entry, start_ms=start, + tests=[test], state=state, end_ms=(None if state == "running" else end), + ) + capturer.send_suites([suite]) + + +def _guarded_execute_script(driver: Any) -> Any: + """An ``execute_script`` that runs WITHOUT command capture β€” for the + adapter's own snapshot injection/readback, which must never show up in the + Actions timeline (the getTraceData/inject scripts are not user commands).""" + + def run(script: str, *args: Any) -> Any: + _internal.active = True + try: + return driver.execute_script(script, *args) + finally: + _internal.active = False + + return run + + +def _capture_source(capturer: SessionCapturer, call_src: Optional[str]) -> None: + """Send the source of the file a command's ``callSource`` points at (once). + + Keyed by the exact ``callSource`` path so the Source tab always matches β€” + and works for any runner (plain script or pytest), not just the plugin. + Guarded: source capture must never break the user's test.""" + if not call_src: + return + path = call_src.rsplit(":", 1)[0] # strip the trailing ":line" + if not path: + return + # Screencast output lands next to the (first) test file, not the cwd. + if _state.get("output_dir") is None: + _state["output_dir"] = os.path.dirname(path) + if path in _sources_sent: + return + try: + text = read_source(path) + except Exception: # noqa: BLE001 + return + if text is None: + return + _sources_sent.add(path) + capturer.send_sources({path: text}) + +# Selenium command names that change the document β€” after these we drain the +# page-side mutation buffer so the snapshot iframe stays current. +_state: dict = { + "installed": False, "cls": None, "orig": None, + "screencast": None, "snapshot": None, "setup_done": False, + "output_dir": None, # dir of the test file β€” where the screencast .webm lands + "default_suite": None, # synthesized suite for non-framework (script) runs +} + + +def _take_screenshot(driver: Any) -> Optional[str]: + """One base64 PNG of the current page on the MAIN thread β€” no background + thread, so we never touch the Selenium session concurrently. Reused for both + the command entry (per-command snapshot view) and the screencast frame. + Best-effort: a transient failure (mid-navigation, dead session) returns + None and never breaks the test. ``screenshot`` is in SKIP_COMMANDS, so this + doesn't appear in the Actions timeline.""" + fn = getattr(driver, "get_screenshot_as_base64", None) + if not callable(fn): + return None + try: + shot = fn() + except Exception: # noqa: BLE001 β€” transient; skip this frame + return None + return shot if isinstance(shot, str) and shot else None + + +def _add_screencast_frame(shot: Optional[str]) -> None: + """Buffer an already-captured screenshot as a screencast frame.""" + recorder = _state.get("screencast") + if recorder is None or not shot: + return + try: + recorder.add_frame(shot) + except Exception as exc: # noqa: BLE001 β€” never break the test + _log.warning("screencast add_frame threw: %s", exc) + + +def _refresh_snapshot(capturer: SessionCapturer) -> None: + """After a command, keep the snapshot current: re-inject the collector if the + page navigated (self-healing β€” a click can submit a form and wipe it), then + drain the mutation buffer. Called after every command, not just explicit + navigations, so the initial full-document snapshot is captured even if it + wasn't ready the instant the navigation returned.""" + snapshot = _state.get("snapshot") + if snapshot is None: + return + try: + snapshot.inject() # no-op if already present; re-installs after navigation + except Exception as exc: # noqa: BLE001 β€” never break the test + _log.warning("snapshot re-inject threw: %s", exc) + _flush_mutations(capturer) + + +def _flush_mutations(capturer: SessionCapturer) -> None: + """Drain the page-side mutation buffer and forward it β€” no-op if the + collector was never injected. Defensive: never breaks the user's test.""" + snapshot: SnapshotCapturer | None = _state.get("snapshot") + if snapshot is None: + return + try: + mutations = snapshot.pull_mutations() + except Exception as exc: # noqa: BLE001 β€” capture must never break the test + _log.warning("mutation flush threw: %s", exc) + return + if mutations: + capturer.send_mutations(mutations) + + +def _enable_bidi_capability(params: Any) -> None: + """Request BiDi at session creation by injecting ``webSocketUrl`` into the + newSession capabilities β€” the one point (before the session exists) where we + can. This makes console/network capture work out-of-box, matching the JS + adapters. Opt out with ``WDIO_DEVTOOLS_BIDI=0``. Never blocks a session.""" + if os.environ.get(ENV_BIDI, "").strip().lower() in ("0", "false", "no", "off"): + return + try: + caps = params.get("capabilities") if isinstance(params, dict) else None + if not isinstance(caps, dict): + return + always = caps.get("alwaysMatch") + if not isinstance(always, dict): + always = {} + caps["alwaysMatch"] = always + always.setdefault(BIDI_CAPABILITY, True) + except Exception: # noqa: BLE001 β€” capability injection is best-effort + pass + + +def _ensure_session_setup(driver: Any, capturer: SessionCapturer) -> None: + """Once per session β€” on the first real command β€” send metadata, attach + BiDi, and start the screencast. + + This CANNOT run in the newSession branch: selenium assigns ``session_id`` / + ``caps`` only *after* the newSession execute() returns, so at that point the + driver has no session yet (BiDi caps missing, screenshots fail). By the first + real command the driver is fully initialized. Each step is independently + defensive β€” a BiDi or screencast failure is a logged no-op. + """ + if _state.get("setup_done"): + return + session_id = getattr(driver, "session_id", None) + if not session_id: + return # driver not ready yet; retry on the next command + _state["setup_done"] = True + capturer.ensure_metadata(session_id, getattr(driver, "caps", None), None) + _log.info("session %s started", session_id) + _send_default_suite(capturer, "running") # tree entry for plain-script runs + try: + if bidi.attach(driver, capturer): + _log.info("BiDi attached β€” capturing console + network") + except Exception as exc: # noqa: BLE001 β€” capture must never break the test + _log.warning("BiDi attach threw: %s", exc) + try: + recorder = ScreencastRecorder() + recorder.start(driver) + _state["screencast"] = recorder + _log.info("screencast recording started") + except Exception as exc: # noqa: BLE001 + _log.warning("screencast start threw: %s", exc) + try: + # Inject the packages/script DOM observer so the snapshot iframe fills. + # Use a capture-bypassing execute_script so injection/readback scripts + # don't pollute the Actions timeline. + snapshot = start_snapshot_capture( + driver, execute_fn=_guarded_execute_script(driver) + ) + _state["snapshot"] = snapshot + if snapshot is not None: + _log.info("DOM snapshot collector injected") + except Exception as exc: # noqa: BLE001 + _log.warning("snapshot start threw: %s", exc) + + +def _on_quit(capturer: SessionCapturer) -> None: + """On driver.quit(): finalize the screencast and forward its metadata. + + ``quit`` is in the skip set, so this is the last hook before the session is + gone β€” encode here while a session id is still known.""" + # Drain any final mutations while the session (and page) still exist. + _flush_mutations(capturer) + _state["snapshot"] = None + if _state.get("default_suite") is not None: + _send_default_suite(capturer, "passed") # mark the script run complete + recorder = _state.get("screencast") + if recorder is None: + return + _state["screencast"] = None + try: + info = recorder.finalize( + capturer.session_id or "session", output_dir=_state.get("output_dir") + ) + except Exception as exc: # noqa: BLE001 + _log.warning("screencast finalize threw: %s", exc) + return + if info is not None: + capturer.send_screencast(**info) + _log.info("screencast saved: %s", info.get("video_path")) + + +def install(capturer: SessionCapturer, webdriver_cls: Optional[type] = None) -> None: + if _state["installed"]: + return + if webdriver_cls is None: + from selenium.webdriver.remote.webdriver import WebDriver # lazy + + webdriver_cls = WebDriver + + orig_execute = webdriver_cls.execute + + def patched_execute(self, driver_command: str, params: Any = None): # noqa: ANN001 + # The adapter's own execute_script (snapshot inject/readback) runs + # transparently β€” never captured as a user command. + if _internal_active(): + return orig_execute(self, driver_command, params) + # Skip capture for noise, but never alter behavior. + if driver_command in SKIP_COMMANDS: + # Finalize the screencast BEFORE quit tears the session down β€” after + # orig_execute the driver's WebSocket/session is gone. + if driver_command == "quit": + _on_quit(capturer) + elif driver_command == "newSession": + _enable_bidi_capability(params) # request BiDi before the session opens + return orig_execute(self, driver_command, params) + + # First real command: the driver is now fully initialized β€” set up + # metadata/BiDi/screencast once, before executing so BiDi sees this cmd. + _ensure_session_setup(self, capturer) + start = now_ms() + src = call_source(_skip_frames()) + _capture_source(capturer, src) # Source tab: send the test file once + try: + result = orig_execute(self, driver_command, params) + except BaseException as exc: # capture then re-raise unchanged + capturer.capture_command( + command=driver_command, + args=params, + error=exc, + start_time=start, + call_source=src, + ) + raise + + # WebDriver.execute returns the full response dict; the useful payload + # is response["value"]. + value = result.get("value") if isinstance(result, dict) else result + # One screenshot per command on this (main) thread β€” attached to the + # command (so selecting it shows the page) AND reused as a screencast + # frame, so we pay for only a single screenshot round-trip either way. + shot = _take_screenshot(self) + capturer.capture_command( + command=driver_command, + args=params, + result=value, + start_time=start, + call_source=src, + screenshot=shot, + ) + _log.debug("command: %s", driver_command) + # Keep the snapshot iframe current after every command (a click can + # navigate too, not just get/back/…), re-injecting if the page changed. + _refresh_snapshot(capturer) + _add_screencast_frame(shot) + return result + + webdriver_cls.execute = patched_execute # type: ignore[assignment] + _sources_sent.clear() + _state.update( + installed=True, cls=webdriver_cls, orig=orig_execute, + setup_done=False, output_dir=None, default_suite=None, + ) + + +def uninstall() -> None: + recorder = _state.get("screencast") + if recorder is not None: + recorder.stop() # never leave a poll thread running past teardown + if not _state["installed"]: + _state.update( + screencast=None, snapshot=None, setup_done=False, + output_dir=None, default_suite=None, + ) + return + _state["cls"].execute = _state["orig"] # type: ignore[union-attr] + _state.update( + installed=False, cls=None, orig=None, screencast=None, snapshot=None, + setup_done=False, output_dir=None, default_suite=None, + ) diff --git a/packages/selenium-py-devtools/src/wdio_selenium_devtools/lifecycle.py b/packages/selenium-py-devtools/src/wdio_selenium_devtools/lifecycle.py new file mode 100644 index 00000000..9e5e3efa --- /dev/null +++ b/packages/selenium-py-devtools/src/wdio_selenium_devtools/lifecycle.py @@ -0,0 +1,363 @@ +"""Dashboard browser-window lifecycle β€” mirror the JS adapters' behavior. + +Three flows, matching `packages/selenium-devtools`: + +1. On capture start, open an external browser window at the dashboard URL and + keep a closable handle to it. +2. When the backend sends a ``clientDisconnected`` control frame (the user + closed the dashboard window/tab), shut capture down and exit the process. +3. When the Python process ends (normal exit / SIGINT / SIGTERM), close the + browser window we opened. + +Everything here is best-effort: a failure to open or close the browser must +never crash the user's test run. All side effects (signal handlers, atexit, +real subprocess spawning) happen only through :func:`register_exit_handlers` +and :func:`open_dashboard`, never at import time, so importing this module is +inert (important for unittest). +""" + +from __future__ import annotations + +import atexit +import os +import shutil +import signal +import subprocess +import sys +import tempfile +import threading +from typing import Callable, Optional + +from .constants import ENV_OPEN + +# ── Local timing constants (lifecycle-specific) ────────────────────────────── +# These live here because constants.py is owned elsewhere; they could move there +# alongside ENV_OPEN if a second module ever needs them. +SHUTDOWN_EXIT_CODE = 0 +SHUTDOWN_GRACE_S = 1.5 # let the WS reader thread unwind before the hard exit +BROWSER_TERM_TIMEOUT_S = 3.0 +DASHBOARD_WINDOW_SIZE = "1600,1200" + +# macOS Chrome/Chromium binaries, most-preferred first. +_MACOS_CHROME_CANDIDATES = ( + "/Applications/Google Chrome.app/Contents/MacOS/Google Chrome", + "/Applications/Google Chrome Beta.app/Contents/MacOS/Google Chrome Beta", + "/Applications/Chromium.app/Contents/MacOS/Chromium", + os.path.expanduser( + "~/Applications/Google Chrome.app/Contents/MacOS/Google Chrome" + ), +) + + +def _log(msg: str) -> None: + print(f"[wdio-devtools] {msg}", file=sys.stderr) + + +# ── Browser handle ─────────────────────────────────────────────────────────── + + +class BrowserHandle: + """A closable reference to the browser window we opened for the dashboard. + + Holds the launched subprocess plus its throwaway ``--user-data-dir`` so + :meth:`close` can terminate exactly this window and clean the profile up β€” + the JS adapter uses ``pkill -f`` on a unique dir; we hold the handle + directly, which is both more precise and unit-testable. + """ + + def __init__( + self, + proc: Optional[subprocess.Popen] = None, + user_data_dir: Optional[str] = None, + ) -> None: + self.proc = proc + self.user_data_dir = user_data_dir + self._closed = False + + def close(self) -> None: + """Terminate the dashboard window and remove its temp profile. Idempotent.""" + if self._closed: + return + self._closed = True + proc = self.proc + if proc is not None and proc.poll() is None: + try: + proc.terminate() + try: + proc.wait(timeout=BROWSER_TERM_TIMEOUT_S) + except subprocess.TimeoutExpired: + proc.kill() + except OSError: + pass # best-effort: never crash on browser teardown + if self.user_data_dir: + shutil.rmtree(self.user_data_dir, ignore_errors=True) + + +# ── Opening the dashboard window ───────────────────────────────────────────── + + +def _find_chrome() -> Optional[str]: + """Path to a Chrome/Chromium binary on this machine, or None.""" + for candidate in _MACOS_CHROME_CANDIDATES: + if os.path.exists(candidate): + return candidate + return None + + +def _default_opener(url: str) -> BrowserHandle: + """Open ``url`` in a dedicated, isolated Chrome window we can later close. + + We spawn the Chrome binary directly rather than stdlib ``webbrowser`` or + macOS ``open``: both hand the URL to the user's already-running Chrome, + which then shows the dashboard as a tab among all their other tabs and + gives us no handle to close it β€” exactly the bug this avoids. + + The isolation guarantee is the throwaway ``--user-data-dir``: launching the + Chrome binary with a distinct profile dir forces a brand-new Chrome + *instance* (a separate process that cannot merge into the user's running + Chrome), so the dashboard always gets its own window. ``--app`` makes that + window chrome-less (no tab strip/omnibox) and ``--new-window`` is a belt- + and-suspenders hint. Holding the subprocess handle lets :meth:`close` + terminate exactly this window β€” more precise than the JS adapter's + ``pkill -f`` on the profile dir, and unit-testable. + """ + chrome = _find_chrome() + if chrome is None: + _log(f"Chrome not found; open the dashboard manually: {url}") + return BrowserHandle() + + user_data_dir = tempfile.mkdtemp(prefix="selenium-py-devtools-ui-") + args = [ + chrome, + f"--user-data-dir={user_data_dir}", # forces a separate Chrome instance + "--no-first-run", + "--no-default-browser-check", + f"--window-size={DASHBOARD_WINDOW_SIZE}", + "--new-window", + f"--app={url}", # dedicated dashboard window, chrome-less + ] + proc = subprocess.Popen(args, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) + _log(f"Opened DevTools UI in a dedicated window: {url}") + return BrowserHandle(proc=proc, user_data_dir=user_data_dir) + + +_FALSY = ("0", "false", "no", "off", "") + + +def auto_open_enabled() -> bool: + """Whether the dashboard window should auto-open. Default ON, opt-out only. + + Rule: open unless ``WDIO_DEVTOOLS_OPEN`` is set to a falsy value + (``0``/``false``/``no``/``off``/empty). This matches the JS adapters, whose + ``openUi`` option defaults true regardless of TTY. + + The previous "default off when stdout isn't a TTY" gate silently disabled + auto-open for the common case β€” running from an IDE or ``python demo.py`` + with no attached TTY β€” so the user opened the URL in their main Chrome + instead. CI/headless runs disable it explicitly with ``WDIO_DEVTOOLS_OPEN=0``. + """ + val = os.environ.get(ENV_OPEN) + if val is None: + return True + return val.strip().lower() not in _FALSY + + +def open_dashboard( + url: Optional[str], + *, + opener: Callable[[str], BrowserHandle] = _default_opener, +) -> Optional[BrowserHandle]: + """Open ``url`` in a closable browser window; return its handle or None. + + ``opener`` is injectable so tests can assert behavior without a real + browser. Never raises β€” a failed open logs one line and returns None. + """ + if not url: + return None + try: + return opener(url) + except (OSError, ValueError) as exc: + _log(f"could not open dashboard window ({exc}); open manually: {url}") + return None + + +# ── Shutdown wiring ────────────────────────────────────────────────────────── + +# Set by register_exit_handlers so the control-frame handler and signal handlers +# can reach the package's disable() and the open window without an import cycle. +_disable: Optional[Callable[[], None]] = None +_handle: Optional[BrowserHandle] = None +_handlers_registered = False +_prev_sigint = None +_prev_sigterm = None +_shutting_down = False +_shutdown_lock = threading.Lock() +# Set when the dashboard is closed / a shutdown is triggered β€” lets a caller +# (e.g. the pytest plugin) block after a run to keep the dashboard open for +# inspection, then exit when the user closes it. +_shutdown_event = threading.Event() +_has_waiter = False + + +def dashboard_window_open() -> bool: + """True if we opened a dashboard window and its process is still alive.""" + h = _handle + return h is not None and h.proc is not None and h.proc.poll() is None + + +def wait_for_shutdown(timeout: Optional[float] = None) -> bool: + """Block until the dashboard is closed (clientDisconnected) or a signal. + + Used to keep the dashboard open for inspection after a run. Returns True on + shutdown, False on timeout. Registers a waiter so the WS handler hands + teardown back to the caller instead of hard-exiting out from under it.""" + global _has_waiter + _has_waiter = True + return _shutdown_event.wait(timeout) + + +def _run_disable() -> None: + """Call the registered disable() once, swallowing errors.""" + fn = _disable + if fn is None: + return + try: + fn() + except Exception as exc: # disable must never re-raise into a handler + _log(f"error during disable(): {exc}") + + +def _close_handle() -> None: + """Close the opened browser window if any.""" + handle = _handle + if handle is not None: + handle.close() + + +def on_control(scope: str, data: dict) -> None: + """WS control-frame handler: shut down when the dashboard client leaves. + + ``clientDisconnected`` means the user closed the dashboard window, so we + tear capture down and exit the process (on a short timer, off the WS reader + thread, so that thread can unwind cleanly). ``clientConnected`` is a no-op. + """ + if scope == "clientDisconnected": + _log("dashboard closed; shutting down") + _trigger_shutdown(exit_after=True) + + +def _trigger_shutdown(*, exit_after: bool, exit_code: int = SHUTDOWN_EXIT_CODE) -> None: + """Run disable() + close the window once; optionally hard-exit afterwards.""" + global _shutting_down + with _shutdown_lock: + if _shutting_down: + return + _shutting_down = True + + _shutdown_event.set() # unblock any wait_for_shutdown() + if _has_waiter: + # A caller is blocked in wait_for_shutdown() and owns teardown β€” don't + # hard-exit out from under it. + return + + _run_disable() + _close_handle() + + if exit_after: + # Defer the hard exit to a daemon timer so the WS reader thread (which + # may be the caller) unwinds first; os._exit avoids re-entering atexit. + def _exit() -> None: + os._exit(exit_code) + + timer = threading.Timer(SHUTDOWN_GRACE_S, _exit) + timer.daemon = True + timer.start() + + +def _on_signal(signum, _frame) -> None: + """SIGINT/SIGTERM: close the window + disable, then re-raise the default.""" + _trigger_shutdown(exit_after=False) + prev = _prev_sigint if signum == signal.SIGINT else _prev_sigterm + if callable(prev): + prev(signum, _frame) + else: + # Restore default disposition and re-raise so the process dies normally. + try: + signal.signal(signum, signal.SIG_DFL) + os.kill(os.getpid(), signum) + except OSError: + os._exit(128 + signum) + + +def register_exit_handlers( + disable: Callable[[], None], + handle: Optional[BrowserHandle], +) -> None: + """Register atexit + SIGINT/SIGTERM handlers to close the window on exit. + + Idempotent: called from enable(). Signal handlers are only installed on the + main thread (Python forbids otherwise) and the previous handlers are chained + so we don't swallow the runner's own Ctrl-C behavior. + """ + global _disable, _handle, _handlers_registered, _prev_sigint, _prev_sigterm + global _shutting_down, _has_waiter + _disable = disable + _handle = handle + # Fresh shutdown state each run (supports enableβ†’disableβ†’enable). + _shutdown_event.clear() + _shutting_down = False + _has_waiter = False + if _handlers_registered: + return + _handlers_registered = True + + atexit.register(_close_handle) + + if threading.current_thread() is threading.main_thread(): + try: + _prev_sigint = signal.getsignal(signal.SIGINT) + _prev_sigterm = signal.getsignal(signal.SIGTERM) + signal.signal(signal.SIGINT, _on_signal) + signal.signal(signal.SIGTERM, _on_signal) + except (ValueError, OSError) as exc: + _log(f"could not install signal handlers ({exc})") + + +def unregister_exit_handlers() -> None: + """Undo register_exit_handlers; close the window. Idempotent β€” for disable().""" + global _disable, _handle, _handlers_registered, _prev_sigint, _prev_sigterm + _close_handle() + if _handlers_registered: + try: + atexit.unregister(_close_handle) + except Exception: + pass + if threading.current_thread() is threading.main_thread(): + for sig, prev in ( + (signal.SIGINT, _prev_sigint), + (signal.SIGTERM, _prev_sigterm), + ): + if prev is not None: + try: + signal.signal(sig, prev) + except (ValueError, OSError): + pass + _disable = None + _handle = None + _handlers_registered = False + _prev_sigint = None + _prev_sigterm = None + + +def _reset_for_tests() -> None: + """Reset module state between unit tests (never used in production).""" + global _disable, _handle, _handlers_registered + global _prev_sigint, _prev_sigterm, _shutting_down, _has_waiter + _disable = None + _handle = None + _handlers_registered = False + _prev_sigint = None + _prev_sigterm = None + _shutting_down = False + _has_waiter = False + _shutdown_event.clear() diff --git a/packages/selenium-py-devtools/src/wdio_selenium_devtools/logcapture.py b/packages/selenium-py-devtools/src/wdio_selenium_devtools/logcapture.py new file mode 100644 index 00000000..3ea0a48b --- /dev/null +++ b/packages/selenium-py-devtools/src/wdio_selenium_devtools/logcapture.py @@ -0,0 +1,70 @@ +"""Forward Python ``logging`` output to the dashboard Console β€” the 'runner' log +stream, the analogue of the JS adapter surfacing ``@wdio/logger`` output +(webdriver/BiDi lines, the adapter's own 'βœ“ Script injected' events, etc.). + +Attaches a handler to the root logger and raises the adapter's + selenium's +loggers to a useful level so their records reach the dashboard. Restores +everything on stop. +""" + +from __future__ import annotations + +import logging +from typing import Any, Dict + +#: The adapter's logger name β€” operational events are logged under it. +LOGGER_NAME = "wdio_selenium_devtools" + +_LEVEL_MAP = { + logging.DEBUG: "debug", + logging.INFO: "info", + logging.WARNING: "warn", + logging.ERROR: "error", + logging.CRITICAL: "error", +} + +# Loggers raised to a useful level so their records reach the dashboard: the +# adapter's own (verbose) and selenium's (info: session/BiDi lines). +_WATCH = {LOGGER_NAME: logging.DEBUG, "selenium": logging.INFO} + + +class _DashboardHandler(logging.Handler): + def __init__(self, capturer: Any) -> None: + super().__init__(level=logging.DEBUG) + self._capturer = capturer + self._reentrant = False + + def emit(self, record: logging.LogRecord) -> None: + if self._reentrant: # forwarding must not re-enter via its own logging + return + self._reentrant = True + try: + level = _LEVEL_MAP.get(record.levelno, "log") + message = f"{record.name}: {record.getMessage()}" + self._capturer.capture_console(level, [message], source="terminal") + except Exception: # noqa: BLE001 β€” logging must never break the test + pass + finally: + self._reentrant = False + + +class LogCapturer: + """Streams Python logging into the dashboard Console for the run's duration.""" + + def __init__(self, capturer: Any) -> None: + self._handler = _DashboardHandler(capturer) + self._prev_levels: Dict[str, int] = {} + + def start(self) -> None: + logging.getLogger().addHandler(self._handler) + for name, level in _WATCH.items(): + logger = logging.getLogger(name) + self._prev_levels[name] = logger.level + if logger.level == logging.NOTSET or logger.level > level: + logger.setLevel(level) + + def stop(self) -> None: + logging.getLogger().removeHandler(self._handler) + for name, prev in self._prev_levels.items(): + logging.getLogger(name).setLevel(prev) + self._prev_levels.clear() diff --git a/packages/selenium-py-devtools/src/wdio_selenium_devtools/pytest_plugin.py b/packages/selenium-py-devtools/src/wdio_selenium_devtools/pytest_plugin.py new file mode 100644 index 00000000..63c83e89 --- /dev/null +++ b/packages/selenium-py-devtools/src/wdio_selenium_devtools/pytest_plugin.py @@ -0,0 +1,124 @@ +"""pytest plugin β€” feeds the suite/test tree to the dashboard. + +The analogue of the JS adapter's mocha/jest hooks. Inert unless the run opts in +via ``WDIO_DEVTOOLS=1`` or ``DEVTOOLS_PORT=...`` so installing the package never +hijacks an unrelated pytest run. On opt-in it enables capture at session start, +stamps per-test timing, and re-sends the ``suites`` frame as each test reports. +""" + +from __future__ import annotations + +import os +from typing import Dict + +import wdio_selenium_devtools as devtools +from . import frames +from .constants import ENV_OPT_IN, ENV_PORT +from .utils import now_ms + + +def _opted_in() -> bool: + return bool(os.environ.get(ENV_OPT_IN) or os.environ.get(ENV_PORT)) + + +class _SuiteRegistry: + """Groups tests by file into the SuiteStats[] the dashboard expects.""" + + def __init__(self) -> None: + self._suites: Dict[str, dict] = {} + self._tests: Dict[str, dict] = {} + self._starts: Dict[str, int] = {} + + def mark_start(self, nodeid: str) -> None: + self._starts.setdefault(nodeid, now_ms()) + + def record(self, nodeid: str, file: str, name: str, line: int, state: str) -> None: + start = self._starts.get(nodeid, now_ms()) + end = now_ms() + suite = self._suites.setdefault( + file, + frames.suite_stats(uid=file, title=file, file=file, + start_ms=start, tests=[]), + ) + self._tests[nodeid] = frames.test_stats( + uid=nodeid, + title=name, + full_title=f"{file} β€Ί {name}", + parent=file, + state=state, + file=file, + start_ms=start, + end_ms=end, + call_source=f"{file}:{line + 1}", + ) + suite["tests"] = [t for nid, t in self._tests.items() + if nid.split("::", 1)[0] == file] + suite["end"] = self._tests[nodeid]["end"] + suite["state"] = "failed" if any( + t["state"] == "failed" for t in suite["tests"] + ) else state + + def snapshot(self) -> list: + return list(self._suites.values()) + + +_registry = _SuiteRegistry() + +# Note: test-file source capture for the Source tab lives in instrumentation.py +# (keyed by each command's callSource path), so it works for plain scripts too β€” +# not just pytest. The plugin only owns the suite/test tree. + + +def pytest_configure(config) -> None: # noqa: ANN001 + if _opted_in(): + capturer = devtools.enable() + # pytest owns the suite tree β€” suppress the adapter's default script suite. + from . import instrumentation + + instrumentation.set_external_suites(True) + url = devtools.dashboard_url() + if capturer is not None and url: + print(f"\n[wdio-devtools] dashboard live at {url}\n") + + +def pytest_runtest_logstart(nodeid, location) -> None: # noqa: ANN001 + if _opted_in(): + _registry.mark_start(nodeid) + + +def pytest_runtest_logreport(report) -> None: # noqa: ANN001 + if not _opted_in(): + return + capturer = devtools.get_capturer() + if capturer is None: + return + # 'call' carries pass/fail; a skip surfaces at 'setup'. + if report.when == "call" or (report.when == "setup" and report.skipped): + state = ( + "skipped" if report.skipped + else "passed" if report.passed + else "failed" + ) + file, line, name = report.location + _registry.record(report.nodeid, file, name, line or 0, state) + capturer.send_suites(_registry.snapshot()) + + +def pytest_sessionfinish(session, exitstatus) -> None: # noqa: ANN001 + if not _opted_in(): + return + capturer = devtools.get_capturer() + if capturer is not None: + capturer.send_suites(_registry.snapshot()) + # Keep the dashboard open for inspection after the run β€” exit when the user + # closes the window (clientDisconnected). Only when we actually opened a + # window; CI (WDIO_DEVTOOLS_OPEN=0) tears down immediately. + from . import lifecycle + + if lifecycle.dashboard_window_open(): + print( + "\n[wdio-devtools] dashboard is live β€” close the window to finish " + "(Ctrl-C also works).\n" + ) + lifecycle.wait_for_shutdown() + devtools.disable() diff --git a/packages/selenium-py-devtools/src/wdio_selenium_devtools/screencast.py b/packages/selenium-py-devtools/src/wdio_selenium_devtools/screencast.py new file mode 100644 index 00000000..4beae282 --- /dev/null +++ b/packages/selenium-py-devtools/src/wdio_selenium_devtools/screencast.py @@ -0,0 +1,213 @@ +"""Screencast recorder β€” the Python analogue of core's ``ScreencastRecorderBase``. + +Frames are captured **synchronously on the main thread, one per command** +(``driver.get_screenshot_as_base64()``), driven from the instrumentation hook. +A background poll thread is deliberately NOT used: Selenium's session is not +thread-safe, and a screenshot fired from a daemon thread races the main thread's +commands and DOM-trace readback on the same connection β€” corrupting both the +video and the snapshot. The reference JS adapters avoid this too (CDP push-mode +screencast / per-command screenshots on the command queue), so we mirror that. + +On ``stop`` the buffered frames are encoded to a ``.webm`` via ffmpeg *if it's on +PATH* β€” ffmpeg is an optional dependency, so its absence is a one-line warning +and a skipped encode, never an error. + +A CDP push-mode fast-path (Chrome ``Page.startScreencast`` via +``execute_cdp_cmd``) is a future optimization β€” noted here, not implemented; +per-command capture works on every browser selenium drives. + +Everything is defensive: a transient screenshot failure (e.g. mid-navigation) is +skipped and recording continues; a recorder that never captured a frame encodes +nothing. Capture never breaks the user's test. +""" + +from __future__ import annotations + +import os +import shutil +import subprocess +import sys +import tempfile +from typing import Any, Callable, List, Optional + +from .constants import SCREENCAST_IMAGE_FORMAT, SCREENCAST_MIN_FRAMES +from .types import ScreencastFrame +from .utils import now_ms + +#: Screenshot fn signature β€” injectable so tests drive the buffer without a +#: real driver. Returns a base64 PNG string, or None on a transient failure. +ScreenshotFn = Callable[[], Optional[str]] + + +def _warn(message: str) -> None: + print(f"[wdio-devtools] screencast: {message}", file=sys.stderr) + + +class ScreencastRecorder: + def __init__(self, *, ffmpeg_path: Optional[str] = None) -> None: + # Resolve ffmpeg once; None means "encoding unavailable, skip it". + self._ffmpeg = ffmpeg_path or shutil.which("ffmpeg") + self._frames: List[ScreencastFrame] = [] + self._screenshot: Optional[ScreenshotFn] = None + self._active = False + + # ── public API ──────────────────────────────────────────────────────────── + + def start(self, driver: Any, screenshot_fn: Optional[ScreenshotFn] = None) -> None: + """Arm the recorder and capture a seed frame. ``screenshot_fn`` is + injectable for tests; by default it reads + ``driver.get_screenshot_as_base64``. Frames are captured on the main + thread via :meth:`capture` β€” no background polling. Safe to call twice.""" + if self._active: + return + if screenshot_fn is not None: + self._screenshot = screenshot_fn + elif driver is not None and hasattr(driver, "get_screenshot_as_base64"): + self._screenshot = driver.get_screenshot_as_base64 + else: + _warn("driver has no get_screenshot_as_base64 β€” recording skipped") + return + self._active = True + self.capture() # best-effort seed frame + + def capture(self) -> bool: + """Capture one frame synchronously (main thread). No-op if not armed. + + A transient failure β€” a screenshot taken mid-navigation, a dead session + β€” is skipped and recording stays armed, so a single miss never truncates + the video the way a background poll loop would. Returns True iff a frame + was buffered.""" + if not self._active: + return False + fn = self._screenshot + if fn is None: + return False + try: + data = fn() + except Exception: # noqa: BLE001 β€” transient miss; keep recording + return False + if isinstance(data, str) and data: + self._frames.append({"data": data, "timestamp": now_ms()}) + return True + return False + + def add_frame(self, data: Optional[str]) -> bool: + """Buffer a frame from an ALREADY-captured base64 screenshot β€” lets the + caller reuse the per-command screenshot it took for the command entry + instead of paying for a second screenshot round-trip. No-op if not armed + or the data is empty. Returns True iff a frame was buffered.""" + if not self._active or not isinstance(data, str) or not data: + return False + self._frames.append({"data": data, "timestamp": now_ms()}) + return True + + def stop(self) -> None: + """Disarm the recorder. Idempotent; safe even if start() never ran.""" + self._active = False + + @property + def frames(self) -> List[ScreencastFrame]: + return list(self._frames) + + @property + def duration(self) -> int: + """ms between first and last frame; 0 with fewer than 2 frames.""" + if len(self._frames) < 2: + return 0 + return self._frames[-1]["timestamp"] - self._frames[0]["timestamp"] + + @property + def is_recording(self) -> bool: + return self._active + + # ── finalize ────────────────────────────────────────────────────────────── + + def finalize( + self, + session_id: str, + output_dir: Optional[str] = None, + *, + min_frames: int = SCREENCAST_MIN_FRAMES, + filename_prefix: str = "selenium-py-video", + ) -> Optional[dict]: + """Stop, encode the buffered frames to a ``.webm``, and return the + ``{video_path, video_file, frame_count, duration, start_time}`` metadata + the capturer forwards β€” or None if there's nothing to encode / ffmpeg is + absent. All failures are caught: screencast is best-effort.""" + self.stop() + frames = self.frames + if len(frames) < min_frames: + return None + if not self._ffmpeg: + _warn("ffmpeg not found on PATH β€” skipping video encode " + f"({len(frames)} frame(s) captured)") + return None + + file_name = f"{filename_prefix}-{session_id}.webm" + target_dir = _writable_dir(output_dir) + video_path = os.path.join(target_dir, file_name) + try: + _encode_webm(frames, video_path, self._ffmpeg) + except Exception as exc: # noqa: BLE001 β€” encode failure must not abort + _warn(f"encode failed: {exc}") + return None + return { + "video_path": video_path, + "video_file": file_name, + "frame_count": len(frames), + "duration": self.duration, + "start_time": frames[0]["timestamp"] if frames else None, + } + + +def _writable_dir(preferred: Optional[str]) -> str: + candidate = preferred or os.getcwd() + if os.path.isdir(candidate) and os.access(candidate, os.W_OK): + return candidate + return tempfile.gettempdir() + + +def _encode_webm(frames: List[ScreencastFrame], output_path: str, ffmpeg: str) -> None: + """Encode base64 frames to a VP8/WebM via ffmpeg's concat demuxer, giving + each frame its real inter-frame duration (VFR reflects command pauses). + Forces CFR 10fps + all-intra so the dashboard ``