Parallel Execution, Isolation, Concurrency, and Test Sharding: Guided Hands-On Workflow

1. Build a concurrency-safe local fixture

Create server.py. It binds only to loopback. The /work endpoint delays its response to simulate controlled application work. That server-side time.sleep() is fixture behavior, not a Selenium synchronization strategy; the test still waits through normal navigation completion and assertions.

from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from urllib.parse import urlparse, parse_qs
from html import escape
import json
import threading
import time

HOST = "127.0.0.1"
PORT = 8765
active_users = set()
lock = threading.Lock()

class Handler(BaseHTTPRequestHandler):
    def log_message(self, format, *args):
        return

    def _html(self, status, body):
        payload = ("<!doctype html><html lang='en'><head><meta charset='utf-8'>"
                   "<title>Chapter 21 Fixture</title></head><body>" + body + "</body></html>").encode()
        self.send_response(status)
        self.send_header("Content-Type", "text/html; charset=utf-8")
        self.send_header("Content-Length", str(len(payload)))
        self.end_headers()
        self.wfile.write(payload)

    def do_GET(self):
        u = urlparse(self.path)
        q = parse_qs(u.query)
        if u.path == "/health":
            payload = json.dumps({"ok": True}).encode()
            self.send_response(200)
            self.send_header("Content-Type", "application/json")
            self.send_header("Content-Length", str(len(payload)))
            self.end_headers()
            self.wfile.write(payload)
            return
        if u.path == "/work":
            case = q.get("case", ["unknown"])[0]
            identity = q.get("identity", ["anonymous"])[0]
            delay_ms = int(q.get("delay_ms", ["300"])[0])
            time.sleep(delay_ms / 1000)  # controlled AUT workload, not a Selenium wait strategy
            self._html(200, f"<main><h1 data-testid='status'>done</h1><p data-testid='case'>{escape(case)}</p><p data-testid='identity'>{escape(identity)}</p></main>")
            return
        if u.path == "/claim":
            identity = q.get("identity", ["shared"])[0]
            hold_ms = int(q.get("hold_ms", ["450"])[0])
            with lock:
                if identity in active_users:
                    self._html(409, f"<main><h1 data-testid='status'>collision</h1><p data-testid='identity'>{escape(identity)}</p></main>")
                    return
                active_users.add(identity)
            try:
                time.sleep(hold_ms / 1000)  # deterministic synthetic service contention
                self._html(200, f"<main><h1 data-testid='status'>claimed</h1><p data-testid='identity'>{escape(identity)}</p></main>")
            finally:
                with lock:
                    active_users.discard(identity)
            return
        self._html(404, "<h1>not found</h1>")

if __name__ == "__main__":
    print(f"fixture listening on http://{HOST}:{PORT}")
    ThreadingHTTPServer((HOST, PORT), Handler).serve_forever()

2. Preflight: environment and fixture state

The following example makes the Preflight: environment and fixture state 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
# Windows PowerShell: .\.venv\Scripts\Activate.ps1
# Linux/macOS: source .venv/bin/activate
python -m pip install selenium==4.47.0
python server.py

In a second terminal, verify http://127.0.0.1:8765/health returns {"ok": true}. Confirm the intended browser is installed. Selenium Manager remains the normal local driver-resolution path; do not add a manually downloaded driver unless your environment intentionally manages drivers that way.

3. Create the isolated serial/parallel runner

Save the following as serial_parallel.py. Each call to run_case() owns its driver from creation through quit(). It also creates a unique synthetic identity, unique evidence directory, and temporary download directory.

from concurrent.futures import ThreadPoolExecutor, as_completed
from pathlib import Path
from tempfile import TemporaryDirectory
from time import perf_counter
import hashlib
import json
import os
import uuid

from selenium import webdriver
from selenium.webdriver.common.by import By

BASE_URL = os.environ.get("LAB_BASE_URL", "http://127.0.0.1:8765")
CASES = [f"case-{n}" for n in range(1, 7)]


def stable_shard(test_id: str, shard_count: int) -> int:
    digest = hashlib.sha256(test_id.encode("utf-8")).digest()
    return int.from_bytes(digest[:8], "big") % shard_count


def run_case(case_id: str, evidence_root: Path) -> dict:
    identity = f"{case_id}@example.test"
    case_dir = evidence_root / case_id
    case_dir.mkdir(parents=True, exist_ok=False)
    with TemporaryDirectory(prefix=f"selenium-{case_id}-") as downloads:
        options = webdriver.ChromeOptions()
        options.add_experimental_option("prefs", {"download.default_directory": str(Path(downloads).resolve())})
        driver = webdriver.Chrome(options=options)
        started = perf_counter()
        try:
            driver.get(f"{BASE_URL}/work?case={case_id}&identity={identity}&delay_ms=300")
            assert driver.find_element(By.CSS_SELECTOR, "[data-testid='status']").text == "done"
            assert driver.find_element(By.CSS_SELECTOR, "[data-testid='identity']").text == identity
            record = {
                "case": case_id,
                "identity": identity,
                "session_id": driver.session_id,
                "browserName": driver.capabilities.get("browserName"),
                "browserVersion": driver.capabilities.get("browserVersion"),
                "duration_s": round(perf_counter() - started, 3),
                "download_dir": downloads,
            }
            (case_dir / "session.json").write_text(json.dumps(record, indent=2), encoding="utf-8")
            driver.save_screenshot(str(case_dir / "viewport.png"))
            return record
        finally:
            driver.quit()


def run_serial(cases, root):
    start = perf_counter()
    rows = [run_case(c, root / "serial") for c in cases]
    return rows, perf_counter() - start


def run_parallel(cases, root, workers):
    start = perf_counter()
    rows = []
    with ThreadPoolExecutor(max_workers=workers, thread_name_prefix="selenium-worker") as pool:
        futures = {pool.submit(run_case, c, root / f"parallel-{workers}"): c for c in cases}
        for future in as_completed(futures):
            rows.append(future.result())
    return rows, perf_counter() - start


if __name__ == "__main__":
    run_id = uuid.uuid4().hex[:10]
    root = Path("evidence") / run_id
    root.mkdir(parents=True)
    print("deterministic shards:", {c: stable_shard(c, 2) for c in CASES})
    serial_rows, serial_s = run_serial(CASES, root)
    parallel_rows, parallel_s = run_parallel(CASES, root, workers=2)
    summary = {
        "run_id": run_id,
        "cpu_count": os.cpu_count(),
        "serial_s": round(serial_s, 3),
        "parallel_2_s": round(parallel_s, 3),
        "speedup": round(serial_s / parallel_s, 3) if parallel_s else None,
        "serial_sessions": [r["session_id"] for r in serial_rows],
        "parallel_sessions": [r["session_id"] for r in parallel_rows],
    }
    (root / "summary.json").write_text(json.dumps(summary, indent=2), encoding="utf-8")
    print(json.dumps(summary, indent=2))

4. Observe serial and parallel state

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

python serial_parallel.py
# Inspect the newest evidence/<run-id>/summary.json
# Compare serial/ and parallel-2/ subdirectories.

Expect 12 browser sessions across the full experiment: six serial sessions and six parallel sessions, because the two runs are separate observations. Inside the parallel run, no more than two run_case() calls are scheduled at once. The exact speedup is environment-dependent because fresh browser startup is intentionally included in the cost.

5. Calculate deterministic two-shard ownership

The script prints the two-shard mapping before it starts browsers. This assignment is independent of completion order. If case-4 finishes before case-2, its shard does not change. In CI, each job should receive a shard index and run only tests whose stable assignment matches that index.

CASES = [f"case-{n}" for n in range(1, 7)]

for shard_index in range(2):
    owned = [case for case in CASES if stable_shard(case, 2) == shard_index]
    print("shard", shard_index, owned)

6. Translate the local experiment to Grid without confusing controls

If you change webdriver.Chrome() to webdriver.Remote(), the runner still owns worker count. Grid independently decides whether compatible slots are available. Setting max_workers=8 does not create eight Grid slots; it can create up to eight competing new-session requests.

from selenium import webdriver

options = webdriver.ChromeOptions()
driver = webdriver.Remote(
    command_executor="http://127.0.0.1:4444",
    options=options,
)
try:
    print(driver.session_id, driver.capabilities.get("browserName"))
finally:
    driver.quit()

7. Challenge: choose the right control

Your local Grid has two Chrome slots. The test runner has max_workers=6. Four tests spend most of their time waiting for a session. Which setting should you change first if the objective is predictable fast feedback without adding infrastructure?

8. Cleanup and verification

Stop server.py with Ctrl+C after tests are complete. Delete only the lab virtual environment/evidence directory if desired. Every browser should already be closed by finally: driver.quit(). Verify no orphan test browser remains before repeating a concurrency benchmark.

# Optional lab cleanup from the lab directory
rm -rf evidence
# Windows PowerShell equivalent: Remove-Item -Recurse -Force evidence
# Keep unrelated browser profiles, downloads, and system processes untouched.

Knowledge check

Official references and current-version notes

• Selenium 4.47 release notes
• Selenium downloads — current stable client and Grid versions
• Avoid sharing state — Selenium test practices
• Test independency — Selenium test practices
• Getting started with Selenium Grid — capacity guidance
• Grid CLI options — max sessions and queue controls
• Python concurrent.futures — ThreadPoolExecutor