Skip to content

Commit 238524f

Browse files
Cover AsyncAdaptedCursor fetchone/fetchmany/fetchall/__iter__ directly
Integration tests go through SQLAlchemy Result, which bypasses the adapter's own fetch entry points; the adapter was only tested for execute/close. Add direct unit tests that preload the internal deque and assert fetchone pops left, fetchmany honours the arraysize default, fetchall drains the deque, and __iter__ yields each row exactly once. Any non-SQLAlchemy consumer that touches these methods now has regression coverage. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
1 parent f35625f commit 238524f

1 file changed

Lines changed: 47 additions & 0 deletions

File tree

tests/test_async_adapter.py

Lines changed: 47 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -254,3 +254,50 @@ def test_iter_returns_iterator(self) -> None:
254254
sig = inspect.signature(AsyncAdaptedCursor.__iter__)
255255
# Accept either the Iterator annotation object or the stringified form.
256256
assert "Iterator" in str(sig.return_annotation)
257+
258+
259+
class TestAioAdapterCursorFetchMethods:
260+
"""The adapter's fetch* methods buffer rows into a deque populated
261+
by ``execute``. The integration tests exercise them via SQLAlchemy
262+
Result, which bypasses these entry points entirely — so direct
263+
unit tests pin the semantics for non-SQLAlchemy consumers.
264+
"""
265+
266+
def _cursor_with_rows(self, rows: list[tuple[object, ...]]) -> object:
267+
from collections import deque
268+
269+
from sqlalchemydqlite.aio import AsyncAdaptedCursor
270+
271+
cursor = AsyncAdaptedCursor.__new__(AsyncAdaptedCursor)
272+
cursor._rows = deque(rows)
273+
cursor.arraysize = 1
274+
return cursor
275+
276+
def test_fetchone_pops_left(self) -> None:
277+
cursor = self._cursor_with_rows([(1,), (2,), (3,)])
278+
assert cursor.fetchone() == (1,) # type: ignore[attr-defined]
279+
assert cursor.fetchone() == (2,) # type: ignore[attr-defined]
280+
assert cursor.fetchone() == (3,) # type: ignore[attr-defined]
281+
assert cursor.fetchone() is None # type: ignore[attr-defined]
282+
283+
def test_fetchmany_default_uses_arraysize(self) -> None:
284+
cursor = self._cursor_with_rows([(1,), (2,), (3,), (4,)])
285+
cursor.arraysize = 2 # type: ignore[attr-defined]
286+
assert list(cursor.fetchmany()) == [(1,), (2,)] # type: ignore[attr-defined]
287+
# Remaining rows are still available.
288+
assert list(cursor.fetchmany()) == [(3,), (4,)] # type: ignore[attr-defined]
289+
290+
def test_fetchmany_explicit_size(self) -> None:
291+
cursor = self._cursor_with_rows([(1,), (2,), (3,)])
292+
assert list(cursor.fetchmany(2)) == [(1,), (2,)] # type: ignore[attr-defined]
293+
assert list(cursor.fetchmany(10)) == [(3,)] # type: ignore[attr-defined]
294+
295+
def test_fetchall_drains_deque(self) -> None:
296+
cursor = self._cursor_with_rows([(1,), (2,), (3,)])
297+
assert cursor.fetchall() == [(1,), (2,), (3,)] # type: ignore[attr-defined]
298+
assert len(cursor._rows) == 0 # type: ignore[attr-defined]
299+
300+
def test_iter_drains_deque(self) -> None:
301+
cursor = self._cursor_with_rows([(1,), (2,), (3,)])
302+
assert list(cursor) == [(1,), (2,), (3,)]
303+
assert len(cursor._rows) == 0 # type: ignore[attr-defined]

0 commit comments

Comments
 (0)