diff --git a/.github/workflows/hyperdrive-tests.yml b/.github/workflows/hyperdrive-tests.yml
new file mode 100644
index 0000000..d77013e
--- /dev/null
+++ b/.github/workflows/hyperdrive-tests.yml
@@ -0,0 +1,74 @@
+name: Hyperdrive Tests
+
+on:
+  push:
+    branches: [ main ]
+  pull_request:
+    branches: [ main ]
+
+permissions:
+  contents: read
+
+jobs:
+  test:
+    strategy:
+      fail-fast: false
+      matrix:
+        python-version: ['3.12', '3.13']
+    # Docker services require Linux runners
+    runs-on: ubuntu-latest
+
+    services:
+      postgres:
+        image: postgres:16
+        env:
+          POSTGRES_USER: testuser
+          POSTGRES_PASSWORD: testpass
+          POSTGRES_DB: testdb
+          POSTGRES_HOST_AUTH_METHOD: md5
+          POSTGRES_INITDB_ARGS: "--auth-host=md5"
+        ports:
+          - 5432:5432
+        options: >-
+          --health-cmd="pg_isready -U testuser -d testdb"
+          --health-interval=10s
+          --health-timeout=5s
+          --health-retries=5
+
+      mysql:
+        image: mysql:8.0
+        env:
+          MYSQL_ROOT_PASSWORD: rootpass
+          MYSQL_USER: testuser
+          MYSQL_PASSWORD: testpass
+          MYSQL_DATABASE: testdb
+        ports:
+          - 3306:3306
+        options: >-
+          --health-cmd="mysqladmin ping -h 127.0.0.1"
+          --health-interval=10s
+          --health-timeout=5s
+          --health-retries=5
+          --default-authentication-plugin=mysql_native_password
+
+    steps:
+    - uses: actions/checkout@v4
+
+    - name: Set up Python ${{ matrix.python-version }}
+      uses: actions/setup-python@v5
+      with:
+        python-version: ${{ matrix.python-version }}
+
+    - name: Install uv
+      run: |
+        pip install uv
+
+    - name: Install project
+      run: |
+        uv pip install --system -e packages/runtime-sdk -e packages/cli
+        uv pip install --system --group dev --project packages/cli
+
+    - name: Run hyperdrive tests
+      working-directory: packages/cli
+      run: |
+        pytest -v --color=yes tests/test_hyperdrive.py
diff --git a/packages/cli/pyproject.toml b/packages/cli/pyproject.toml
index 75bf8f9..24289f2 100644
--- a/packages/cli/pyproject.toml
+++ b/packages/cli/pyproject.toml
@@ -65,9 +65,10 @@ lint.flake8-comprehensions.allow-dict-calls-with-keyword-arguments = true
 lint.per-file-ignores."tests/**" = ["PLR2004", "PLR0913"]
 lint.per-file-ignores."tests/workerd-test/**" = ["C901", "PLR0911", "PLR0912", "PLR2004", "PLW0603"]
 lint.per-file-ignores."tests/bindings-test/**" = ["C901", "PLR0911", "PLR0912", "PLR2004", "PLW0603"]
+lint.per-file-ignores."tests/hyperdrive-test/**" = ["C901", "PLR0911", "PLR0912", "PLR2004", "PLW0603"]
 
 [tool.pytest.ini_options]
-addopts = ["--ignore=tests/workerd-test", "--ignore=tests/bindings-test"]
+addopts = ["--ignore=tests/workerd-test", "--ignore=tests/bindings-test", "--ignore=tests/hyperdrive-test"]
 
 [tool.mypy]
 packages = ["pywrangler"]
diff --git a/packages/cli/tests/hyperdrive-test/pyproject.toml b/packages/cli/tests/hyperdrive-test/pyproject.toml
new file mode 100644
index 0000000..4a75098
--- /dev/null
+++ b/packages/cli/tests/hyperdrive-test/pyproject.toml
@@ -0,0 +1,5 @@
+[project]
+name = "hyperdrive-test"
+version = "0.1.0"
+requires-python = ">=3.12"
+dependencies = ["pytest", "pytest-asyncio<1.2.0", "pymysql", "pg8000"]
diff --git a/packages/cli/tests/hyperdrive-test/src/conftest.py b/packages/cli/tests/hyperdrive-test/src/conftest.py
new file mode 100644
index 0000000..f18c4a6
--- /dev/null
+++ b/packages/cli/tests/hyperdrive-test/src/conftest.py
@@ -0,0 +1,9 @@
+# pyright: reportMissingImports=false
+
+import pytest
+from workers import env as _env
+
+
+@pytest.fixture
+def env():
+    return _env
diff --git a/packages/cli/tests/hyperdrive-test/src/test_mysql.py b/packages/cli/tests/hyperdrive-test/src/test_mysql.py
new file mode 100644
index 0000000..d54b3e3
--- /dev/null
+++ b/packages/cli/tests/hyperdrive-test/src/test_mysql.py
@@ -0,0 +1,165 @@
+import uuid
+
+import pymysql
+import pytest
+
+
+def _table_name():
+    return f"test_{uuid.uuid4().hex[:10]}"
+
+
+def _connect(env):
+    hd = env.HYPERDRIVE_MYSQL
+    return pymysql.connect(
+        host=hd.host,
+        port=int(hd.port),
+        user=hd.user,
+        password=hd.password,
+        database=hd.database,
+        unix_socket=False,
+    )
+
+
+@pytest.mark.asyncio
+async def test_connect(env):
+    conn = _connect(env)
+    cur = conn.cursor()
+    cur.execute("SELECT 1")
+    assert cur.fetchone() == (1,)
+    conn.close()
+
+
+@pytest.mark.asyncio
+async def test_create_insert_select(env):
+    conn = _connect(env)
+    table = _table_name()
+    cur = conn.cursor()
+    try:
+        cur.execute(f"CREATE TABLE {table} (id INT AUTO_INCREMENT PRIMARY KEY, name VARCHAR(100), value INT)")
+        cur.execute(f"INSERT INTO {table} (name, value) VALUES (%s, %s)", ("alpha", 1))
+        cur.execute(f"INSERT INTO {table} (name, value) VALUES (%s, %s)", ("beta", 2))
+        conn.commit()
+
+        cur.execute(f"SELECT name, value FROM {table} ORDER BY name")
+        rows = cur.fetchall()
+        assert rows == (("alpha", 1), ("beta", 2))
+    finally:
+        cur.execute(f"DROP TABLE IF EXISTS {table}")
+        conn.commit()
+        conn.close()
+
+
+@pytest.mark.asyncio
+async def test_update(env):
+    conn = _connect(env)
+    table = _table_name()
+    cur = conn.cursor()
+    try:
+        cur.execute(f"CREATE TABLE {table} (id INT AUTO_INCREMENT PRIMARY KEY, name VARCHAR(100), value INT)")
+        cur.execute(f"INSERT INTO {table} (name, value) VALUES (%s, %s)", ("alpha", 1))
+        conn.commit()
+
+        cur.execute(f"UPDATE {table} SET value = value + 10 WHERE name = %s", ("alpha",))
+        conn.commit()
+
+        cur.execute(f"SELECT value FROM {table} WHERE name = %s", ("alpha",))
+        assert cur.fetchone() == (11,)
+    finally:
+        cur.execute(f"DROP TABLE IF EXISTS {table}")
+        conn.commit()
+        conn.close()
+
+
+@pytest.mark.asyncio
+async def test_delete(env):
+    conn = _connect(env)
+    table = _table_name()
+    cur = conn.cursor()
+    try:
+        cur.execute(f"CREATE TABLE {table} (id INT AUTO_INCREMENT PRIMARY KEY, name VARCHAR(100))")
+        cur.execute(f"INSERT INTO {table} (name) VALUES (%s)", ("alpha",))
+        cur.execute(f"INSERT INTO {table} (name) VALUES (%s)", ("beta",))
+        conn.commit()
+
+        cur.execute(f"DELETE FROM {table} WHERE name = %s", ("alpha",))
+        conn.commit()
+
+        cur.execute(f"SELECT name FROM {table}")
+        rows = cur.fetchall()
+        assert rows == (("beta",),)
+    finally:
+        cur.execute(f"DROP TABLE IF EXISTS {table}")
+        conn.commit()
+        conn.close()
+
+
+@pytest.mark.asyncio
+async def test_transaction_rollback(env):
+    conn = _connect(env)
+    table = _table_name()
+    cur = conn.cursor()
+    try:
+        cur.execute(f"CREATE TABLE {table} (id INT AUTO_INCREMENT PRIMARY KEY, name VARCHAR(100)) ENGINE=InnoDB")
+        conn.commit()
+
+        cur.execute(f"INSERT INTO {table} (name) VALUES (%s)", ("should_disappear",))
+        conn.rollback()
+
+        cur.execute(f"SELECT COUNT(*) FROM {table}")
+        assert cur.fetchone() == (0,)
+    finally:
+        cur.execute(f"DROP TABLE IF EXISTS {table}")
+        conn.commit()
+        conn.close()
+
+
+@pytest.mark.asyncio
+async def test_multiple_data_types(env):
+    conn = _connect(env)
+    table = _table_name()
+    cur = conn.cursor()
+    try:
+        cur.execute(
+            f"CREATE TABLE {table} ("
+            "id INT AUTO_INCREMENT PRIMARY KEY, "
+            "text_col VARCHAR(255), "
+            "int_col INT, "
+            "float_col DOUBLE, "
+            "bool_col BOOLEAN)"
+        )
+        cur.execute(
+            f"INSERT INTO {table} (text_col, int_col, float_col, bool_col) "
+            "VALUES (%s, %s, %s, %s)",
+            ("hello", 42, 3.14, True),
+        )
+        conn.commit()
+
+        cur.execute(f"SELECT text_col, int_col, float_col, bool_col FROM {table}")
+        row = cur.fetchone()
+        assert row[0] == "hello"
+        assert row[1] == 42
+        assert abs(row[2] - 3.14) < 0.001
+        assert row[3] == 1
+    finally:
+        cur.execute(f"DROP TABLE IF EXISTS {table}")
+        conn.commit()
+        conn.close()
+
+
+@pytest.mark.asyncio
+async def test_executemany(env):
+    conn = _connect(env)
+    table = _table_name()
+    cur = conn.cursor()
+    try:
+        cur.execute(f"CREATE TABLE {table} (id INT AUTO_INCREMENT PRIMARY KEY, name VARCHAR(100))")
+        cur.executemany(f"INSERT INTO {table} (name) VALUES (%s)", [("a",), ("b",), ("c",)])
+        conn.commit()
+
+        cur.execute(f"SELECT name FROM {table} ORDER BY name")
+        rows = cur.fetchall()
+        assert rows == (("a",), ("b",), ("c",))
+    finally:
+        cur.execute(f"DROP TABLE IF EXISTS {table}")
+        conn.commit()
+        conn.close()
diff --git a/packages/cli/tests/hyperdrive-test/src/test_postgresql.py b/packages/cli/tests/hyperdrive-test/src/test_postgresql.py
new file mode 100644
index 0000000..cd9e9f3
--- /dev/null
+++ b/packages/cli/tests/hyperdrive-test/src/test_postgresql.py
@@ -0,0 +1,145 @@
+import uuid
+
+import pg8000
+import pytest
+
+
+def _table_name():
+    return f"test_{uuid.uuid4().hex[:10]}"
+
+
+def _connect(env):
+    hd = env.HYPERDRIVE_PG
+    return pg8000.connect(
+        host=hd.host,
+        port=int(hd.port),
+        user=hd.user,
+        password=hd.password,
+        database=hd.database,
+    )
+
+
+@pytest.mark.asyncio
+async def test_connect(env):
+    conn = _connect(env)
+    cur = conn.cursor()
+    cur.execute("SELECT 1")
+    assert cur.fetchone() == [1]
+    conn.close()
+
+
+@pytest.mark.asyncio
+async def test_create_insert_select(env):
+    conn = _connect(env)
+    table = _table_name()
+    cur = conn.cursor()
+    try:
+        cur.execute(f"CREATE TABLE {table} (id SERIAL PRIMARY KEY, name TEXT, value INT)")
+        cur.execute(f"INSERT INTO {table} (name, value) VALUES (%s, %s)", ("alpha", 1))
+        cur.execute(f"INSERT INTO {table} (name, value) VALUES (%s, %s)", ("beta", 2))
+        conn.commit()
+
+        cur.execute(f"SELECT name, value FROM {table} ORDER BY name")
+        rows = cur.fetchall()
+        assert rows == [["alpha", 1], ["beta", 2]]
+    finally:
+        cur.execute(f"DROP TABLE IF EXISTS {table}")
+        conn.commit()
+        conn.close()
+
+
+@pytest.mark.asyncio
+async def test_update(env):
+    conn = _connect(env)
+    table = _table_name()
+    cur = conn.cursor()
+    try:
+        cur.execute(f"CREATE TABLE {table} (id SERIAL PRIMARY KEY, name TEXT, value INT)")
+        cur.execute(f"INSERT INTO {table} (name, value) VALUES (%s, %s)", ("alpha", 1))
+        conn.commit()
+
+        cur.execute(f"UPDATE {table} SET value = value + 10 WHERE name = %s", ("alpha",))
+        conn.commit()
+
+        cur.execute(f"SELECT value FROM {table} WHERE name = %s", ("alpha",))
+        assert cur.fetchone() == [11]
+    finally:
+        cur.execute(f"DROP TABLE IF EXISTS {table}")
+        conn.commit()
+        conn.close()
+
+
+@pytest.mark.asyncio
+async def test_delete(env):
+    conn = _connect(env)
+    table = _table_name()
+    cur = conn.cursor()
+    try:
+        cur.execute(f"CREATE TABLE {table} (id SERIAL PRIMARY KEY, name TEXT)")
+        cur.execute(f"INSERT INTO {table} (name) VALUES (%s)", ("alpha",))
+        cur.execute(f"INSERT INTO {table} (name) VALUES (%s)", ("beta",))
+        conn.commit()
+
+        cur.execute(f"DELETE FROM {table} WHERE name = %s", ("alpha",))
+        conn.commit()
+
+        cur.execute(f"SELECT name FROM {table}")
+        rows = cur.fetchall()
+        assert rows == [["beta"]]
+    finally:
+        cur.execute(f"DROP TABLE IF EXISTS {table}")
+        conn.commit()
+        conn.close()
+
+
+@pytest.mark.asyncio
+async def test_transaction_rollback(env):
+    conn = _connect(env)
+    table = _table_name()
+    cur = conn.cursor()
+    try:
+        cur.execute(f"CREATE TABLE {table} (id SERIAL PRIMARY KEY, name TEXT)")
+        conn.commit()
+
+        cur.execute(f"INSERT INTO {table} (name) VALUES (%s)", ("should_disappear",))
+        conn.rollback()
+
+        cur.execute(f"SELECT COUNT(*) FROM {table}")
+        assert cur.fetchone() == [0]
+    finally:
+        cur.execute(f"DROP TABLE IF EXISTS {table}")
+        conn.commit()
+        conn.close()
+
+
+@pytest.mark.asyncio
+async def test_multiple_data_types(env):
+    conn = _connect(env)
+    table = _table_name()
+    cur = conn.cursor()
+    try:
+        cur.execute(
+            f"CREATE TABLE {table} ("
+            "id SERIAL PRIMARY KEY, "
+            "text_col TEXT, "
+            "int_col INT, "
+            "float_col DOUBLE PRECISION, "
+            "bool_col BOOLEAN)"
+        )
+        cur.execute(
+            f"INSERT INTO {table} (text_col, int_col, float_col, bool_col) "
+            "VALUES (%s, %s, %s, %s)",
+            ("hello", 42, 3.14, True),
+        )
+        conn.commit()
+
+        cur.execute(f"SELECT text_col, int_col, float_col, bool_col FROM {table}")
+        row = cur.fetchone()
+        assert row[0] == "hello"
+        assert row[1] == 42
+        assert abs(row[2] - 3.14) < 0.001
+        assert row[3] is True
+    finally:
+        cur.execute(f"DROP TABLE IF EXISTS {table}")
+        conn.commit()
+        conn.close()
diff --git a/packages/cli/tests/hyperdrive-test/src/worker.py b/packages/cli/tests/hyperdrive-test/src/worker.py
new file mode 100644
index 0000000..a036aca
--- /dev/null
+++ b/packages/cli/tests/hyperdrive-test/src/worker.py
@@ -0,0 +1,163 @@
+"""Hyperdrive test worker.
+
+Each database driver suite lives in a `test_<suite>.py` module written as ordinary pytest
+tests (see test_postgresql.py, test_mysql.py). The `/run-tests/<suite>` endpoint runs pytest
+against that module inside workerd and returns per-test results as JSON, which the host-side
+test_hyperdrive.py maps onto individual pytest cases.
+
+To add a new driver: create `src/test_<driver>.py` with pytest tests.
+"""
+
+import asyncio
+import importlib.util
+import sys
+from asyncio import InvalidStateError
+
+import pytest
+from pyodide.webloop import WebLoop
+from workers import Response, WorkerEntrypoint
+
+
+async def _noop(*args):
+    pass
+
+
+# pytest-asyncio relies on these but in Pyodide < 0.29 WebLoop does not implement them.
+WebLoop.shutdown_asyncgens = _noop
+WebLoop.shutdown_default_executor = _noop
+
+# Pyodide 0.26.0a2's WebLoop causes InvalidStateError when the
+# _cancel_all_tasks calls task.exception() on done-but-not-cancelled tasks.
+# Replace with a version that cancels tasks but tolerates that error.
+if sys.version_info < (3, 13):
+
+    def _cancel_all_tasks(loop):
+        to_cancel = asyncio.tasks.all_tasks(loop)
+        if not to_cancel:
+            return
+        for task in to_cancel:
+            task.cancel()
+        loop.run_until_complete(
+            asyncio.tasks.gather(*to_cancel, return_exceptions=True)
+        )
+        for task in to_cancel:
+            if task.cancelled():
+                continue
+            try:
+                if task.exception() is not None:
+                    loop.call_exception_handler(
+                        {
+                            "message": "unhandled exception during asyncio.run() shutdown",
+                            "exception": task.exception(),
+                            "task": task,
+                        }
+                    )
+            # Note: This exception catch is added from the original implementation
+            except (InvalidStateError, RuntimeError):
+                pass
+
+    asyncio.runners._cancel_all_tasks = _cancel_all_tasks  # type: ignore[attr-defined]
+
+
+class ResultCollector:
+    """pytest plugin that records each test's outcome keyed by its short name.
+
+    The "test_" prefix is stripped so keys match the names registered in
+    tests/test_hyperdrive.py (e.g. test_basic_query -> "basic_query").
+    """
+
+    def __init__(self):
+        self.results = {}
+
+    @staticmethod
+    def _key(item):
+        name = item.name
+        return name[len("test_") :] if name.startswith("test_") else name
+
+    @pytest.hookimpl(hookwrapper=True)
+    def pytest_runtest_makereport(self, item, call):
+        outcome = yield
+        report = outcome.get_result()
+        key = self._key(item)
+
+        if report.when == "call":
+            if report.passed:
+                self.results[key] = {"status": "passed"}
+            elif report.skipped:
+                self.results[key] = {
+                    "status": "skipped",
+                    "reason": str(report.longrepr),
+                }
+            elif report.failed:
+                excinfo = call.excinfo
+                if excinfo is not None and excinfo.errisinstance(AssertionError):
+                    self.results[key] = {
+                        "status": "failed",
+                        "error": str(excinfo.value),
+                    }
+                else:
+                    self.results[key] = {
+                        "status": "error",
+                        "error": f"{excinfo.typename}: {excinfo.value}"
+                        if excinfo is not None
+                        else "unknown error",
+                        "traceback": report.longreprtext,
+                    }
+        elif report.when in ("setup", "teardown") and report.skipped:
+            self.results[key] = {
+                "status": "skipped",
+                "reason": str(report.longrepr),
+            }
+        elif report.when in ("setup", "teardown") and report.failed:
+            self.results[key] = {
+                "status": "error",
+                "error": report.longreprtext,
+                "traceback": report.longreprtext,
+            }
+
+
+class EnvPlugin:
+    def __init__(self, env):
+        self._env = env
+
+    @pytest.fixture
+    def env(self):
+        return self._env
+
+
+class Default(WorkerEntrypoint):
+    async def fetch(self, request):
+
+        # TODO: remove these when workerd releases a new version with this feature enabled by default
+        import pyodide_js
+        from workers import import_from_javascript
+
+        await pyodide_js._api.initializeNodeSockFS(
+            import_from_javascript("cloudflare:sockets").connect
+        )
+
+        from urllib.parse import urlparse
+
+        path = urlparse(request.url).path
+
+        if path.startswith("/run-tests/"):
+            suite_name = path[len("/run-tests/") :]
+            return self._run_suite(suite_name)
+        if path == "/health":
+            return Response.json({"ok": True})
+        return Response.json({"error": "not found"}, status=404)
+
+    def _run_suite(self, suite_name):
+        module = f"test_{suite_name}"
+        if importlib.util.find_spec(module) is None:
+            return Response.json(
+                {"error": f"Unknown suite '{suite_name}' (no module '{module}')"},
+                status=404,
+            )
+
+        collector = ResultCollector()
+        pytest.main(
+            ["--pyargs", module, "-p", "no:cacheprovider"],
+            plugins=[collector, EnvPlugin(self.env)],
+        )
+        return Response.json(collector.results)
diff --git a/packages/cli/tests/hyperdrive-test/wrangler.jsonc b/packages/cli/tests/hyperdrive-test/wrangler.jsonc
new file mode 100644
index 0000000..21871b2
--- /dev/null
+++ b/packages/cli/tests/hyperdrive-test/wrangler.jsonc
@@ -0,0 +1,23 @@
+{
+    "name": "hyperdrive-test",
+    "main": "src/worker.py",
+    "compatibility_date": "%COMPAT_DATE",
+    "compatibility_flags": [
+        "python_workers"
+    ],
+    // Hyperdrive bindings for local development.
+    // localConnectionString lets pywrangler dev connect directly to a local database.
+    // In CI, these point to Docker-hosted MySQL and PostgreSQL instances.
+    "hyperdrive": [
+        {
+            "binding": "HYPERDRIVE_PG",
+            "id": "00000000-0000-0000-0000-000000000001",
+            "localConnectionString": "postgres://testuser:testpass@127.0.0.1:5432/testdb"
+        },
+        {
+            "binding": "HYPERDRIVE_MYSQL",
+            "id": "00000000-0000-0000-0000-000000000002",
+            "localConnectionString": "mysql://testuser:testpass@127.0.0.1:3306/testdb"
+        }
+    ]
+}
diff --git a/packages/cli/tests/test_hyperdrive.py b/packages/cli/tests/test_hyperdrive.py
new file mode 100644
index 0000000..a1cc731
--- /dev/null
+++ b/packages/cli/tests/test_hyperdrive.py
@@ -0,0 +1,213 @@
+"""Tests for Hyperdrive database connectivity.
+
+These tests require running database instances. They are disabled locally by default
+because the database setup is non-trivial. In CI, databases are provided via Docker services.
+
+To run locally:
+
+    1. Start MySQL and PostgreSQL (e.g. via Docker):
+
+        docker run -d --name test-mysql -p 3306:3306 \
+            -e MYSQL_ROOT_PASSWORD=rootpass \
+            -e MYSQL_USER=testuser \
+            -e MYSQL_PASSWORD=testpass \
+            -e MYSQL_DATABASE=testdb \
+            mysql:8.0 --default-authentication-plugin=mysql_native_password
+
+        docker run -d --name test-postgres -p 5432:5432 \
+            -e POSTGRES_USER=testuser \
+            -e POSTGRES_PASSWORD=testpass \
+            -e POSTGRES_DB=testdb \
+            -e POSTGRES_HOST_AUTH_METHOD=md5 \
+            -e POSTGRES_INITDB_ARGS="--auth-host=md5" \
+            postgres:16
+
+    2. Run the tests:
+
+        uv run pytest tests/test_hyperdrive.py -v
+
+    3. Clean up:
+
+        docker rm -f test-mysql test-postgres
+"""
+
+import ast
+import functools
+import os
+import shutil
+import socket
+import subprocess
+import time
+from collections.abc import Callable, Generator
+from pathlib import Path
+from typing import Any, Literal, TypedDict
+
+import pytest
+import requests
+from conftest import COMPAT_CONFIGS, inject_compat_flags, replace_compat_date
+
+TEST_DIR: Path = Path(__file__).parent
+HYPERDRIVE_TEST_DIR: Path = TEST_DIR / "hyperdrive-test"
+HYPERDRIVE_SRC_DIR: Path = HYPERDRIVE_TEST_DIR / "src"
+WORKERS_PY: Path = TEST_DIR.parent
+WORKERS_RUNTIME_SDK: Path = WORKERS_PY.parent / "runtime-sdk" / "src"
+
+DEV_STARTUP_TIMEOUT: int = 120
+DEV_POLL_INTERVAL: float = 0.5
+
+COMPAT_CONFIGS_SOCKET_SUPPORT = [c for c in COMPAT_CONFIGS if c.compat_date not in  ("2025-09-01", "2026-01-01")]
+
+
+class HyperdriveTestResult(TypedDict):
+    status: Literal["passed", "failed", "error", "skipped"]
+    error: str
+    traceback: str
+    reason: str
+
+
+SuiteResults = dict[str, HyperdriveTestResult]
+
+
+def _get_free_port() -> int:
+    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
+        s.bind(("127.0.0.1", 0))
+        return s.getsockname()[1]
+
+
+def _wait_for_ready(process: subprocess.Popen[str], base_url: str) -> None:
+    deadline = time.time() + DEV_STARTUP_TIMEOUT
+    while time.time() < deadline:
+        if process.poll() is not None:
+            stdout = process.stdout.read() if process.stdout else ""
+            pytest.fail(
+                f"pywrangler dev exited early with code {process.returncode}\n"
+                f"stdout: {stdout}"
+            )
+        try:
+            resp = requests.get(f"{base_url}/health", timeout=2)
+            if resp.ok:
+                return
+        except (requests.ConnectionError, requests.Timeout):
+            time.sleep(DEV_POLL_INTERVAL)
+
+    process.kill()
+    process.wait()
+    pytest.fail(f"pywrangler dev did not become ready within {DEV_STARTUP_TIMEOUT}s")
+
+
+@pytest.fixture(
+    scope="module",
+    params=COMPAT_CONFIGS_SOCKET_SUPPORT,
+    ids=[c.python_version for c in COMPAT_CONFIGS_SOCKET_SUPPORT],
+)
+def compat_config(request: pytest.FixtureRequest) -> CompatConfig:
+    return request.param
+
+
+@pytest.fixture(scope="module")
+def dev_server(
+    tmp_path_factory: pytest.TempPathFactory, compat_config: CompatConfig
+) -> Generator[str]:
+    tmp_path = tmp_path_factory.mktemp("hyperdrive_test")
+    target = tmp_path / "hyperdrive-test"
+    shutil.copytree(HYPERDRIVE_TEST_DIR, target)
+    env = os.environ | {"_PYODIDE_EXTRA_MOUNTS": str(tmp_path)}
+
+    wrangler_jsonc = target / "wrangler.jsonc"
+    replace_compat_date(wrangler_jsonc, compat_config.compat_date)
+    inject_compat_flags(wrangler_jsonc, compat_config.extra_compat_flags)
+
+    subprocess.run(
+        ["uv", "run", "--with", WORKERS_PY, "pywrangler", "sync"],
+        cwd=target,
+        check=True,
+        env=env,
+    )
+
+    shutil.copytree(WORKERS_RUNTIME_SDK, target / "python_modules", dirs_exist_ok=True)
+
+    port: int = _get_free_port()
+    base_url: str = f"http://127.0.0.1:{port}"
+
+    process = subprocess.Popen(
+        [
+            "uv",
+            "run",
+            "--with",
+            WORKERS_PY,
+            "pywrangler",
+            "dev",
+            "--port",
+            str(port),
+            "--persist-to",
+            str(tmp_path / "state"),
+        ],
+        cwd=target,
+        stdout=subprocess.PIPE,
+        stderr=subprocess.STDOUT,
+        text=True,
+        env=env,
+    )
+
+    _wait_for_ready(process, base_url)
+    yield base_url
+
+    process.terminate()
+    try:
+        process.wait(timeout=10)
+    except subprocess.TimeoutExpired:
+        process.kill()
+        process.wait()
+
+
+@functools.cache
+def _get_test_results(dev_server: str, suite: str) -> SuiteResults:
+    resp = requests.get(f"{dev_server}/run-tests/{suite}", timeout=60)
+    assert resp.ok, f"Suite '{suite}' returned {resp.status_code}: {resp.text}"
+    return resp.json()
+
+
+def _make_test(suite: str, test_name: str) -> Callable:
+    def test_fn(self: Any, dev_server: str) -> None:
+        results = _get_test_results(dev_server, suite)
+        result: HyperdriveTestResult | None = results.get(test_name)
+        assert result is not None, f"Test {suite}::{test_name} not found in results"
+        if result["status"] == "skipped":
+            pytest.skip(result.get("reason", ""))
+        elif result["status"] == "failed":
+            pytest.fail(result["error"])
+        elif result["status"] == "error":
+            pytest.fail(f"{result['error']}\n{result.get('traceback', '')}")
+
+    test_fn.__name__ = f"test_{test_name}"
+    return test_fn
+
+
+def binding_suite(suite: str, tests: list[str]) -> type:
+    return type(
+        f"Test{suite.upper()}",
+        (),
+        {f"test_{name}": _make_test(suite, name) for name in tests},
+    )
+
+
+def _discover_test_names(module_path: Path) -> list[str]:
+    tree = ast.parse(module_path.read_text())
+    return [
+        node.name[len("test_") :]
+        for node in tree.body
+        if isinstance(node, ast.FunctionDef | ast.AsyncFunctionDef)
+        and node.name.startswith("test_")
+    ]
+
+
+def _discover_suites() -> dict[str, list[str]]:
+    return {
+        module_path.stem[len("test_") :]: _discover_test_names(module_path)
+        for module_path in sorted(HYPERDRIVE_SRC_DIR.glob("test_*.py"))
+    }
+
+
+for _suite, _test_names in _discover_suites().items():
+    _suite_cls = binding_suite(_suite, _test_names)
+    globals()[_suite_cls.__name__] = _suite_cls