Ryan Malloy
ac9075ca9a
Last of the untested surface: pipelined executemany, scrollable cursors, and smart LOBs. One bug, in the first. executemany builds all its BIND+EXECUTE PDUs after the PREPARE and before anything is drained — that batching is what makes the pipeline fast. If encoding a row raises there (a value the connection's codec cannot represent), the exception escaped without sending the RELEASE, leaking the prepared statement. The next PREPARE collided with it and every later call on the connection failed with an error pointing at the previous SQL. Same failure as 2026.08.31.2, in a sibling path. _execute_dml_with_params has guarded this exact case for the single-row path for a long time; the pipelined path was never given the same treatment. That is the recurring shape of these bugs: a hazard understood in one place and not carried to the code beside it. What the fuzzer found sound, which is the larger part of the result: executemany constraint failures — duplicate-key and NOT NULL at first/mid/last of batches from 2 to 1000 rows all recover, and COUNT(*) always agrees with a full fetch, so the drain-N-responses invariant holds under partial failure. Scrollable cursors — fetch_first/last/prior/relative/absolute correct at 0, 1, 2, 5, 50, 300 rows, including off both ends (None, not a wrap or a crash) and a full forward walk after arbitrary positioning. Twenty abandoned scroll cursors leak nothing. Smart LOBs — round-trip at 0, 1, 255, 256, 1023, 1024, 4095, 4096, 65535, 65536 bytes, straddling the 4096-byte SQ_FILE chunk and the 64K mark, plus recovery from failed reads. Still not fuzzed, stated plainly: TLS is handshake-tested against a self-signed local socket, not a real Informix TLS listener (that needs server-side keystore + onconfig SSL setup absent from the test containers). The SQLI layer above the socket is identical either way. 399/399 integration on 15, 14.10 and 12.10 (was 356).
259 lines
9.9 KiB
Python
259 lines
9.9 KiB
Python
"""Regression tests for pipelined executemany, scrollable cursors, and LOBs.
|
|
|
|
Found here by fuzzing: **a client-side encoding failure inside an
|
|
``executemany`` batch leaked the prepared statement.** The PDUs are built
|
|
after the PREPARE, so a value the connection's codec cannot represent
|
|
raises there — before anything is drained — and the exception escaped
|
|
without the RELEASE. The leaked statement then collided with the next
|
|
PREPARE and every later call on that connection failed with an error
|
|
pointing at the *previous* SQL.
|
|
|
|
``_execute_dml_with_params`` already guarded exactly this case for the
|
|
single-row path. The pipelined path was simply missed, which is the
|
|
recurring shape of these bugs: a hazard understood in one place and not
|
|
carried to its sibling.
|
|
|
|
The rest of this file is the coverage that proved the neighbouring paths
|
|
sound — batch failures at every position, scroll boundaries, LOB sizes —
|
|
kept so they stay that way. Each failure case ends in a health check,
|
|
because the interesting damage is never in the statement that failed.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import contextlib
|
|
|
|
import pytest
|
|
|
|
import informix_db
|
|
from tests.conftest import ConnParams
|
|
|
|
pytestmark = pytest.mark.integration
|
|
|
|
|
|
def _connect(conn_params: ConnParams, **kw) -> informix_db.Connection:
|
|
return informix_db.connect(
|
|
host=conn_params.host,
|
|
port=conn_params.port,
|
|
user=conn_params.user,
|
|
password=conn_params.password,
|
|
database=conn_params.database,
|
|
server=conn_params.server,
|
|
connect_timeout=15.0,
|
|
read_timeout=30.0,
|
|
**kw,
|
|
)
|
|
|
|
|
|
def _assert_healthy(cur) -> None:
|
|
cur.execute("SELECT FIRST 1 tabid FROM systables")
|
|
assert cur.fetchone() is not None, "connection unusable"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# executemany — the encoding-failure leak, and the neighbours
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.mark.parametrize("n", [3, 50])
|
|
@pytest.mark.parametrize("position", ["first", "mid", "last"])
|
|
def test_encoding_failure_in_batch_does_not_leak(
|
|
conn_params: ConnParams, n: int, position: str
|
|
) -> None:
|
|
"""The bug. A value the codec can't encode raises after the PREPARE;
|
|
escaping without the RELEASE bricked the connection."""
|
|
idx = {"first": 0, "mid": n // 2, "last": n - 1}[position]
|
|
with _connect(conn_params, autocommit=True) as conn:
|
|
cur = conn.cursor()
|
|
cur.execute("CREATE TEMP TABLE t_em_enc (s VARCHAR(32))")
|
|
rows: list[tuple] = [(f"s{i}",) for i in range(n)]
|
|
rows[idx] = ("中文",) # not representable in iso-8859-1
|
|
|
|
with pytest.raises(Exception): # noqa: B017 — DataError or UnicodeError
|
|
cur.executemany("INSERT INTO t_em_enc VALUES (?)", rows)
|
|
|
|
# The connection must survive, repeatedly.
|
|
for _ in range(3):
|
|
_assert_healthy(cur)
|
|
with pytest.raises(Exception): # noqa: B017
|
|
cur.executemany("INSERT INTO t_em_enc VALUES (?)", rows)
|
|
_assert_healthy(cur)
|
|
|
|
|
|
@pytest.mark.parametrize("n", [2, 3, 10, 100])
|
|
@pytest.mark.parametrize("position", ["first", "mid", "last"])
|
|
def test_constraint_violation_in_batch_recovers(
|
|
conn_params: ConnParams, n: int, position: str
|
|
) -> None:
|
|
idx = {"first": 0, "mid": n // 2, "last": n - 1}[position]
|
|
with _connect(conn_params, autocommit=True) as conn:
|
|
cur = conn.cursor()
|
|
cur.execute("CREATE TEMP TABLE t_em_dup (k INT PRIMARY KEY)")
|
|
cur.execute("INSERT INTO t_em_dup VALUES (?)", (10_000,))
|
|
rows = [(i,) for i in range(n)]
|
|
rows[idx] = (10_000,)
|
|
|
|
with pytest.raises(informix_db.Error):
|
|
cur.executemany("INSERT INTO t_em_dup VALUES (?)", rows)
|
|
_assert_healthy(cur)
|
|
|
|
# Whatever the partial-batch semantics, COUNT(*) and a full fetch
|
|
# must agree — a disagreement means the row decoder and the server
|
|
# have different ideas about what is in the table.
|
|
cur.execute("SELECT COUNT(*) FROM t_em_dup")
|
|
(counted,) = cur.fetchone()
|
|
cur.execute("SELECT k FROM t_em_dup")
|
|
assert counted == len(cur.fetchall())
|
|
|
|
|
|
@pytest.mark.parametrize("n", [1, 2, 3, 10, 100, 1000])
|
|
def test_executemany_inserts_exactly_n_rows(
|
|
conn_params: ConnParams, n: int
|
|
) -> None:
|
|
with _connect(conn_params, autocommit=True) as conn:
|
|
cur = conn.cursor()
|
|
cur.execute("CREATE TEMP TABLE t_em_ok (k INT, v VARCHAR(16))")
|
|
cur.executemany(
|
|
"INSERT INTO t_em_ok VALUES (?, ?)",
|
|
[(i, f"v{i}") for i in range(n)],
|
|
)
|
|
cur.execute("SELECT COUNT(*) FROM t_em_ok")
|
|
assert cur.fetchone() == (n,)
|
|
|
|
|
|
def test_executemany_empty_and_single(conn_params: ConnParams) -> None:
|
|
with _connect(conn_params, autocommit=True) as conn:
|
|
cur = conn.cursor()
|
|
cur.execute("CREATE TEMP TABLE t_em_edge (k INT)")
|
|
cur.executemany("INSERT INTO t_em_edge VALUES (?)", [])
|
|
_assert_healthy(cur)
|
|
cur.executemany("INSERT INTO t_em_edge VALUES (?)", [(1,)])
|
|
cur.execute("SELECT COUNT(*) FROM t_em_edge")
|
|
assert cur.fetchone() == (1,)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Scrollable cursors — boundaries get their own round-trip each
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.mark.parametrize("n", [0, 1, 2, 5, 50, 300])
|
|
def test_scroll_boundaries(conn_params: ConnParams, n: int) -> None:
|
|
with _connect(conn_params, autocommit=True) as conn:
|
|
setup = conn.cursor()
|
|
with contextlib.suppress(Exception):
|
|
setup.execute("DROP TABLE t_scroll")
|
|
setup.execute("CREATE TABLE t_scroll (k INT)")
|
|
try:
|
|
if n:
|
|
setup.executemany(
|
|
"INSERT INTO t_scroll VALUES (?)", [(i,) for i in range(n)]
|
|
)
|
|
|
|
cur = conn.cursor(scrollable=True)
|
|
try:
|
|
cur.execute("SELECT k FROM t_scroll ORDER BY k")
|
|
assert cur.fetch_first() == ((0,) if n else None)
|
|
assert cur.fetch_last() == ((n - 1,) if n else None)
|
|
|
|
if n >= 3:
|
|
assert cur.fetch_absolute(2) == (2,)
|
|
assert cur.fetch_prior() == (1,)
|
|
assert cur.fetch_relative(2) == (3,)
|
|
|
|
# Off both ends must be None — not a crash, not a wrap.
|
|
assert cur.fetch_absolute(n + 50) is None
|
|
cur.fetch_first()
|
|
assert cur.fetch_prior() is None
|
|
|
|
if n:
|
|
cur.fetch_first()
|
|
seen = [(0,)]
|
|
while (row := cur.fetchone()) is not None:
|
|
seen.append(row)
|
|
assert seen == [(i,) for i in range(n)]
|
|
finally:
|
|
cur.close()
|
|
_assert_healthy(setup)
|
|
finally:
|
|
with contextlib.suppress(Exception):
|
|
setup.execute("DROP TABLE t_scroll")
|
|
|
|
|
|
def test_abandoned_scroll_cursors_do_not_leak(
|
|
conn_params: ConnParams,
|
|
) -> None:
|
|
"""Scroll cursors stay open server-side, so abandoning one leaks
|
|
unless the finalizer runs."""
|
|
with _connect(conn_params, autocommit=True) as conn:
|
|
setup = conn.cursor()
|
|
with contextlib.suppress(Exception):
|
|
setup.execute("DROP TABLE t_scroll_ab")
|
|
setup.execute("CREATE TABLE t_scroll_ab (k INT)")
|
|
try:
|
|
setup.executemany(
|
|
"INSERT INTO t_scroll_ab VALUES (?)", [(i,) for i in range(20)]
|
|
)
|
|
for i in range(20):
|
|
c = conn.cursor(scrollable=True)
|
|
c.execute("SELECT k FROM t_scroll_ab ORDER BY k")
|
|
c.fetch_first()
|
|
if i % 2:
|
|
c.close()
|
|
else:
|
|
del c # rely on the finalizer
|
|
_assert_healthy(setup)
|
|
setup.execute("SELECT COUNT(*) FROM t_scroll_ab")
|
|
assert setup.fetchone() == (20,)
|
|
finally:
|
|
with contextlib.suppress(Exception):
|
|
setup.execute("DROP TABLE t_scroll_ab")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Smart LOBs
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"size", [0, 1, 255, 256, 1023, 1024, 4095, 4096, 65535, 65536]
|
|
)
|
|
def test_blob_round_trip_sizes(conn_params: ConnParams, size: int) -> None:
|
|
"""Sizes straddle the 4096-byte SQ_FILE chunk and the 64K mark."""
|
|
payload = bytes((i * 7 + size) % 256 for i in range(size))
|
|
with _connect(conn_params, autocommit=True) as conn:
|
|
cur = conn.cursor()
|
|
with contextlib.suppress(Exception):
|
|
cur.execute("DROP TABLE t_blob_rt")
|
|
try:
|
|
cur.execute("CREATE TABLE t_blob_rt (k INT, b BLOB)")
|
|
except informix_db.Error:
|
|
pytest.skip("no sbspace configured; see make ifx-spaces")
|
|
try:
|
|
cur.write_blob_column(
|
|
"INSERT INTO t_blob_rt VALUES (?, BLOB_PLACEHOLDER)",
|
|
payload, (1,),
|
|
)
|
|
got = cur.read_blob_column(
|
|
"SELECT b FROM t_blob_rt WHERE k = ?", (1,)
|
|
)
|
|
if size == 0:
|
|
assert got in (b"", None)
|
|
else:
|
|
assert got == payload
|
|
_assert_healthy(cur)
|
|
finally:
|
|
with contextlib.suppress(Exception):
|
|
cur.execute("DROP TABLE t_blob_rt")
|
|
|
|
|
|
def test_failed_blob_read_recovers(conn_params: ConnParams) -> None:
|
|
"""The SQ_FILE path involves a server-side temp file; a failure part
|
|
way through must not strand the connection."""
|
|
with _connect(conn_params, autocommit=True) as conn:
|
|
cur = conn.cursor()
|
|
for _ in range(5):
|
|
with contextlib.suppress(Exception):
|
|
cur.read_blob_column("SELECT b FROM t_no_such_blob_tbl", ())
|
|
_assert_healthy(cur)
|