Ryan Malloy
d508a489fd
Three Phase 4 follow-ups in one push, all with empirical wire analysis:
1. PARAMETERIZED SELECT
cur.execute('SELECT tabname FROM systables WHERE tabid = ?', (1,))
→ ('systables',)
Wire flow: PREPARE → DESCRIBE → SQ_BIND-only (no EXECUTE) →
CURNAME+NFETCH → TUPLE+DONE → drain → CLOSE+RELEASE.
The cursor open is what executes the prepared query; SQ_BIND just
binds values into scope. No need for the IDESCRIBE handshake JDBC
does for type discovery — server accepts our typed bind directly.
2. NULL ROW DECODING — per-type sentinel detection
Each IDS type has its own NULL sentinel in tuple data:
INT → 0x80000000 (INT_MIN)
BIGINT → 0x8000000000000000 (LONG_MIN)
SMALLINT→ 0x8000 (SHORT_MIN)
REAL → all 0xFF (NaN bit pattern)
FLOAT → all 0xFF
DATE → 0x80000000 (same as INT)
VARCHAR → [byte 1][byte 0] (length=1, single nul) — distinguishable
from empty '' which is [byte 0] (length=0)
Verified by wire capture against the dev container — see
docs/CAPTURES/19-py-null-vs-onechar.socat.log and
docs/CAPTURES/20-py-int-null.socat.log.
The VARCHAR null marker is the trickiest because it LOOKS like a
1-byte string of nul, but VARCHAR can't contain embedded nuls
anyway, so the byte-0 within length-1 is unambiguous.
3. executemany(sql, seq_of_params) — PEP 249 batched DML
PREPARE once, loop SQ_BIND+SQ_EXECUTE per param set, RELEASE once.
Performance: only ~1.06x faster than execute() loop for 200 INSERTs
(dominated by per-row round trips). Phase 4.x optimization opportunity:
chain BIND+EXECUTE in one PDU without intermediate flush+read for
true bulk performance (would likely give 5-10x). Documented in
DECISION_LOG.md as a follow-up.
Module changes:
src/informix_db/converters.py:
+ Per-type NULL sentinel constants and detection in each decoder
+ Decoders now return None for sentinel values
src/informix_db/cursors.py:
+ _execute_select_with_params() — SQ_BIND alone, then cursor open
+ _build_bind_only_pdu() — SQ_BIND without trailing SQ_EXECUTE
+ executemany() — loop BIND+EXECUTE, accumulate rowcount
+ execute() now dispatches to _execute_select_with_params for
parameterized SELECT (was: NotSupportedError)
Tests: 40 unit + 47 integration (was 32; added 15 new) = 87 total,
all green, ruff clean. New test files / cases:
tests/test_nulls.py (7) — NULL decoding for INT, BIGINT, FLOAT,
REAL, VARCHAR, empty-vs-null, mixed columns
tests/test_params.py — added 4 parameterized SELECT tests, 5
executemany tests
tests/test_smoke.py — updated cursor-with-params test (was Phase 1
"raises", now Phase 4 "works")
Discovered captures kept for next-session debugging:
docs/CAPTURES/18-py-null-rows.socat.log
docs/CAPTURES/19-py-null-vs-onechar.socat.log
docs/CAPTURES/20-py-int-null.socat.log
208 lines
8.1 KiB
Python
208 lines
8.1 KiB
Python
"""Phase 4 integration tests — parameter binding (SQ_BIND).
|
|
|
|
Tests cover ``?`` and ``:N`` placeholder styles, the supported Python
|
|
type set (int, float, str, bool, None), and round-tripping through INSERT
|
|
+ SELECT to verify both encode AND decode paths.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import pytest
|
|
|
|
import informix_db
|
|
from tests.conftest import ConnParams
|
|
|
|
pytestmark = pytest.mark.integration
|
|
|
|
|
|
def _connect(conn_params: ConnParams) -> 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=10.0,
|
|
read_timeout=10.0,
|
|
)
|
|
|
|
|
|
def test_insert_with_qmark_params(conn_params: ConnParams) -> None:
|
|
"""``?`` placeholder style."""
|
|
with _connect(conn_params) as conn:
|
|
cur = conn.cursor()
|
|
cur.execute("CREATE TEMP TABLE t_p_a (id INTEGER, name VARCHAR(50))")
|
|
cur.execute("INSERT INTO t_p_a VALUES (?, ?)", (42, "hello"))
|
|
assert cur.rowcount == 1
|
|
|
|
cur.execute("SELECT id, name FROM t_p_a")
|
|
assert cur.fetchall() == [(42, "hello")]
|
|
|
|
|
|
def test_insert_with_numeric_params(conn_params: ConnParams) -> None:
|
|
"""``:1`` placeholder style (paramstyle="numeric")."""
|
|
with _connect(conn_params) as conn:
|
|
cur = conn.cursor()
|
|
cur.execute("CREATE TEMP TABLE t_p_b (id INTEGER, name VARCHAR(50))")
|
|
cur.execute("INSERT INTO t_p_b VALUES (:1, :2)", (99, "world"))
|
|
assert cur.rowcount == 1
|
|
|
|
cur.execute("SELECT id, name FROM t_p_b")
|
|
assert cur.fetchall() == [(99, "world")]
|
|
|
|
|
|
def test_int_float_str_round_trip(conn_params: ConnParams) -> None:
|
|
"""All three core types in one INSERT, verified via SELECT."""
|
|
with _connect(conn_params) as conn:
|
|
cur = conn.cursor()
|
|
cur.execute("CREATE TEMP TABLE t_p_c (i INTEGER, f FLOAT, s VARCHAR(20))")
|
|
cur.execute("INSERT INTO t_p_c VALUES (?, ?, ?)", (123, 4.5, "alpha"))
|
|
cur.execute("INSERT INTO t_p_c VALUES (?, ?, ?)", (-7, -1.25, "beta"))
|
|
|
|
cur.execute("SELECT i, f, s FROM t_p_c ORDER BY i")
|
|
rows = cur.fetchall()
|
|
assert rows == [(-7, -1.25, "beta"), (123, 4.5, "alpha")]
|
|
|
|
|
|
def test_update_with_params(conn_params: ConnParams) -> None:
|
|
"""UPDATE with parameter values in both SET and WHERE clauses."""
|
|
with _connect(conn_params) as conn:
|
|
cur = conn.cursor()
|
|
cur.execute("CREATE TEMP TABLE t_p_d (id INTEGER, name VARCHAR(50))")
|
|
cur.execute("INSERT INTO t_p_d VALUES (?, ?)", (1, "old"))
|
|
cur.execute("INSERT INTO t_p_d VALUES (?, ?)", (2, "old"))
|
|
|
|
cur.execute("UPDATE t_p_d SET name = ? WHERE id = ?", ("new", 2))
|
|
|
|
cur.execute("SELECT id, name FROM t_p_d ORDER BY id")
|
|
assert cur.fetchall() == [(1, "old"), (2, "new")]
|
|
|
|
|
|
def test_delete_with_param(conn_params: ConnParams) -> None:
|
|
"""DELETE with a parameter in the WHERE clause."""
|
|
with _connect(conn_params) as conn:
|
|
cur = conn.cursor()
|
|
cur.execute("CREATE TEMP TABLE t_p_e (id INTEGER, name VARCHAR(50))")
|
|
for i in range(1, 6):
|
|
cur.execute("INSERT INTO t_p_e VALUES (?, ?)", (i, f"row{i}"))
|
|
|
|
cur.execute("DELETE FROM t_p_e WHERE id = ?", (3,))
|
|
cur.execute("SELECT id FROM t_p_e ORDER BY id")
|
|
assert cur.fetchall() == [(1,), (2,), (4,), (5,)]
|
|
|
|
|
|
def test_unsupported_param_type_raises(conn_params: ConnParams) -> None:
|
|
"""Phase 4 supports int/float/str/bool/None; other types raise."""
|
|
with _connect(conn_params) as conn:
|
|
cur = conn.cursor()
|
|
cur.execute("CREATE TEMP TABLE t_p_f (id INTEGER)")
|
|
with pytest.raises(NotImplementedError, match="bytes"):
|
|
cur.execute("INSERT INTO t_p_f VALUES (?)", (b"raw bytes",))
|
|
|
|
|
|
def test_parameterized_select_int(conn_params: ConnParams) -> None:
|
|
"""Parameterized SELECT with an int parameter."""
|
|
with _connect(conn_params) as conn:
|
|
cur = conn.cursor()
|
|
cur.execute("SELECT tabname FROM systables WHERE tabid = ?", (1,))
|
|
assert cur.fetchone() == ("systables",)
|
|
|
|
|
|
def test_parameterized_select_multiple_params(conn_params: ConnParams) -> None:
|
|
"""Parameterized SELECT with two int parameters bounding a range."""
|
|
with _connect(conn_params) as conn:
|
|
cur = conn.cursor()
|
|
cur.execute(
|
|
"SELECT tabname FROM systables WHERE tabid >= ? AND tabid <= ? ORDER BY tabid",
|
|
(1, 3),
|
|
)
|
|
assert cur.fetchall() == [("systables",), ("syscolumns",), ("sysindices",)]
|
|
|
|
|
|
def test_parameterized_select_string_param(conn_params: ConnParams) -> None:
|
|
"""Parameterized SELECT with a string parameter."""
|
|
with _connect(conn_params) as conn:
|
|
cur = conn.cursor()
|
|
cur.execute("SELECT tabid FROM systables WHERE tabname = ?", ("systables",))
|
|
assert cur.fetchone() == (1,)
|
|
|
|
|
|
def test_parameterized_select_numeric_style(conn_params: ConnParams) -> None:
|
|
"""``:1`` style works for SELECT too."""
|
|
with _connect(conn_params) as conn:
|
|
cur = conn.cursor()
|
|
cur.execute("SELECT tabname FROM systables WHERE tabid = :1", (2,))
|
|
assert cur.fetchone() == ("syscolumns",)
|
|
|
|
|
|
def test_executemany_basic_insert(conn_params: ConnParams) -> None:
|
|
"""``executemany`` for batched INSERT — PREPARE once, BIND/EXECUTE per row."""
|
|
with _connect(conn_params) as conn:
|
|
cur = conn.cursor()
|
|
cur.execute("CREATE TEMP TABLE t_em_a (id INTEGER, name VARCHAR(50))")
|
|
rows = [(1, "alpha"), (2, "beta"), (3, "gamma"), (4, "delta")]
|
|
cur.executemany("INSERT INTO t_em_a VALUES (?, ?)", rows)
|
|
assert cur.rowcount == 4
|
|
|
|
cur.execute("SELECT id, name FROM t_em_a ORDER BY id")
|
|
assert cur.fetchall() == rows
|
|
|
|
|
|
def test_executemany_update(conn_params: ConnParams) -> None:
|
|
"""``executemany`` works for UPDATE too — per-row WHERE matches."""
|
|
with _connect(conn_params) as conn:
|
|
cur = conn.cursor()
|
|
cur.execute("CREATE TEMP TABLE t_em_u (id INTEGER, name VARCHAR(50))")
|
|
cur.executemany(
|
|
"INSERT INTO t_em_u VALUES (?, ?)",
|
|
[(1, "old"), (2, "old"), (3, "old")],
|
|
)
|
|
cur.executemany(
|
|
"UPDATE t_em_u SET name = ? WHERE id = ?",
|
|
[("A", 1), ("B", 2), ("C", 3)],
|
|
)
|
|
cur.execute("SELECT id, name FROM t_em_u ORDER BY id")
|
|
assert cur.fetchall() == [(1, "A"), (2, "B"), (3, "C")]
|
|
|
|
|
|
def test_executemany_empty_list(conn_params: ConnParams) -> None:
|
|
"""Empty parameter list is a no-op with rowcount=0."""
|
|
with _connect(conn_params) as conn:
|
|
cur = conn.cursor()
|
|
cur.execute("CREATE TEMP TABLE t_em_e (id INTEGER)")
|
|
cur.executemany("INSERT INTO t_em_e VALUES (?)", [])
|
|
assert cur.rowcount == 0
|
|
|
|
|
|
def test_executemany_inconsistent_param_lens_raises(conn_params: ConnParams) -> None:
|
|
"""Mismatched parameter-set lengths must raise ProgrammingError."""
|
|
with _connect(conn_params) as conn:
|
|
cur = conn.cursor()
|
|
cur.execute("CREATE TEMP TABLE t_em_bad (id INTEGER, name VARCHAR(50))")
|
|
with pytest.raises(informix_db.ProgrammingError, match="parameter set"):
|
|
cur.executemany(
|
|
"INSERT INTO t_em_bad VALUES (?, ?)",
|
|
[(1, "ok"), (2,)], # second has only 1 value
|
|
)
|
|
|
|
|
|
def test_executemany_select_unsupported(conn_params: ConnParams) -> None:
|
|
"""``executemany`` on SELECT doesn't make sense — must raise."""
|
|
with _connect(conn_params) as conn:
|
|
cur = conn.cursor()
|
|
with pytest.raises(informix_db.NotSupportedError, match="SELECT"):
|
|
cur.executemany(
|
|
"SELECT tabname FROM systables WHERE tabid = ?",
|
|
[(1,), (2,)],
|
|
)
|
|
|
|
|
|
def test_dict_params_unsupported(conn_params: ConnParams) -> None:
|
|
"""Named parameters aren't supported — paramstyle is ``numeric``."""
|
|
with _connect(conn_params) as conn:
|
|
cur = conn.cursor()
|
|
cur.execute("CREATE TEMP TABLE t_p_g (id INTEGER)")
|
|
with pytest.raises(informix_db.NotSupportedError, match="positional"):
|
|
cur.execute("INSERT INTO t_p_g VALUES (:id)", {"id": 1})
|