Update PRAGMA deny-list integration tests for busy_timeout interception

The pre-existing pin asserted PRAGMA busy_timeout (both setter and
getter forms) raises DatabaseError("not authorized") from the
server's VFS authorizer. With the dbapi-layer interception now in
place, the PRAGMA is intercepted at the cursor's execute path
BEFORE the wire send — the server never sees it, so the
"not authorized" error never fires.

Drop the busy_timeout entries from _DENIED_PRAGMAS_SYNC and add
positive pins for the new behaviour: the setter form updates
connection._busy_timeout and emits the value as a row; the getter
form emits the current value. Tests run against the live cluster
to verify the user-observable behaviour end-to-end (the unit-test
test_pragma_busy_timeout_intercept.py covers the interception
logic against a fake cursor).

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: antoineleclair

tests/integration/test_pragma_deny_list.py

@@ -30,8 +30,18 @@
     # Both-form denials (read + write)
     "PRAGMA wal_checkpoint",
     "PRAGMA wal_checkpoint(FULL)",
-    "PRAGMA busy_timeout = 1000",
-    "PRAGMA busy_timeout",
+    # ``PRAGMA busy_timeout`` was historically in this deny list
+    # (C VFS rejects it server-side via SQLITE_AUTH), but the
+    # dbapi-layer now intercepts the PRAGMA at the cursor's
+    # execute path BEFORE the wire send — the interception updates
+    # the connection's busy_timeout (stdlib parity) without ever
+    # reaching the server. See
+    # ``test_pragma_busy_timeout_intercept.py`` for the
+    # interception pin and ``_pragma_intercept.py`` for the
+    # implementation. The busy_timeout entries are removed from
+    # this deny-list-pin because the user-observable behaviour is
+    # no longer ``DatabaseError("not authorized")`` — it's a
+    # successful PRAGMA that returns the current value as a row.
     "PRAGMA read_uncommitted = 1",
     "PRAGMA read_uncommitted",
     "PRAGMA locking_mode = EXCLUSIVE",
@@ -58,3 +68,55 @@ async def test_denied_pragma_raises_database_error_async(cluster_address: str, p
             await cur.execute(pragma)
     finally:
         await conn.close()
+
+
+# Pre-existing pin was "PRAGMA busy_timeout raises DatabaseError(not
+# authorized)" — that pin pre-dated the dbapi-layer interception.
+# The busy_timeout entries are now intercepted client-side: the
+# setter updates the connection's busy_timeout and emits the value
+# as a row; the getter emits the current value. These tests pin the
+# new user-observable behaviour against the live cluster (the
+# unit-test ``test_pragma_busy_timeout_intercept.py`` pins it
+# against a fake cursor).
+
+
+@pytest.mark.integration
+@pytest.mark.parametrize("pragma", ["PRAGMA busy_timeout = 1000", "PRAGMA busy_timeout"])
+def test_pragma_busy_timeout_intercepted_sync(cluster_address: str, pragma: str) -> None:
+    """``PRAGMA busy_timeout`` is intercepted at the cursor layer
+    BEFORE the wire send — the server never sees it. The setter
+    updates the connection's busy_timeout; both forms emit the
+    (possibly updated) value as a row."""
+    with dqlitedbapi.connect(cluster_address, timeout=2.0) as conn:
+        cur = conn.cursor()
+        cur.execute(pragma)
+        row = cur.fetchone()
+        assert row is not None
+        # Setter sets to 1000ms (1.0s); getter returns whatever the
+        # connection's default is (5000ms from the kwarg default).
+        if "=" in pragma:
+            assert row[0] == 1000
+            # Verify the connection state was actually updated.
+            assert conn._busy_timeout == 1.0
+        else:
+            # Getter — returns the current value (default 5000ms).
+            assert row[0] == 5000
+
+
+@pytest.mark.integration
+@pytest.mark.parametrize("pragma", ["PRAGMA busy_timeout = 1000", "PRAGMA busy_timeout"])
+async def test_pragma_busy_timeout_intercepted_async(cluster_address: str, pragma: str) -> None:
+    """Async sibling — same interception behaviour."""
+    conn = await aconnect(cluster_address)
+    try:
+        cur = conn.cursor()
+        await cur.execute(pragma)
+        row = await cur.fetchone()
+        assert row is not None
+        if "=" in pragma:
+            assert row[0] == 1000
+            assert conn._busy_timeout == 1.0
+        else:
+            assert row[0] == 5000
+    finally:
+        await conn.close()