"""End-of-row reconciliation, and the two bugs it found immediately. Fourteen framing bugs reached users before this check existed. Every one was the same defect — a column consuming the wrong number of bytes — and not one raised at the point of the mistake. The wrong width produced a plausible value and corrupted whatever came *next*, so the damage always surfaced in a different column, row, or statement. All fourteen were detectable for free. ``payload`` is a fully extracted ``bytes`` object of known length, so after decoding N columns the offset must land exactly on the end. ``_row_not_consumed`` raises when it doesn't, naming the column shape and the byte delta. Turning the check on found two more bugs on its first run against the existing suite: * **Smart LOBs were read as a flat 72-byte field.** They use the UDT envelope: 149 bytes populated (``[ind=0][len=144][144 hex chars]``), 5 bytes NULL. We consumed 72 and left 77 behind, so any LOB not in the final column position corrupted everything after it. The 144 bytes are the 72-byte locator *hex-encoded*, which means ``BlobLocator.raw`` had never actually held a locator — only the first half of the hex text. Nobody noticed because ``read_blob_column`` resolves LOBs through the server-side ``lotofile`` function and never uses the locator. * **NULL composite UDTs skipped their length field**, byte-for-byte the LVARCHAR NULL bug in the branch twelve lines away. Both were the same ``[indicator][int32 length][data]`` envelope, written out by hand a third and fourth time. ``_read_udt_envelope`` is now the single implementation. """ from __future__ import annotations import contextlib import io import struct import pytest import informix_db from informix_db._protocol import IfxStreamReader, ProtocolError from informix_db._resultset import ( ColumnInfo, _read_udt_envelope, parse_tuple_payload, ) from tests.conftest import ConnParams # --------------------------------------------------------------------------- # The check itself — unit level, no server # --------------------------------------------------------------------------- def _tuple_pdu(body: bytes) -> IfxStreamReader: """Wrap a payload in the SQ_TUPLE framing parse_tuple_payload expects.""" return IfxStreamReader( io.BytesIO(struct.pack("!h", 0) + struct.pack("!i", len(body)) + body) ) def test_under_read_is_detected() -> None: cols = [ColumnInfo(name="k", type_code=2, raw_type_code=2, encoded_length=4)] with pytest.raises(ProtocolError, match="under-read by 6"): parse_tuple_payload(_tuple_pdu(struct.pack("!i", 5) + b"LEFTOV"), cols) def test_exact_consumption_passes() -> None: cols = [ColumnInfo(name="k", type_code=2, raw_type_code=2, encoded_length=4)] assert parse_tuple_payload(_tuple_pdu(struct.pack("!i", 5)), cols) == (5,) def test_error_names_the_column_shape() -> None: """"row decode failed" is not actionable. The message has to say which shape and by how much, so the next report arrives with the answer.""" cols = [ ColumnInfo(name="ident", type_code=2, raw_type_code=2, encoded_length=4), ColumnInfo(name="qty", type_code=2, raw_type_code=2, encoded_length=4), ] body = struct.pack("!i", 1) + struct.pack("!i", 2) + b"UNCONSUMED" with pytest.raises(ProtocolError) as exc: parse_tuple_payload(_tuple_pdu(body), cols) msg = str(exc.value) assert "ident" in msg and "qty" in msg, "message must name the columns" assert "tc=2" in msg assert "consumed 8 of 18" in msg, "message must give the byte counts" assert "under-read by 10" in msg # --------------------------------------------------------------------------- # The shared UDT envelope # --------------------------------------------------------------------------- def test_envelope_reads_length_even_when_null() -> None: """The length belongs to the envelope, not the value. Returning early on the indicator is the bug that hit LVARCHAR and then composites.""" payload = bytes([1]) + struct.pack("!i", 0) + b"NEXT" offset, body = _read_udt_envelope(payload, 0) assert body is None assert offset == 5, "NULL envelope must still consume its length field" assert payload[offset:] == b"NEXT" def test_envelope_reads_body_when_present() -> None: payload = bytes([0]) + struct.pack("!i", 3) + b"abc" + b"NEXT" offset, body = _read_udt_envelope(payload, 0) assert body == b"abc" assert offset == 8 assert payload[offset:] == b"NEXT" def test_envelope_rejects_negative_length() -> None: """A negative length off the wire would rewind the offset and silently re-decode earlier bytes as a later column.""" payload = bytes([0]) + struct.pack("!i", -4) + b"abc" with pytest.raises(ProtocolError, match="negative length"): _read_udt_envelope(payload, 0) # --------------------------------------------------------------------------- # Against a real server # --------------------------------------------------------------------------- pytestmark_integration = 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=15.0, read_timeout=30.0, autocommit=True, ) @pytest.mark.integration def test_null_composite_does_not_shift_next_column( conn_params: ConnParams, ) -> None: """A NULL SET made the following VARCHAR return '' instead of its value.""" with _connect(conn_params) as conn: cur = conn.cursor() with contextlib.suppress(Exception): cur.execute("DROP TABLE t_nullset") cur.execute( "CREATE TABLE t_nullset (a INT, s SET(INT NOT NULL), b VARCHAR(10))" ) try: cur.execute("INSERT INTO t_nullset VALUES (7, SET{1,2}, 'abc')") cur.execute("INSERT INTO t_nullset VALUES (8, NULL, 'xyz')") cur.execute("SELECT a, s, b FROM t_nullset ORDER BY a") rows = cur.fetchall() assert rows[0][0] == 7 and rows[0][2] == "abc" assert rows[1] == (8, None, "xyz"), ( "NULL composite shifted the following column" ) finally: with contextlib.suppress(Exception): cur.execute("DROP TABLE t_nullset") @pytest.mark.integration @pytest.mark.parametrize("populated", [False, True]) def test_lob_in_non_final_position( conn_params: ConnParams, populated: bool ) -> None: """A LOB was read as 72 bytes when it occupies 149 (or 5 when NULL), so anything selected after it was garbage.""" with _connect(conn_params) as conn: cur = conn.cursor() with contextlib.suppress(Exception): cur.execute("DROP TABLE t_lobmid") try: cur.execute( "CREATE TABLE t_lobmid (a INT, b BLOB, c INT, d VARCHAR(10))" ) except informix_db.Error: pytest.skip("no sbspace configured; see make ifx-spaces") try: if populated: cur.write_blob_column( "INSERT INTO t_lobmid VALUES (?, BLOB_PLACEHOLDER, ?, ?)", b"hello", (1, 222222, "tail"), ) else: cur.execute( "INSERT INTO t_lobmid VALUES (1, NULL, 222222, 'tail')" ) cur.execute("SELECT a, b, c, d FROM t_lobmid") a, b, c, d = cur.fetchone() assert (a, c, d) == (1, 222222, "tail"), ( "columns after the LOB were shifted" ) if populated: assert isinstance(b, informix_db.BlobLocator) assert len(b.raw) == 72, ( "locator must be the decoded 72 bytes, not hex text" ) else: assert b is None finally: with contextlib.suppress(Exception): cur.execute("DROP TABLE t_lobmid") @pytest.mark.integration def test_wide_mixed_row_reconciles(conn_params: ConnParams) -> None: """Belt and braces: a row using most of the tricky types must consume its payload exactly. The check runs on every fetch, so merely getting a row back proves reconciliation held.""" with _connect(conn_params) as conn: cur = conn.cursor() cur.execute( "CREATE TEMP TABLE t_recon (" " a INT8 NOT NULL, k LVARCHAR(512), n NCHAR(8), v VARCHAR(32)," " d DECIMAL(16), b BOOLEAN, t DATETIME YEAR TO FRACTION(5)," " tail INT)" ) cur.execute( "INSERT INTO t_recon VALUES (?, ?, ?, ?, ?, ?, CURRENT, ?)", (1, "PackageRoot", "nch", "v", None, True, 424242), ) cur.execute( "SELECT a, k, n, v, d, b, t, tail FROM t_recon" ) row = cur.fetchone() assert row[0] == 1 assert row[-1] == 424242