Pin SAVEPOINT round-trip and concurrent-cursor visibility at the dbapi layer

Two integration test additions:

- SAVEPOINT round-trip with partial rollback (sync) and RELEASE
  (async). ISSUE-701 added the equivalent at the SQLAlchemy adapter
  layer; the dbapi layer had no analogous pin and a future change
  to the SQL classifier or in-transaction tracking could silently
  break SAVEPOINT routing.

- Concurrent-cursor visibility per PEP 249 §6.1.2. Two cursors on the
  same connection share that connection's transaction visibility:
  cursor B sees cursor A's uncommitted INSERT, and B sees A's
  ROLLBACK. The pin guards against a future regression that would
  route cursors through separate underlying client connections.

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

Author: antoineleclair

tests/integration/test_concurrent_cursors_visibility.py

@@ -0,0 +1,55 @@
+"""PEP 249 §6.1.2: cursors created from the same connection share that
+connection's transaction visibility. Pin that two cursors on one
+connection see each other's uncommitted writes inside a transaction,
+and that ROLLBACK on one cursor is observed by the other.
+"""
+
+from __future__ import annotations
+
+import pytest
+
+from dqlitedbapi import connect
+from dqlitedbapi.aio import aconnect
+
+
+def test_sync_two_cursors_share_uncommitted_writes(cluster_address: str) -> None:
+    conn = connect(cluster_address, timeout=2.0)
+    try:
+        setup = conn.cursor()
+        setup.execute("DROP TABLE IF EXISTS test_two_cursors")
+        setup.execute("CREATE TABLE test_two_cursors (n INTEGER PRIMARY KEY)")
+        cur_a = conn.cursor()
+        cur_b = conn.cursor()
+
+        cur_a.execute("BEGIN")
+        cur_a.execute("INSERT INTO test_two_cursors (n) VALUES (1)")
+        cur_b.execute("SELECT n FROM test_two_cursors WHERE n = 1")
+        assert cur_b.fetchone() == (1,)
+        cur_a.execute("ROLLBACK")
+        cur_b.execute("SELECT n FROM test_two_cursors WHERE n = 1")
+        assert cur_b.fetchone() is None
+    finally:
+        conn.close()
+
+
+@pytest.mark.asyncio
+async def test_async_two_cursors_share_uncommitted_writes(
+    cluster_address: str,
+) -> None:
+    conn = await aconnect(cluster_address, timeout=2.0)
+    try:
+        setup = conn.cursor()
+        await setup.execute("DROP TABLE IF EXISTS test_two_cursors_aio")
+        await setup.execute("CREATE TABLE test_two_cursors_aio (n INTEGER PRIMARY KEY)")
+        cur_a = conn.cursor()
+        cur_b = conn.cursor()
+
+        await cur_a.execute("BEGIN")
+        await cur_a.execute("INSERT INTO test_two_cursors_aio (n) VALUES (5)")
+        await cur_b.execute("SELECT n FROM test_two_cursors_aio WHERE n = 5")
+        assert await cur_b.fetchone() == (5,)
+        await cur_a.execute("ROLLBACK")
+        await cur_b.execute("SELECT n FROM test_two_cursors_aio WHERE n = 5")
+        assert await cur_b.fetchone() is None
+    finally:
+        await conn.close()

tests/integration/test_savepoint_dbapi.py

@@ -0,0 +1,54 @@
+"""SAVEPOINT round-trip pins at the dbapi layer.
+
+ISSUE-701 added SAVEPOINT integration coverage for the SQLAlchemy
+adapter. The dbapi layer had no equivalent — a future change to the
+SQL classifier or to the in-transaction flag tracking could silently
+break SAVEPOINT routing.
+"""
+
+from __future__ import annotations
+
+import pytest
+
+from dqlitedbapi import connect
+from dqlitedbapi.aio import aconnect
+
+
+def test_sync_savepoint_roundtrip_partial_rollback(cluster_address: str) -> None:
+    conn = connect(cluster_address, timeout=2.0)
+    try:
+        cur = conn.cursor()
+        cur.execute("DROP TABLE IF EXISTS test_sp_dbapi")
+        cur.execute("CREATE TABLE test_sp_dbapi (n INTEGER PRIMARY KEY)")
+        cur.execute("BEGIN")
+        cur.execute("INSERT INTO test_sp_dbapi (n) VALUES (1)")
+        cur.execute("SAVEPOINT sp1")
+        cur.execute("INSERT INTO test_sp_dbapi (n) VALUES (2)")
+        cur.execute("ROLLBACK TO SAVEPOINT sp1")
+        cur.execute("RELEASE SAVEPOINT sp1")
+        cur.execute("COMMIT")
+
+        cur.execute("SELECT n FROM test_sp_dbapi ORDER BY n")
+        assert cur.fetchall() == [(1,)]
+    finally:
+        conn.close()
+
+
+@pytest.mark.asyncio
+async def test_async_savepoint_roundtrip_release(cluster_address: str) -> None:
+    conn = await aconnect(cluster_address, timeout=2.0)
+    try:
+        cur = conn.cursor()
+        await cur.execute("DROP TABLE IF EXISTS test_sp_dbapi_aio")
+        await cur.execute("CREATE TABLE test_sp_dbapi_aio (n INTEGER PRIMARY KEY)")
+        await cur.execute("BEGIN")
+        await cur.execute("INSERT INTO test_sp_dbapi_aio (n) VALUES (10)")
+        await cur.execute("SAVEPOINT sp_a")
+        await cur.execute("INSERT INTO test_sp_dbapi_aio (n) VALUES (20)")
+        await cur.execute("RELEASE SAVEPOINT sp_a")
+        await cur.execute("COMMIT")
+
+        await cur.execute("SELECT n FROM test_sp_dbapi_aio ORDER BY n")
+        assert await cur.fetchall() == [(10,), (20,)]
+    finally:
+        await conn.close()