From 6a1c33cbe25dcec073bfca3e1e2d98bbe80cc2a1 Mon Sep 17 00:00:00 2001 From: Vishnu Vardhan Date: Wed, 1 Jul 2026 14:36:22 +0530 Subject: [PATCH 01/12] feat(selenium-py-devtools): add Python Selenium adapter --- .github/workflows/python.yml | 55 +++++ examples/python-spike/README.md | 77 +++++++ examples/python-spike/spike.py | 213 ++++++++++++++++++ examples/python-spike/verify-client.mjs | 57 +++++ packages/selenium-py-devtools/.gitignore | 8 + packages/selenium-py-devtools/README.md | 148 ++++++++++++ .../selenium-py-devtools/e2e/test_smoke.py | 35 +++ packages/selenium-py-devtools/e2e_check.py | 48 ++++ packages/selenium-py-devtools/pyproject.toml | 28 +++ .../scripts/gen_contract.py | 95 ++++++++ .../src/wdio_selenium_devtools/__init__.py | 92 ++++++++ .../src/wdio_selenium_devtools/_contract.py | 12 + .../src/wdio_selenium_devtools/backend.py | 114 ++++++++++ .../src/wdio_selenium_devtools/capturer.py | 100 ++++++++ .../src/wdio_selenium_devtools/constants.py | 37 +++ .../src/wdio_selenium_devtools/frames.py | 144 ++++++++++++ .../wdio_selenium_devtools/instrumentation.py | 83 +++++++ .../wdio_selenium_devtools/pytest_plugin.py | 102 +++++++++ .../src/wdio_selenium_devtools/transport.py | 185 +++++++++++++++ .../src/wdio_selenium_devtools/types.py | 93 ++++++++ .../src/wdio_selenium_devtools/utils.py | 52 +++++ .../tests/test_backend.py | 57 +++++ .../tests/test_capturer.py | 58 +++++ .../selenium-py-devtools/tests/test_frames.py | 57 +++++ .../tests/test_instrumentation.py | 81 +++++++ 25 files changed, 2031 insertions(+) create mode 100644 .github/workflows/python.yml create mode 100644 examples/python-spike/README.md create mode 100644 examples/python-spike/spike.py create mode 100644 examples/python-spike/verify-client.mjs create mode 100644 packages/selenium-py-devtools/.gitignore create mode 100644 packages/selenium-py-devtools/README.md create mode 100644 packages/selenium-py-devtools/e2e/test_smoke.py create mode 100644 packages/selenium-py-devtools/e2e_check.py create mode 100644 packages/selenium-py-devtools/pyproject.toml create mode 100644 packages/selenium-py-devtools/scripts/gen_contract.py create mode 100644 packages/selenium-py-devtools/src/wdio_selenium_devtools/__init__.py create mode 100644 packages/selenium-py-devtools/src/wdio_selenium_devtools/_contract.py create mode 100644 packages/selenium-py-devtools/src/wdio_selenium_devtools/backend.py create mode 100644 packages/selenium-py-devtools/src/wdio_selenium_devtools/capturer.py create mode 100644 packages/selenium-py-devtools/src/wdio_selenium_devtools/constants.py create mode 100644 packages/selenium-py-devtools/src/wdio_selenium_devtools/frames.py create mode 100644 packages/selenium-py-devtools/src/wdio_selenium_devtools/instrumentation.py create mode 100644 packages/selenium-py-devtools/src/wdio_selenium_devtools/pytest_plugin.py create mode 100644 packages/selenium-py-devtools/src/wdio_selenium_devtools/transport.py create mode 100644 packages/selenium-py-devtools/src/wdio_selenium_devtools/types.py create mode 100644 packages/selenium-py-devtools/src/wdio_selenium_devtools/utils.py create mode 100644 packages/selenium-py-devtools/tests/test_backend.py create mode 100644 packages/selenium-py-devtools/tests/test_capturer.py create mode 100644 packages/selenium-py-devtools/tests/test_frames.py create mode 100644 packages/selenium-py-devtools/tests/test_instrumentation.py diff --git a/.github/workflows/python.yml b/.github/workflows/python.yml new file mode 100644 index 00000000..d72aaf5a --- /dev/null +++ b/.github/workflows/python.yml @@ -0,0 +1,55 @@ +name: Python Adapter + +on: + push: + branches: + - main + tags: + - python-v[0-9]+.[0-9]+.[0-9]+* + 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 + + publish: + needs: test + if: startsWith(github.ref, 'refs/tags/python-v') + runs-on: ubuntu-latest + environment: pypi + permissions: + id-token: write # PyPI trusted publishing (OIDC) β€” no token needed + steps: + - uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd # v6.0.2 + - uses: actions/setup-python@a26af69be951a213d495a4c3e4e4022e16d87065 # v5 + with: + python-version: '3.12' + - name: πŸ“¦ Build sdist + wheel + run: | + python -m pip install --upgrade build + python -m build + - uses: pypa/gh-action-pypi-publish@cef221092ed1bacb1cc03d23a2d87d1d172e277b # release/v1 + with: + packages-dir: packages/selenium-py-devtools/dist diff --git a/examples/python-spike/README.md b/examples/python-spike/README.md new file mode 100644 index 00000000..f931930a --- /dev/null +++ b/examples/python-spike/README.md @@ -0,0 +1,77 @@ +# Python spike β€” Phase 0 feasibility proof + +Proves that a **non-JS client** can drive the DevTools dashboard with no +backend or UI changes. A dependency-free Python script speaks the adapter side +of the wire (raw socket + RFC-6455 handshake) and pushes the same +`{ "scope", "data" }` frames the JS Selenium adapter emits. + +This is a throwaway spike, not the adapter. Its job is to de-risk everything +downstream and to capture the **golden frames** the real Python adapter must +reproduce. + +## Result + +βœ… **Boundary proven end to end.** Every data scope the dashboard consumes +(`metadata`, `suites`, `commands`, `consoleLogs`, `networkRequests`) was sent +from Python and observed arriving at a dashboard-role `/client` subscriber. + +## Run it + +```bash +# 1. build once (from repo root) +pnpm --filter @wdio/devtools-app --filter @wdio/devtools-backend build + +# 2. start the backend + dashboard +node packages/backend/dist/index.js +# note the port it logs β€” 3000 if free, else a negotiated port e.g. 63763 + +# 3a. visual check: open the dashboard, then run the spike +# (export the port if it isn't 3000) +DEVTOOLS_PORT=63763 python3 examples/python-spike/spike.py +# β†’ a "Python spike" suite, a command timeline, console + network appear live + +# 3b. headless proof (no browser): run the listener, then the spike +DEVTOOLS_PORT=63763 node examples/python-spike/verify-client.mjs & +DEVTOOLS_PORT=63763 python3 examples/python-spike/spike.py +# β†’ listener prints "βœ“ all expected scopes received β€” boundary proven" +``` + +## What this established (the contract a Python adapter must honor) + +- **Endpoint.** Adapters connect to `ws://:/worker`. The dashboard + subscribes to `/client`. No handshake, auth, or session registration β€” the + backend parses each frame and broadcasts it verbatim + (`packages/backend/src/worker-message-handler.ts`). +- **Frame envelope.** `{ "scope": string, "data": }`, one JSON object + per WS text frame. `data` must be truthy or the UI drops it + (`DataManager.ts:308`). +- **Data scopes the UI renders** (`DataManager.ts:283-302`): + | scope | payload | notes | + |---|---|---| + | `metadata` | `Metadata` object | `type: "testrunner"`, carries `sessionId` | + | `suites` | `SuiteStats[]` | the test tree; re-send to update final state | + | `commands` | `CommandLog[]` | send incrementally for a live timeline | + | `consoleLogs` | `ConsoleLog[]` | `source: "browser" \| "test" \| "terminal"` | + | `networkRequests` | `NetworkRequest[]` | | +- **Field shapes** mirror `packages/shared/src/types.ts`. Two encoding rules + learned here: + - timestamps are **epoch milliseconds** (numbers). + - `SuiteStats.start/end` are TS `Date`s β€” they cross the wire as **ISO + strings**. +- **Port negotiation.** If 3000 is busy the backend uses `get-port` and logs the + real port. The JS adapters read it from `start()`'s return value; the Python + adapter will need the same (env var / discovery). +- **Replay buffer.** The backend buffers broadcast frames and replays them to + late-connecting clients β€” a dashboard opened mid-run catches up. The Python + adapter gets this for free. + +## Files + +- `spike.py` β€” dependency-free worker client + golden frames. +- `verify-client.mjs` β€” `/client` subscriber that asserts the frames fan out. + +## Not covered by this spike (later phases) + +Driver instrumentation (`execute()` wrap), BiDi console/network, screencast, +pytest lifecycle, trace export β€” see the Phase roadmap. This spike only proves +the transport boundary, which everything else depends on. diff --git a/examples/python-spike/spike.py b/examples/python-spike/spike.py new file mode 100644 index 00000000..11d9ae52 --- /dev/null +++ b/examples/python-spike/spike.py @@ -0,0 +1,213 @@ +#!/usr/bin/env python3 +""" +Phase 0 spike β€” prove a non-JS client can drive the DevTools dashboard. + +This script speaks the adapter side of the wire with ZERO third-party +dependencies (raw socket + RFC-6455 handshake + masked text frames). It +connects to the backend's /worker WebSocket and pushes the same +`{ "scope": ..., "data": ... }` frames the JS Selenium adapter emits. + +The real Python adapter would use the `websockets` package instead of this +hand-rolled client β€” the point here is only to prove the boundary is +language-agnostic, with nothing to install. + +Run: + 1. node packages/backend/dist/index.js # backend + dashboard on :3000 + 2. open http://localhost:3000 + 3. python3 examples/python-spike/spike.py # watch the run appear, live + +Frame shapes mirror packages/shared/src/types.ts. See README.md. +""" + +import base64 +import json +import os +import socket +import struct +import sys +import time + +HOST = os.environ.get("DEVTOOLS_HOST", "localhost") +PORT = int(os.environ.get("DEVTOOLS_PORT", "3000")) +WORKER_PATH = "/worker" +SESSION_ID = "spike-py-0001" + + +# ── Minimal RFC-6455 client (text frames, clientβ†’server masked) ────────────── + +def ws_connect(host: str, port: int, path: str) -> socket.socket: + sock = socket.create_connection((host, port), timeout=5) + key = base64.b64encode(os.urandom(16)).decode() + handshake = ( + f"GET {path} HTTP/1.1\r\n" + f"Host: {host}:{port}\r\n" + "Upgrade: websocket\r\n" + "Connection: Upgrade\r\n" + f"Sec-WebSocket-Key: {key}\r\n" + "Sec-WebSocket-Version: 13\r\n" + "\r\n" + ) + sock.sendall(handshake.encode()) + + # Read response headers up to the blank line. + buf = b"" + while b"\r\n\r\n" not in buf: + chunk = sock.recv(1024) + if not chunk: + raise ConnectionError("backend closed during handshake") + buf += chunk + status_line = buf.split(b"\r\n", 1)[0].decode(errors="replace") + if "101" not in status_line: + raise ConnectionError(f"upgrade failed: {status_line!r}") + return sock + + +def ws_send_text(sock: socket.socket, text: str) -> None: + payload = text.encode("utf-8") + header = bytearray() + header.append(0x81) # FIN + opcode 0x1 (text) + mask_bit = 0x80 # client frames MUST be masked + n = len(payload) + if n < 126: + header.append(mask_bit | n) + elif n < 65536: + header.append(mask_bit | 126) + header += struct.pack(">H", n) + else: + header.append(mask_bit | 127) + header += struct.pack(">Q", n) + mask = os.urandom(4) + header += mask + masked = bytes(b ^ mask[i % 4] for i, b in enumerate(payload)) + sock.sendall(bytes(header) + masked) + + +def send_frame(sock: socket.socket, scope: str, data) -> None: + ws_send_text(sock, json.dumps({"scope": scope, "data": data})) + print(f" β†’ sent {scope:<16} ({json.dumps(data)[:60]}…)") + + +def now_ms() -> int: + return int(time.time() * 1000) + + +def iso(ms: int) -> str: + # SuiteStats.start/end are Dates in TS β€” they cross the wire as ISO strings. + return time.strftime("%Y-%m-%dT%H:%M:%S", time.gmtime(ms / 1000)) + \ + f".{ms % 1000:03d}Z" + + +# ── The golden frames (mirror packages/shared/src/types.ts) ────────────────── + +def metadata_frame() -> dict: + return { + "type": "testrunner", # TraceType.Testrunner + "sessionId": SESSION_ID, + "url": "https://example.com/", + "capabilities": {"browserName": "chrome", "browserVersion": "126.0"}, + "desiredCapabilities": {"browserName": "chrome"}, + "viewport": {"width": 1280, "height": 720, + "offsetLeft": 0, "offsetTop": 0, "scale": 1}, + "testEnv": "python-spike", + } + + +def suite_frame(start: int) -> list: + test = { + "uid": "test-1", "cid": "0-0", + "title": "loads the homepage", + "fullTitle": "Python spike β€Ί loads the homepage", + "parent": "Python spike", + "state": "passed", + "start": iso(start), "end": iso(start + 1200), + "type": "test", "file": "examples/python-spike/spike.py", + "retries": 0, "_duration": 1200, + "callSource": "spike.py:loads_the_homepage", + } + suite = { + "uid": "suite-1", "cid": "0-0", + "title": "Python spike", "fullTitle": "Python spike", + "type": "suite", "file": "examples/python-spike/spike.py", + "start": iso(start), "end": iso(start + 1200), "state": "passed", + "tests": [test], "suites": [], "hooks": [], "_duration": 1200, + } + return [suite] + + +def command_frames(start: int) -> list: + base = [ + ("navigateTo", ["https://example.com/"], None), + ("findElement", [{"using": "css selector", "value": "h1"}], + {"ELEMENT": "elem-h1"}), + ("getText", [], "Example Domain"), + ("click", [{"using": "css selector", "value": "a"}], None), + ] + frames = [] + for i, (cmd, args, result) in enumerate(base): + ts = start + 100 + i * 250 + frames.append({ + "command": cmd, "args": args, "result": result, + "timestamp": ts, "startTime": ts - 40, + "callSource": f"spike.py:{40 + i}", "id": i + 1, + }) + return frames + + +def console_frames(start: int) -> list: + return [ + {"type": "info", "args": ["Hello from the Python spike 🐍"], + "timestamp": start + 120, "source": "browser"}, + {"type": "warn", "args": ["this is a synthetic console line"], + "timestamp": start + 480, "source": "browser"}, + ] + + +def network_frames(start: int) -> list: + return [{ + "id": "net-1", "url": "https://example.com/", "method": "GET", + "status": 200, "statusText": "OK", + "timestamp": start + 90, "startTime": start + 90, + "endTime": start + 240, "time": 150, "type": "document", + "response": {"fromCache": False, "headers": {}, "mimeType": "text/html", + "status": 200}, + }] + + +def main() -> int: + print(f"connecting to ws://{HOST}:{PORT}{WORKER_PATH} …") + try: + sock = ws_connect(HOST, PORT, WORKER_PATH) + except OSError as exc: + print(f"\n βœ— could not connect: {exc}") + print(" is the backend running? node packages/backend/dist/index.js") + return 1 + + print(" βœ“ connected β€” streaming a synthetic test run\n") + start = now_ms() + + send_frame(sock, "metadata", metadata_frame()) + time.sleep(0.3) + send_frame(sock, "suites", suite_frame(start)) + time.sleep(0.3) + send_frame(sock, "networkRequests", network_frames(start)) + send_frame(sock, "consoleLogs", console_frames(start)) + time.sleep(0.3) + + # Stream commands one at a time so the timeline fills in "live". + for frame in command_frames(start): + send_frame(sock, "commands", [frame]) + time.sleep(0.4) + + # Re-send the suite with final state so the tree settles green. + send_frame(sock, "suites", suite_frame(start)) + + print("\n βœ“ done β€” check the dashboard at " + f"http://{HOST}:{PORT}") + print(" keeping the socket open for 2s so the backend flushes…") + time.sleep(2) + sock.close() + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/examples/python-spike/verify-client.mjs b/examples/python-spike/verify-client.mjs new file mode 100644 index 00000000..e3b49e5e --- /dev/null +++ b/examples/python-spike/verify-client.mjs @@ -0,0 +1,57 @@ +// Proof harness: plays the role the dashboard plays β€” subscribes to the +// backend's /client WebSocket and records every frame the backend broadcasts. +// If the Python spike's frames arrive here, the language-agnostic boundary is +// proven end to end without needing to eyeball a browser. +// +// Run (from repo root, after starting the backend): +// node examples/python-spike/verify-client.mjs +// It listens for COLLECT_MS, prints the scopes it saw, and exits non-zero if +// the expected set didn't arrive. + +import { createRequire } from 'node:module' +import { fileURLToPath, pathToFileURL } from 'node:url' +import { dirname, resolve } from 'node:path' + +// `ws` is a dependency of packages/backend, not hoisted to the repo root β€” +// anchor module resolution there so this runs from any cwd. +const here = dirname(fileURLToPath(import.meta.url)) +const require = createRequire( + pathToFileURL(resolve(here, '../../packages/backend/package.json')) +) +const WebSocket = require('ws') + +const HOST = process.env.DEVTOOLS_HOST || 'localhost' +const PORT = process.env.DEVTOOLS_PORT || '3000' +const COLLECT_MS = Number(process.env.COLLECT_MS || 6000) +const EXPECTED = ( + process.env.EXPECT || 'metadata,suites,commands,consoleLogs,networkRequests' +) + .split(',') + .map((s) => s.trim()) + .filter(Boolean) + +const seen = new Map() +const ws = new WebSocket(`ws://${HOST}:${PORT}/client`) + +ws.on('open', () => console.log(`[client] subscribed to ws://${HOST}:${PORT}/client`)) +ws.on('error', (e) => { console.error('[client] error:', e.message); process.exit(2) }) +ws.on('message', (buf) => { + let msg + try { msg = JSON.parse(buf.toString()) } catch { return } + if (!msg || !msg.scope) return + seen.set(msg.scope, (seen.get(msg.scope) || 0) + 1) + const preview = JSON.stringify(msg.data).slice(0, 70) + console.log(`[client] β—€ ${msg.scope.padEnd(16)} ${preview}…`) +}) + +setTimeout(() => { + console.log('\n[client] ── summary ──') + for (const [scope, n] of seen) console.log(` ${scope.padEnd(18)} Γ—${n}`) + const missing = EXPECTED.filter((s) => !seen.has(s)) + if (missing.length) { + console.log(`\n βœ— missing expected scopes: ${missing.join(', ')}`) + process.exit(1) + } + console.log('\n βœ“ all expected scopes received β€” boundary proven') + process.exit(0) +}, COLLECT_MS) 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..101aa10e --- /dev/null +++ b/packages/selenium-py-devtools/README.md @@ -0,0 +1,148 @@ +# 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 proven in `examples/python-spike`. + +**Status: Phase 1 (MVP).** Live command capture + test tree, 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 + +`enable()` launches the dashboard backend for you (see +[Backend & publishing](#backend--publishing)) β€” no separate step needed. + +**With pytest** (auto-wired; opt in per run so it never hijacks other runs): + +```bash +WDIO_DEVTOOLS=1 pytest tests/ # DEVTOOLS_PORT= also opts in (and attaches) +``` + +**Without pytest** (any script / unittest): + +```python +import wdio_selenium_devtools as devtools + +devtools.enable() # launch-or-attach backend + instrument selenium +# ... your normal selenium code ... +devtools.disable() # terminates the backend if we launched it +``` + +If the backend can't be launched or reached, `enable()` warns and returns +`None` β€” capture is skipped, your tests still run. + +## What Phase 1 captures + +| Data | How | Scope sent | +|---|---|---| +| Commands (driver + element) | wrap `WebDriver.execute()` β€” the single chokepoint all commands flow through | `commands` | +| Session metadata | read `session_id` + `caps` after `newSession` | `metadata` | +| Test / suite tree | pytest plugin (`pytest_runtest_logreport` / `sessionfinish`) | `suites` | + +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. + +## 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 (patch target injectable for tests) + backend.py launch-or-attach the Node backend + port discovery + 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) +``` + +## 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 +``` + +Pair either with `examples/python-spike/verify-client.mjs` to watch the frames +fan out to a dashboard-role client. + +## Release (approach A) + +CI (`.github/workflows/python.yml`) runs the unit tests on Python 3.9 + 3.12 and +fails if the generated contract has drifted from `shared`. Tagging +`python-vX.Y.Z` builds the sdist + wheel and publishes to PyPI via trusted +publishing (one-time setup: a PyPI trusted-publisher entry + a `pypi` GitHub +environment). 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. + +## Roadmap + +- **Phase 2** β€” screencast (CDP `Page.startScreencast` via `execute_cdp_cmd` + + screenshot-poll fallback), per-command screenshots, 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..de5c0726 --- /dev/null +++ b/packages/selenium-py-devtools/e2e_check.py @@ -0,0 +1,48 @@ +#!/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 + +Pair with examples/python-spike/verify-client.mjs to confirm the captured +commands fan out to a dashboard client. 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..798542a1 --- /dev/null +++ b/packages/selenium-py-devtools/scripts/gen_contract.py @@ -0,0 +1,95 @@ +#!/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", +} + + +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..f4074fba --- /dev/null +++ b/packages/selenium-py-devtools/src/wdio_selenium_devtools/__init__.py @@ -0,0 +1,92 @@ +"""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 sys +from typing import Optional + +from . import backend, instrumentation +from ._contract import CONTRACT_VERSION +from .capturer import SessionCapturer +from .constants import DEFAULT_HOST, DEFAULT_PORT, ENV_HOST, ENV_PORT +from .transport import WSClient + +__version__ = "0.1.0" +__all__ = ["enable", "disable", "get_capturer", "CONTRACT_VERSION"] + +_active: dict = {"capturer": None, "transport": None, "process": 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) + 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) + _active.update(capturer=capturer, transport=transport, process=process) + return capturer + + +def disable() -> None: + instrumentation.uninstall() + 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() + _active.update(capturer=None, transport=None, process=None) + + +def get_capturer() -> Optional[SessionCapturer]: + return _active["capturer"] 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..13d37d00 --- /dev/null +++ b/packages/selenium-py-devtools/src/wdio_selenium_devtools/_contract.py @@ -0,0 +1,12 @@ +# 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" + +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..eb426545 --- /dev/null +++ b/packages/selenium-py-devtools/src/wdio_selenium_devtools/backend.py @@ -0,0 +1,114 @@ +"""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, +) + +_PORT_RE = re.compile(r"on port (\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/capturer.py b/packages/selenium-py-devtools/src/wdio_selenium_devtools/capturer.py new file mode 100644 index 00000000..ec6a2f5c --- /dev/null +++ b/packages/selenium-py-devtools/src/wdio_selenium_devtools/capturer.py @@ -0,0 +1,100 @@ +"""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_NETWORK_REQUESTS, + SCOPE_SUITES, +) +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], + ) -> 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, + ) + 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)]) + + # ── suites ─────────────────────────────────────────────────────────────────── + + def send_suites(self, suites: List[dict]) -> None: + self._tx.send_json(SCOPE_SUITES, suites) + + 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..2aa1fece --- /dev/null +++ b/packages/selenium-py-devtools/src/wdio_selenium_devtools/constants.py @@ -0,0 +1,37 @@ +"""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" + +# ── 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. +SKIP_COMMANDS = frozenset( + {"newSession", "quit", "status", "getLog", "getAllSessions", "getSessions"} +) + +# Stack-frame path fragments to skip when resolving a command's call source β€” +# the adapter's own package and selenium internals. +_PACKAGE_DIR = os.path.dirname(__file__) +SKIP_STACK_FRAMES = (_PACKAGE_DIR, f"{os.sep}selenium{os.sep}") 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..c55daa77 --- /dev/null +++ b/packages/selenium-py-devtools/src/wdio_selenium_devtools/frames.py @@ -0,0 +1,144 @@ +"""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`` and are pinned by the Phase-0 +golden frames (see examples/python-spike). 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, 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, +) -> 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), + } + 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, + status: Optional[int], + timestamp: int, + start_time: int, + request_type: str = "fetch", + end_time: Optional[int] = None, +) -> NetworkRequest: + return { + "id": request_id, + "url": url, + "method": method, + "status": status, + "timestamp": timestamp, + "startTime": start_time, + "endTime": end_time, + "type": request_type, + } + + +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..6b4ca019 --- /dev/null +++ b/packages/selenium-py-devtools/src/wdio_selenium_devtools/instrumentation.py @@ -0,0 +1,83 @@ +"""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 + +from typing import Any, Optional + +from .capturer import SessionCapturer +from .constants import SKIP_COMMANDS, SKIP_STACK_FRAMES +from .utils import call_source, now_ms + +_state: dict = {"installed": False, "cls": None, "orig": None} + + +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 + # Skip capture for noise, but never alter behavior. + if driver_command in SKIP_COMMANDS: + result = orig_execute(self, driver_command, params) + if driver_command == "newSession": + capturer.ensure_metadata( + getattr(self, "session_id", None), + getattr(self, "caps", None), + None, + ) + return result + + start = now_ms() + src = call_source(SKIP_STACK_FRAMES) + 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 + + capturer.ensure_metadata( + getattr(self, "session_id", None), getattr(self, "caps", None), None + ) + # WebDriver.execute returns the full response dict; the useful payload + # is response["value"]. + value = result.get("value") if isinstance(result, dict) else result + capturer.capture_command( + command=driver_command, + args=params, + result=value, + start_time=start, + call_source=src, + ) + return result + + webdriver_cls.execute = patched_execute # type: ignore[assignment] + _state.update(installed=True, cls=webdriver_cls, orig=orig_execute) + + +def uninstall() -> None: + if not _state["installed"]: + return + _state["cls"].execute = _state["orig"] # type: ignore[union-attr] + _state.update(installed=False, cls=None, orig=None) 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..8ec439b2 --- /dev/null +++ b/packages/selenium-py-devtools/src/wdio_selenium_devtools/pytest_plugin.py @@ -0,0 +1,102 @@ +"""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() + + +def pytest_configure(config) -> None: # noqa: ANN001 + if _opted_in(): + devtools.enable() + + +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()) + devtools.disable() diff --git a/packages/selenium-py-devtools/src/wdio_selenium_devtools/transport.py b/packages/selenium-py-devtools/src/wdio_selenium_devtools/transport.py new file mode 100644 index 00000000..f097e1db --- /dev/null +++ b/packages/selenium-py-devtools/src/wdio_selenium_devtools/transport.py @@ -0,0 +1,185 @@ +"""Dependency-free WebSocket client for the adapterβ†’backend ``/worker`` link. + +A hardened version of the Phase-0 spike's client: RFC-6455 handshake, masked +client text frames, plus a background reader thread that handles ping/pong and +surfaces the backend's control frames (``clientConnected`` / ``clientDisconnected``) +to a callback. No third-party dependency β€” the only runtime requirement the +adapter adds on top of selenium itself. +""" + +from __future__ import annotations + +import base64 +import json +import os +import socket +import struct +import threading +from typing import Callable, Optional + +from .constants import CONNECT_TIMEOUT_S, WORKER_PATH + + +class WSClient: + def __init__( + self, + host: str, + port: int, + path: str = WORKER_PATH, + on_control: Optional[Callable[[str, dict], None]] = None, + ) -> None: + self.host = host + self.port = port + self.path = path + self._on_control = on_control + self._sock: Optional[socket.socket] = None + self._send_lock = threading.Lock() + self._reader: Optional[threading.Thread] = None + self._stop = threading.Event() + self.connected = False + + # ── lifecycle ──────────────────────────────────────────────────────────── + + def connect(self, timeout: float = CONNECT_TIMEOUT_S) -> None: + sock = socket.create_connection((self.host, self.port), timeout=timeout) + key = base64.b64encode(os.urandom(16)).decode() + sock.sendall( + ( + f"GET {self.path} HTTP/1.1\r\n" + f"Host: {self.host}:{self.port}\r\n" + "Upgrade: websocket\r\n" + "Connection: Upgrade\r\n" + f"Sec-WebSocket-Key: {key}\r\n" + "Sec-WebSocket-Version: 13\r\n" + "\r\n" + ).encode() + ) + buf = b"" + while b"\r\n\r\n" not in buf: + chunk = sock.recv(1024) + if not chunk: + raise ConnectionError("backend closed during handshake") + buf += chunk + status = buf.split(b"\r\n", 1)[0].decode(errors="replace") + if "101" not in status: + raise ConnectionError(f"websocket upgrade failed: {status!r}") + + sock.settimeout(1.0) # so the reader can poll the stop flag + self._sock = sock + self.connected = True + self._reader = threading.Thread(target=self._read_loop, daemon=True) + self._reader.start() + + def close(self) -> None: + self._stop.set() + sock = self._sock + if sock is not None: + try: + self._send_frame(0x8, b"") # close + except OSError: + pass + try: + sock.close() + except OSError: + pass + if self._reader is not None: + self._reader.join(timeout=2.0) + self.connected = False + + # ── sending ────────────────────────────────────────────────────────────── + + def send_json(self, scope: str, data) -> bool: + """Send one ``{scope, data}`` frame. Returns False if not connected β€” + mirrors the JS capturer's silent-drop-on-disconnect behavior.""" + if not self.connected or self._sock is None: + return False + try: + self._send_frame(0x1, json.dumps({"scope": scope, "data": data}).encode()) + return True + except OSError: + self.connected = False + return False + + def _send_frame(self, opcode: int, payload: bytes) -> None: + header = bytearray([0x80 | opcode]) + n = len(payload) + if n < 126: + header.append(0x80 | n) + elif n < 65536: + header.append(0x80 | 126) + header += struct.pack(">H", n) + else: + header.append(0x80 | 127) + header += struct.pack(">Q", n) + mask = os.urandom(4) + header += mask + masked = bytes(b ^ mask[i % 4] for i, b in enumerate(payload)) + with self._send_lock: + assert self._sock is not None + self._sock.sendall(bytes(header) + masked) + + # ── receiving ────────────────────────────────────────────────────────────── + + def _recv_exact(self, n: int) -> Optional[bytes]: + buf = b"" + while len(buf) < n: + if self._stop.is_set(): + return None + try: + chunk = self._sock.recv(n - len(buf)) # type: ignore[union-attr] + except socket.timeout: + continue + except OSError: + return None + if not chunk: + return None + buf += chunk + return buf + + def _read_loop(self) -> None: + while not self._stop.is_set(): + head = self._recv_exact(2) + if head is None: + break + opcode = head[0] & 0x0F + masked = bool(head[1] & 0x80) + length = head[1] & 0x7F + if length == 126: + ext = self._recv_exact(2) + if ext is None: + break + length = struct.unpack(">H", ext)[0] + elif length == 127: + ext = self._recv_exact(8) + if ext is None: + break + length = struct.unpack(">Q", ext)[0] + mask = self._recv_exact(4) if masked else None + payload = self._recv_exact(length) if length else b"" + if payload is None: + break + if mask: + payload = bytes(b ^ mask[i % 4] for i, b in enumerate(payload)) + + if opcode == 0x8: # close + break + if opcode == 0x9: # ping β†’ pong + try: + self._send_frame(0xA, payload) + except OSError: + break + continue + if opcode in (0x1, 0x2): # text / binary + self._dispatch(payload) + self.connected = False + + def _dispatch(self, payload: bytes) -> None: + if self._on_control is None: + return + try: + msg = json.loads(payload.decode("utf-8")) + except (ValueError, UnicodeDecodeError): + return + scope = msg.get("scope") + if isinstance(scope, str): + self._on_control(scope, msg.get("data") or {}) diff --git a/packages/selenium-py-devtools/src/wdio_selenium_devtools/types.py b/packages/selenium-py-devtools/src/wdio_selenium_devtools/types.py new file mode 100644 index 00000000..709c768b --- /dev/null +++ b/packages/selenium-py-devtools/src/wdio_selenium_devtools/types.py @@ -0,0 +1,93 @@ +"""Wire payload types β€” the Python mirror of ``packages/shared/src/types.ts``. + +TypedDicts document the ``data`` payload shapes behind each ``{scope, data}`` +frame the dashboard consumes. They're structural (plain dicts at runtime); the +value is a single typed definition per concept, checkable by mypy. +""" + +from __future__ import annotations + +from typing import Any, Dict, List, Optional, TypedDict, Union + +#: Anything that survives ``json.dumps``. Payloads must reduce to this. +JSONValue = Union[ + None, bool, int, float, str, List["JSONValue"], Dict[str, "JSONValue"] +] + +#: A ``{scope, data}`` frame's scope β€” a value from the generated ``_contract``. +Scope = str + + +class SerializedError(TypedDict): + name: str + message: str + + +class CommandLog(TypedDict, total=False): + command: str + args: List[Any] + result: Any + error: SerializedError + timestamp: int + startTime: int + callSource: Optional[str] + id: int + + +class ConsoleLog(TypedDict): + type: str + args: List[Any] + timestamp: int + source: str + + +class NetworkRequest(TypedDict, total=False): + id: str + url: str + method: str + status: Optional[int] + timestamp: int + startTime: int + endTime: Optional[int] + type: str + + +class Metadata(TypedDict, total=False): + type: str + sessionId: str + url: Optional[str] + capabilities: Dict[str, Any] + desiredCapabilities: Dict[str, Any] + testEnv: str + + +class TestStats(TypedDict, total=False): + uid: str + cid: str + title: str + fullTitle: str + parent: str + state: str + start: str + end: str + type: str + file: str + retries: int + _duration: int + callSource: Optional[str] + + +class SuiteStats(TypedDict, total=False): + uid: str + cid: str + title: str + fullTitle: str + type: str + file: str + start: str + end: Optional[str] + state: Optional[str] + tests: List[TestStats] + suites: List["SuiteStats"] + hooks: List[Any] + _duration: int diff --git a/packages/selenium-py-devtools/src/wdio_selenium_devtools/utils.py b/packages/selenium-py-devtools/src/wdio_selenium_devtools/utils.py new file mode 100644 index 00000000..b9cfdbe2 --- /dev/null +++ b/packages/selenium-py-devtools/src/wdio_selenium_devtools/utils.py @@ -0,0 +1,52 @@ +"""Small framework-agnostic helpers. No third-party imports.""" + +from __future__ import annotations + +import json +import time +import traceback +from typing import Any, Iterable + + +def now_ms() -> int: + """Epoch milliseconds β€” the timestamp unit every frame uses.""" + return int(time.time() * 1000) + + +def iso(ms: int) -> str: + """ISO-8601 string. SuiteStats.start/end are TS Dates on the wire.""" + base = time.strftime("%Y-%m-%dT%H:%M:%S", time.gmtime(ms / 1000)) + return f"{base}.{ms % 1000:03d}Z" + + +def to_jsonable(value: Any, _depth: int = 0) -> Any: + """Coerce an arbitrary value into something json.dumps can handle. + + Selenium command params/results may carry WebElement refs or other + non-serializable objects; the wire only speaks JSON, so anything exotic + degrades to its string form rather than blowing up the send. + """ + if value is None or isinstance(value, (bool, int, float, str)): + return value + if _depth > 6: + return str(value) + if isinstance(value, dict): + return {str(k): to_jsonable(v, _depth + 1) for k, v in value.items()} + if isinstance(value, (list, tuple)): + return [to_jsonable(v, _depth + 1) for v in value] + try: + json.dumps(value) + return value + except (TypeError, ValueError): + return str(value) + + +def call_source(skip_substrings: Iterable[str]) -> str | None: + """First stack frame outside the adapter + selenium internals, as + ``path:line`` β€” what the dashboard shows as the command's origin.""" + skips = tuple(skip_substrings) + # Innermost-last; walk outward (reversed) to find the user's frame. + for frame in reversed(traceback.extract_stack()[:-1]): + if not any(s in frame.filename for s in skips): + return f"{frame.filename}:{frame.lineno}" + return None diff --git a/packages/selenium-py-devtools/tests/test_backend.py b/packages/selenium-py-devtools/tests/test_backend.py new file mode 100644 index 00000000..b4b22e36 --- /dev/null +++ b/packages/selenium-py-devtools/tests/test_backend.py @@ -0,0 +1,57 @@ +import os +import tempfile +import unittest +from pathlib import Path + +from wdio_selenium_devtools import backend + + +class TestBackendResolution(unittest.TestCase): + def setUp(self): + self._saved = {k: os.environ.get(k) for k in + ("DEVTOOLS_PORT", "DEVTOOLS_HOST", "DEVTOOLS_BACKEND_CMD")} + for k in self._saved: + os.environ.pop(k, None) + + def tearDown(self): + for k, v in self._saved.items(): + if v is None: + os.environ.pop(k, None) + else: + os.environ[k] = v + + def test_port_regex(self): + self.assertEqual( + backend._PORT_RE.search("… application on port 63763").group(1), "63763" + ) + + def test_attaches_when_port_env_set_without_spawning(self): + os.environ["DEVTOOLS_PORT"] = "4321" + os.environ["DEVTOOLS_HOST"] = "example.test" + host, port, proc = backend.launch_or_attach() + self.assertEqual((host, port), ("example.test", 4321)) + self.assertIsNone(proc) # attached, not owned + + def test_finds_monorepo_backend_when_present(self): + with tempfile.TemporaryDirectory() as tmp: + dist = Path(tmp) / "packages" / "backend" / "dist" + dist.mkdir(parents=True) + (dist / "index.js").write_text("//") + start = Path(tmp) / "a" / "b" / "mod.py" + start.parent.mkdir(parents=True) + self.assertEqual( + backend._find_monorepo_backend(start=start), dist / "index.js" + ) + + def test_no_monorepo_backend_when_absent(self): + with tempfile.TemporaryDirectory() as tmp: + start = Path(tmp) / "x" / "y.py" + start.parent.mkdir(parents=True) + self.assertIsNone(backend._find_monorepo_backend(start=start)) + + def test_pinned_backend_version_is_set(self): + self.assertRegex(backend.BACKEND_NPM_VERSION, r"^\d+\.\d+\.\d+$") + + +if __name__ == "__main__": + unittest.main() diff --git a/packages/selenium-py-devtools/tests/test_capturer.py b/packages/selenium-py-devtools/tests/test_capturer.py new file mode 100644 index 00000000..5a2ab272 --- /dev/null +++ b/packages/selenium-py-devtools/tests/test_capturer.py @@ -0,0 +1,58 @@ +import unittest + +from wdio_selenium_devtools.capturer import SessionCapturer + + +class FakeTransport: + connected = True + + def __init__(self): + self.sent = [] + + def send_json(self, scope, data): + self.sent.append((scope, data)) + return True + + def close(self): + pass + + +class TestSessionCapturer(unittest.TestCase): + def setUp(self): + self.tx = FakeTransport() + self.cap = SessionCapturer(self.tx) + + def test_metadata_sent_once(self): + self.cap.ensure_metadata("sess-1", {"browserName": "chrome"}, None) + self.cap.ensure_metadata("sess-1", {"browserName": "chrome"}, None) + meta_frames = [d for s, d in self.tx.sent if s == "metadata"] + self.assertEqual(len(meta_frames), 1) + self.assertEqual(meta_frames[0]["sessionId"], "sess-1") + + def test_capture_command_increments_id_and_wraps_in_array(self): + self.cap.capture_command(command="get", args={"url": "x"}, + result={"v": 1}, start_time=1, call_source=None) + self.cap.capture_command(command="click", args=None, + result=None, start_time=2, call_source=None) + cmds = [d for s, d in self.tx.sent if s == "commands"] + self.assertEqual(len(cmds), 2) + self.assertEqual(cmds[0][0]["id"], 1) + self.assertEqual(cmds[1][0]["id"], 2) + # non-list args are normalized to a list + self.assertEqual(cmds[0][0]["args"], [{"url": "x"}]) + self.assertEqual(cmds[1][0]["args"], []) + + def test_capture_command_records_error(self): + self.cap.capture_command(command="boom", args=[], + error=RuntimeError("x"), start_time=1, + call_source=None) + cmd = [d for s, d in self.tx.sent if s == "commands"][0][0] + self.assertEqual(cmd["error"]["name"], "RuntimeError") + + def test_suites_passthrough(self): + self.cap.send_suites([{"uid": "s"}]) + self.assertIn(("suites", [{"uid": "s"}]), self.tx.sent) + + +if __name__ == "__main__": + unittest.main() diff --git a/packages/selenium-py-devtools/tests/test_frames.py b/packages/selenium-py-devtools/tests/test_frames.py new file mode 100644 index 00000000..eba72ddc --- /dev/null +++ b/packages/selenium-py-devtools/tests/test_frames.py @@ -0,0 +1,57 @@ +import json +import unittest + +from wdio_selenium_devtools import frames +from wdio_selenium_devtools.utils import to_jsonable + + +class TestFrames(unittest.TestCase): + def test_metadata_shape(self): + m = frames.metadata("sess-1", {"browserName": "chrome"}, "https://x/") + self.assertEqual(m["type"], "testrunner") + self.assertEqual(m["sessionId"], "sess-1") + self.assertEqual(m["capabilities"]["browserName"], "chrome") + + def test_command_log_includes_error_only_when_present(self): + ok = frames.command_log( + command="get", args=["u"], result=None, timestamp=2, start_time=1, + call_source="f.py:1", command_id=3, + ) + self.assertNotIn("error", ok) + self.assertEqual(ok["id"], 3) + bad = frames.command_log( + command="get", args=[], error=ValueError("nope"), + timestamp=2, start_time=1, call_source=None, command_id=4, + ) + self.assertEqual(bad["error"], {"name": "ValueError", "message": "nope"}) + + def test_suite_and_test_stats_are_json_serializable(self): + t = frames.test_stats( + uid="n::t", title="t", full_title="m β€Ί t", parent="m", + state="passed", file="m.py", start_ms=1000, end_ms=2200, + ) + s = frames.suite_stats(uid="m.py", title="m.py", file="m.py", + start_ms=1000, tests=[t], end_ms=2200, state="passed") + self.assertEqual(t["_duration"], 1200) + self.assertEqual(s["type"], "suite") + self.assertEqual(s["tests"][0]["state"], "passed") + # start/end must be ISO strings, not numbers (TS Date on the wire). + self.assertRegex(t["start"], r"^\d{4}-\d{2}-\d{2}T") + json.dumps(s) # raises if any field is non-serializable + + +class TestJsonable(unittest.TestCase): + def test_passes_through_primitives_and_containers(self): + self.assertEqual(to_jsonable({"a": [1, "b", True, None]}), + {"a": [1, "b", True, None]}) + + def test_falls_back_to_str_for_exotic(self): + class Weird: + def __repr__(self): + return "" + + self.assertEqual(to_jsonable(Weird()), "") + + +if __name__ == "__main__": + unittest.main() diff --git a/packages/selenium-py-devtools/tests/test_instrumentation.py b/packages/selenium-py-devtools/tests/test_instrumentation.py new file mode 100644 index 00000000..e8bd3d28 --- /dev/null +++ b/packages/selenium-py-devtools/tests/test_instrumentation.py @@ -0,0 +1,81 @@ +import unittest + +from wdio_selenium_devtools import instrumentation +from wdio_selenium_devtools.capturer import SessionCapturer + + +class FakeTransport: + connected = True + + def __init__(self): + self.sent = [] + + def send_json(self, scope, data): + self.sent.append((scope, data)) + return True + + def close(self): + pass + + +class FakeDriver: + """Stand-in for selenium's WebDriver β€” same execute() chokepoint.""" + + def __init__(self): + self.session_id = None + self.caps = {"browserName": "chrome"} + + def execute(self, command, params=None): + if command == "newSession": + self.session_id = "sess-9" + if command == "boom": + raise ValueError("kaboom") + return {"value": f"ok:{command}"} + + +class TestInstrumentation(unittest.TestCase): + def setUp(self): + instrumentation.uninstall() + self.tx = FakeTransport() + self.cap = SessionCapturer(self.tx) + instrumentation.install(self.cap, FakeDriver) + self.driver = FakeDriver() + + def tearDown(self): + instrumentation.uninstall() + + def _commands(self): + return [d[0] for s, d in self.tx.sent if s == "commands"] + + def test_captures_command_and_unwraps_value(self): + out = self.driver.execute("get", {"url": "https://x/"}) + self.assertEqual(out, {"value": "ok:get"}) # behavior unchanged + cmds = self._commands() + self.assertEqual(len(cmds), 1) + self.assertEqual(cmds[0]["command"], "get") + self.assertEqual(cmds[0]["result"], "ok:get") # unwrapped from .value + self.assertEqual(cmds[0]["args"], [{"url": "https://x/"}]) + + def test_skip_commands_not_captured_but_metadata_sent(self): + self.driver.execute("newSession") + self.assertEqual(self._commands(), []) + metas = [d for s, d in self.tx.sent if s == "metadata"] + self.assertEqual(len(metas), 1) + self.assertEqual(metas[0]["sessionId"], "sess-9") + + def test_error_is_captured_then_reraised(self): + with self.assertRaises(ValueError): + self.driver.execute("boom") + cmds = self._commands() + self.assertEqual(len(cmds), 1) + self.assertEqual(cmds[0]["error"]["name"], "ValueError") + + def test_uninstall_restores_original(self): + instrumentation.uninstall() + self.tx.sent.clear() + self.driver.execute("get") + self.assertEqual(self._commands(), []) # no capture after uninstall + + +if __name__ == "__main__": + unittest.main() From d441d6afa6b2b1a745b949b4ab87a7db136b7870 Mon Sep 17 00:00:00 2001 From: Vishnu Vardhan Date: Thu, 2 Jul 2026 16:31:53 +0530 Subject: [PATCH 02/12] chore(selenium-py-devtools): remove python-spike proof-of-concept --- examples/python-spike/README.md | 77 --------- examples/python-spike/spike.py | 213 ------------------------ examples/python-spike/verify-client.mjs | 57 ------- 3 files changed, 347 deletions(-) delete mode 100644 examples/python-spike/README.md delete mode 100644 examples/python-spike/spike.py delete mode 100644 examples/python-spike/verify-client.mjs diff --git a/examples/python-spike/README.md b/examples/python-spike/README.md deleted file mode 100644 index f931930a..00000000 --- a/examples/python-spike/README.md +++ /dev/null @@ -1,77 +0,0 @@ -# Python spike β€” Phase 0 feasibility proof - -Proves that a **non-JS client** can drive the DevTools dashboard with no -backend or UI changes. A dependency-free Python script speaks the adapter side -of the wire (raw socket + RFC-6455 handshake) and pushes the same -`{ "scope", "data" }` frames the JS Selenium adapter emits. - -This is a throwaway spike, not the adapter. Its job is to de-risk everything -downstream and to capture the **golden frames** the real Python adapter must -reproduce. - -## Result - -βœ… **Boundary proven end to end.** Every data scope the dashboard consumes -(`metadata`, `suites`, `commands`, `consoleLogs`, `networkRequests`) was sent -from Python and observed arriving at a dashboard-role `/client` subscriber. - -## Run it - -```bash -# 1. build once (from repo root) -pnpm --filter @wdio/devtools-app --filter @wdio/devtools-backend build - -# 2. start the backend + dashboard -node packages/backend/dist/index.js -# note the port it logs β€” 3000 if free, else a negotiated port e.g. 63763 - -# 3a. visual check: open the dashboard, then run the spike -# (export the port if it isn't 3000) -DEVTOOLS_PORT=63763 python3 examples/python-spike/spike.py -# β†’ a "Python spike" suite, a command timeline, console + network appear live - -# 3b. headless proof (no browser): run the listener, then the spike -DEVTOOLS_PORT=63763 node examples/python-spike/verify-client.mjs & -DEVTOOLS_PORT=63763 python3 examples/python-spike/spike.py -# β†’ listener prints "βœ“ all expected scopes received β€” boundary proven" -``` - -## What this established (the contract a Python adapter must honor) - -- **Endpoint.** Adapters connect to `ws://:/worker`. The dashboard - subscribes to `/client`. No handshake, auth, or session registration β€” the - backend parses each frame and broadcasts it verbatim - (`packages/backend/src/worker-message-handler.ts`). -- **Frame envelope.** `{ "scope": string, "data": }`, one JSON object - per WS text frame. `data` must be truthy or the UI drops it - (`DataManager.ts:308`). -- **Data scopes the UI renders** (`DataManager.ts:283-302`): - | scope | payload | notes | - |---|---|---| - | `metadata` | `Metadata` object | `type: "testrunner"`, carries `sessionId` | - | `suites` | `SuiteStats[]` | the test tree; re-send to update final state | - | `commands` | `CommandLog[]` | send incrementally for a live timeline | - | `consoleLogs` | `ConsoleLog[]` | `source: "browser" \| "test" \| "terminal"` | - | `networkRequests` | `NetworkRequest[]` | | -- **Field shapes** mirror `packages/shared/src/types.ts`. Two encoding rules - learned here: - - timestamps are **epoch milliseconds** (numbers). - - `SuiteStats.start/end` are TS `Date`s β€” they cross the wire as **ISO - strings**. -- **Port negotiation.** If 3000 is busy the backend uses `get-port` and logs the - real port. The JS adapters read it from `start()`'s return value; the Python - adapter will need the same (env var / discovery). -- **Replay buffer.** The backend buffers broadcast frames and replays them to - late-connecting clients β€” a dashboard opened mid-run catches up. The Python - adapter gets this for free. - -## Files - -- `spike.py` β€” dependency-free worker client + golden frames. -- `verify-client.mjs` β€” `/client` subscriber that asserts the frames fan out. - -## Not covered by this spike (later phases) - -Driver instrumentation (`execute()` wrap), BiDi console/network, screencast, -pytest lifecycle, trace export β€” see the Phase roadmap. This spike only proves -the transport boundary, which everything else depends on. diff --git a/examples/python-spike/spike.py b/examples/python-spike/spike.py deleted file mode 100644 index 11d9ae52..00000000 --- a/examples/python-spike/spike.py +++ /dev/null @@ -1,213 +0,0 @@ -#!/usr/bin/env python3 -""" -Phase 0 spike β€” prove a non-JS client can drive the DevTools dashboard. - -This script speaks the adapter side of the wire with ZERO third-party -dependencies (raw socket + RFC-6455 handshake + masked text frames). It -connects to the backend's /worker WebSocket and pushes the same -`{ "scope": ..., "data": ... }` frames the JS Selenium adapter emits. - -The real Python adapter would use the `websockets` package instead of this -hand-rolled client β€” the point here is only to prove the boundary is -language-agnostic, with nothing to install. - -Run: - 1. node packages/backend/dist/index.js # backend + dashboard on :3000 - 2. open http://localhost:3000 - 3. python3 examples/python-spike/spike.py # watch the run appear, live - -Frame shapes mirror packages/shared/src/types.ts. See README.md. -""" - -import base64 -import json -import os -import socket -import struct -import sys -import time - -HOST = os.environ.get("DEVTOOLS_HOST", "localhost") -PORT = int(os.environ.get("DEVTOOLS_PORT", "3000")) -WORKER_PATH = "/worker" -SESSION_ID = "spike-py-0001" - - -# ── Minimal RFC-6455 client (text frames, clientβ†’server masked) ────────────── - -def ws_connect(host: str, port: int, path: str) -> socket.socket: - sock = socket.create_connection((host, port), timeout=5) - key = base64.b64encode(os.urandom(16)).decode() - handshake = ( - f"GET {path} HTTP/1.1\r\n" - f"Host: {host}:{port}\r\n" - "Upgrade: websocket\r\n" - "Connection: Upgrade\r\n" - f"Sec-WebSocket-Key: {key}\r\n" - "Sec-WebSocket-Version: 13\r\n" - "\r\n" - ) - sock.sendall(handshake.encode()) - - # Read response headers up to the blank line. - buf = b"" - while b"\r\n\r\n" not in buf: - chunk = sock.recv(1024) - if not chunk: - raise ConnectionError("backend closed during handshake") - buf += chunk - status_line = buf.split(b"\r\n", 1)[0].decode(errors="replace") - if "101" not in status_line: - raise ConnectionError(f"upgrade failed: {status_line!r}") - return sock - - -def ws_send_text(sock: socket.socket, text: str) -> None: - payload = text.encode("utf-8") - header = bytearray() - header.append(0x81) # FIN + opcode 0x1 (text) - mask_bit = 0x80 # client frames MUST be masked - n = len(payload) - if n < 126: - header.append(mask_bit | n) - elif n < 65536: - header.append(mask_bit | 126) - header += struct.pack(">H", n) - else: - header.append(mask_bit | 127) - header += struct.pack(">Q", n) - mask = os.urandom(4) - header += mask - masked = bytes(b ^ mask[i % 4] for i, b in enumerate(payload)) - sock.sendall(bytes(header) + masked) - - -def send_frame(sock: socket.socket, scope: str, data) -> None: - ws_send_text(sock, json.dumps({"scope": scope, "data": data})) - print(f" β†’ sent {scope:<16} ({json.dumps(data)[:60]}…)") - - -def now_ms() -> int: - return int(time.time() * 1000) - - -def iso(ms: int) -> str: - # SuiteStats.start/end are Dates in TS β€” they cross the wire as ISO strings. - return time.strftime("%Y-%m-%dT%H:%M:%S", time.gmtime(ms / 1000)) + \ - f".{ms % 1000:03d}Z" - - -# ── The golden frames (mirror packages/shared/src/types.ts) ────────────────── - -def metadata_frame() -> dict: - return { - "type": "testrunner", # TraceType.Testrunner - "sessionId": SESSION_ID, - "url": "https://example.com/", - "capabilities": {"browserName": "chrome", "browserVersion": "126.0"}, - "desiredCapabilities": {"browserName": "chrome"}, - "viewport": {"width": 1280, "height": 720, - "offsetLeft": 0, "offsetTop": 0, "scale": 1}, - "testEnv": "python-spike", - } - - -def suite_frame(start: int) -> list: - test = { - "uid": "test-1", "cid": "0-0", - "title": "loads the homepage", - "fullTitle": "Python spike β€Ί loads the homepage", - "parent": "Python spike", - "state": "passed", - "start": iso(start), "end": iso(start + 1200), - "type": "test", "file": "examples/python-spike/spike.py", - "retries": 0, "_duration": 1200, - "callSource": "spike.py:loads_the_homepage", - } - suite = { - "uid": "suite-1", "cid": "0-0", - "title": "Python spike", "fullTitle": "Python spike", - "type": "suite", "file": "examples/python-spike/spike.py", - "start": iso(start), "end": iso(start + 1200), "state": "passed", - "tests": [test], "suites": [], "hooks": [], "_duration": 1200, - } - return [suite] - - -def command_frames(start: int) -> list: - base = [ - ("navigateTo", ["https://example.com/"], None), - ("findElement", [{"using": "css selector", "value": "h1"}], - {"ELEMENT": "elem-h1"}), - ("getText", [], "Example Domain"), - ("click", [{"using": "css selector", "value": "a"}], None), - ] - frames = [] - for i, (cmd, args, result) in enumerate(base): - ts = start + 100 + i * 250 - frames.append({ - "command": cmd, "args": args, "result": result, - "timestamp": ts, "startTime": ts - 40, - "callSource": f"spike.py:{40 + i}", "id": i + 1, - }) - return frames - - -def console_frames(start: int) -> list: - return [ - {"type": "info", "args": ["Hello from the Python spike 🐍"], - "timestamp": start + 120, "source": "browser"}, - {"type": "warn", "args": ["this is a synthetic console line"], - "timestamp": start + 480, "source": "browser"}, - ] - - -def network_frames(start: int) -> list: - return [{ - "id": "net-1", "url": "https://example.com/", "method": "GET", - "status": 200, "statusText": "OK", - "timestamp": start + 90, "startTime": start + 90, - "endTime": start + 240, "time": 150, "type": "document", - "response": {"fromCache": False, "headers": {}, "mimeType": "text/html", - "status": 200}, - }] - - -def main() -> int: - print(f"connecting to ws://{HOST}:{PORT}{WORKER_PATH} …") - try: - sock = ws_connect(HOST, PORT, WORKER_PATH) - except OSError as exc: - print(f"\n βœ— could not connect: {exc}") - print(" is the backend running? node packages/backend/dist/index.js") - return 1 - - print(" βœ“ connected β€” streaming a synthetic test run\n") - start = now_ms() - - send_frame(sock, "metadata", metadata_frame()) - time.sleep(0.3) - send_frame(sock, "suites", suite_frame(start)) - time.sleep(0.3) - send_frame(sock, "networkRequests", network_frames(start)) - send_frame(sock, "consoleLogs", console_frames(start)) - time.sleep(0.3) - - # Stream commands one at a time so the timeline fills in "live". - for frame in command_frames(start): - send_frame(sock, "commands", [frame]) - time.sleep(0.4) - - # Re-send the suite with final state so the tree settles green. - send_frame(sock, "suites", suite_frame(start)) - - print("\n βœ“ done β€” check the dashboard at " - f"http://{HOST}:{PORT}") - print(" keeping the socket open for 2s so the backend flushes…") - time.sleep(2) - sock.close() - return 0 - - -if __name__ == "__main__": - sys.exit(main()) diff --git a/examples/python-spike/verify-client.mjs b/examples/python-spike/verify-client.mjs deleted file mode 100644 index e3b49e5e..00000000 --- a/examples/python-spike/verify-client.mjs +++ /dev/null @@ -1,57 +0,0 @@ -// Proof harness: plays the role the dashboard plays β€” subscribes to the -// backend's /client WebSocket and records every frame the backend broadcasts. -// If the Python spike's frames arrive here, the language-agnostic boundary is -// proven end to end without needing to eyeball a browser. -// -// Run (from repo root, after starting the backend): -// node examples/python-spike/verify-client.mjs -// It listens for COLLECT_MS, prints the scopes it saw, and exits non-zero if -// the expected set didn't arrive. - -import { createRequire } from 'node:module' -import { fileURLToPath, pathToFileURL } from 'node:url' -import { dirname, resolve } from 'node:path' - -// `ws` is a dependency of packages/backend, not hoisted to the repo root β€” -// anchor module resolution there so this runs from any cwd. -const here = dirname(fileURLToPath(import.meta.url)) -const require = createRequire( - pathToFileURL(resolve(here, '../../packages/backend/package.json')) -) -const WebSocket = require('ws') - -const HOST = process.env.DEVTOOLS_HOST || 'localhost' -const PORT = process.env.DEVTOOLS_PORT || '3000' -const COLLECT_MS = Number(process.env.COLLECT_MS || 6000) -const EXPECTED = ( - process.env.EXPECT || 'metadata,suites,commands,consoleLogs,networkRequests' -) - .split(',') - .map((s) => s.trim()) - .filter(Boolean) - -const seen = new Map() -const ws = new WebSocket(`ws://${HOST}:${PORT}/client`) - -ws.on('open', () => console.log(`[client] subscribed to ws://${HOST}:${PORT}/client`)) -ws.on('error', (e) => { console.error('[client] error:', e.message); process.exit(2) }) -ws.on('message', (buf) => { - let msg - try { msg = JSON.parse(buf.toString()) } catch { return } - if (!msg || !msg.scope) return - seen.set(msg.scope, (seen.get(msg.scope) || 0) + 1) - const preview = JSON.stringify(msg.data).slice(0, 70) - console.log(`[client] β—€ ${msg.scope.padEnd(16)} ${preview}…`) -}) - -setTimeout(() => { - console.log('\n[client] ── summary ──') - for (const [scope, n] of seen) console.log(` ${scope.padEnd(18)} Γ—${n}`) - const missing = EXPECTED.filter((s) => !seen.has(s)) - if (missing.length) { - console.log(`\n βœ— missing expected scopes: ${missing.join(', ')}`) - process.exit(1) - } - console.log('\n βœ“ all expected scopes received β€” boundary proven') - process.exit(0) -}, COLLECT_MS) From 6bd69a9601aaf9b380089f2075d22706e3ebf297 Mon Sep 17 00:00:00 2001 From: Vishnu Vardhan Date: Thu, 2 Jul 2026 16:32:18 +0530 Subject: [PATCH 03/12] ci(selenium-py-devtools): add path-filtered test + OIDC release workflows --- .github/workflows/python-release.yml | 52 ++++++++++++++++++++++++++++ .github/workflows/python.yml | 26 +++----------- 2 files changed, 56 insertions(+), 22 deletions(-) create mode 100644 .github/workflows/python-release.yml 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 index d72aaf5a..7a56aca4 100644 --- a/.github/workflows/python.yml +++ b/.github/workflows/python.yml @@ -4,8 +4,10 @@ on: push: branches: - main - tags: - - python-v[0-9]+.[0-9]+.[0-9]+* + paths: + - packages/selenium-py-devtools/** + - packages/shared/** + - .github/workflows/python.yml pull_request: paths: - packages/selenium-py-devtools/** @@ -33,23 +35,3 @@ jobs: git diff --exit-code src/wdio_selenium_devtools/_contract.py - name: πŸ§ͺ Unit tests run: PYTHONPATH=src python -m unittest discover -s tests - - publish: - needs: test - if: startsWith(github.ref, 'refs/tags/python-v') - runs-on: ubuntu-latest - environment: pypi - permissions: - id-token: write # PyPI trusted publishing (OIDC) β€” no token needed - steps: - - uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd # v6.0.2 - - uses: actions/setup-python@a26af69be951a213d495a4c3e4e4022e16d87065 # v5 - with: - python-version: '3.12' - - name: πŸ“¦ Build sdist + wheel - run: | - python -m pip install --upgrade build - python -m build - - uses: pypa/gh-action-pypi-publish@cef221092ed1bacb1cc03d23a2d87d1d172e277b # release/v1 - with: - packages-dir: packages/selenium-py-devtools/dist From 2aa29c569510ba14e386152a99a1b2911ead44c2 Mon Sep 17 00:00:00 2001 From: Vishnu Vardhan Date: Thu, 2 Jul 2026 16:35:01 +0530 Subject: [PATCH 04/12] feat(selenium-py-devtools): WS transport, backend launch, dashboard window --- .../src/wdio_selenium_devtools/backend.py | 7 +- .../src/wdio_selenium_devtools/lifecycle.py | 363 ++++++++++++++++++ .../src/wdio_selenium_devtools/transport.py | 7 +- .../tests/test_backend.py | 18 +- .../tests/test_lifecycle.py | 295 ++++++++++++++ .../tests/test_transport.py | 24 ++ 6 files changed, 709 insertions(+), 5 deletions(-) create mode 100644 packages/selenium-py-devtools/src/wdio_selenium_devtools/lifecycle.py create mode 100644 packages/selenium-py-devtools/tests/test_lifecycle.py create mode 100644 packages/selenium-py-devtools/tests/test_transport.py diff --git a/packages/selenium-py-devtools/src/wdio_selenium_devtools/backend.py b/packages/selenium-py-devtools/src/wdio_selenium_devtools/backend.py index eb426545..0693cb99 100644 --- a/packages/selenium-py-devtools/src/wdio_selenium_devtools/backend.py +++ b/packages/selenium-py-devtools/src/wdio_selenium_devtools/backend.py @@ -35,7 +35,12 @@ ENV_PORT, ) -_PORT_RE = re.compile(r"on port (\d+)") +# 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]: 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/transport.py b/packages/selenium-py-devtools/src/wdio_selenium_devtools/transport.py index f097e1db..6cc3f1b3 100644 --- a/packages/selenium-py-devtools/src/wdio_selenium_devtools/transport.py +++ b/packages/selenium-py-devtools/src/wdio_selenium_devtools/transport.py @@ -82,8 +82,11 @@ def close(self) -> None: sock.close() except OSError: pass - if self._reader is not None: - self._reader.join(timeout=2.0) + reader = self._reader + # Don't join the reader from within itself β€” close() can be called on the + # reader thread (a clientDisconnected control frame triggering shutdown). + if reader is not None and reader is not threading.current_thread(): + reader.join(timeout=2.0) self.connected = False # ── sending ────────────────────────────────────────────────────────────── diff --git a/packages/selenium-py-devtools/tests/test_backend.py b/packages/selenium-py-devtools/tests/test_backend.py index b4b22e36..a64a1312 100644 --- a/packages/selenium-py-devtools/tests/test_backend.py +++ b/packages/selenium-py-devtools/tests/test_backend.py @@ -20,9 +20,23 @@ def tearDown(self): else: os.environ[k] = v - def test_port_regex(self): + def test_port_regex_uses_actual_listening_port(self): + # The preferred "Starting … on port 3000" line must NOT match β€” it's not + # the bound port when 3000 is busy. + self.assertIsNone( + backend._PORT_RE.search("Starting application on port 3000") + ) + # The Fastify "Server listening at …:PORT" line is the real port, + # including the IPv6 form. + self.assertEqual( + backend._PORT_RE.search( + '{"msg":"Server listening at http://[::1]:63763"}' + ).group(1), + "63763", + ) self.assertEqual( - backend._PORT_RE.search("… application on port 63763").group(1), "63763" + backend._PORT_RE.search("Server listening at http://127.0.0.1:3000").group(1), + "3000", ) def test_attaches_when_port_env_set_without_spawning(self): diff --git a/packages/selenium-py-devtools/tests/test_lifecycle.py b/packages/selenium-py-devtools/tests/test_lifecycle.py new file mode 100644 index 00000000..1d069f1e --- /dev/null +++ b/packages/selenium-py-devtools/tests/test_lifecycle.py @@ -0,0 +1,295 @@ +"""Unit tests for the dashboard browser-window lifecycle. + +These never spawn a real browser or exit the process: the opener is injected, +and the shutdown path's hard-exit timer is asserted on rather than run. +""" + +import os +import threading +import unittest +from unittest import mock + +from wdio_selenium_devtools import lifecycle +from wdio_selenium_devtools.lifecycle import BrowserHandle + + +class FakeProc: + """Minimal subprocess.Popen stand-in for BrowserHandle.close().""" + + def __init__(self, alive=True): + self._alive = alive + self.terminated = False + self.killed = False + + def poll(self): + return None if self._alive else 0 + + def terminate(self): + self.terminated = True + self._alive = False + + def wait(self, timeout=None): + return 0 + + def kill(self): + self.killed = True + self._alive = False + + +class TestBrowserHandle(unittest.TestCase): + def test_close_terminates_proc_and_removes_profile(self): + proc = FakeProc() + with mock.patch.object(lifecycle.shutil, "rmtree") as rmtree: + handle = BrowserHandle(proc=proc, user_data_dir="/tmp/fake-dir") + handle.close() + self.assertTrue(proc.terminated) + rmtree.assert_called_once_with("/tmp/fake-dir", ignore_errors=True) + + def test_close_is_idempotent(self): + proc = FakeProc() + handle = BrowserHandle(proc=proc) + handle.close() + proc.terminated = False + handle.close() # second call is a no-op + self.assertFalse(proc.terminated) + + def test_close_on_empty_handle_is_safe(self): + BrowserHandle().close() # no proc, no dir β€” must not raise + + +class TestOpenDashboard(unittest.TestCase): + def test_uses_injected_opener(self): + sentinel = BrowserHandle() + opener = mock.Mock(return_value=sentinel) + result = lifecycle.open_dashboard("http://localhost:3000", opener=opener) + opener.assert_called_once_with("http://localhost:3000") + self.assertIs(result, sentinel) + + def test_none_url_returns_none_without_opening(self): + opener = mock.Mock() + self.assertIsNone(lifecycle.open_dashboard(None, opener=opener)) + opener.assert_not_called() + + def test_opener_failure_is_swallowed(self): + opener = mock.Mock(side_effect=OSError("boom")) + self.assertIsNone( + lifecycle.open_dashboard("http://x", opener=opener) + ) + + +class TestDefaultOpener(unittest.TestCase): + """Guard the launch flags that keep the dashboard in its own Chrome window. + + Never spawns a real browser: Popen, mkdtemp, and Chrome discovery are mocked. + """ + + def test_launches_isolated_dedicated_window(self): + fake_proc = FakeProc() + with mock.patch.object( + lifecycle, "_find_chrome", return_value="/fake/Chrome"), \ + mock.patch.object( + lifecycle.tempfile, "mkdtemp", + return_value="/tmp/selenium-py-devtools-ui-x"), \ + mock.patch.object( + lifecycle.subprocess, "Popen", + return_value=fake_proc) as popen: + handle = lifecycle._default_opener("http://localhost:3000") + + args = popen.call_args.args[0] + # A distinct --user-data-dir is what forces a separate Chrome instance + # (cannot merge into the user's running Chrome). + self.assertIn( + "--user-data-dir=/tmp/selenium-py-devtools-ui-x", args) + self.assertIn("--new-window", args) + self.assertIn("--app=http://localhost:3000", args) + self.assertEqual(args[0], "/fake/Chrome") + # The URL is not passed as a bare arg (that would open an extra tab). + self.assertNotIn("http://localhost:3000", args) + self.assertIs(handle.proc, fake_proc) + self.assertEqual( + handle.user_data_dir, "/tmp/selenium-py-devtools-ui-x") + + def test_chrome_not_found_returns_empty_handle_without_spawning(self): + with mock.patch.object(lifecycle, "_find_chrome", return_value=None), \ + mock.patch.object(lifecycle.subprocess, "Popen") as popen: + handle = lifecycle._default_opener("http://localhost:3000") + popen.assert_not_called() # never crash, never spawn + self.assertIsNone(handle.proc) + self.assertIsNone(handle.user_data_dir) + handle.close() # empty handle stays safe to close + + +class TestAutoOpenEnabled(unittest.TestCase): + def setUp(self): + self._saved = os.environ.get(lifecycle.ENV_OPEN) + os.environ.pop(lifecycle.ENV_OPEN, None) + + def tearDown(self): + if self._saved is None: + os.environ.pop(lifecycle.ENV_OPEN, None) + else: + os.environ[lifecycle.ENV_OPEN] = self._saved + + def test_env_falsy_disables(self): + for val in ("0", "false", "no", "off", ""): + os.environ[lifecycle.ENV_OPEN] = val + self.assertFalse(lifecycle.auto_open_enabled(), val) + + def test_env_truthy_enables(self): + for val in ("1", "true", "yes", "on", " YES "): + os.environ[lifecycle.ENV_OPEN] = val + self.assertTrue(lifecycle.auto_open_enabled(), val) + + def test_defaults_on_when_unset(self): + # Regardless of TTY: unset means open (opt-out design). A non-TTY IDE / + # `python demo.py` run must still auto-open into a dedicated window. + os.environ.pop(lifecycle.ENV_OPEN, None) + with mock.patch.object(lifecycle.sys.stdout, "isatty", return_value=False): + self.assertTrue(lifecycle.auto_open_enabled()) + with mock.patch.object(lifecycle.sys.stdout, "isatty", return_value=True): + self.assertTrue(lifecycle.auto_open_enabled()) + + +class TestShutdownFlow(unittest.TestCase): + def setUp(self): + lifecycle._reset_for_tests() + + def tearDown(self): + lifecycle._reset_for_tests() + + def test_client_disconnected_disables_and_schedules_exit(self): + disable = mock.Mock() + handle = BrowserHandle(proc=FakeProc()) + # register without touching real signals: not on the main thread guard + # not needed here β€” we call the internals directly. + lifecycle._disable = disable + lifecycle._handle = handle + + with mock.patch.object(lifecycle.threading, "Timer") as Timer: + lifecycle.on_control("clientDisconnected", {}) + Timer.assert_called_once() # a hard-exit was scheduled + timer = Timer.return_value + self.assertTrue(timer.start.called) + + disable.assert_called_once() # capture torn down + self.assertTrue(handle._closed) # window closed + + def test_client_connected_is_noop(self): + disable = mock.Mock() + lifecycle._disable = disable + with mock.patch.object(lifecycle.threading, "Timer") as Timer: + lifecycle.on_control("clientConnected", {}) + Timer.assert_not_called() + disable.assert_not_called() + + def test_shutdown_runs_once(self): + disable = mock.Mock() + lifecycle._disable = disable + with mock.patch.object(lifecycle.threading, "Timer"): + lifecycle.on_control("clientDisconnected", {}) + lifecycle.on_control("clientDisconnected", {}) + disable.assert_called_once() # guard prevents re-entry + + def test_disable_errors_do_not_propagate(self): + lifecycle._disable = mock.Mock(side_effect=RuntimeError("bad")) + with mock.patch.object(lifecycle.threading, "Timer"): + lifecycle.on_control("clientDisconnected", {}) # must not raise + + +class TestWaitForShutdown(unittest.TestCase): + def setUp(self): + lifecycle._reset_for_tests() + + def tearDown(self): + lifecycle._reset_for_tests() + + def test_dashboard_window_open(self): + self.assertFalse(lifecycle.dashboard_window_open()) # no handle + lifecycle._handle = BrowserHandle(proc=FakeProc(alive=True)) + self.assertTrue(lifecycle.dashboard_window_open()) + lifecycle._handle = BrowserHandle(proc=FakeProc(alive=False)) + self.assertFalse(lifecycle.dashboard_window_open()) + + def test_wait_returns_false_on_timeout(self): + self.assertFalse(lifecycle.wait_for_shutdown(timeout=0.01)) + + def test_wait_unblocks_on_close_without_hard_exit(self): + # With a waiter registered, clientDisconnected releases it instead of + # hard-exiting or tearing down from the WS thread. + import time + + disable = mock.Mock() + lifecycle._disable = disable + unblocked = threading.Event() + + def waiter(): + lifecycle.wait_for_shutdown(timeout=2) + unblocked.set() + + t = threading.Thread(target=waiter) + t.start() + time.sleep(0.05) # let the waiter register + with mock.patch.object(lifecycle.threading, "Timer") as Timer: + lifecycle.on_control("clientDisconnected", {}) + Timer.assert_not_called() # no hard-exit scheduled + self.assertTrue(unblocked.wait(2)) # waiter released + disable.assert_not_called() # handler left teardown to the waiter + t.join(2) + + +class TestExitHandlerRegistration(unittest.TestCase): + def setUp(self): + lifecycle._reset_for_tests() + + def tearDown(self): + lifecycle._reset_for_tests() + + def test_register_off_main_thread_skips_signals_but_wires_atexit(self): + disable = mock.Mock() + handle = BrowserHandle(proc=FakeProc()) + results = {} + + def worker(): + with mock.patch.object(lifecycle.atexit, "register") as reg, \ + mock.patch.object(lifecycle.signal, "signal") as sig: + lifecycle.register_exit_handlers(disable, handle) + results["atexit"] = reg.called + results["signal"] = sig.called + + t = threading.Thread(target=worker) + t.start() + t.join() + self.assertTrue(results["atexit"]) + self.assertFalse(results["signal"]) # signals only on main thread + + def test_register_is_idempotent(self): + disable = mock.Mock() + with mock.patch.object(lifecycle.atexit, "register") as reg, \ + mock.patch.object(lifecycle.signal, "signal"), \ + mock.patch.object(lifecycle.signal, "getsignal"), \ + mock.patch.object( + lifecycle.threading, "main_thread", + return_value=threading.current_thread()): + lifecycle.register_exit_handlers(disable, None) + lifecycle.register_exit_handlers(disable, None) + self.assertEqual(reg.call_count, 1) + + def test_unregister_closes_handle_and_restores(self): + disable = mock.Mock() + handle = BrowserHandle(proc=FakeProc()) + with mock.patch.object(lifecycle.atexit, "register"), \ + mock.patch.object(lifecycle.atexit, "unregister") as unreg, \ + mock.patch.object(lifecycle.signal, "signal"), \ + mock.patch.object(lifecycle.signal, "getsignal"), \ + mock.patch.object( + lifecycle.threading, "main_thread", + return_value=threading.current_thread()): + lifecycle.register_exit_handlers(disable, handle) + lifecycle.unregister_exit_handlers() + self.assertTrue(unreg.called) + self.assertTrue(handle._closed) # window closed on unregister + self.assertFalse(lifecycle._handlers_registered) + + +if __name__ == "__main__": + unittest.main() diff --git a/packages/selenium-py-devtools/tests/test_transport.py b/packages/selenium-py-devtools/tests/test_transport.py new file mode 100644 index 00000000..32563bec --- /dev/null +++ b/packages/selenium-py-devtools/tests/test_transport.py @@ -0,0 +1,24 @@ +import threading +import unittest + +from wdio_selenium_devtools.transport import WSClient + + +class TestClose(unittest.TestCase): + def test_close_from_reader_thread_does_not_join_self(self): + # close() can run ON the reader thread (a clientDisconnected control + # frame triggering shutdown) β€” it must not try to join itself. + client = WSClient("localhost", 1) + client._reader = threading.current_thread() + client.connected = True + client.close() # must not raise "cannot join current thread" + self.assertFalse(client.connected) + + def test_close_is_safe_when_never_connected(self): + client = WSClient("localhost", 1) + client.close() # no socket, no reader β€” no raise + self.assertFalse(client.connected) + + +if __name__ == "__main__": + unittest.main() From 7046401895eab8b6ede3ea8ad5b80819e5200006 Mon Sep 17 00:00:00 2001 From: Vishnu Vardhan Date: Thu, 2 Jul 2026 16:35:26 +0530 Subject: [PATCH 05/12] feat(selenium-py-devtools): BiDi console + network capture --- .../src/wdio_selenium_devtools/bidi.py | 412 ++++++++++++++++++ .../tests/test_snapshot.py | 189 ++++++++ 2 files changed, 601 insertions(+) create mode 100644 packages/selenium-py-devtools/src/wdio_selenium_devtools/bidi.py create mode 100644 packages/selenium-py-devtools/tests/test_snapshot.py 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/tests/test_snapshot.py b/packages/selenium-py-devtools/tests/test_snapshot.py new file mode 100644 index 00000000..7edb37a2 --- /dev/null +++ b/packages/selenium-py-devtools/tests/test_snapshot.py @@ -0,0 +1,189 @@ +import os +import tempfile +import unittest + +from wdio_selenium_devtools import snapshot +from wdio_selenium_devtools.snapshot import ( + SnapshotCapturer, + load_injectable_script, + normalize_mutations, + resolve_script_path, + start_snapshot_capture, + wrap_injectable, +) + + +class FakeExec: + """Records execute_script calls; returns queued values in order.""" + + def __init__(self, *returns): + self.calls = [] + self._returns = list(returns) + + def __call__(self, script, *args): + self.calls.append((script, args)) + if self._returns: + return self._returns.pop(0) + return None + + +class TestScriptResolution(unittest.TestCase): + def test_resolves_monorepo_script(self): + path = resolve_script_path() + # The built runtime exists in this repo; resolution must find it. + self.assertIsNotNone(path) + self.assertTrue(path.endswith(os.path.join("script", "dist", "script.js"))) + self.assertTrue(os.path.isfile(path)) + + +class TestWrapAndLoad(unittest.TestCase): + def test_wrap_produces_async_iife(self): + wrapped = wrap_injectable("doThing();") + self.assertEqual(wrapped, "(async function() { doThing(); })()") + + def test_load_reads_and_wraps(self): + with tempfile.NamedTemporaryFile("w", suffix=".js", delete=False) as fh: + fh.write("BODY") + temp = fh.name + try: + out = load_injectable_script(temp) + finally: + os.unlink(temp) + self.assertEqual(out, "(async function() { BODY })()") + + def test_load_missing_file_returns_none(self): + self.assertIsNone(load_injectable_script("/no/such/script.js")) + + +class TestNormalizeMutations(unittest.TestCase): + def test_extracts_mutation_list(self): + payload = { + "mutations": [{"type": "childList", "target": "1"}], + "consoleLogs": [], + "metadata": {"url": "https://x/"}, + } + self.assertEqual(normalize_mutations(payload), [{"type": "childList", "target": "1"}]) + + def test_missing_key_returns_empty(self): + self.assertEqual(normalize_mutations({"consoleLogs": []}), []) + + def test_non_list_mutations_returns_empty(self): + self.assertEqual(normalize_mutations({"mutations": "oops"}), []) + + def test_none_payload_returns_empty(self): + self.assertEqual(normalize_mutations(None), []) + + def test_non_dict_payload_returns_empty(self): + self.assertEqual(normalize_mutations([1, 2, 3]), []) + + +class TestSnapshotCapturerInject(unittest.TestCase): + def _tmp_script(self): + fh = tempfile.NamedTemporaryFile("w", suffix=".js", delete=False) + fh.write("COLLECTOR") + fh.close() + self.addCleanup(lambda: os.path.exists(fh.name) and os.unlink(fh.name)) + return fh.name + + def test_inject_appends_script_and_probes(self): + # probe (absent) -> install -> probe (present) + exec_fn = FakeExec(False, None, True) + cap = SnapshotCapturer(exec_fn, script_path=self._tmp_script()) + self.assertTrue(cap.inject()) + self.assertTrue(cap.injected) + install = [c for c in exec_fn.calls if "createElement('script')" in c[0]] + self.assertEqual(len(install), 1) + self.assertEqual(install[0][1], ("(async function() { COLLECTOR })()",)) + self.assertTrue(any("wdioTraceCollector" in c[0] for c in exec_fn.calls)) + + def test_inject_skips_install_when_collector_present(self): + # Self-healing: collector already on the page β†’ probe True β†’ no install. + exec_fn = FakeExec(True) + cap = SnapshotCapturer(exec_fn, script_path=self._tmp_script()) + self.assertTrue(cap.inject()) + self.assertEqual([c for c in exec_fn.calls if "createElement" in c[0]], []) + + def test_inject_reinjects_after_navigation(self): + # Navigation wipes the collector: probe absent both times β†’ 2 installs. + exec_fn = FakeExec(False, None, True, False, None, True) + cap = SnapshotCapturer(exec_fn, script_path=self._tmp_script()) + cap.inject() + cap.inject() + installs = [c for c in exec_fn.calls if "createElement" in c[0]] + self.assertEqual(len(installs), 2) + + def test_inject_missing_script_is_noop(self): + exec_fn = FakeExec() + cap = SnapshotCapturer(exec_fn, script_path="/no/such.js") + self.assertFalse(cap.inject()) + self.assertFalse(cap.injected) + self.assertEqual(exec_fn.calls, []) # script missing β†’ no execute at all + + def test_inject_swallows_execute_errors(self): + def boom(script, *args): + raise RuntimeError("no such session") + + cap = SnapshotCapturer(boom, script_path=self._tmp_script()) + self.assertFalse(cap.inject()) # no raise + self.assertFalse(cap.injected) + + +class TestSnapshotCapturerPull(unittest.TestCase): + def test_pull_returns_normalized_mutations(self): + trace = {"mutations": [{"type": "attributes", "target": "9"}]} + exec_fn = FakeExec(trace) + cap = SnapshotCapturer(exec_fn) + self.assertEqual(cap.pull_mutations(), [{"type": "attributes", "target": "9"}]) + # The read uses the atomic getTraceData expression. + self.assertIn("getTraceData", exec_fn.calls[0][0]) + + def test_pull_null_trace_returns_empty(self): + cap = SnapshotCapturer(FakeExec(None)) + self.assertEqual(cap.pull_mutations(), []) + + def test_pull_swallows_execute_errors(self): + def boom(script, *args): + raise RuntimeError("boom") + + cap = SnapshotCapturer(boom) + self.assertEqual(cap.pull_mutations(), []) # no raise + + +class TestStartSnapshotCapture(unittest.TestCase): + def _tmp_script(self): + fh = tempfile.NamedTemporaryFile("w", suffix=".js", delete=False) + fh.write("C") + fh.close() + self.addCleanup(lambda: os.path.exists(fh.name) and os.unlink(fh.name)) + return fh.name + + def test_returns_capturer_when_driver_can_execute(self): + class Driver: + def __init__(self): + self.calls = [] + + def execute_script(self, script, *args): + self.calls.append(script) + return True # ready probe + + driver = Driver() + cap = start_snapshot_capture(driver, script_path=self._tmp_script()) + self.assertIsInstance(cap, SnapshotCapturer) + self.assertTrue(cap.injected) + + def test_none_when_driver_has_no_execute_script(self): + self.assertIsNone(start_snapshot_capture(object())) + + def test_none_when_injection_fails(self): + class Driver: + def execute_script(self, script, *args): + return None + + # No script on disk β†’ injection returns None. + self.assertIsNone( + start_snapshot_capture(Driver(), script_path="/no/such.js") + ) + + +if __name__ == "__main__": + unittest.main() From 2fec768d25c2793bbbd5caa8ec292c1e3c031bb4 Mon Sep 17 00:00:00 2001 From: Vishnu Vardhan Date: Thu, 2 Jul 2026 16:35:50 +0530 Subject: [PATCH 06/12] feat(selenium-py-devtools): DOM snapshot capture via injected collector --- .../src/wdio_selenium_devtools/snapshot.py | 197 +++++++++ .../selenium-py-devtools/tests/test_bidi.py | 374 ++++++++++++++++++ 2 files changed, 571 insertions(+) create mode 100644 packages/selenium-py-devtools/src/wdio_selenium_devtools/snapshot.py create mode 100644 packages/selenium-py-devtools/tests/test_bidi.py diff --git a/packages/selenium-py-devtools/src/wdio_selenium_devtools/snapshot.py b/packages/selenium-py-devtools/src/wdio_selenium_devtools/snapshot.py new file mode 100644 index 00000000..e199e9d0 --- /dev/null +++ b/packages/selenium-py-devtools/src/wdio_selenium_devtools/snapshot.py @@ -0,0 +1,197 @@ +"""DOM snapshot capture β€” the Python analogue of core's page-side trace path. + +The dashboard's center "browser preview" panel replays the page by applying a +stream of DOM mutations captured in the browser. Those mutations come from the +``packages/script`` runtime (``window.wdioTraceCollector``): we inject it once, +then periodically read the buffered ``getTraceData()`` back and forward the +``mutations`` array via ``capturer.send_mutations``. + +Mirrors ``core/script-loader.ts`` (``loadInjectableScript``) and +``selenium-devtools/session.ts`` (``injectScript`` / ``captureTrace``). The JS +adapter injects via ``document.createElement('script')`` rather than a BiDi +preload, and reads back with a single atomic ``executeScript`` β€” we do the same. + +The script-path resolution and ``execute_script`` calls are injectable so the +pure logic (path resolution, IIFE wrapping, payload normalization) unit-tests +without selenium or a real browser. Everything is defensive: an injection or +readback failure is a logged no-op β€” capture never breaks the user's test. +""" + +from __future__ import annotations + +import os +import sys +from typing import Any, Callable, List, Optional + +#: A ``driver.execute_script(script, *args)`` shaped callable β€” injectable so +#: tests drive injection/readback without a real driver. +ExecuteFn = Callable[..., Any] + +#: Installs the collector: append a