Ryan Malloy
509af9efa4
cur.execute("INSERT INTO t VALUES (?, ?, ?)", (42, "hello", 3.14))
cur.execute("INSERT INTO t VALUES (:1, :2)", (99, "world"))
cur.execute("UPDATE t SET name = ? WHERE id = ?", ("new", 2))
cur.execute("DELETE FROM t WHERE id = ?", (5,))
# all work end-to-end against a real Informix server
Two breakthroughs decoded from JDBC:
1. SQ_BIND PDU shape (chained with SQ_EXECUTE in one PDU, no separate
round trip):
[short SQ_ID=4][int SQ_BIND=5][short numparams]
for each param:
[short type][short indicator][short prec_or_encLen]
writePadded(rawbytes)
[short SQ_EXECUTE=7][short SQ_EOT]
2. Strings are sent as CHAR (type=0) not VARCHAR (type=13). The server
handles conversion to the actual column type via internal CIDESCRIBE
— we don't need to do it explicitly.
Per-type encoding (Phase 4 MVP):
int (32-bit) → IDS INT (type=2), prec=0x0a00 (packed width=10/scale=0),
4-byte BE
int (64-bit) → IDS BIGINT (type=52), prec=0x1300, 8-byte BE
str → IDS CHAR (type=0), prec=0, [short len][bytes][pad]
float → IDS FLOAT (type=3), prec=0, 8-byte IEEE 754
bool → IDS BOOL (type=45), prec=0, 1 byte
None → indicator=-1, no data
The integer "precision" field is PACKED — initially looked like a bug
(why would precision be 2560?) until I realized 0x0a00 = (10 << 8) | 0
= packed display-width and scale. Captured this surprise in
DECISION_LOG.md.
Critical fix to execute-path branching: parameterized INSERT also
returns nfields > 0 (server describes the would-be inserted row).
Switched from "branch on nfields" to "branch on SQL keyword" — JDBC
does the same via its IfxStatement / IfxPreparedStatement subclassing.
Numeric paramstyle support: cur.execute("... :1 ...", (val,)) works
by rewriting :N → ? before sending PREPARE. Trivial regex (doesn't
escape strings/comments — Phase 5 can add a proper SQL tokenizer).
Module changes:
src/informix_db/converters.py:
+ encode_param() dispatcher
+ _encode_int / _encode_bigint / _encode_str / _encode_float / _encode_bool
src/informix_db/cursors.py:
+ _build_bind_execute_pdu() — chains SQ_BIND + SQ_EXECUTE in one PDU
+ _execute_dml_with_params() — sends bind PDU, drains, releases
+ execute() now accepts parameters; rewrites :N → ?; branches by
SQL keyword (SELECT vs DML)
+ _NUMERIC_PLACEHOLDER_RE for paramstyle="numeric" support
Tests: 40 unit + 32 integration (8 new parameter tests + 1 updated
smoke) = 72 total, all green, ruff clean. New tests cover:
- INSERT with ? params
- INSERT with :N params
- INT + FLOAT + str round trip via INSERT then SELECT
- UPDATE with params in SET and WHERE
- DELETE with parameter in WHERE
- Unsupported param type (bytes) raises NotImplementedError
- Parameterized SELECT raises NotSupportedError (Phase 4.x)
- Dict/named params raise NotSupportedError
Known gaps (Phase 4.x / Phase 5):
- Parameterized SELECT (needs SQ_BIND before CURNAME+NFETCH)
- NULL row decoding for VARCHAR (currently surfaces empty string)
- Proper SQL tokenizer (so :N inside string literals is preserved)
- bytes/datetime/Decimal parameter types
119 lines
4.6 KiB
Python
119 lines
4.6 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_not_yet_supported(conn_params: ConnParams) -> None:
|
|
"""Parameterized SELECT lands in Phase 4.x — currently raises."""
|
|
with _connect(conn_params) as conn:
|
|
cur = conn.cursor()
|
|
with pytest.raises(informix_db.NotSupportedError, match=r"Phase 4\.x"):
|
|
cur.execute("SELECT 1 FROM systables WHERE tabid = ?", (1,))
|
|
|
|
|
|
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})
|