diff --git a/.changeset/backend-server-entry.md b/.changeset/backend-server-entry.md new file mode 100644 index 00000000..66c4a738 --- /dev/null +++ b/.changeset/backend-server-entry.md @@ -0,0 +1,9 @@ +--- +"@wdio/devtools-backend": minor +--- + +The backend ships a runnable entry, so a non-Node adapter can start the dashboard itself. + +- `dist/server.js` is a new CLI entry, exposed as the `devtools-backend` bin. `node packages/backend/dist/server.js` and `npx @wdio/devtools-backend` both start the live dashboard, and `--port` / `--hostname` / `--help` are accepted. +- `dist/index.js` stays the library entry the JS adapters import in-process. Its "start if run directly" guard is gone rather than repaired: `show-trace.ts` imports `start` from index, which makes index a shared module whose body tsup hoists into `dist/chunk-*.js`, and there `import.meta.url` is the chunk's path and can never equal `process.argv[1]`. The guard was therefore dead in every build, which is why `node dist/index.js` imported a module and exited 0 without serving. A leaf entry keeps its body in its own output file, so the new file needs no guard at all. +- `dev:app` now runs `dist/server.js`, since it was watching an entry that could not start. diff --git a/.github/workflows/python-release.yml b/.github/workflows/python-release.yml new file mode 100644 index 00000000..9ad973d8 --- /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-devtools-py/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-devtools-py + +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-devtools-py/dist + - name: πŸš€ Publish to TestPyPI + if: ${{ inputs.target == 'testpypi' }} + uses: pypa/gh-action-pypi-publish@cef221092ed1bacb1cc03d23a2d87d1d172e277b # release/v1 + with: + packages-dir: packages/selenium-devtools-py/dist + repository-url: https://test.pypi.org/legacy/ diff --git a/.github/workflows/python.yml b/.github/workflows/python.yml new file mode 100644 index 00000000..9d3102e9 --- /dev/null +++ b/.github/workflows/python.yml @@ -0,0 +1,46 @@ +name: Python Adapter + +# Deliberately NOT triggered by packages/shared/**, even though the adapter's +# wire contract is generated from it. Doing so put a Python check on every PR +# that touched shared, which is most of them, and a red Python job would block +# work unrelated to this adapter. The trade-off: a shared change that breaks the +# contract is not caught here at the moment it lands. It is still caught by the +# drift check below on the next PR that touches this package, which regenerates +# _contract.py and fails on any diff. Revisit once the adapter is published and +# the coupling is worth the noise. +on: + push: + branches: + - main + paths: + - packages/selenium-devtools-py/** + - .github/workflows/python.yml + pull_request: + paths: + - packages/selenium-devtools-py/** + - .github/workflows/python.yml + +permissions: + contents: read + +defaults: + run: + working-directory: packages/selenium-devtools-py + +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/selenium_devtools/_contract.py + - name: πŸ§ͺ Unit tests + run: PYTHONPATH=src python -m unittest discover -s tests diff --git a/ARCHITECTURE.md b/ARCHITECTURE.md index aab07153..ef0c5766 100644 --- a/ARCHITECTURE.md +++ b/ARCHITECTURE.md @@ -106,6 +106,7 @@ The server adapters connect to and the app talks to. Contains: +- Two build entries: `index.ts` is the **library** entry (`dist/index.js`) the adapters import in-process for `start`/`stop`, and `server.ts` is the **CLI** entry (`dist/server.js`, executable, shipped as the `devtools-backend` bin) that starts the same server standalone. `server.ts` is a leaf entry on purpose: a self-start guard inside `index.ts` is dead, because `show-trace.ts` imports it and tsup hoists a shared entry's body into `dist/chunk-*.js`, where `import.meta.url` never matches `process.argv[1]`. - Fastify HTTP server. - WebSocket gateway: one connection per adapter worker, one per app client. - Baseline store (in-memory) for preserve-and-rerun; reuses `shared` types directly via thin `*Like` aliases (`baseline/types.ts`). diff --git a/CLAUDE.md b/CLAUDE.md index 31abe165..8f475542 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -287,6 +287,7 @@ Documented divergences from the conventions above. They exist today as debt to b - **Assert outcomes are correlated off the assertion's own promise, because the results bag does not exist.** Cucumber's Nightwatch client is built by `createClient` with **no reporter**, so `SimplifiedReporter.logAssertResult` no-ops and `results.assertions`/`results.testcases` are *never* populated β€” no scenario-level reconcile can recover them, which is why `currentTest: undefined` was a dead end. `nativeAssertions.ts` `observedAssertOutcome` reads the outcome from the returned promise instead (`lib/core/asynctree.js` `shouldRejectNodePromise`: a failing `assert.*` rejects its deferred, a failing `verify.*` resolves *with* the AssertionError, a pass resolves with the command value). A fulfilment of `undefined` stays **neutral** β€” that is an assertion enqueued but never executed after an earlier `assert.*` emptied the queue, and reading it as a pass would paint a never-run assertion green. The results bag still wins where it exists and the row's window comes from whichever source supplied the outcome, so the describe/it timeline is byte-stable. Measured: **4 of 4 assert rows with a real pass/fail** (was 2 of 4 rows, 0 correlated), spanning real 44–372 ms windows instead of a synthetic 1 ms. - Relatedly, cucumber's **per-step** `resetCommandTracking()` was wiping the native-assert buffer, so each scenario kept only its *last* step's assertions (measured 1 of 2). The buffer is per **test unit** β€” `resetTestTracking()` at `wrapBrowserOnce` now clears it. - **DOM `mutations` ARE captured**, and the old `ECONNRESET` / "collector not found" attribution is **wrong** β€” neither appears any more. The gap closed itself with document-start injection (`core/bidi-preload.ts`), confirmed on a *baseline* build so the credit is A10's. Measured per scenario with `traceGranularity:'test'` + `webSocketUrl: true`: 39 and 19 mutation entries, 2 DOM anchors each, **0 of 19 rows on the wrong document**. What remains is noise, not a gap: the screencast poller issues `/screenshot` every 200 ms into the session Nightwatch quits per scenario, logging 9–26 `WARN webdriverHttp: … socket hang up` per run. Fix is to stop the recorder before the per-scenario quit, or suppress the warn for a session in teardown. +- **A tsup entry that another entry imports is not a leaf, so an `import.meta.url === process.argv[1]` self-start check inside it is dead code; a CLI has to be its own leaf entry.** tsup hoists a module body shared by two entries into `dist/chunk-*.js`, and there `import.meta.url` is the chunk's path, which can never equal `process.argv[1]`. `packages/backend/src/show-trace.ts` imports `start` from `index.ts`, which is what made index shared, so index's old "start if run directly" guard was dead in **every** build: `node dist/index.js` exited 0 without ever serving, while `dist/show-trace.js` self-started correctly for exactly the same reason inverted, being a leaf whose body stays in its own output file (it also compares realpaths, because the invoked path is the `node_modules/.bin` symlink). The live dashboard server is therefore its own leaf entry, `packages/backend/src/server.ts` (shebang, built to an executable `dist/server.js`, shipped as the `devtools-backend` bin, accepting `--port`, `--hostname`, `-h`/`--help`), and `index.ts` stays library-only for the three adapters' in-process `start`/`stop`. Same family as the `cucumberHooks.cts` entry above: the tsup entry list is part of the contract, and in both cases the symptom was silence rather than an error. - **A captured text locator is generated in the recording runner's dialect; every other branch is portable CSS.** `shared/locator-dialect.ts` `locatorDialect(runner)` is the one fact table β€” WDIO runners (`mocha`/`jasmine`/`cucumber`) get `a*=Logout`, `nightwatch*`/`selenium-webdriver` and an **unidentified recorder** get `//a[contains(., "Logout")]`. The id reaches the page script as `CaptureActionSnapshotInput.runner` and the zip as an extension field on `context-options`, read back through `isTestRunnerId` onto `Metadata.runner`; absent in older and foreign zips, where the player shows no hint. Under WDIO a text carrying a `"` still emits XPath β€” WDIO compiles `tag*=` to XPath with `"` quoting and would build a broken expression β€” and WDIO resolves `//` itself. `locatorsMatch` decomposes **both** sides from either dialect, so `after.point` survives whichever way round the two grammars fall; the `concat()`-stitched literal is still left to exact comparison. `@wdio/elements`' standalone `getSnapshot` deliberately keeps the portable XPath default (its output is pasted into arbitrary tools), so a WDIO run's `browser.getSnapshot()` and its trace A11y tab disagree on that one branch. - `Metadata.runner` (typed `TestRunnerId`) and `metadata.options.framework` (untyped `string`, read by the sidebar's `getFramework`) are two carriers of the same fact. All three adapters now set both; the next change to either should collapse `getFramework` onto `Metadata.runner`. - Selenium's `ctx.runner` (the *detected JS runner* β€” mocha/jest/cucumber, from `detectRunner()`) shadows the new `Metadata.runner` concept. Renamed to `detectedRunner` at the `driverMetadata` boundary only, where the two met inside one function and a swap would have been a silent dialect bug; the repo-wide rename in `plugin-internals.ts`/`session-lifecycle.ts` is still open. diff --git a/examples/selenium/python-test/.gitignore b/examples/selenium/python-test/.gitignore new file mode 100644 index 00000000..58341d59 --- /dev/null +++ b/examples/selenium/python-test/.gitignore @@ -0,0 +1,3 @@ +# Run output goes to test-results/, which the root .gitignore already covers. +# *.webm stays for videos written by older versions of the adapter. +*.webm diff --git a/examples/selenium/python-test/web_form.py b/examples/selenium/python-test/web_form.py new file mode 100644 index 00000000..bf13aa37 --- /dev/null +++ b/examples/selenium/python-test/web_form.py @@ -0,0 +1,35 @@ +"""The smallest useful DevTools example: three devtools lines, one Selenium test. + +This is the script reproduced in the integration one-pager, so keep the two in +sync when either changes. + +Run it: + + pip install -e packages/selenium-devtools-py + python examples/selenium/python-test/web_form.py + +``enable()`` starts the dashboard backend itself when none is running. Set +DEVTOOLS_PORT instead to attach to one you already have open. Run output (the +screencast .webm) lands in ``test-results/`` beside this file, matching the JS +adapters. +""" + +import selenium_devtools as devtools +from selenium import webdriver +from selenium.webdriver.common.by import By +from selenium.webdriver.chrome.options import Options + +devtools.enable() # opens the dashboard, starts capturing + +options = Options() +options.add_argument("--headless=new") # drop this line to watch the browser +options.add_argument("--window-size=1280,1024") # bigger viewport, fuller screencast +driver = webdriver.Chrome(options=options) +try: + driver.get("https://www.selenium.dev/selenium/web/web-form.html") + driver.find_element(By.NAME, "my-text").send_keys("Selenium") + driver.find_element(By.CSS_SELECTOR, "button").click() + print(driver.find_element(By.ID, "message").text) +finally: + driver.quit() + devtools.wait_for_dashboard_close() # hold the UI open to inspect diff --git a/package.json b/package.json index b6bc9ee3..b52459b4 100644 --- a/package.json +++ b/package.json @@ -6,12 +6,14 @@ "scripts": { "build": "pnpm -r build", "show-trace": "node packages/backend/dist/show-trace.js", + "backend": "node packages/backend/dist/server.js", "demo:wdio": "wdio run ./examples/wdio/cucumber/wdio.conf.ts", "demo:wdio:mocha": "wdio run ./examples/wdio/mocha/wdio.conf.ts", "demo:wdio:retry": "wdio run ./examples/wdio/mocha/wdio.retry.conf.ts", "demo:nightwatch": "pnpm --filter @wdio/nightwatch-devtools example", "demo:nightwatch:retry": "pnpm --filter @wdio/nightwatch-devtools example:retry", "demo:selenium": "pnpm --filter @wdio/selenium-devtools example", + "demo:python": "python3 examples/selenium/python-test/web_form.py", "dev": "pnpm --parallel dev", "preview": "pnpm --parallel preview", "test": "vitest run", diff --git a/packages/backend/README.md b/packages/backend/README.md index ef3110a4..58d5b3c0 100644 --- a/packages/backend/README.md +++ b/packages/backend/README.md @@ -1,6 +1,6 @@ # @wdio/devtools-backend -The server that the three adapter packages connect to and the dashboard UI talks to. Internal to the monorepo β€” not published. +The server that the three adapter packages connect to and the dashboard UI talks to. Published to npm as `@wdio/devtools-backend` (it's on [CONTRIBUTING.md](../../CONTRIBUTING.md)'s list of published packages, has a `prepublishOnly` build, and ships the `devtools-backend` and `show-trace` bins); the adapters depend on it rather than vendoring it. ## Responsibilities @@ -25,7 +25,7 @@ Trace mode (see the [root README](../../README.md#-trace-mode-tracezip)) writes pnpm show-trace path/to/trace.zip ``` - The same entry is shipped as a `show-trace` **bin** by the backend (`./dist/show-trace.js`) and by each adapter (`@wdio/devtools-service`, `@wdio/selenium-devtools`, `@wdio/nightwatch-devtools` each ship a thin `bin/show-trace.mjs`), so `npx show-trace ` works in any project that installs an adapter β€” backend need only be a transitive dependency. In trace-serve mode `start({ trace })` exposes the reconstructed payload at `TRACE_API.get` and skips the worker/rerun machinery. + The same entry is shipped as a `show-trace` **bin** by the backend (`./dist/show-trace.js`) and by each adapter (`@wdio/devtools-service`, `@wdio/selenium-devtools`, `@wdio/nightwatch-devtools` each ship a thin `bin/show-trace.mjs`), so `npx show-trace ` works in any project that installs an adapter β€” backend need only be a transitive dependency. In trace-serve mode `start({ trace })` exposes the reconstructed payload at `TRACE_API.get` and skips the worker/rerun machinery. It is one of the backend's two bins; the other, `devtools-backend` (`./dist/server.js`), starts the live dashboard server described under [Public API](#public-api). - **Reader** (`src/trace-reader.ts`, with sibling `trace-reader-{constants,types,utils,groups}.ts`) β€” `parseTraceZip` / `readTraceZip` reconstruct a `TracePlayerData` payload from the archive. It accepts this repo's own exporter output **and foreign zips** written by other tools (every `.trace` entry is an action-event stream, every `.network` a HAR stream, `.stacks` sidecars carry call stacks). It rebuilds: - **commands** β€” from `before`/`after` action events, with call source, result, error, nearest screenshot frame, and pointer hit point; @@ -40,6 +40,20 @@ Because the archive uses that portable, standard trace-viewer format, the same ` ## Public API -The backend is consumed only by other workspace packages. Adapter launchers call `start({ port, hostname })` and receive the bound port. The dashboard accesses it via the documented HTTP routes (`packages/shared/src/baseline.ts`, `packages/shared/src/runner.ts`) and WS scopes (`packages/shared/src/ws.ts`, `packages/shared/src/routes.ts`). +Two build entries, and they are deliberately separate files. + +- **Library entry** (`src/index.ts` β†’ `dist/index.js`), consumed in-process by the other workspace packages: adapter launchers call `start({ port, hostname })` and receive the bound port, then `stop()` on teardown. The dashboard accesses the running server via the documented HTTP routes (`packages/shared/src/baseline.ts`, `packages/shared/src/runner.ts`) and WS scopes (`packages/shared/src/ws.ts`, `packages/shared/src/routes.ts`). + +- **CLI entry** (`src/server.ts` β†’ the executable `dist/server.js`), shipped as the `devtools-backend` bin, which starts the same server on its own instead of from an adapter's launcher: + + ```sh + npx @wdio/devtools-backend --port 8080 --hostname 0.0.0.0 + ``` + + - `--port ` (or `--port=`): preferred port; a free one is chosen if it's taken. + - `--hostname ` (or `--hostname=`): host to bind, `localhost` by default. + - `-h`, `--help`: print usage and exit without starting. + +Don't collapse the CLI back into `index.ts` behind a "start if run directly" guard: `show-trace.ts` imports `start` from index, which makes index a shared module whose body tsup hoists into `dist/chunk-*.js`, and there `import.meta.url` is the chunk's own path and can never equal `process.argv[1]`, so the guard is dead in every build (`node dist/index.js` exited 0 without ever serving); a leaf entry keeps its body in its own output file, which is why `show-trace.js` self-starts correctly. For the full picture of how events flow adapter β†’ backend β†’ dashboard, see [ARCHITECTURE.md](../../ARCHITECTURE.md). diff --git a/packages/backend/package.json b/packages/backend/package.json index 46a1395c..b1d66f0f 100644 --- a/packages/backend/package.json +++ b/packages/backend/package.json @@ -18,15 +18,16 @@ } }, "bin": { + "devtools-backend": "./dist/server.js", "show-trace": "./dist/show-trace.js" }, "types": "./dist/index.d.ts", "typeScriptVersion": "^5.0.0", "scripts": { "dev": "run-p dev:*", - "dev:ts": "tsup src/index.ts src/show-trace.ts --format esm --dts --watch", - "dev:app": "nodemon --watch ./dist ./dist/index.js", - "build": "tsup src/index.ts src/show-trace.ts --format esm --dts --clean", + "dev:ts": "tsup src/index.ts src/show-trace.ts src/server.ts --format esm --dts --watch", + "dev:app": "nodemon --watch ./dist ./dist/server.js", + "build": "tsup src/index.ts src/show-trace.ts src/server.ts --format esm --dts --clean", "lint": "eslint .", "prepublishOnly": "pnpm build" }, diff --git a/packages/backend/src/index.ts b/packages/backend/src/index.ts index 173e0a0f..e5061121 100644 --- a/packages/backend/src/index.ts +++ b/packages/backend/src/index.ts @@ -1,5 +1,4 @@ import fs from 'node:fs' -import url from 'node:url' import Fastify, { type FastifyInstance, @@ -443,13 +442,3 @@ export async function stop() { await server.close() } - -/** - * start server if this file is called directly - */ -if (import.meta.url.startsWith('file:')) { - const modulePath = url.fileURLToPath(import.meta.url) - if (process.argv[1] === modulePath) { - start() - } -} diff --git a/packages/backend/src/server-args.ts b/packages/backend/src/server-args.ts new file mode 100644 index 00000000..5bad9e2d --- /dev/null +++ b/packages/backend/src/server-args.ts @@ -0,0 +1,44 @@ +/** + * Argv parsing for the `devtools-backend` bin. Split from `server.ts` because + * that file starts a server as its module body, so a unit test cannot import it. + */ + +export const SERVER_USAGE = ` + Usage: devtools-backend [options] + + Options: + --port Preferred port; a free one is chosen if it is taken + --hostname Host to bind (default: localhost) + -h, --help Show this message +` + +export interface ServerArgs { + help: boolean + port?: number + hostname?: string +} + +/** Read `--name value` or `--name=value`, whichever the caller used. */ +function option(args: string[], name: string): string | undefined { + const flag = `--${name}` + const at = args.indexOf(flag) + if (at !== -1) { + return args[at + 1] + } + return args.find((arg) => arg.startsWith(`${flag}=`))?.slice(flag.length + 1) +} + +/** + * A port is kept only when it parses as a positive integer, so `--port` with no + * value, `--port abc` and `--port 0` all fall through to the server's own + * default rather than binding something the caller did not ask for. + */ +export function parseServerArgs(args: string[]): ServerArgs { + const port = Number(option(args, 'port')) + const hostname = option(args, 'hostname') + return { + help: args.includes('-h') || args.includes('--help'), + ...(Number.isInteger(port) && port > 0 ? { port } : {}), + ...(hostname ? { hostname } : {}) + } +} diff --git a/packages/backend/src/server.ts b/packages/backend/src/server.ts new file mode 100644 index 00000000..5ec3b551 --- /dev/null +++ b/packages/backend/src/server.ts @@ -0,0 +1,35 @@ +#!/usr/bin/env node +// CLI entry for the live dashboard server. `index.ts` is the LIBRARY entry the +// JS adapters import in-process; a "start if run directly" guard cannot live +// there, because `show-trace.ts` imports `start` from it, which makes index a +// shared module whose body tsup hoists into `dist/chunk-*.js` β€” there +// `import.meta.url` is the chunk's path and never equals `process.argv[1]`, so +// the guard silently never fires. A leaf entry keeps its body in its own output +// file, which is why this file needs no guard at all. + +import { start } from './index.js' +import { parseServerArgs, SERVER_USAGE } from './server-args.js' + +async function run(args: string[]): Promise { + const { help, ...opts } = parseServerArgs(args) + if (help) { + console.log(SERVER_USAGE) + return + } + try { + const { port } = await start(opts) + // Our own line, so a consumer scraping stdout for the bound port (the Python + // adapter does) depends on this file rather than on Fastify's logger staying + // enabled. The port is the one actually bound, which differs from any + // `--port` when that was taken. + const host = opts.hostname ?? 'localhost' + console.log(`devtools-backend listening at http://${host}:${port}`) + } catch (error) { + console.error( + `Failed to start WebdriverIO Devtools: ${(error as Error).message}` + ) + process.exit(1) + } +} + +void run(process.argv.slice(2)) diff --git a/packages/backend/tests/server-args.test.ts b/packages/backend/tests/server-args.test.ts new file mode 100644 index 00000000..f309a3fb --- /dev/null +++ b/packages/backend/tests/server-args.test.ts @@ -0,0 +1,64 @@ +import { describe, it, expect } from 'vitest' +import { parseServerArgs, SERVER_USAGE } from '../src/server-args.js' + +describe('parseServerArgs', () => { + it('reads a port given as two arguments', () => { + expect(parseServerArgs(['--port', '4123'])).toEqual({ + help: false, + port: 4123 + }) + }) + + it('reads a port given with an equals sign', () => { + expect(parseServerArgs(['--port=4123'])).toEqual({ + help: false, + port: 4123 + }) + }) + + it('reads a hostname in either form', () => { + expect(parseServerArgs(['--hostname', '0.0.0.0'])).toEqual({ + help: false, + hostname: '0.0.0.0' + }) + expect(parseServerArgs(['--hostname=0.0.0.0'])).toEqual({ + help: false, + hostname: '0.0.0.0' + }) + }) + + it('takes both together, in any order', () => { + expect( + parseServerArgs(['--hostname', '127.0.0.1', '--port', '5000']) + ).toEqual({ help: false, hostname: '127.0.0.1', port: 5000 }) + }) + + it('omits the port when it is absent, so the server keeps its default', () => { + expect(parseServerArgs([])).toEqual({ help: false }) + }) + + // `--port` as the final argument reads `undefined`, and a flag as its value + // must not be swallowed as a port either. + it.each([ + ['no value', ['--port']], + ['the next flag as its value', ['--port', '--hostname', 'localhost']], + ['a non-numeric value', ['--port', 'abc']], + ['a fractional value', ['--port', '80.5']], + ['zero', ['--port', '0']], + ['a negative value', ['--port', '-1']] + ])('drops a port given as %s', (_label, args) => { + expect(parseServerArgs(args as string[]).port).toBeUndefined() + }) + + it('reports help for -h and --help', () => { + expect(parseServerArgs(['-h']).help).toBe(true) + expect(parseServerArgs(['--help']).help).toBe(true) + expect(parseServerArgs(['--port', '3000']).help).toBe(false) + }) + + it('documents every flag it parses', () => { + for (const flag of ['--port', '--hostname', '-h', '--help']) { + expect(SERVER_USAGE).toContain(flag) + } + }) +}) diff --git a/packages/selenium-devtools-py/.gitignore b/packages/selenium-devtools-py/.gitignore new file mode 100644 index 00000000..69b17f32 --- /dev/null +++ b/packages/selenium-devtools-py/.gitignore @@ -0,0 +1,8 @@ +__pycache__/ +*.py[cod] +.pytest_cache/ +.venv/ +venv/ +build/ +dist/ +*.egg-info/ diff --git a/packages/selenium-devtools-py/README.md b/packages/selenium-devtools-py/README.md new file mode 100644 index 00000000..8ad4f1cc --- /dev/null +++ b/packages/selenium-devtools-py/README.md @@ -0,0 +1,202 @@ +# selenium-devtools-py (Python) + +Python Selenium adapter for the WebdriverIO DevTools dashboard β€” the fourth +adapter alongside the JS WebdriverIO / Nightwatch / Selenium-JS ones. It feeds +the **same backend and UI**, unchanged, over the language-neutral +`{scope, data}` WebSocket contract. + +**Status: Phase 1 + 2.** Live command capture + test tree; browser console & +network via BiDi; screencast video; and a dashboard window that auto-opens and +tears down with the run. Verified against real headless Chrome. See +[Roadmap](#roadmap) for what's deferred. + +## Install (dev) + +```bash +pip install -e packages/selenium-devtools-py # or: pip install selenium-devtools-py (when published) +``` + +The transport is **dependency-free** (stdlib WebSocket client). The only thing +on top of your own `selenium` install is this package; `pytest` is optional. + +## Use + +**With pytest (recommended) β€” no code changes to your tests:** + +```bash +DEVTOOLS_ENABLE=1 pytest tests/ # DEVTOOLS_PORT= also opts in (and attaches) +``` + +The bundled plugin auto-captures the run, opens the dashboard in a dedicated +window, and β€” after the run β€” **keeps it open so you can inspect it**; close the +window (or Ctrl-C) to finish. Nothing devtools-specific goes in your test files. + +**Without pytest** (any script / unittest) β€” add two lines to a normal Selenium +script (`devtools.enable()` + `devtools.wait_for_dashboard_close()`): + +```python +import selenium_devtools as devtools + +devtools.enable() # open dashboard + capture every command +# ... your normal selenium code, ending with driver.quit() ... +devtools.wait_for_dashboard_close() # keep the UI open to inspect (no-op if headless) +devtools.disable() +``` + +Runnable example: [`web_form.py`](../../examples/selenium/python-test/web_form.py), +the three-line version above. From the repo root, after `pip install -e` above and +a `pnpm build` so the backend exists: + +```bash +pnpm demo:python +``` + +Unlike `demo:wdio` and friends, this needs the Python package installed and uses +whatever `python3` resolves to on your PATH. +If the backend can't be launched or reached, `enable()` warns and returns +`None` β€” capture is skipped, your tests still run. + +**ChromeDriver:** you need one matching your Chrome (a mismatch breaks all +Selenium, not just this). Selenium 4.6+ auto-manages it when no `chromedriver` +is on `PATH`; otherwise keep it current (`brew upgrade chromedriver`). + +## What it captures + +| Data | How | Scope | Phase | +|---|---|---|---| +| Commands (driver + element) | wrap `WebDriver.execute()` β€” the single chokepoint all commands flow through | `commands` | 1 | +| Session metadata | read `session_id` + `caps` on the first ready command | `metadata` | 1 | +| Test / suite tree | pytest plugin (`pytest_runtest_logreport` / `sessionfinish`) | `suites` | 1 | +| Browser console + JS errors | Selenium **BiDi** (`driver.script` handlers) | `consoleLogs` | 2 | +| Network requests | Selenium **BiDi** (low-level subscribe, no interception) | `networkRequests` | 2 | +| DOM snapshot (preview iframe) | inject `packages/script`, re-inject per navigation, drain mutations | `mutations` | 2 | +| Screencast video | screenshot polling β†’ ffmpeg-encoded `.webm` | `screencast` | 2 | + +Element actions (`click`, `send_keys`, `text`, …) are captured for free: they +delegate to `self._parent.execute`, so the one wrapper sees them as +`clickElement`, `getElementText`, etc. + +**BiDi is auto-enabled** β€” the adapter injects the `webSocketUrl` capability +into the `newSession` request so console/network work out-of-box (opt out with +`DEVTOOLS_BIDI=0`). **Screencast** needs `ffmpeg` on PATH to encode the +`.webm`; without it, recording is skipped (one warning, no error). + +## Dashboard window lifecycle + +Like the JS adapters, `enable()` opens the dashboard in a dedicated, closable +Chrome window; closing that window (backend `clientDisconnected`) shuts the run +down, and ending the process (exit / Ctrl-C) closes the window. Auto-open is on +when stdout is a TTY; force it with `DEVTOOLS_OPEN=1` or disable with `=0`. + +## Layout + +``` +src/selenium_devtools/ + __init__.py public API β€” enable() / disable() / get_capturer() + constants.py defaults, env-var names, skip sets, pinned backend version + types.py TypedDicts for the wire payloads (mirror packages/shared) + _contract.py GENERATED from packages/shared β€” scope names + CONTRACT_VERSION + utils.py framework-agnostic helpers (now_ms, iso, to_jsonable, call_source) + frames.py pure builders for each {scope,data} payload + transport.py stdlib WebSocket client (handshake, masked frames, ping/pong, control reader) + capturer.py SessionCapturer: command IDs, normalizeβ†’send, metadata-once + instrumentation.py execute() wrap + BiDi auto-enable + session-setup hook + bidi.py BiDi console/JS-error + network capture (pure mapping + wiring) + screencast.py screenshot-polling recorder + ffmpeg webm encode + backend.py launch-or-attach the Node backend + port discovery + lifecycle.py dashboard window open/close + shutdown-on-disconnect + pytest_plugin.py suite/test tree feeder (opt-in) +scripts/gen_contract.py regenerate _contract.py from shared (dev-time; also a drift-guard) +tests/ stdlib-unittest unit tests (no selenium/pytest needed) +e2e_check.py real-Chrome smoke (plain script) +e2e/test_smoke.py real-Chrome smoke (pytest + plugin) +(example lives at repo root: examples/selenium/python-test/web_form.py) +``` + +## Backend & publishing + +Two artifacts, two registries β€” pip can't resolve the Node backend, so each +coupling is handled explicitly rather than via a `workspace:^`-style resolver: + +| | Local (monorepo) | Published | +|---|---|---| +| **Adapter** (this package) | `pip install -e` | PyPI: `pip install selenium-devtools-py` | +| **Backend + UI** (Node) | `node packages/backend/dist/server.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/server.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-devtools-py/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 selenium_devtools.pytest_plugin -q +``` + +## Release (approach A) + +Two workflows, mirroring the JS split (`ci.yml` tests / `release.yml` publish): + +- **`python.yml`** β€” runs on PRs + pushes touching this package or `shared`: + unit tests on Python 3.9 + 3.12, and a contract-drift check (regenerate + `_contract.py`, fail on any diff). Zero repo config needed. +- **`python-release.yml`** β€” **manual** (`workflow_dispatch`, like the JS + "Manual NPM Publish"), target `pypi` or `testpypi`. Builds the sdist + wheel + and publishes via **trusted publishing (OIDC)** β€” no token/secret. + +The wheel does **not** bundle the backend β€” approach A fetches a pinned +`@wdio/devtools-backend` via `npx` at runtime (Node 18+ required). Bundling it +(approach B/C) is a GA-time change. + +**One-time setup before the first publish** (this is what claims the PyPI name): + +1. On PyPI, add a **pending trusted publisher** for project + `selenium-devtools-py` β†’ owner `webdriverio`, repo `devtools`, workflow + `python-release.yml`, environment `pypi` (repeat on TestPyPI with env + `testpypi` if you want a dry run first). +2. Create matching GitHub **Environments** `pypi` (and `testpypi`). +3. Run the workflow β€” the first successful publish creates and claims the name. + +Each release: bump `version` in `pyproject.toml`, then run the workflow (PyPI +rejects re-uploading an existing version). + +## Roadmap + +- **Phase 2 (done)** β€” BiDi console/network + screenshot-polling screencast. + Not yet: a CDP `Page.startScreencast` push-mode fast-path, per-command + screenshots, and performance capture. +- **Phase 3** β€” trace export, preserve-and-rerun, action snapshots. Per the + architecture, the heavy post-processing is a candidate to live server-side in + the backend (written once) rather than re-implemented here. + +## Design notes + +- **Backend/UI unchanged.** This adapter only produces the wire frames; the + server routes and renders them exactly as for the JS adapters. +- **Capture never breaks tests.** Commands are recorded around the real call; + errors are captured *and re-raised* unchanged; a missing dashboard is a no-op. +- **Contract drift** is the main long-term risk (see the integration artifact). + Mitigated two ways: `_contract.py` is generated from `packages/shared` (scope + names + `CONTRACT_VERSION`), and the generator fails if a required scope + vanishes. Full field-level type generation is a future step. diff --git a/packages/selenium-devtools-py/e2e/test_smoke.py b/packages/selenium-devtools-py/e2e/test_smoke.py new file mode 100644 index 00000000..4b4d647d --- /dev/null +++ b/packages/selenium-devtools-py/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 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-devtools-py/e2e_check.py b/packages/selenium-devtools-py/e2e_check.py new file mode 100644 index 00000000..7e64c2db --- /dev/null +++ b/packages/selenium-devtools-py/e2e_check.py @@ -0,0 +1,47 @@ +#!/usr/bin/env python3 +"""End-to-end smoke: real headless Chrome β†’ adapter β†’ backend. + +Run against a running backend (set DEVTOOLS_PORT if not 3000): + DEVTOOLS_PORT=63763 PYTHONPATH=src python3 e2e_check.py + +Uses a data: URL so it needs no network. +""" + +import sys +import time + +import 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-devtools-py/pyproject.toml b/packages/selenium-devtools-py/pyproject.toml new file mode 100644 index 00000000..8b78481b --- /dev/null +++ b/packages/selenium-devtools-py/pyproject.toml @@ -0,0 +1,28 @@ +[build-system] +requires = ["hatchling"] +build-backend = "hatchling.build" + +[project] +name = "selenium-devtools-py" +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 DEVTOOLS_ENABLE / DEVTOOLS_PORT is set. +[project.entry-points.pytest11] +selenium_devtools = "selenium_devtools.pytest_plugin" + +[tool.hatch.build.targets.wheel] +packages = ["src/selenium_devtools"] diff --git a/packages/selenium-devtools-py/scripts/gen_contract.py b/packages/selenium-devtools-py/scripts/gen_contract.py new file mode 100644 index 00000000..925eb089 --- /dev/null +++ b/packages/selenium-devtools-py/scripts/gen_contract.py @@ -0,0 +1,98 @@ +#!/usr/bin/env python3 +"""Generate ``_contract.py`` from ``packages/shared`` β€” the Python side of the +wire contract, derived from the single TS source of truth. + +Runs only in the monorepo (dev time); the generated file is committed and ships +inside the wheel, so published installs need neither this script nor `shared`. + +It doubles as a drift-guard: if a scope the adapter relies on disappears from +shared's ``TraceLog`` / ``WS_SCOPE``, generation fails loudly rather than +letting the Python side silently send a name the UI no longer understands. + +Run: python3 scripts/gen_contract.py +""" + +from __future__ import annotations + +import json +import re +import sys +from pathlib import Path + +# Data scopes the Python adapter emits β€” each must exist as a TraceLog key. +REQUIRED_DATA_SCOPES = { + "SCOPE_METADATA": "metadata", + "SCOPE_COMMANDS": "commands", + "SCOPE_CONSOLE_LOGS": "consoleLogs", + "SCOPE_NETWORK_REQUESTS": "networkRequests", + "SCOPE_SUITES": "suites", + "SCOPE_SCREENCAST": "screencast", + "SCOPE_SOURCES": "sources", + "SCOPE_MUTATIONS": "mutations", +} + + +def _repo_root() -> Path: + for parent in Path(__file__).resolve().parents: + if (parent / "packages" / "shared" / "package.json").exists(): + return parent + raise SystemExit("could not locate the monorepo root (packages/shared)") + + +def _shared_version(shared: Path) -> str: + pkg = json.loads((shared / "package.json").read_text()) + return pkg["version"] + + +def _trace_log_keys(types_ts: str) -> list[str]: + m = re.search(r"export interface TraceLog \{(.*?)\n\}", types_ts, re.DOTALL) + if not m: + raise SystemExit("could not find `interface TraceLog` in shared/types.ts") + return re.findall(r"^\s*(\w+)\??:", m.group(1), re.MULTILINE) + + +def _ws_scopes(routes_ts: str) -> dict[str, str]: + m = re.search(r"export const WS_SCOPE = \{(.*?)\n\} as const", routes_ts, re.DOTALL) + if not m: + raise SystemExit("could not find `WS_SCOPE` in shared/routes.ts") + return dict(re.findall(r"(\w+):\s*'([^']+)'", m.group(1))) + + +def main() -> int: + root = _repo_root() + shared = root / "packages" / "shared" + version = _shared_version(shared) + data_keys = _trace_log_keys((shared / "src" / "types.ts").read_text()) + control = _ws_scopes((shared / "src" / "routes.ts").read_text()) + + # Drift-guard. + missing = [v for v in REQUIRED_DATA_SCOPES.values() if v not in data_keys] + if missing: + raise SystemExit( + f"contract drift: scope(s) {missing} no longer in shared TraceLog " + f"(present: {data_keys}). Update the adapter or shared." + ) + + lines = [ + "# GENERATED by scripts/gen_contract.py from packages/shared.", + "# Do not edit by hand β€” run the script to regenerate.", + f'CONTRACT_VERSION = "{version}"', + "", + ] + for const, value in REQUIRED_DATA_SCOPES.items(): + lines.append(f'{const} = "{value}"') + lines += [ + "", + f"DATA_SCOPES = frozenset({sorted(data_keys)!r})", + f"CONTROL_SCOPES = frozenset({sorted(control.values())!r})", + "", + ] + out = shared.parent / "selenium-devtools-py" / "src" / "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-devtools-py/src/selenium_devtools/__init__.py b/packages/selenium-devtools-py/src/selenium_devtools/__init__.py new file mode 100644 index 00000000..4a7632d4 --- /dev/null +++ b/packages/selenium-devtools-py/src/selenium_devtools/__init__.py @@ -0,0 +1,148 @@ +"""selenium-devtools-py β€” Python Selenium adapter for the DevTools dashboard. + +Public API: + + import 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 +``DEVTOOLS_ENABLE`` / ``DEVTOOLS_PORT`` env vars). The transport has no +third-party dependency; the only requirement on top is selenium itself. +""" + +from __future__ import annotations + +import os +import subprocess +import sys +from typing import Optional + +from . import backend, instrumentation, lifecycle +from ._contract import CONTRACT_VERSION +from .capturer import SessionCapturer +from .constants import DEFAULT_HOST, DEFAULT_PORT, ENV_HOST, ENV_PORT +from .logcapture import LogCapturer +from .terminal import TerminalCapturer +from .transport import WSClient + +__version__ = "0.1.0" +__all__ = [ + "enable", "disable", "get_capturer", "dashboard_url", + "wait_for_dashboard_close", "CONTRACT_VERSION", +] + +_active: dict = { + "capturer": None, "transport": None, "process": None, "url": None, + "handle": None, "terminal": None, "logs": None, +} + + +def enable( + host: Optional[str] = None, + port: Optional[int] = None, + *, + webdriver_cls: Optional[type] = None, +) -> Optional[SessionCapturer]: + """Connect to the backend and instrument Selenium. Idempotent. + + With no host/port and no ``DEVTOOLS_PORT``, the backend is launched + automatically (see :mod:`.backend`). Returns the SessionCapturer, or None if + the dashboard can't be reached/launched β€” a missing dashboard must never + break the user's test run. + """ + if _active["capturer"] is not None: + return _active["capturer"] + + process = None + try: + if host is not None or port is not None: + host = host or os.environ.get(ENV_HOST, DEFAULT_HOST) + port = int(port or os.environ.get(ENV_PORT, DEFAULT_PORT)) + else: + host, port, process = backend.launch_or_attach() + except (OSError, RuntimeError, TimeoutError) as exc: + print(f"[devtools] could not start dashboard ({exc}); " + f"continuing without capture", file=sys.stderr) + return None + + transport = WSClient(host, port, on_control=lifecycle.on_control) + try: + transport.connect() + except OSError as exc: + print( + f"[devtools] dashboard not reachable at {host}:{port} " + f"({exc}); continuing without capture", + file=sys.stderr, + ) + if process is not None: + process.terminate() + return None + + capturer = SessionCapturer(transport) + instrumentation.install(capturer, webdriver_cls) + # Surface the runner's output in the dashboard Console: Python logging + # (selenium + the adapter's own events) and the test's stdout. + logs = LogCapturer(capturer) + logs.start() + term = TerminalCapturer(capturer) + term.start() + url = f"http://{host}:{port}" + _active.update( + capturer=capturer, transport=transport, process=process, url=url, + terminal=term, logs=logs, + ) + + # Open the dashboard window and wire exit/signal + control-frame teardown so + # closing the window (clientDisconnected) or ending the process both tidy up. + handle = lifecycle.open_dashboard(url) if lifecycle.auto_open_enabled() else None + _active["handle"] = handle + lifecycle.register_exit_handlers(disable, handle) + return capturer + + +def disable() -> None: + # Close the dashboard window + unregister exit/signal handlers first, so a + # re-enable() starts clean. Idempotent and defensive β€” never raises. + lifecycle.unregister_exit_handlers() + instrumentation.uninstall() + term = _active["terminal"] + if term is not None: # restore stdout/stderr before tearing the transport down + term.stop() + logs = _active["logs"] + if logs is not None: # detach the logging handler + restore logger levels + logs.stop() + transport = _active["transport"] + if transport is not None: + transport.close() + process = _active["process"] + if process is not None: # only set when we launched it ourselves + process.terminate() + try: + process.wait(timeout=5) + except subprocess.TimeoutExpired: + process.kill() # backend ignored SIGTERM β€” force it + _active.update( + capturer=None, transport=None, process=None, url=None, handle=None, + terminal=None, logs=None, + ) + + +def get_capturer() -> Optional[SessionCapturer]: + return _active["capturer"] + + +def dashboard_url() -> Optional[str]: + """URL of the connected dashboard, or None if capture isn't active.""" + return _active["url"] + + +def wait_for_dashboard_close() -> None: + """Block until the user closes the dashboard window, so you can inspect the + run after your test finishes. Returns immediately if no dashboard window is + open (headless/CI) β€” safe to always call before ``disable()``.""" + if lifecycle.dashboard_window_open(): + print(f"[devtools] dashboard live at {dashboard_url()} β€” " + "close the window to finish.", file=sys.stderr) + lifecycle.wait_for_shutdown() diff --git a/packages/selenium-devtools-py/src/selenium_devtools/_contract.py b/packages/selenium-devtools-py/src/selenium_devtools/_contract.py new file mode 100644 index 00000000..973c8fb9 --- /dev/null +++ b/packages/selenium-devtools-py/src/selenium_devtools/_contract.py @@ -0,0 +1,15 @@ +# GENERATED by scripts/gen_contract.py from packages/shared. +# Do not edit by hand β€” run the script to regenerate. +CONTRACT_VERSION = "1.0.0" + +SCOPE_METADATA = "metadata" +SCOPE_COMMANDS = "commands" +SCOPE_CONSOLE_LOGS = "consoleLogs" +SCOPE_NETWORK_REQUESTS = "networkRequests" +SCOPE_SUITES = "suites" +SCOPE_SCREENCAST = "screencast" +SCOPE_SOURCES = "sources" +SCOPE_MUTATIONS = "mutations" + +DATA_SCOPES = frozenset(['actionSnapshots', 'commands', 'config', 'consoleLogs', 'logs', 'metadata', 'mutations', 'networkRequests', 'screencast', 'screencastFrames', 'sources', 'suites']) +CONTROL_SCOPES = frozenset(['clearCommands', 'clearExecutionData', 'clientConnected', 'clientDisconnected', 'config', 'replaceCommand', 'testStopped']) diff --git a/packages/selenium-devtools-py/src/selenium_devtools/backend.py b/packages/selenium-devtools-py/src/selenium_devtools/backend.py new file mode 100644 index 00000000..9f1bfe17 --- /dev/null +++ b/packages/selenium-devtools-py/src/selenium_devtools/backend.py @@ -0,0 +1,122 @@ +"""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/server.js (LOCAL dev) + 4. else β†’ npx @wdio/devtools-backend@ (PUBLISHED) + +The pinned version below is bumped deliberately alongside a contract change β€” +there is no auto-resolution, so this constant *is* the version link. +""" + +from __future__ import annotations + +import os +import re +import shlex +import shutil +import subprocess +import threading +import time +from pathlib import Path +from typing import List, Optional, Tuple + +from .constants import ( + BACKEND_NPM_PACKAGE, + BACKEND_NPM_VERSION, + BACKEND_SPAWN_TIMEOUT_S, + DEFAULT_HOST, + ENV_BACKEND_CMD, + ENV_HOST, + ENV_PORT, +) + +# Match the ACTUAL bound port from Fastify's "Server listening at http://…:PORT" +# line β€” NOT the earlier "Starting … on port 3000" line, which is only the +# *preferred* port. When 3000 is busy the backend negotiates a different port, +# so keying off the preferred port connects to the wrong (or a dead) socket. +# Greedy `.*` so the IPv6 form (http://[::1]:PORT) resolves to the final :PORT. +_PORT_RE = re.compile(r"listening at .*:(\d+)") + + +def _find_monorepo_backend(start: Optional[Path] = None) -> Optional[Path]: + """Walk up from ``start`` (default: this module) for a built backend. Present + only in a monorepo checkout; None from an installed wheel. + + Targets ``server.js``, the backend's CLI entry, NOT ``index.js``: that one is + the library entry the JS adapters import, and running it starts nothing.""" + base = start or Path(__file__).resolve() + for parent in base.parents: + candidate = parent / "packages" / "backend" / "dist" / "server.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-devtools-py/src/selenium_devtools/bidi.py b/packages/selenium-devtools-py/src/selenium_devtools/bidi.py new file mode 100644 index 00000000..9c311529 --- /dev/null +++ b/packages/selenium-devtools-py/src/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"[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-devtools-py/src/selenium_devtools/capturer.py b/packages/selenium-devtools-py/src/selenium_devtools/capturer.py new file mode 100644 index 00000000..9cc65798 --- /dev/null +++ b/packages/selenium-devtools-py/src/selenium_devtools/capturer.py @@ -0,0 +1,159 @@ +"""Session capturer β€” the Python analogue of core's ``SessionCapturerBase``. + +Owns the transport, a per-session command counter, and the normalizeβ†’send +path. Deliberately thin: it turns already-captured data into wire frames and +pushes them. No driver knowledge lives here (that's instrumentation.py); no +post-processing lives here (that's the backend's job, per the architecture). +""" + +from __future__ import annotations + +import threading +from typing import Any, List, Optional, Protocol + +from . import frames +from ._contract import ( + SCOPE_COMMANDS, + SCOPE_CONSOLE_LOGS, + SCOPE_METADATA, + SCOPE_MUTATIONS, + SCOPE_NETWORK_REQUESTS, + SCOPE_SCREENCAST, + SCOPE_SOURCES, + SCOPE_SUITES, +) +from .types import SuiteStats +from .utils import now_ms, to_jsonable + + +class Transport(Protocol): + connected: bool + + def send_json(self, scope: str, data: Any) -> bool: ... + def close(self) -> None: ... + + +class SessionCapturer: + def __init__(self, transport: Transport) -> None: + self._tx = transport + self._command_counter = 0 + self._lock = threading.Lock() + self._metadata_sent: set = set() # session ids already announced + self.session_id: Optional[str] = None + + # ── metadata ─────────────────────────────────────────────────────────────── + + def ensure_metadata( + self, session_id: str, capabilities: Optional[dict], url: Optional[str] + ) -> None: + """Announce a session once. Keyed by id, not a boolean: one process can + drive several sessions (a function-scoped pytest fixture makes a driver + per test), and the UI keys metadata by sessionId so each one is kept + rather than overwriting the first.""" + if not session_id or session_id in self._metadata_sent: + return + self._metadata_sent.add(session_id) + self.session_id = session_id + self._tx.send_json( + SCOPE_METADATA, + frames.metadata(session_id, to_jsonable(capabilities or {}), url), + ) + + # ── commands ─────────────────────────────────────────────────────────────── + + def capture_command( + self, + *, + command: str, + args: Any, + result: Any = None, + error: Optional[BaseException] = None, + start_time: int, + call_source: Optional[str], + screenshot: Optional[str] = None, + ) -> None: + with self._lock: + self._command_counter += 1 + command_id = self._command_counter + norm_args = args if isinstance(args, list) else ([] if args is None else [args]) + entry = frames.command_log( + command=command, + args=to_jsonable(norm_args), + result=to_jsonable(result), + error=error, + timestamp=now_ms(), + start_time=start_time, + call_source=call_source, + command_id=command_id, + screenshot=screenshot, + ) + self._tx.send_json(SCOPE_COMMANDS, [entry]) + + # ── console / network ──────────────────────────────────────────────────────── + + def capture_console(self, level: str, args: List[Any], source: str = "browser") -> None: + self._tx.send_json( + SCOPE_CONSOLE_LOGS, + [frames.console_log(level=level, args=to_jsonable(args), + timestamp=now_ms(), source=source)], + ) + + def capture_network(self, **kwargs: Any) -> None: + self._tx.send_json(SCOPE_NETWORK_REQUESTS, [frames.network_request(**kwargs)]) + + # ── screencast ───────────────────────────────────────────────────────────── + + def send_screencast( + self, + *, + video_path: str, + video_file: str, + frame_count: int, + duration: int, + start_time: Optional[int], + session_id: Optional[str] = None, + ) -> None: + # Keyed to the session the frames came from. `session_id` is passed + # explicitly when finalizing a session that has already been replaced, + # so the video is not filed under whichever session is current. + target = session_id or self.session_id + if not target: + return + self._tx.send_json( + SCOPE_SCREENCAST, + frames.screencast( + session_id=target, + video_path=video_path, + video_file=video_file, + frame_count=frame_count, + duration=duration, + start_time=start_time, + ), + ) + + # ── sources / mutations ────────────────────────────────────────────────────── + + def send_sources(self, sources: dict) -> None: + """Send a ``{absolute_path: source_text}`` map β€” the Source tab shows the + file a command's ``callSource`` points at. Empty map is a no-op.""" + if sources: + self._tx.send_json(SCOPE_SOURCES, sources) + + def send_mutations(self, mutations: List[Any]) -> None: + """Forward DOM mutations captured by the injected browser script + (packages/script) β€” the UI renders them into the snapshot iframe.""" + if mutations: + self._tx.send_json(SCOPE_MUTATIONS, mutations) + + # ── suites ─────────────────────────────────────────────────────────────────── + + def send_suites(self, suites: List[SuiteStats]) -> None: + # The UI expects Record[] β€” one single-key record per + # suite β€” mirroring core's TestReporterBase.sendUpstream. A plain array + # of suites won't render. Empty-payload guard matches the JS behavior. + payload = [{s["uid"]: s} for s in suites if s.get("uid")] + if payload: + self._tx.send_json(SCOPE_SUITES, payload) + + def close(self) -> None: + self._tx.close() diff --git a/packages/selenium-devtools-py/src/selenium_devtools/constants.py b/packages/selenium-devtools-py/src/selenium_devtools/constants.py new file mode 100644 index 00000000..10046d1f --- /dev/null +++ b/packages/selenium-devtools-py/src/selenium_devtools/constants.py @@ -0,0 +1,81 @@ +"""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 = "DEVTOOLS_ENABLE" +ENV_BIDI = "DEVTOOLS_BIDI" # "0"/"false"/"no"/"off" disables BiDi auto-enable +ENV_OPEN = "DEVTOOLS_OPEN" # "0"/"false"/"no"/"off" disables dashboard auto-open + +# ── Backend launch ─────────────────────────────────────────────────────────── +# Pinned backend version fetched via npx from a published install. There is no +# cross-ecosystem resolver, so this constant *is* the version link β€” bump it in +# the same change that regenerates _contract.py. 1.10.0 is the first version +# carrying the `devtools-backend` bin; `npx` cannot start anything older. +BACKEND_NPM_VERSION = "1.10.0" +BACKEND_NPM_PACKAGE = "@wdio/devtools-backend" +BACKEND_SPAWN_TIMEOUT_S = 40.0 + +# ── Instrumentation ────────────────────────────────────────────────────────── +# Selenium commands that are bookkeeping/noise rather than user-meaningful. +# `screenshot`/`elementScreenshot` are skipped so the screencast recorder's +# per-command frame capture (get_screenshot_as_base64) doesn't flood the +# Actions timeline. +SKIP_COMMANDS = frozenset( + {"newSession", "quit", "status", "getLog", "getAllSessions", "getSessions", + "screenshot", "elementScreenshot"} +) + +# Stack-frame path fragment to skip when resolving a command's call source β€” +# the adapter's own package. The selenium library dir is added at runtime by +# instrumentation (resolved from selenium.__file__), NOT matched by the +# substring "/selenium/" β€” that would wrongly skip a user's own test file living +# under a path like examples/selenium/... . +_PACKAGE_DIR = os.path.dirname(__file__) +SKIP_STACK_FRAMES = (_PACKAGE_DIR,) + +# ── BiDi ────────────────────────────────────────────────────────────────────── +# The capability the driver must advertise for selenium's BiDi channel to open +# (set via ``options.web_socket_url = True`` at build time). Without it, +# accessing ``driver.script`` / ``driver.network`` raises β€” attach() degrades. +BIDI_CAPABILITY = "webSocketUrl" +# BiDi network event names we subscribe to WITHOUT interception β€” a plain +# session.subscribe, so requests are observed but never paused (interception +# would stall the user's page loads if a callback failed to continue them). +BIDI_NET_BEFORE_REQUEST = "network.beforeRequestSent" +BIDI_NET_RESPONSE_COMPLETED = "network.responseCompleted" +# selenium's BiDi log entries already carry lowercase levels; this normalizes +# the stragglers to the shared LogLevel union. Unmapped levels fall back to log. +BIDI_LEVEL_MAP = { + "debug": "debug", + "info": "info", + "warn": "warn", + "warning": "warn", + "error": "error", + "severe": "error", + "log": "log", + "trace": "trace", +} + +# ── Screencast ──────────────────────────────────────────────────────────────── +# Frames are captured synchronously (one per command) on the main thread β€” see +# screencast.py for why a background poll thread is avoided. Screenshots via +# WebDriver are always PNG. +SCREENCAST_IMAGE_FORMAT = "png" +# Skip encoding below this many frames β€” a single still isn't a video. +SCREENCAST_MIN_FRAMES = 2 +# Output filename stem; the session id + .webm suffix are appended. +SCREENCAST_FILENAME_PREFIX = "selenium-py-video" +# The `screencast` wire scope is generated into _contract.py (SCOPE_SCREENCAST). diff --git a/packages/selenium-devtools-py/src/selenium_devtools/frames.py b/packages/selenium-devtools-py/src/selenium_devtools/frames.py new file mode 100644 index 00000000..b5deeb48 --- /dev/null +++ b/packages/selenium-devtools-py/src/selenium_devtools/frames.py @@ -0,0 +1,193 @@ +"""Builders for the wire frames the dashboard renders. + +Each function returns the ``data`` payload for a ``{scope, data}`` frame. +Shapes mirror ``packages/shared/src/types.ts``. Keeping them here β€” pure and +side-effect free β€” makes them unit-testable and the single place the contract +lives on the Python side. +""" + +from __future__ import annotations + +from typing import Any, List, Optional + +from .types import ( + CommandLog, + ConsoleLog, + Metadata, + NetworkRequest, + ScreencastInfo, + SuiteStats, + TestStats, +) +from .utils import iso + + +def metadata( + session_id: str, + capabilities: Optional[dict] = None, + url: Optional[str] = None, +) -> Metadata: + caps = capabilities or {} + return { + "type": "testrunner", # TraceType.Testrunner + "sessionId": session_id, + "url": url, + "capabilities": caps, + "desiredCapabilities": caps, + "testEnv": "python-selenium", + } + + +def command_log( + *, + command: str, + args: List[Any], + result: Any = None, + error: Optional[BaseException] = None, + timestamp: int, + start_time: int, + call_source: Optional[str], + command_id: int, + screenshot: Optional[str] = None, +) -> CommandLog: + entry: CommandLog = { + "command": command, + "args": args, + "result": result, + "timestamp": timestamp, + "startTime": start_time, + "callSource": call_source, + "id": command_id, + } + if error is not None: + entry["error"] = { + "name": type(error).__name__, + "message": str(error), + } + if screenshot: + entry["screenshot"] = screenshot + return entry + + +def console_log( + *, level: str, args: List[Any], timestamp: int, source: str = "browser" +) -> ConsoleLog: + return {"type": level, "args": args, "timestamp": timestamp, "source": source} + + +def network_request( + *, + request_id: str, + url: str, + method: str, + timestamp: int, + start_time: int, + status: Optional[int] = None, + request_type: str = "fetch", + end_time: Optional[int] = None, + status_text: Optional[str] = None, + time: Optional[int] = None, + size: Optional[int] = None, + request_headers: Optional[dict] = None, + response_headers: Optional[dict] = None, +) -> NetworkRequest: + entry: NetworkRequest = { + "id": request_id, + "url": url, + "method": method, + "status": status, + "timestamp": timestamp, + "startTime": start_time, + "endTime": end_time, + "type": request_type, + } + # Response-phase fields are absent on the initial request frame; omit rather + # than send nulls so the dashboard's "pending" state renders correctly. + if status_text is not None: + entry["statusText"] = status_text + if time is not None: + entry["time"] = time + if size is not None: + entry["size"] = size + if request_headers is not None: + entry["requestHeaders"] = request_headers + if response_headers is not None: + entry["responseHeaders"] = response_headers + return entry + + +def screencast( + *, + session_id: str, + video_path: str, + video_file: str, + frame_count: int, + duration: int, + start_time: Optional[int], +) -> ScreencastInfo: + info: ScreencastInfo = { + "sessionId": session_id, + "videoPath": video_path, + "videoFile": video_file, + "frameCount": frame_count, + "duration": duration, + } + if start_time is not None: + info["startTime"] = start_time + return info + + +def test_stats( + *, + uid: str, + title: str, + full_title: str, + parent: str, + state: str, + file: str, + start_ms: int, + end_ms: int, + call_source: Optional[str] = None, +) -> TestStats: + return { + "uid": uid, + "cid": "0-0", + "title": title, + "fullTitle": full_title, + "parent": parent, + "state": state, + "start": iso(start_ms), + "end": iso(end_ms), + "type": "test", + "file": file, + "retries": 0, + "_duration": max(0, end_ms - start_ms), + "callSource": call_source, + } + + +def suite_stats( + *, + uid: str, + title: str, + file: str, + start_ms: int, + tests: List[TestStats], + end_ms: Optional[int] = None, + state: Optional[str] = None, +) -> SuiteStats: + return { + "uid": uid, + "cid": "0-0", + "title": title, + "fullTitle": title, + "type": "suite", + "file": file, + "start": iso(start_ms), + "end": iso(end_ms) if end_ms is not None else None, + "state": state, + "tests": tests, + "suites": [], + "hooks": [], + "_duration": max(0, (end_ms - start_ms)) if end_ms is not None else 0, + } diff --git a/packages/selenium-devtools-py/src/selenium_devtools/instrumentation.py b/packages/selenium-devtools-py/src/selenium_devtools/instrumentation.py new file mode 100644 index 00000000..4a8dc104 --- /dev/null +++ b/packages/selenium-devtools-py/src/selenium_devtools/instrumentation.py @@ -0,0 +1,411 @@ +"""Driver instrumentation β€” the one genuinely per-language piece. + +Every Selenium command (driver- *and* element-level, since element methods +delegate to ``self._parent.execute``) funnels through +``WebDriver.execute(driver_command, params)``. Wrapping that single chokepoint +captures the whole command stream from one place β€” cleaner than the JS +adapter's prototype patching. + +The patch target is injected so the module imports and unit-tests without +selenium present; ``install`` defaults to the real selenium class. +""" + +from __future__ import annotations + +import logging +import os +import sys +import threading +from typing import Any, Optional + +from . import bidi, frames +from .capturer import SessionCapturer +from .constants import BIDI_CAPABILITY, ENV_BIDI, SKIP_COMMANDS, SKIP_STACK_FRAMES +from .output_dir import resolve_adapter_output_dir +from .screencast import ScreencastRecorder +from .snapshot import SnapshotCapturer, start_snapshot_capture +from .sources import read_source +from .utils import call_source, now_ms + +# Operational logging β€” surfaced in the dashboard Console (the 'runner' stream). +_log = logging.getLogger("selenium_devtools") + +# Marks the adapter's OWN execute_script calls (snapshot inject/readback) so +# patched_execute skips capturing them as user commands. +_internal = threading.local() + +# User-file paths whose source we've already sent (once per file per run). +_sources_sent: set = set() + + +_skip_frames_cache: Optional[tuple] = None + + +def _skip_frames() -> tuple: + """Call-source skip fragments: the adapter package + the REAL selenium + library dir (resolved from selenium.__file__), cached. Resolving the actual + package dir avoids skipping a user test file whose path merely contains + 'selenium' (e.g. examples/selenium/...).""" + global _skip_frames_cache + if _skip_frames_cache is None: + try: + import selenium + + extra = (os.path.dirname(os.path.abspath(selenium.__file__)) + os.sep,) + except Exception: # noqa: BLE001 β€” narrow fallback if selenium isn't importable + extra = (f"{os.sep}selenium{os.sep}webdriver{os.sep}",) + _skip_frames_cache = tuple(SKIP_STACK_FRAMES) + extra + return _skip_frames_cache + + +def _internal_active() -> bool: + return getattr(_internal, "active", False) + + +# When a test framework (the pytest plugin) reports the suite tree, the adapter +# must NOT also synthesize a default one. The plugin flips this in its configure. +_external_suites = False + + +def set_external_suites(value: bool = True) -> None: + """Tell the adapter a test framework owns the suite tree (suppresses the + default single-session suite used for plain scripts).""" + global _external_suites + _external_suites = value + + +def _send_default_suite(capturer: SessionCapturer, state: str) -> None: + """Report the run as a single suite/test named after the entry script, so a + plain-script run (no test framework) still shows in the TESTS tree. No-op if + a framework is reporting suites.""" + if _external_suites: + return + entry = os.path.abspath(sys.argv[0]) if sys.argv and sys.argv[0] else "" + title = os.path.basename(entry) or "Selenium session" + ds = _state.get("default_suite") + start = ds["start"] if ds else now_ms() + _state["default_suite"] = {"start": start} + end = start if state == "running" else now_ms() + test = frames.test_stats( + uid=f"{entry or title}::session", title=title, full_title=title, + parent=title, state=state, file=entry, start_ms=start, end_ms=end, + ) + suite = frames.suite_stats( + uid=entry or title, title=title, file=entry, start_ms=start, + tests=[test], state=state, end_ms=(None if state == "running" else end), + ) + capturer.send_suites([suite]) + + +def _guarded_execute_script(driver: Any) -> Any: + """An ``execute_script`` that runs WITHOUT command capture β€” for the + adapter's own snapshot injection/readback, which must never show up in the + Actions timeline (the getTraceData/inject scripts are not user commands).""" + + def run(script: str, *args: Any) -> Any: + _internal.active = True + try: + return driver.execute_script(script, *args) + finally: + _internal.active = False + + return run + + +def _capture_source(capturer: SessionCapturer, call_src: Optional[str]) -> None: + """Send the source of the file a command's ``callSource`` points at (once). + + Keyed by the exact ``callSource`` path so the Source tab always matches β€” + and works for any runner (plain script or pytest), not just the plugin. + Guarded: source capture must never break the user's test.""" + if not call_src: + return + path = call_src.rsplit(":", 1)[0] # strip the trailing ":line" + if not path: + return + # Screencast output lands next to the (first) test file, not the cwd. + if _state.get("output_dir") is None: + _state["output_dir"] = resolve_adapter_output_dir(test_file_path=path) + if path in _sources_sent: + return + try: + text = read_source(path) + except Exception: # noqa: BLE001 + return + if text is None: + return + _sources_sent.add(path) + capturer.send_sources({path: text}) + +# Selenium command names that change the document β€” after these we drain the +# page-side mutation buffer so the snapshot iframe stays current. +_state: dict = { + "installed": False, "cls": None, "orig": None, + "screencast": None, "snapshot": None, + # Session id the per-session bringup last ran for. A boolean here silently + # skipped bringup for every session after the first. + "armed_session": None, + "output_dir": None, # test-results dir beside the test file (see output_dir.py) + "default_suite": None, # synthesized suite for non-framework (script) runs +} + + +def _take_screenshot(driver: Any) -> Optional[str]: + """One base64 PNG of the current page on the MAIN thread β€” no background + thread, so we never touch the Selenium session concurrently. Reused for both + the command entry (per-command snapshot view) and the screencast frame. + Best-effort: a transient failure (mid-navigation, dead session) returns + None and never breaks the test. ``screenshot`` is in SKIP_COMMANDS, so this + doesn't appear in the Actions timeline.""" + fn = getattr(driver, "get_screenshot_as_base64", None) + if not callable(fn): + return None + try: + shot = fn() + except Exception: # noqa: BLE001 β€” transient; skip this frame + return None + return shot if isinstance(shot, str) and shot else None + + +def _add_screencast_frame(shot: Optional[str]) -> None: + """Buffer an already-captured screenshot as a screencast frame.""" + recorder = _state.get("screencast") + if recorder is None or not shot: + return + try: + recorder.add_frame(shot) + except Exception as exc: # noqa: BLE001 β€” never break the test + _log.warning("screencast add_frame threw: %s", exc) + + +def _refresh_snapshot(capturer: SessionCapturer) -> None: + """After a command, keep the snapshot current: re-inject the collector if the + page navigated (self-healing β€” a click can submit a form and wipe it), then + drain the mutation buffer. Called after every command, not just explicit + navigations, so the initial full-document snapshot is captured even if it + wasn't ready the instant the navigation returned.""" + snapshot = _state.get("snapshot") + if snapshot is None: + return + try: + snapshot.inject() # no-op if already present; re-installs after navigation + except Exception as exc: # noqa: BLE001 β€” never break the test + _log.warning("snapshot re-inject threw: %s", exc) + _flush_mutations(capturer) + + +def _flush_mutations(capturer: SessionCapturer) -> None: + """Drain the page-side mutation buffer and forward it β€” no-op if the + collector was never injected. Defensive: never breaks the user's test.""" + snapshot: SnapshotCapturer | None = _state.get("snapshot") + if snapshot is None: + return + try: + mutations = snapshot.pull_mutations() + except Exception as exc: # noqa: BLE001 β€” capture must never break the test + _log.warning("mutation flush threw: %s", exc) + return + if mutations: + capturer.send_mutations(mutations) + + +def _enable_bidi_capability(params: Any) -> None: + """Request BiDi at session creation by injecting ``webSocketUrl`` into the + newSession capabilities β€” the one point (before the session exists) where we + can. This makes console/network capture work out-of-box, matching the JS + adapters. Opt out with ``DEVTOOLS_BIDI=0``. Never blocks a session.""" + if os.environ.get(ENV_BIDI, "").strip().lower() in ("0", "false", "no", "off"): + return + try: + caps = params.get("capabilities") if isinstance(params, dict) else None + if not isinstance(caps, dict): + return + always = caps.get("alwaysMatch") + if not isinstance(always, dict): + always = {} + caps["alwaysMatch"] = always + always.setdefault(BIDI_CAPABILITY, True) + except Exception: # noqa: BLE001 β€” capability injection is best-effort + pass + + +def _finalize_screencast(capturer: SessionCapturer, session_id: Optional[str]) -> None: + """Encode and forward the recording for ``session_id``. Attributed explicitly, + because a rotation finalizes the session that just ended while a newer one is + already current.""" + recorder = _state.get("screencast") + if recorder is None: + return + _state["screencast"] = None + try: + info = recorder.finalize( + session_id or "session", output_dir=_state.get("output_dir") + ) + except Exception as exc: # noqa: BLE001 + _log.warning("screencast finalize threw: %s", exc) + return + if info is not None: + capturer.send_screencast(**info, session_id=session_id) + _log.info("screencast saved: %s", info.get("video_path")) + + +def _ensure_session_setup(driver: Any, capturer: SessionCapturer) -> None: + """Once per session β€” on the first real command β€” send metadata, attach + BiDi, and start the screencast. + + This CANNOT run in the newSession branch: selenium assigns ``session_id`` / + ``caps`` only *after* the newSession execute() returns, so at that point the + driver has no session yet (BiDi caps missing, screenshots fail). By the first + real command the driver is fully initialized. Each step is independently + defensive β€” a BiDi or screencast failure is a logged no-op. + """ + session_id = getattr(driver, "session_id", None) + if not session_id or session_id == _state.get("armed_session"): + return # not ready yet, or already armed for this session + previous = _state.get("armed_session") + if previous: + # A second driver in the same process. Close out the previous session + # before arming this one, or its video and DOM would be attributed here. + _flush_mutations(capturer) + _state["snapshot"] = None + _finalize_screencast(capturer, previous) + _log.info("session %s replaced by %s; re-arming capture", previous, session_id) + _state["armed_session"] = session_id + capturer.ensure_metadata(session_id, getattr(driver, "caps", None), None) + _log.info("session %s started", session_id) + _send_default_suite(capturer, "running") # tree entry for plain-script runs + try: + if bidi.attach(driver, capturer): + _log.info("BiDi attached β€” capturing console + network") + except Exception as exc: # noqa: BLE001 β€” capture must never break the test + _log.warning("BiDi attach threw: %s", exc) + try: + recorder = ScreencastRecorder() + recorder.start(driver) + _state["screencast"] = recorder + _log.info("screencast recording started") + except Exception as exc: # noqa: BLE001 + _log.warning("screencast start threw: %s", exc) + try: + # Inject the packages/script DOM observer so the snapshot iframe fills. + # Use a capture-bypassing execute_script so injection/readback scripts + # don't pollute the Actions timeline. + snapshot = start_snapshot_capture( + driver, execute_fn=_guarded_execute_script(driver) + ) + _state["snapshot"] = snapshot + if snapshot is not None: + _log.info("DOM snapshot collector injected") + except Exception as exc: # noqa: BLE001 + _log.warning("snapshot start threw: %s", exc) + + +def _on_quit(capturer: SessionCapturer, driver: Any = None) -> None: + """On driver.quit(): finalize the screencast and forward its metadata. + + ``quit`` is in the skip set, so this is the last hook before the session is + gone β€” encode here while a session id is still known.""" + quitting = getattr(driver, "session_id", None) if driver is not None else None + armed = _state.get("armed_session") + if quitting and armed and quitting != armed: + # Another driver is the live one; leave its capture alone. + return + # Drain any final mutations while the session (and page) still exist. + _flush_mutations(capturer) + _state["snapshot"] = None + if _state.get("default_suite") is not None: + _send_default_suite(capturer, "passed") # mark the script run complete + _finalize_screencast(capturer, quitting or armed or capturer.session_id) + # Cleared so a driver created later re-arms instead of being treated as this + # session and losing its metadata, BiDi, DOM and video. + _state["armed_session"] = 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 + # The adapter's own execute_script (snapshot inject/readback) runs + # transparently β€” never captured as a user command. + if _internal_active(): + return orig_execute(self, driver_command, params) + # Skip capture for noise, but never alter behavior. + if driver_command in SKIP_COMMANDS: + # Finalize the screencast BEFORE quit tears the session down β€” after + # orig_execute the driver's WebSocket/session is gone. + if driver_command == "quit": + _on_quit(capturer, self) + elif driver_command == "newSession": + _enable_bidi_capability(params) # request BiDi before the session opens + return orig_execute(self, driver_command, params) + + # First real command: the driver is now fully initialized β€” set up + # metadata/BiDi/screencast once, before executing so BiDi sees this cmd. + _ensure_session_setup(self, capturer) + start = now_ms() + src = call_source(_skip_frames()) + _capture_source(capturer, src) # Source tab: send the test file once + try: + result = orig_execute(self, driver_command, params) + except BaseException as exc: # capture then re-raise unchanged + capturer.capture_command( + command=driver_command, + args=params, + error=exc, + start_time=start, + call_source=src, + ) + raise + + # WebDriver.execute returns the full response dict; the useful payload + # is response["value"]. + value = result.get("value") if isinstance(result, dict) else result + # One screenshot per command on this (main) thread β€” attached to the + # command (so selecting it shows the page) AND reused as a screencast + # frame, so we pay for only a single screenshot round-trip either way. + shot = _take_screenshot(self) + capturer.capture_command( + command=driver_command, + args=params, + result=value, + start_time=start, + call_source=src, + screenshot=shot, + ) + _log.debug("command: %s", driver_command) + # Keep the snapshot iframe current after every command (a click can + # navigate too, not just get/back/…), re-injecting if the page changed. + _refresh_snapshot(capturer) + _add_screencast_frame(shot) + return result + + webdriver_cls.execute = patched_execute # type: ignore[assignment] + _sources_sent.clear() + _state.update( + installed=True, cls=webdriver_cls, orig=orig_execute, + armed_session=None, output_dir=None, default_suite=None, + ) + + +def uninstall() -> None: + recorder = _state.get("screencast") + if recorder is not None: + recorder.stop() # never leave a poll thread running past teardown + if not _state["installed"]: + _state.update( + screencast=None, snapshot=None, armed_session=None, + output_dir=None, default_suite=None, + ) + return + _state["cls"].execute = _state["orig"] # type: ignore[union-attr] + _state.update( + installed=False, cls=None, orig=None, screencast=None, snapshot=None, + armed_session=None, output_dir=None, default_suite=None, + ) diff --git a/packages/selenium-devtools-py/src/selenium_devtools/lifecycle.py b/packages/selenium-devtools-py/src/selenium_devtools/lifecycle.py new file mode 100644 index 00000000..b7ffb08d --- /dev/null +++ b/packages/selenium-devtools-py/src/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"[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-devtools-py-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 ``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 ``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-devtools-py/src/selenium_devtools/logcapture.py b/packages/selenium-devtools-py/src/selenium_devtools/logcapture.py new file mode 100644 index 00000000..1984ad0b --- /dev/null +++ b/packages/selenium-devtools-py/src/selenium_devtools/logcapture.py @@ -0,0 +1,70 @@ +"""Forward Python ``logging`` output to the dashboard Console β€” the 'runner' log +stream, the analogue of the JS adapter surfacing ``@wdio/logger`` output +(webdriver/BiDi lines, the adapter's own 'βœ“ Script injected' events, etc.). + +Attaches a handler to the root logger and raises the adapter's + selenium's +loggers to a useful level so their records reach the dashboard. Restores +everything on stop. +""" + +from __future__ import annotations + +import logging +from typing import Any, Dict + +#: The adapter's logger name β€” operational events are logged under it. +LOGGER_NAME = "selenium_devtools" + +_LEVEL_MAP = { + logging.DEBUG: "debug", + logging.INFO: "info", + logging.WARNING: "warn", + logging.ERROR: "error", + logging.CRITICAL: "error", +} + +# Loggers raised to a useful level so their records reach the dashboard: the +# adapter's own (verbose) and selenium's (info: session/BiDi lines). +_WATCH = {LOGGER_NAME: logging.DEBUG, "selenium": logging.INFO} + + +class _DashboardHandler(logging.Handler): + def __init__(self, capturer: Any) -> None: + super().__init__(level=logging.DEBUG) + self._capturer = capturer + self._reentrant = False + + def emit(self, record: logging.LogRecord) -> None: + if self._reentrant: # forwarding must not re-enter via its own logging + return + self._reentrant = True + try: + level = _LEVEL_MAP.get(record.levelno, "log") + message = f"{record.name}: {record.getMessage()}" + self._capturer.capture_console(level, [message], source="terminal") + except Exception: # noqa: BLE001 β€” logging must never break the test + pass + finally: + self._reentrant = False + + +class LogCapturer: + """Streams Python logging into the dashboard Console for the run's duration.""" + + def __init__(self, capturer: Any) -> None: + self._handler = _DashboardHandler(capturer) + self._prev_levels: Dict[str, int] = {} + + def start(self) -> None: + logging.getLogger().addHandler(self._handler) + for name, level in _WATCH.items(): + logger = logging.getLogger(name) + self._prev_levels[name] = logger.level + if logger.level == logging.NOTSET or logger.level > level: + logger.setLevel(level) + + def stop(self) -> None: + logging.getLogger().removeHandler(self._handler) + for name, prev in self._prev_levels.items(): + logging.getLogger(name).setLevel(prev) + self._prev_levels.clear() diff --git a/packages/selenium-devtools-py/src/selenium_devtools/output_dir.py b/packages/selenium-devtools-py/src/selenium_devtools/output_dir.py new file mode 100644 index 00000000..f22346a3 --- /dev/null +++ b/packages/selenium-devtools-py/src/selenium_devtools/output_dir.py @@ -0,0 +1,80 @@ +"""Where the adapter writes run output. Mirrors ``core/output-dir.ts`` so all four +adapters group artifacts under the same ``test-results/`` subfolder, wherever the +base directory resolves to. + +Base directory, by priority: + + 1. ``user_configured_dir`` β€” explicit opt-in, honored as-is. + 2. ``dirname(test_file_path)`` β€” the folder holding the test that just ran. + 3. ``fallback_dir`` (default: cwd). + +A candidate inside an installed-package directory is skipped: a test file +resolved through site-packages must not receive artifacts. That is the Python +counterpart of core skipping ``node_modules``. Each candidate must also be +writable, or resolution falls through to the next. +""" + +from __future__ import annotations + +import os +from typing import List, Optional + +#: All run output is grouped under this subfolder, matching the JS adapters. +OUTPUT_SUBDIR = "test-results" + +#: Installed-package segments: the analogue of core's ``node_modules`` filter. +_INSTALLED_SEGMENTS = ( + f"{os.sep}site-packages{os.sep}", + f"{os.sep}dist-packages{os.sep}", +) + + +def _is_writable(directory: str) -> bool: + return os.path.isdir(directory) and os.access(directory, os.W_OK) + + +def _resolve_base_dir( + user_configured_dir: Optional[str], + test_file_path: Optional[str], + fallback_dir: Optional[str], +) -> str: + fallback = fallback_dir or os.getcwd() + # An explicit directory bypasses the installed-package and writability + # filters: the user opted into it, and a surprising override is worse than + # failing where they pointed us. + if user_configured_dir: + return user_configured_dir + candidates: List[str] = [] + if test_file_path: + candidates.append(os.path.dirname(os.path.abspath(test_file_path))) + candidates.append(fallback) + for directory in candidates: + if not directory: + continue + if any(seg in f"{directory}{os.sep}" for seg in _INSTALLED_SEGMENTS): + continue + if _is_writable(directory): + return directory + return fallback + + +def resolve_adapter_output_dir( + *, + user_configured_dir: Optional[str] = None, + test_file_path: Optional[str] = None, + fallback_dir: Optional[str] = None, +) -> str: + """Absolute path of the ``test-results`` directory to write into. Does not + create it; ``ensure_output_dir`` does that at write time.""" + base = _resolve_base_dir(user_configured_dir, test_file_path, fallback_dir) + return os.path.join(os.path.abspath(base), OUTPUT_SUBDIR) + + +def ensure_output_dir(directory: str) -> Optional[str]: + """Create ``directory`` if needed. Returns it, or None when it cannot be + created, so a caller can fall back rather than crash a user's test run.""" + try: + os.makedirs(directory, exist_ok=True) + except OSError: + return None + return directory if _is_writable(directory) else None diff --git a/packages/selenium-devtools-py/src/selenium_devtools/pytest_plugin.py b/packages/selenium-devtools-py/src/selenium_devtools/pytest_plugin.py new file mode 100644 index 00000000..783770c4 --- /dev/null +++ b/packages/selenium-devtools-py/src/selenium_devtools/pytest_plugin.py @@ -0,0 +1,124 @@ +"""pytest plugin β€” feeds the suite/test tree to the dashboard. + +The analogue of the JS adapter's mocha/jest hooks. Inert unless the run opts in +via ``DEVTOOLS_ENABLE=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 selenium_devtools as devtools +from . import frames +from .constants import ENV_OPT_IN, ENV_PORT +from .utils import now_ms + + +def _opted_in() -> bool: + return bool(os.environ.get(ENV_OPT_IN) or os.environ.get(ENV_PORT)) + + +class _SuiteRegistry: + """Groups tests by file into the SuiteStats[] the dashboard expects.""" + + def __init__(self) -> None: + self._suites: Dict[str, dict] = {} + self._tests: Dict[str, dict] = {} + self._starts: Dict[str, int] = {} + + def mark_start(self, nodeid: str) -> None: + self._starts.setdefault(nodeid, now_ms()) + + def record(self, nodeid: str, file: str, name: str, line: int, state: str) -> None: + start = self._starts.get(nodeid, now_ms()) + end = now_ms() + suite = self._suites.setdefault( + file, + frames.suite_stats(uid=file, title=file, file=file, + start_ms=start, tests=[]), + ) + self._tests[nodeid] = frames.test_stats( + uid=nodeid, + title=name, + full_title=f"{file} β€Ί {name}", + parent=file, + state=state, + file=file, + start_ms=start, + end_ms=end, + call_source=f"{file}:{line + 1}", + ) + suite["tests"] = [t for nid, t in self._tests.items() + if nid.split("::", 1)[0] == file] + suite["end"] = self._tests[nodeid]["end"] + suite["state"] = "failed" if any( + t["state"] == "failed" for t in suite["tests"] + ) else state + + def snapshot(self) -> list: + return list(self._suites.values()) + + +_registry = _SuiteRegistry() + +# Note: test-file source capture for the Source tab lives in instrumentation.py +# (keyed by each command's callSource path), so it works for plain scripts too β€” +# not just pytest. The plugin only owns the suite/test tree. + + +def pytest_configure(config) -> None: # noqa: ANN001 + if _opted_in(): + capturer = devtools.enable() + # pytest owns the suite tree β€” suppress the adapter's default script suite. + from . import instrumentation + + instrumentation.set_external_suites(True) + url = devtools.dashboard_url() + if capturer is not None and url: + print(f"\n[devtools] dashboard live at {url}\n") + + +def pytest_runtest_logstart(nodeid, location) -> None: # noqa: ANN001 + if _opted_in(): + _registry.mark_start(nodeid) + + +def pytest_runtest_logreport(report) -> None: # noqa: ANN001 + if not _opted_in(): + return + capturer = devtools.get_capturer() + if capturer is None: + return + # 'call' carries pass/fail; a skip surfaces at 'setup'. + if report.when == "call" or (report.when == "setup" and report.skipped): + state = ( + "skipped" if report.skipped + else "passed" if report.passed + else "failed" + ) + file, line, name = report.location + _registry.record(report.nodeid, file, name, line or 0, state) + capturer.send_suites(_registry.snapshot()) + + +def pytest_sessionfinish(session, exitstatus) -> None: # noqa: ANN001 + if not _opted_in(): + return + capturer = devtools.get_capturer() + if capturer is not None: + capturer.send_suites(_registry.snapshot()) + # Keep the dashboard open for inspection after the run β€” exit when the user + # closes the window (clientDisconnected). Only when we actually opened a + # window; CI (DEVTOOLS_OPEN=0) tears down immediately. + from . import lifecycle + + if lifecycle.dashboard_window_open(): + print( + "\n[devtools] dashboard is live β€” close the window to finish " + "(Ctrl-C also works).\n" + ) + lifecycle.wait_for_shutdown() + devtools.disable() diff --git a/packages/selenium-devtools-py/src/selenium_devtools/screencast.py b/packages/selenium-devtools-py/src/selenium_devtools/screencast.py new file mode 100644 index 00000000..5c515659 --- /dev/null +++ b/packages/selenium-devtools-py/src/selenium_devtools/screencast.py @@ -0,0 +1,218 @@ +"""Screencast recorder β€” the Python analogue of core's ``ScreencastRecorderBase``. + +Frames are captured **synchronously on the main thread, one per command** +(``driver.get_screenshot_as_base64()``), driven from the instrumentation hook. +A background poll thread is deliberately NOT used: Selenium's session is not +thread-safe, and a screenshot fired from a daemon thread races the main thread's +commands and DOM-trace readback on the same connection β€” corrupting both the +video and the snapshot. The reference JS adapters avoid this too (CDP push-mode +screencast / per-command screenshots on the command queue), so we mirror that. + +On ``stop`` the buffered frames are encoded to a ``.webm`` via ffmpeg *if it's on +PATH* β€” ffmpeg is an optional dependency, so its absence is a one-line warning +and a skipped encode, never an error. + +A CDP push-mode fast-path (Chrome ``Page.startScreencast`` via +``execute_cdp_cmd``) is a future optimization β€” noted here, not implemented; +per-command capture works on every browser selenium drives. + +Everything is defensive: a transient screenshot failure (e.g. mid-navigation) is +skipped and recording continues; a recorder that never captured a frame encodes +nothing. Capture never breaks the user's test. +""" + +from __future__ import annotations + +import os +import shutil +import subprocess +import sys +import tempfile +from typing import Any, Callable, List, Optional + +from .constants import SCREENCAST_IMAGE_FORMAT, SCREENCAST_MIN_FRAMES +from .output_dir import OUTPUT_SUBDIR, ensure_output_dir +from .types import ScreencastFrame +from .utils import now_ms + +#: Screenshot fn signature β€” injectable so tests drive the buffer without a +#: real driver. Returns a base64 PNG string, or None on a transient failure. +ScreenshotFn = Callable[[], Optional[str]] + + +def _warn(message: str) -> None: + print(f"[devtools] screencast: {message}", file=sys.stderr) + + +class ScreencastRecorder: + def __init__(self, *, ffmpeg_path: Optional[str] = None) -> None: + # Resolve ffmpeg once; None means "encoding unavailable, skip it". + self._ffmpeg = ffmpeg_path or shutil.which("ffmpeg") + self._frames: List[ScreencastFrame] = [] + self._screenshot: Optional[ScreenshotFn] = None + self._active = False + + # ── public API ──────────────────────────────────────────────────────────── + + def start(self, driver: Any, screenshot_fn: Optional[ScreenshotFn] = None) -> None: + """Arm the recorder and capture a seed frame. ``screenshot_fn`` is + injectable for tests; by default it reads + ``driver.get_screenshot_as_base64``. Frames are captured on the main + thread via :meth:`capture` β€” no background polling. Safe to call twice.""" + if self._active: + return + if screenshot_fn is not None: + self._screenshot = screenshot_fn + elif driver is not None and hasattr(driver, "get_screenshot_as_base64"): + self._screenshot = driver.get_screenshot_as_base64 + else: + _warn("driver has no get_screenshot_as_base64 β€” recording skipped") + return + self._active = True + self.capture() # best-effort seed frame + + def capture(self) -> bool: + """Capture one frame synchronously (main thread). No-op if not armed. + + A transient failure β€” a screenshot taken mid-navigation, a dead session + β€” is skipped and recording stays armed, so a single miss never truncates + the video the way a background poll loop would. Returns True iff a frame + was buffered.""" + if not self._active: + return False + fn = self._screenshot + if fn is None: + return False + try: + data = fn() + except Exception: # noqa: BLE001 β€” transient miss; keep recording + return False + if isinstance(data, str) and data: + self._frames.append({"data": data, "timestamp": now_ms()}) + return True + return False + + def add_frame(self, data: Optional[str]) -> bool: + """Buffer a frame from an ALREADY-captured base64 screenshot β€” lets the + caller reuse the per-command screenshot it took for the command entry + instead of paying for a second screenshot round-trip. No-op if not armed + or the data is empty. Returns True iff a frame was buffered.""" + if not self._active or not isinstance(data, str) or not data: + return False + self._frames.append({"data": data, "timestamp": now_ms()}) + return True + + def stop(self) -> None: + """Disarm the recorder. Idempotent; safe even if start() never ran.""" + self._active = False + + @property + def frames(self) -> List[ScreencastFrame]: + return list(self._frames) + + @property + def duration(self) -> int: + """ms between first and last frame; 0 with fewer than 2 frames.""" + if len(self._frames) < 2: + return 0 + return self._frames[-1]["timestamp"] - self._frames[0]["timestamp"] + + @property + def is_recording(self) -> bool: + return self._active + + # ── finalize ────────────────────────────────────────────────────────────── + + def finalize( + self, + session_id: str, + output_dir: Optional[str] = None, + *, + min_frames: int = SCREENCAST_MIN_FRAMES, + filename_prefix: str = "selenium-py-video", + ) -> Optional[dict]: + """Stop, encode the buffered frames to a ``.webm``, and return the + ``{video_path, video_file, frame_count, duration, start_time}`` metadata + the capturer forwards β€” or None if there's nothing to encode / ffmpeg is + absent. All failures are caught: screencast is best-effort.""" + self.stop() + frames = self.frames + if len(frames) < min_frames: + return None + if not self._ffmpeg: + _warn("ffmpeg not found on PATH β€” skipping video encode " + f"({len(frames)} frame(s) captured)") + return None + + file_name = f"{filename_prefix}-{session_id}.webm" + target_dir = _writable_dir(output_dir) + video_path = os.path.join(target_dir, file_name) + try: + _encode_webm(frames, video_path, self._ffmpeg) + except Exception as exc: # noqa: BLE001 β€” encode failure must not abort + _warn(f"encode failed: {exc}") + return None + return { + "video_path": video_path, + "video_file": file_name, + "frame_count": len(frames), + "duration": self.duration, + "start_time": frames[0]["timestamp"] if frames else None, + } + + +def _writable_dir(preferred: Optional[str]) -> str: + """The directory to encode into. ``preferred`` is the resolved + ``test-results`` path, which will not exist on a first run, so it is created + here; a failure falls back to the temp dir rather than losing the video.""" + candidate = preferred or os.path.join(os.getcwd(), OUTPUT_SUBDIR) + created = ensure_output_dir(candidate) + if created: + return created + return tempfile.gettempdir() + + +def _encode_webm(frames: List[ScreencastFrame], output_path: str, ffmpeg: str) -> None: + """Encode base64 frames to a VP8/WebM via ffmpeg's concat demuxer, giving + each frame its real inter-frame duration (VFR reflects command pauses). + Forces CFR 10fps + all-intra so the dashboard ``