Ryan Malloy
461c62c8d3
Adds scroll/random-access methods on Cursor: * scroll(value, mode='relative'|'absolute') — PEP 249 compatible * fetch_first() / fetch_last() — jump to result-set ends * fetch_prior() — step backward (SQL-standard: from past-end yields the last row, matching JDBC ResultSet.previous() semantics) * fetch_absolute(n) — 0-indexed jump; negative n indexes from end * fetch_relative(n) — n-step from current position * rownumber property — current 0-indexed position Implementation: replaced _row_iter (single-pass iterator) with _row_index (random-access index) on the cursor. The result set is already materialized in _rows during execute(); scroll just repositions the index. No new wire protocol needed. For server-side scroll over genuinely huge result sets, SQ_SFETCH (tag 23) would be needed — JDBC has executeScrollFetch (line 3908) but we only need it if someone hits the in-memory materialization ceiling. Phase 18 if so. Out-of-range scroll raises IndexError per PEP 249. Invalid mode strings raise ProgrammingError. fetchall() now correctly returns only the rows from the current position to end (not all rows). 14 new integration tests in test_scroll_cursor.py covering: * fetchone advancing rownumber sequentially * fetch_first reset * fetch_last * fetch_prior including the past-end-to-last-row semantics * fetch_absolute with positive and negative indexes * fetch_relative * PEP 249 scroll(value, mode='relative'/'absolute') * IndexError on out-of-range * ProgrammingError on bad mode * Empty-result-set edge cases * fetchall after partial iteration Total: 69 unit + 177 integration = 246 tests.
234 lines
8.3 KiB
Python
234 lines
8.3 KiB
Python
"""Phase 17 integration tests — scroll cursor API.
|
|
|
|
The cursor materializes all rows on ``execute()`` (current behavior),
|
|
and Phase 17 adds index-based scroll methods on top of that:
|
|
``scroll()``, ``fetch_first()``, ``fetch_last()``, ``fetch_prior()``,
|
|
``fetch_absolute()``, ``fetch_relative()``, plus a ``rownumber`` property.
|
|
|
|
No new wire protocol — pure-Python feature on top of the existing
|
|
materialized result set.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import pytest
|
|
|
|
import informix_db
|
|
from tests.conftest import ConnParams
|
|
|
|
pytestmark = pytest.mark.integration
|
|
|
|
|
|
def _connect(params: ConnParams) -> informix_db.Connection:
|
|
return informix_db.connect(
|
|
host=params.host,
|
|
port=params.port,
|
|
user=params.user,
|
|
password=params.password,
|
|
database=params.database,
|
|
server=params.server,
|
|
)
|
|
|
|
|
|
# -------- Basic scroll API --------
|
|
|
|
|
|
def test_fetchone_advances_index(conn_params: ConnParams) -> None:
|
|
"""Sequential ``fetchone`` advances ``rownumber`` 0, 1, 2, ..."""
|
|
with _connect(conn_params) as conn:
|
|
cur = conn.cursor()
|
|
cur.execute("SELECT FIRST 4 tabname FROM systables ORDER BY tabid")
|
|
assert cur.rownumber is None # before-first
|
|
for expected_idx in range(4):
|
|
row = cur.fetchone()
|
|
assert row is not None
|
|
assert cur.rownumber == expected_idx
|
|
# After last row
|
|
assert cur.fetchone() is None
|
|
|
|
|
|
def test_fetch_first_resets_position(conn_params: ConnParams) -> None:
|
|
"""``fetch_first`` repositions to row 0 regardless of current position."""
|
|
with _connect(conn_params) as conn:
|
|
cur = conn.cursor()
|
|
cur.execute("SELECT FIRST 5 tabname FROM systables ORDER BY tabid")
|
|
# Advance a few rows
|
|
cur.fetchone()
|
|
cur.fetchone()
|
|
cur.fetchone()
|
|
assert cur.rownumber == 2
|
|
# Reset
|
|
first = cur.fetch_first()
|
|
assert first is not None
|
|
assert cur.rownumber == 0
|
|
|
|
|
|
def test_fetch_last(conn_params: ConnParams) -> None:
|
|
"""``fetch_last`` jumps to the final row."""
|
|
with _connect(conn_params) as conn:
|
|
cur = conn.cursor()
|
|
cur.execute("SELECT FIRST 5 tabname FROM systables ORDER BY tabid")
|
|
last = cur.fetch_last()
|
|
assert last is not None
|
|
assert cur.rownumber == 4
|
|
# No more rows after
|
|
assert cur.fetchone() is None
|
|
|
|
|
|
def test_fetch_prior(conn_params: ConnParams) -> None:
|
|
"""``fetch_prior`` moves backward one row.
|
|
|
|
Semantics match SQL-standard FETCH PRIOR: from "past-end" (after
|
|
fetchone returned None), the first fetch_prior returns the *last*
|
|
row. From row N, fetch_prior returns row N-1.
|
|
"""
|
|
with _connect(conn_params) as conn:
|
|
cur = conn.cursor()
|
|
cur.execute("SELECT FIRST 5 tabname FROM systables ORDER BY tabid")
|
|
rows = []
|
|
# Fetch all forward
|
|
while (row := cur.fetchone()) is not None:
|
|
rows.append(row)
|
|
# Now scroll back — 5 fetch_priors take us from past-end to row 0
|
|
for expected_idx in (4, 3, 2, 1, 0):
|
|
row = cur.fetch_prior()
|
|
assert row == rows[expected_idx]
|
|
assert cur.rownumber == expected_idx
|
|
# One more prior → before-first → None
|
|
assert cur.fetch_prior() is None
|
|
assert cur.rownumber is None
|
|
|
|
|
|
def test_fetch_absolute(conn_params: ConnParams) -> None:
|
|
"""``fetch_absolute(n)`` jumps to row ``n`` (0-indexed)."""
|
|
with _connect(conn_params) as conn:
|
|
cur = conn.cursor()
|
|
cur.execute(
|
|
"SELECT FIRST 5 tabid FROM systables ORDER BY tabid"
|
|
)
|
|
# Pre-collect all rows for comparison
|
|
all_rows = cur.fetchall()
|
|
cur.fetch_first() # rewind via the pre-fetch path
|
|
# Re-execute since fetchall consumed
|
|
cur.execute(
|
|
"SELECT FIRST 5 tabid FROM systables ORDER BY tabid"
|
|
)
|
|
|
|
# Jump around
|
|
assert cur.fetch_absolute(0) == all_rows[0]
|
|
assert cur.fetch_absolute(4) == all_rows[4]
|
|
assert cur.fetch_absolute(2) == all_rows[2]
|
|
assert cur.rownumber == 2
|
|
|
|
|
|
def test_fetch_absolute_negative(conn_params: ConnParams) -> None:
|
|
"""Negative absolute indexes count from the end (Python-style)."""
|
|
with _connect(conn_params) as conn:
|
|
cur = conn.cursor()
|
|
cur.execute("SELECT FIRST 5 tabid FROM systables ORDER BY tabid")
|
|
rows = cur.fetchall()
|
|
cur.execute("SELECT FIRST 5 tabid FROM systables ORDER BY tabid")
|
|
assert cur.fetch_absolute(-1) == rows[-1]
|
|
assert cur.rownumber == 4
|
|
assert cur.fetch_absolute(-2) == rows[-2]
|
|
|
|
|
|
def test_fetch_relative(conn_params: ConnParams) -> None:
|
|
"""``fetch_relative(n)`` moves ``n`` rows from current position."""
|
|
with _connect(conn_params) as conn:
|
|
cur = conn.cursor()
|
|
cur.execute("SELECT FIRST 5 tabid FROM systables ORDER BY tabid")
|
|
cur.fetchone() # at row 0
|
|
cur.fetchone() # at row 1
|
|
# Jump forward 2
|
|
cur.fetch_relative(2)
|
|
assert cur.rownumber == 3
|
|
# Jump back 3
|
|
cur.fetch_relative(-3)
|
|
assert cur.rownumber == 0
|
|
|
|
|
|
# -------- PEP 249 scroll() --------
|
|
|
|
|
|
def test_scroll_relative(conn_params: ConnParams) -> None:
|
|
"""``scroll(value)`` defaults to relative mode."""
|
|
with _connect(conn_params) as conn:
|
|
cur = conn.cursor()
|
|
cur.execute("SELECT FIRST 5 tabid FROM systables ORDER BY tabid")
|
|
cur.fetchone() # row 0
|
|
cur.scroll(2) # relative +2 → row 2
|
|
assert cur.rownumber == 2
|
|
cur.scroll(-1) # relative -1 → row 1
|
|
assert cur.rownumber == 1
|
|
|
|
|
|
def test_scroll_absolute(conn_params: ConnParams) -> None:
|
|
"""``scroll(value, mode='absolute')`` jumps to row N (1-indexed per PEP 249)."""
|
|
with _connect(conn_params) as conn:
|
|
cur = conn.cursor()
|
|
cur.execute("SELECT FIRST 5 tabid FROM systables ORDER BY tabid")
|
|
cur.scroll(3, mode="absolute") # to row 3 (1-indexed)
|
|
assert cur.rownumber == 2 # 0-indexed: position 2
|
|
# The next fetchone advances to position 3
|
|
cur.fetchone()
|
|
assert cur.rownumber == 3
|
|
|
|
|
|
def test_scroll_out_of_range_raises(conn_params: ConnParams) -> None:
|
|
"""``scroll`` past the result set raises IndexError per PEP 249."""
|
|
with _connect(conn_params) as conn:
|
|
cur = conn.cursor()
|
|
cur.execute("SELECT FIRST 3 tabid FROM systables ORDER BY tabid")
|
|
with pytest.raises(IndexError):
|
|
cur.scroll(100)
|
|
with pytest.raises(IndexError):
|
|
cur.scroll(-5)
|
|
|
|
|
|
def test_scroll_invalid_mode_raises(conn_params: ConnParams) -> None:
|
|
"""Unknown scroll mode raises ProgrammingError."""
|
|
with _connect(conn_params) as conn:
|
|
cur = conn.cursor()
|
|
cur.execute("SELECT FIRST 3 tabid FROM systables ORDER BY tabid")
|
|
with pytest.raises(informix_db.ProgrammingError, match="scroll mode"):
|
|
cur.scroll(1, mode="forward") # type: ignore[arg-type]
|
|
|
|
|
|
# -------- Edge cases --------
|
|
|
|
|
|
def test_scroll_on_empty_result_set(conn_params: ConnParams) -> None:
|
|
"""Scroll methods on empty result return None gracefully."""
|
|
with _connect(conn_params) as conn:
|
|
cur = conn.cursor()
|
|
cur.execute("SELECT tabid FROM systables WHERE tabid = -999")
|
|
assert cur.fetch_first() is None
|
|
assert cur.fetch_last() is None
|
|
assert cur.fetch_absolute(0) is None
|
|
assert cur.fetch_relative(1) is None
|
|
assert cur.fetch_prior() is None
|
|
|
|
|
|
def test_fetchall_after_partial_iteration(conn_params: ConnParams) -> None:
|
|
"""``fetchall()`` returns only remaining rows after partial iteration."""
|
|
with _connect(conn_params) as conn:
|
|
cur = conn.cursor()
|
|
cur.execute("SELECT FIRST 5 tabid FROM systables ORDER BY tabid")
|
|
cur.fetchone() # consume row 0
|
|
cur.fetchone() # consume row 1
|
|
remaining = cur.fetchall()
|
|
assert len(remaining) == 3 # rows 2, 3, 4
|
|
|
|
|
|
def test_fetchall_then_fetch_first_restarts(conn_params: ConnParams) -> None:
|
|
"""After ``fetchall``, ``fetch_first`` repositions back to row 0."""
|
|
with _connect(conn_params) as conn:
|
|
cur = conn.cursor()
|
|
cur.execute("SELECT FIRST 3 tabid FROM systables ORDER BY tabid")
|
|
all_rows = cur.fetchall()
|
|
assert len(all_rows) == 3
|
|
# Cursor now at past-end. Rewind.
|
|
first = cur.fetch_first()
|
|
assert first == all_rows[0]
|