Fix stale rows and cursor leak in async adapter

Clear _rows deque after non-query execute() and executemany() to prevent
stale data from a previous SELECT leaking through fetchone(). Wrap cursor
operations in try/finally to ensure the underlying cursor is always closed.

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

Author: antoineleclair

src/sqlalchemydqlite/aio.py

@@ -39,29 +39,34 @@ def close(self) -> None:
 
     def execute(self, operation: str, parameters: Any = None) -> Any:
         cursor = self._connection.cursor()
-        if parameters is not None:
-            await_only(cursor.execute(operation, parameters))
-        else:
-            await_only(cursor.execute(operation))
-
-        if cursor.description:
-            self.description = cursor.description
-            self.lastrowid = self.rowcount = -1
-            self._rows = deque(await_only(cursor.fetchall()))
-        else:
-            self.description = None
-            self.lastrowid = cursor.lastrowid
-            self.rowcount = cursor.rowcount
-
-        await_only(cursor.close())
+        try:
+            if parameters is not None:
+                await_only(cursor.execute(operation, parameters))
+            else:
+                await_only(cursor.execute(operation))
+
+            if cursor.description:
+                self.description = cursor.description
+                self.lastrowid = self.rowcount = -1
+                self._rows = deque(await_only(cursor.fetchall()))
+            else:
+                self.description = None
+                self.lastrowid = cursor.lastrowid
+                self.rowcount = cursor.rowcount
+                self._rows.clear()
+        finally:
+            await_only(cursor.close())
 
     def executemany(self, operation: str, seq_of_parameters: Any) -> Any:
         cursor = self._connection.cursor()
-        await_only(cursor.executemany(operation, seq_of_parameters))
-        self.description = None
-        self.lastrowid = cursor.lastrowid
-        self.rowcount = cursor.rowcount
-        await_only(cursor.close())
+        try:
+            await_only(cursor.executemany(operation, seq_of_parameters))
+            self.description = None
+            self.lastrowid = cursor.lastrowid
+            self.rowcount = cursor.rowcount
+            self._rows.clear()
+        finally:
+            await_only(cursor.close())
 
     def fetchone(self) -> Any:
         if self._rows:

tests/test_async_adapter.py

@@ -0,0 +1,103 @@
+"""Tests for async adapter cursor behavior."""
+
+import ast
+import inspect
+import textwrap
+from collections import deque
+from unittest.mock import MagicMock, patch
+
+from sqlalchemydqlite.aio import AsyncAdaptedConnection, AsyncAdaptedCursor
+
+
+def _make_cursor() -> AsyncAdaptedCursor:
+    """Create an AsyncAdaptedCursor with a mocked connection."""
+    mock_conn = MagicMock()
+    adapted_conn = AsyncAdaptedConnection.__new__(AsyncAdaptedConnection)
+    adapted_conn._connection = mock_conn
+    cursor = AsyncAdaptedCursor(adapted_conn)
+    return cursor
+
+
+def _run_sync(coro_or_value: object) -> object:
+    """Replacement for await_only that resolves coroutines synchronously."""
+    if hasattr(coro_or_value, "__await__") or hasattr(coro_or_value, "cr_await"):
+        # It's a coroutine -- we can't actually await it in sync context,
+        # but our mocks return plain values, so send(None) is enough.
+        try:
+            coro_or_value.send(None)  # type: ignore[union-attr]
+        except StopIteration as e:
+            return e.value
+    return coro_or_value
+
+
+class TestAsyncAdaptedCursorRowsCleared:
+    def test_rows_cleared_after_non_query_execute(self) -> None:
+        """After a SELECT then an INSERT, fetchone() must return None."""
+        cursor = _make_cursor()
+
+        # Simulate that a previous SELECT populated _rows
+        cursor._rows = deque([(1, "alice")])
+
+        # Set up a mock inner cursor that returns no description (DML)
+        mock_inner = MagicMock()
+        mock_inner.description = None
+        mock_inner.lastrowid = 1
+        mock_inner.rowcount = 1
+        mock_inner.execute.return_value = None
+        mock_inner.close.return_value = None
+        cursor._connection.cursor.return_value = mock_inner
+
+        with patch("sqlalchemydqlite.aio.await_only", side_effect=_run_sync):
+            cursor.execute("INSERT INTO t VALUES (1)")
+
+        result = cursor.fetchone()
+        assert result is None, f"Expected None after non-query execute, got {result}"
+
+    def test_rows_cleared_after_executemany(self) -> None:
+        """After executemany(), fetchone() must return None."""
+        cursor = _make_cursor()
+
+        # Simulate that a previous SELECT populated _rows
+        cursor._rows = deque([(1, "alice"), (2, "bob")])
+
+        mock_inner = MagicMock()
+        mock_inner.lastrowid = 3
+        mock_inner.rowcount = 2
+        mock_inner.executemany.return_value = None
+        mock_inner.close.return_value = None
+        cursor._connection.cursor.return_value = mock_inner
+
+        with patch("sqlalchemydqlite.aio.await_only", side_effect=_run_sync):
+            cursor.executemany("INSERT INTO t VALUES (?)", [(1,), (2,)])
+
+        result = cursor.fetchone()
+        assert result is None, f"Expected None after executemany, got {result}"
+
+
+def _has_finally_with_close(func: object) -> bool:
+    """Check if a function has cursor.close() inside a finally block."""
+    source = textwrap.dedent(inspect.getsource(func))
+    tree = ast.parse(source)
+
+    for node in ast.walk(tree):
+        if isinstance(node, ast.Try) and node.finalbody:
+            for stmt in ast.walk(node):
+                if isinstance(stmt, ast.Call):
+                    func_node = stmt.func
+                    if isinstance(func_node, ast.Attribute) and func_node.attr == "close":
+                        return True
+    return False
+
+
+class TestAsyncAdaptedCursorCleanup:
+    def test_cursor_closed_on_execute_error(self) -> None:
+        """Underlying cursor must be closed even if execute() raises."""
+        assert _has_finally_with_close(AsyncAdaptedCursor.execute), (
+            "cursor.close() should be in a finally block to prevent leaks on error"
+        )
+
+    def test_executemany_cursor_closed_on_error(self) -> None:
+        """Underlying cursor must be closed even if executemany() raises."""
+        assert _has_finally_with_close(AsyncAdaptedCursor.executemany), (
+            "cursor.close() should be in a finally block to prevent leaks on error"
+        )