"""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)