Checkpoint Lab — WebDriver BiDi: Events, Network, Logs, Script, and Bidirectional Control

1. Checkpoint scenario and predictions

Use the same loopback fixture. Predict before running:

1. Enabling BiDi should produce a returned WebSocket capability on a supported session.
2. Installing the console handler before the click should append one correlated console record.
3. Installing the network request handler before the fetch should append a record for /api/ping.
4. Removing both handlers should stop later events from changing the evidence lists.
5. If one high-level domain is unavailable, the run must record fallback mode rather than claiming full live capture.

2. Setup and preflight

The following example makes the Setup and preflight behavior concrete. Read it with the stated assumptions, then compare its observable output or state changes with the explanation that follows.

from pathlib import Path

ROOT = Path("chapter18_fixture")
SITE = ROOT / "site"
SITE.mkdir(parents=True, exist_ok=True)

(SITE / "index.html").write_text(r'''<!doctype html>
<html lang="en"><head><meta charset="utf-8"><meta name="viewport" content="width=device-width,initial-scale=1">
<title>BiDi Local Lab</title><style>body{font-family:system-ui;margin:30px;max-width:850px}button{margin:6px;padding:10px 14px}.panel{border:1px solid #999;padding:16px;border-radius:12px;margin:16px 0}</style></head>
<body><h1 data-testid="heading">BiDi Local Lab</h1><p data-testid="context">ready</p>
<div class="panel"><button data-testid="console" type="button">Emit console</button><button data-testid="request" type="button">Make request</button><button data-testid="fail" type="button">Fail request</button></div>
<div class="panel"><p data-testid="status" data-state="idle">idle</p><code data-testid="result"></code></div>
<script>
const q=id=>document.querySelector(`[data-testid="${id}"]`);
const testId=new URLSearchParams(location.search).get('test_id')||'manual';
q('context').textContent=`test_id=${testId}`;
q('console').onclick=()=>{console.log(`bidi-console test_id=${testId}`);q('status').dataset.state='console';q('status').textContent='console emitted'};
q('request').onclick=async()=>{q('status').dataset.state='pending';const r=await fetch(`/api/ping?test_id=${encodeURIComponent(testId)}`,{headers:{'X-Test-Id':testId}});const data=await r.json();q('result').textContent=JSON.stringify(data);q('status').dataset.state='ok';q('status').textContent='request complete'};
q('fail').onclick=async()=>{q('status').dataset.state='pending';try{const r=await fetch(`/api/fail?test_id=${encodeURIComponent(testId)}`);q('result').textContent=String(r.status);q('status').dataset.state='http-error';q('status').textContent='expected HTTP error'}catch(e){q('status').dataset.state='transport-error';q('status').textContent='transport error'}};
</script></body></html>''', encoding="utf-8")

(ROOT / "serve.py").write_text(r'''from http.server import ThreadingHTTPServer, SimpleHTTPRequestHandler
from pathlib import Path
from urllib.parse import urlsplit, parse_qs
import json
ROOT=Path(__file__).resolve().parent; SITE=ROOT/"site"; HOST="127.0.0.1"; PORT=8782
class H(SimpleHTTPRequestHandler):
    def translate_path(self,path):
        raw=super().translate_path(path); rel=Path(raw).relative_to(Path.cwd()); return str(SITE/rel)
    def log_message(self,fmt,*args): pass
    def do_GET(self):
        parts=urlsplit(self.path)
        if parts.path=="/api/ping":
            body=json.dumps({"ok":True,"test_id":parse_qs(parts.query).get("test_id",[""])[0]}).encode()
            self.send_response(200); self.send_header("Content-Type","application/json"); self.send_header("Content-Length",str(len(body))); self.end_headers(); self.wfile.write(body); return
        if parts.path=="/api/fail":
            body=b'{"ok":false,"reason":"synthetic"}'
            self.send_response(503); self.send_header("Content-Type","application/json"); self.send_header("Content-Length",str(len(body))); self.end_headers(); self.wfile.write(body); return
        return super().do_GET()
server=ThreadingHTTPServer((HOST,PORT),H); print(f"BiDi fixture: http://{HOST}:{PORT}/")
try: server.serve_forever()
except KeyboardInterrupt: pass
finally: server.server_close()
''', encoding="utf-8")
print(ROOT.resolve())

The following example makes the Setup and preflight behavior concrete. Read it with the stated assumptions, then compare its observable output or state changes with the explanation that follows.

python -m venv .venv
# Activate the virtual environment for your platform.
python -m pip install "selenium==4.47.0"
python build_fixture.py
python chapter18_fixture/serve.py

Preflight: Python 3.10+, Selenium 4.47.0, supported local browser, Selenium Manager normal resolution, fixture reachable only on 127.0.0.1:8782. Grid/browser cloud is optional. No real credential or production endpoint is permitted.

3. Build the diagnostic helper

The following example makes the Build the diagnostic helper behavior concrete. Read it with the stated assumptions, then compare its observable output or state changes with the explanation that follows.

from dataclasses import dataclass, field
from pathlib import Path
from datetime import datetime, timezone
import json
import selenium
from selenium import webdriver
from selenium.webdriver.common.by import By
from selenium.webdriver.support.ui import WebDriverWait

@dataclass
class BidiEvidence:
    mode: str
    console: list = field(default_factory=list)
    network: list = field(default_factory=list)
    support: dict = field(default_factory=dict)

class BidiDiagnostic:
    def __init__(self, driver):
        self.driver = driver
        self.console_id = None
        self.network_id = None
        self.evidence = BidiEvidence(mode="live")

    def support_snapshot(self):
        caps = self.driver.capabilities
        self.evidence.support = {
            "selenium": selenium.__version__,
            "session_id": self.driver.session_id,
            "browser": caps.get("browserName"),
            "browserVersion": caps.get("browserVersion"),
            "platformName": caps.get("platformName"),
            "webSocketUrl": bool(caps.get("webSocketUrl")),
            "classic_window_handle": self.driver.current_window_handle,
            "script_api": hasattr(type(self.driver), "script"),
            "network_api": hasattr(type(self.driver), "network"),
        }
        return self.evidence.support

    def start(self):
        support = self.support_snapshot()
        if not support["webSocketUrl"] or not support["script_api"]:
            self.evidence.mode = "fallback"
            return False

        self.console_id = self.driver.script.add_console_message_handler(self._console)
        if support["network_api"]:
            self.network_id = self.driver.network.add_request_handler("before_request", self._network)
        else:
            self.evidence.mode = "partial-live"
        return True

    def _console(self, entry):
        self.evidence.console.append({"text": getattr(entry, "text", str(entry))})

    def _network(self, request):
        url = getattr(request, "url", "")
        if "/api/ping" in url:
            self.evidence.network.append({"url": url, "method": getattr(request, "method", None)})

    def stop(self):
        if self.console_id is not None:
            self.driver.script.remove_console_message_handler(self.console_id)
            self.console_id = None
        if self.network_id is not None:
            self.driver.network.remove_request_handler("before_request", self.network_id)
            self.network_id = None

4. Execute the live path with bounded waits

The following example makes the Execute the live path with bounded waits behavior concrete. Read it with the stated assumptions, then compare its observable output or state changes with the explanation that follows.

OUT = Path("checkpoint-bidi")
OUT.mkdir(exist_ok=True)
options = webdriver.ChromeOptions()
options.enable_bidi = True
driver = webdriver.Chrome(options=options)
collector = BidiDiagnostic(driver)
try:
    driver.get("http://127.0.0.1:8782/?test_id=checkpoint")
    live = collector.start()
    if live:
        driver.find_element(By.CSS_SELECTOR, '[data-testid="console"]').click()
        WebDriverWait(driver, 3).until(lambda _: collector.evidence.console)

        driver.find_element(By.CSS_SELECTOR, '[data-testid="request"]').click()
        if collector.evidence.support.get("network_api"):
            WebDriverWait(driver, 3).until(lambda _: collector.evidence.network)
        WebDriverWait(driver, 3).until(
            lambda d: d.find_element(By.CSS_SELECTOR, '[data-testid="status"]').get_attribute("data-state") == "ok"
        )
finally:
    collector.stop()
    driver.quit()

stamp = datetime.now(timezone.utc).isoformat()
payload = {"captured_at": stamp, **collector.evidence.__dict__}
(OUT / "evidence.json").write_text(json.dumps(payload, indent=2), encoding="utf-8")
print(json.dumps(payload, indent=2))

A supported script-only session is labeled partial-live; it is not promoted to full success. The network event remains optional only because current browser/binding parity is part of what the lab measures.

5. Graceful fallback for an unsupported runtime

The following example makes the Graceful fallback for an unsupported runtime behavior concrete. Read it with the stated assumptions, then compare its observable output or state changes with the explanation that follows.

from pathlib import Path
import json
from datetime import datetime, timezone

OUT=Path("bidi-simulation")
OUT.mkdir(exist_ok=True)
context_id="simulated-context-01"
rows=[
    {"ts":datetime.now(timezone.utc).isoformat(),"domain":"log","event":"log.entryAdded","context":context_id,"text":"bidi-console test_id=checkpoint"},
    {"ts":datetime.now(timezone.utc).isoformat(),"domain":"network","event":"network.beforeRequestSent","context":context_id,"url":"http://127.0.0.1:8782/api/ping?test_id=checkpoint","method":"GET"},
]
(OUT/"events.jsonl").write_text("".join(json.dumps(x)+"\n" for x in rows),encoding="utf-8")
print(OUT/"events.jsonl")

If the live helper reports fallback, run the simulation and copy its explicitly labeled event packet beside the capability snapshot. The conclusion must say “simulated schema exercise,” not “browser event captured.”

6. Verification checklist

• evidence.json records Selenium/browser versions, session ID, BiDi capability state, and live/partial/fallback mode.
• Live mode has at least one console message from the fixture.
• When the network API is available, live mode contains the /api/ping request.
• Callbacks were installed before triggers and removed before session teardown.
• No fixed sleeps are used.
• No CDP API is used as a Firefox/cross-browser fallback.
• No headers, cookies, bodies, real tokens, or public endpoints are persisted.

7. Prove handler cleanup conceptually

The helper sets stored handler IDs back to None after removal and then quits the session. In a reused-session architecture, add a contract test that triggers another console/request event after stop() and asserts the collector lists do not change. This checkpoint uses a fresh session, so teardown provides the stronger isolation boundary.

8. Inject one lifecycle error

Move collector.start() after the console click. The callback list should remain empty because the event already occurred. Diagnose it as a subscription-order race, restore the original ordering, and rerun only the console portion. Do not add a retry or longer timeout.

9. Write diagnostic conclusions

For the live run, write one paragraph covering: negotiated BiDi capability, console event proof, network event proof or explicit partial support, session/context correlation, and cleanup. For fallback, state exactly which capability/domain was unavailable and that the event rows are simulated schema examples.

10. Production retry/quarantine rule

BiDi evidence failures need policy just like test failures. Rule: an unsupported domain may use a declared fallback lane; a subscription exception on a supposedly supported lane is a real compatibility/infrastructure failure and must preserve first evidence. Never blanket-retry event setup, never overwrite attempt-1 evidence, and never switch to CDP without an explicitly Chromium-only test contract.

11. Cleanup and rollback

The following example makes the Cleanup and rollback behavior concrete. Read it with the stated assumptions, then compare its observable output or state changes with the explanation that follows.

# Stop chapter18_fixture/serve.py with Ctrl+C.
python - <<'PY'
from pathlib import Path
import shutil
for p in [Path("chapter18_fixture"), Path("checkpoint-bidi"), Path("bidi-simulation")]:
    if p.exists(): shutil.rmtree(p)
print("Chapter 18 checkpoint removed")
PY

12. What Chapter 18 adds to the production operating model

Your automation platform now has an event-channel contract: negotiate and record BiDi support, subscribe before triggering, scope evidence narrowly, keep context identifiers explicit, remove handlers deterministically, preserve live failures, label fallbacks, and keep CDP quarantined to explicit browser-specific adapters. This turns browser events into governed observability instead of incidental debugging output.

13. Bridge to Chapter 19 — Selenium Grid architecture

BiDi becomes more operationally interesting when sessions move off the test runner. Chapter 19 will decompose Grid roles, routing, session distribution, queues, nodes, observability, and the remote session lifecycle. The WebSocket path introduced here becomes one of the channels Grid must route correctly.

Knowledge check

Official references and version notes

• Selenium 4.47 release notes — current stable client/Grid baseline and BiDi changes.
• WebDriver BiDi overview — high-level Selenium direction and CDP relationship.
• W3C-compliant BiDirectional API — cross-browser WebSocket event model and domains.
• BiDi logging features — console and JavaScript error handlers.
• BiDi network features — request/response/auth handler concepts.
• BiDi script features — high-level script namespace.
• Python Options API — enable_bidi.
• Selenium Python API 4.47 — Python 3.10+ and current binding surface.
• Selenium Grid — remote sessions and the next chapter boundary.