From 5ad296419c8692d1b2083800a6320ff12440f175 Mon Sep 17 00:00:00 2001 From: Ryan Malloy Date: Wed, 2 Sep 2026 00:54:12 -0600 Subject: [PATCH] Two readers shared a stream without agreeing, and a length was taken on trust IfxSocket owns a read-ahead buffer that BufferedSocketReader fills and drains. Connection._drain_to_eot, _raise_sq_err and the login path bypass that reader and call IfxSocket.read_exact directly, which recv'd from the socket without looking at the buffer. Bytes sitting in the buffer were skipped, and skipped bytes in a length-framed protocol do not announce themselves -- the next read lands mid-field and every read after it is wrong. Nothing triggers it today. The server sends one response per request, so recv returns exactly that response and the buffered reader consumes all of it before control returns to a direct read. That is a property of the traffic, not of the code, and the buffer is connection-scoped precisely so read-ahead can cross response boundaries -- pipelined executemany already puts several responses in flight. read_exact now drains the buffer first, which costs one branch on a cold path and makes the two paths agree by construction rather than by luck. fill_recv_buf believed whatever byte count it was handed, and that count is almost always a length field straight off the wire. A garbage 0x7FFFFFFF reads as a 2 GB request and the fill loop sits in recv until the read timeout while the buffer grows. It now refuses above IFX_MAX_READ_BYTES (256 MiB default) with an error naming the number, which is the actual diagnostic: a length that absurd means framing was already lost upstream. BufferedSocketReader.skip advanced the cursor arithmetically with no guard, so a negative count rewound it and re-decoded consumed bytes as the next field. The base reader's skip delegates to read_exact and does guard; this one diverged. Transaction control run as SQL desynced Connection._in_transaction, which is what commit() and rollback() are guarded by and what the pool reads to decide whether a returned connection needs cleaning up. With autocommit on, cursor.execute("BEGIN WORK") opened a real transaction while the flag stayed False, so rollback() returned successfully having sent nothing and the rows it was asked to discard survived. The connection then went back to the pool holding an open transaction and its locks. With autocommit off it failed instead: the driver's implicit SQ_BEGIN fired first and the caller's BEGIN WORK got -535. The server labels these -- 34 BEGIN, 35 COMMIT, 36 ROLLBACK, with and without the WORK keyword, measured on all three versions. JDBC reads the same values off the describe and calls setTxBeginState/setTxEndState. _ensure_transaction moves to after the describe, matching JDBC's initiateTransaction placement, which is what makes it possible to skip for transaction control at all -- until the describe lands you cannot know that is what the statement is. --- src/informix_db/_protocol.py | 7 ++ src/informix_db/_socket.py | 76 ++++++++++++- src/informix_db/cursors.py | 68 ++++++++++-- tests/test_socket_reads.py | 172 ++++++++++++++++++++++++++++ tests/test_transaction_state.py | 191 ++++++++++++++++++++++++++++++++ 5 files changed, 502 insertions(+), 12 deletions(-) create mode 100644 tests/test_socket_reads.py create mode 100644 tests/test_transaction_state.py diff --git a/src/informix_db/_protocol.py b/src/informix_db/_protocol.py index e9df9c1..b5fb043 100644 --- a/src/informix_db/_protocol.py +++ b/src/informix_db/_protocol.py @@ -292,6 +292,13 @@ class BufferedSocketReader(IfxStreamReader): return b def skip(self, n: int) -> None: + # The base reader's skip delegates to read_exact, which returns + # b"" for n <= 0. This one advances the cursor arithmetically, so + # without the guard a negative n *rewinds* it — silently + # re-decoding bytes already consumed as if they were the next + # field. Same divergence as read_exact, which does guard. + if n <= 0: + return sock = self._sock sock.fill_recv_buf(n) sock._recv_pos += n diff --git a/src/informix_db/_socket.py b/src/informix_db/_socket.py index e77faf9..5b7b226 100644 --- a/src/informix_db/_socket.py +++ b/src/informix_db/_socket.py @@ -17,11 +17,35 @@ the rest of the protocol layer. from __future__ import annotations import contextlib +import os import socket import ssl +from ._protocol import ProtocolError from .exceptions import InterfaceError, OperationalError + +def _max_read_bytes() -> int: + """Ceiling for a single length-prefixed read, from IFX_MAX_READ_BYTES. + + 256 MiB by default: comfortably above any row, string table, or blob + chunk a real query produces, and far below the values a desynced + stream invents. A genuinely larger single value is possible (a very + large TEXT column read in one go), which is why the knob exists. + """ + raw = os.environ.get("IFX_MAX_READ_BYTES") + if raw: + try: + value = int(raw) + except ValueError: + value = 0 + if value > 0: + return value + return 256 * 1024 * 1024 + + +MAX_READ_BYTES = _max_read_bytes() + # A ``tls`` parameter to ``IfxSocket`` accepts: # False (default) — plain TCP # True — TLS with verification disabled (dev / self-signed) @@ -124,10 +148,42 @@ class IfxSocket: raise OperationalError(f"write failed: {e}") from e def read_exact(self, n: int) -> bytes: - """Read exactly ``n`` bytes or raise on EOF / timeout.""" + """Read exactly ``n`` bytes or raise on EOF / timeout. + + Consumes the read-ahead buffer before touching the socket. This + matters because two readers share one stream: ``BufferedSocketReader`` + fills ``_recv_buf`` (over-reading by design, up to ``_recv_size``), + while ``Connection._drain_to_eot``, ``_raise_sq_err`` and the + login path call this method directly. Recv'ing here while bytes + sat unconsumed in the buffer would skip them, and skipped bytes + in a length-framed protocol don't announce themselves — the next + read lands mid-field and every subsequent one is wrong. + + No workload triggers it today: the server sends one response per + request, so recv returns exactly that response and the buffered + reader consumes all of it before control returns here. That is a + property of the traffic, not of the code. Pipelined executemany + already puts multiple responses in flight, and the buffer is + connection-scoped precisely so read-ahead can cross response + boundaries. Making the two paths agree by construction costs one + branch on a cold path. + """ if self._sock is None: raise InterfaceError("socket is closed") + if n <= 0: + return b"" + wanted = n chunks: list[bytes] = [] + buffered = len(self._recv_buf) - self._recv_pos + if buffered > 0: + take = min(buffered, n) + chunks.append( + bytes(self._recv_buf[self._recv_pos : self._recv_pos + take]) + ) + self._recv_pos += take + n -= take + if n == 0: + return chunks[0] remaining = n while remaining > 0: try: @@ -138,7 +194,8 @@ class IfxSocket: if not chunk: self._force_close() raise OperationalError( - f"server closed connection mid-read (wanted {n} bytes, got {n - remaining})" + f"server closed connection mid-read " + f"(wanted {wanted} bytes, got {wanted - remaining})" ) chunks.append(chunk) remaining -= len(chunk) @@ -155,6 +212,21 @@ class IfxSocket: """ if self._sock is None: raise InterfaceError("socket is closed") + if need > MAX_READ_BYTES: + # ``need`` is almost always a length field straight off the + # wire. Trusting it means a corrupt or desynced stream turns + # into an allocation of whatever that field happened to say — + # a garbage 0x7FFFFFFF reads as a 2 GB request, and the loop + # below sits in recv until the read timeout while the buffer + # grows. Refusing names the number instead, which is the + # diagnostic: a length that absurd means framing is already + # lost upstream, not that the row is genuinely that big. + raise ProtocolError( + f"refusing to read {need} bytes in one field (limit " + f"{MAX_READ_BYTES}). Either the wire has desynced, or a " + f"single value really is this large — raise the limit " + f"with the IFX_MAX_READ_BYTES environment variable." + ) avail = len(self._recv_buf) - self._recv_pos if avail >= need: return diff --git a/src/informix_db/cursors.py b/src/informix_db/cursors.py index c8240d3..fde6080 100644 --- a/src/informix_db/cursors.py +++ b/src/informix_db/cursors.py @@ -108,6 +108,15 @@ def _make_socket_reader(sock): _ST_SELECT = 2 _ST_ROUTINE = 56 # EXECUTE PROCEDURE / EXECUTE FUNCTION +# Transaction control run as SQL. JDBC reads the same three values off +# the describe and calls setTxBeginState / setTxEndState (IfxSqli, the +# TxStmt field). Both ``BEGIN`` and ``BEGIN WORK`` report 34; likewise +# the WORK-less spellings of the other two. +_ST_TX_BEGIN = 34 +_ST_TX_COMMIT = 35 +_ST_TX_ROLLBACK = 36 +_TX_CONTROL_TYPES = frozenset({_ST_TX_BEGIN, _ST_TX_COMMIT, _ST_TX_ROLLBACK}) + def _produces_result_set(statement_type: int, ncolumns: int) -> bool: """Whether a prepared statement needs a cursor opened for it. @@ -423,12 +432,6 @@ class Cursor: self._statement_type = 0 self._statement_already_done = False - # On a logged DB in non-autocommit mode, the server requires an - # explicit SQ_BEGIN before the first DML in each transaction. - # _ensure_transaction is a no-op for autocommit / unlogged DBs, - # and idempotent within an open transaction. - self._conn._ensure_transaction() - # Step 1: PREPARE — send SQL with numQmarks = len(params). # statement_boundary: nothing is open server-side yet, so this is # the one safe moment to flush a finalizer's deferred cleanup. @@ -438,6 +441,21 @@ class Cursor: ) self._read_describe_response() + # On a logged DB in non-autocommit mode, the server requires an + # explicit SQ_BEGIN before the first DML in each transaction. + # _ensure_transaction is a no-op for autocommit / unlogged DBs, + # and idempotent within an open transaction. + # + # This runs *after* the describe, matching JDBC's + # initiateTransaction placement, because until the describe lands + # we don't know whether the caller's statement is itself + # transaction control. Opening a transaction on their behalf and + # then executing their BEGIN WORK gets -535, "already in + # transaction" — the driver competing with the user for the same + # job and the user losing. + if self._statement_type not in _TX_CONTROL_TYPES: + self._conn._ensure_transaction() + # Ask the server what it just prepared, rather than guessing from # the first word of the SQL. if _produces_result_set(self._statement_type, len(self._columns)): @@ -450,6 +468,11 @@ class Cursor: else: self._execute_dml() + # The statement succeeded. If it was transaction control, the + # server's transaction state just changed and the connection has + # to know, or commit() and rollback() silently do nothing. + self._note_transaction_control() + # SELECT path: position cursor before the first row so the next # ``fetchone()`` returns ``rows[0]``. DML paths leave _row_index # at -1 too (no rows to iterate). @@ -458,6 +481,30 @@ class Cursor: if self._description is not None: self._row_index = -1 + def _note_transaction_control(self) -> None: + """Sync the connection's transaction flag after a successful execute. + + ``BEGIN WORK`` run through ``execute()`` opens a real transaction + on the server, and the connection had no idea. With autocommit on + nothing stopped it reaching the server, so ``_in_transaction`` + stayed False while a transaction was open — and both ``commit()`` + and ``rollback()`` are guarded by that flag. ``rollback()`` + returned successfully having sent nothing, and the rows it was + asked to discard were still there. + + The pool reads the same flag to decide whether a returned + connection needs cleaning up, so the connection went back into + circulation holding an open transaction and its locks. + + Called only on the success path: a statement that failed did not + change the server's transaction state. + """ + statement_type = self._statement_type + if statement_type == _ST_TX_BEGIN: + self._conn._in_transaction = True + elif statement_type in (_ST_TX_COMMIT, _ST_TX_ROLLBACK): + self._conn._in_transaction = False + def _close_server_cursor(self) -> None: """Free the server-side scrollable cursor. Caller MUST hold ``_wire_lock``. @@ -1132,10 +1179,6 @@ class Cursor: self._statement_type = 0 self._statement_already_done = False - # Logged-DB transaction guard — same as execute(). Idempotent - # within an open transaction. - self._conn._ensure_transaction() - # PREPARE once. self._conn._send_pdu( self._build_prepare_pdu(sql, num_qmarks=first_len), @@ -1155,6 +1198,11 @@ class Cursor: "supported" ) + # Logged-DB transaction guard — same as execute(), and for the + # same reason placed after the describe rather than before it. + if self._statement_type not in _TX_CONTROL_TYPES: + self._conn._ensure_transaction() + # Phase 33: pipeline — build all BIND+EXECUTE PDUs first # (Python work, no I/O), then send them back-to-back, then # drain all responses. Eliminates the per-row round-trip diff --git a/tests/test_socket_reads.py b/tests/test_socket_reads.py new file mode 100644 index 0000000..c34acde --- /dev/null +++ b/tests/test_socket_reads.py @@ -0,0 +1,172 @@ +"""Two readers, one stream, and a length field taken on trust. + +``IfxSocket`` owns a read-ahead buffer that ``BufferedSocketReader`` +fills and drains. But ``Connection._drain_to_eot``, ``_raise_sq_err`` +and the login path bypass that reader and call ``IfxSocket.read_exact`` +directly, which recv'd from the socket without ever looking at the +buffer. Bytes sitting in the buffer would simply be skipped, and skipped +bytes in a length-framed protocol don't announce themselves — the next +read lands mid-field and every read after it is wrong. + +Nothing triggers it today. The server sends one response per request, so +recv returns exactly that response and the buffered reader consumes all +of it before control returns to a direct read. That is a property of the +traffic, not of the code, and the buffer is connection-scoped precisely +so read-ahead *can* cross response boundaries — pipelined executemany +already puts several responses in flight. A latent desync waiting on a +timing change is not a good thing to leave in a wire protocol. + +Separately, ``fill_recv_buf`` took its byte count on trust, and that +count is almost always a length field straight off the wire. A corrupt +or desynced stream turned into an allocation of whatever the field +happened to say: a garbage ``0x7FFFFFFF`` reads as a 2 GB request, and +the fill loop sits in recv until the read timeout while the buffer +grows. The limit turns that into an error that names the number, which +is the actual diagnostic — a length that absurd means framing was +already lost upstream. +""" + +from __future__ import annotations + +import pytest + +from informix_db._protocol import BufferedSocketReader, ProtocolError +from informix_db._socket import MAX_READ_BYTES, IfxSocket + + +class _FakeSocket: + """Stands in for the raw socket. Records what recv actually asked for.""" + + def __init__(self, data: bytes = b"") -> None: + self.data = data + self.pos = 0 + self.recv_calls: list[int] = [] + + def recv(self, n: int) -> bytes: + self.recv_calls.append(n) + chunk = self.data[self.pos : self.pos + n] + self.pos += len(chunk) + return chunk + + def close(self) -> None: + # The EOF path force-closes; a stand-in has to survive that. + pass + + +def _socket_with(buffered: bytes, on_wire: bytes = b"") -> IfxSocket: + """An IfxSocket with ``buffered`` already read ahead into _recv_buf.""" + sock = IfxSocket.__new__(IfxSocket) + sock._sock = _FakeSocket(on_wire) + sock._recv_buf = bytearray(buffered) + sock._recv_pos = 0 + sock._recv_size = 65536 + sock._read_timeout = None + return sock + + +# --------------------------------------------------------------------------- +# read_exact must not step over the buffer +# --------------------------------------------------------------------------- + + +def test_read_exact_consumes_the_buffer_first() -> None: + sock = _socket_with(b"BUFFERED", on_wire=b"SOCKET") + assert sock.read_exact(8) == b"BUFFERED" + assert sock._sock.recv_calls == [], "must not touch the socket at all" + assert sock._recv_pos == 8 + + +def test_read_exact_spans_buffer_then_socket() -> None: + """The interesting case: a read that starts in the buffer and + finishes on the wire. Getting this wrong reorders the stream.""" + sock = _socket_with(b"HEAD", on_wire=b"TAIL") + assert sock.read_exact(8) == b"HEADTAIL" + assert sock._sock.recv_calls == [4], "only the shortfall comes from recv" + + +def test_read_exact_respects_a_partly_consumed_buffer() -> None: + sock = _socket_with(b"XXABCD") + sock._recv_pos = 2 # first two bytes already decoded + assert sock.read_exact(4) == b"ABCD" + assert sock._sock.recv_calls == [] + + +def test_read_exact_of_zero_is_empty() -> None: + sock = _socket_with(b"DATA") + assert sock.read_exact(0) == b"" + assert sock.read_exact(-5) == b"", "a negative count must not rewind" + assert sock._recv_pos == 0 + + +def test_short_read_error_reports_the_original_request() -> None: + """The message counts bytes; taking some from the buffer must not make + it lie about how many were asked for.""" + from informix_db.exceptions import OperationalError + + sock = _socket_with(b"AB", on_wire=b"") # 2 buffered, nothing on the wire + with pytest.raises(OperationalError, match="wanted 10 bytes"): + sock.read_exact(10) + + +def test_buffered_reader_and_direct_read_agree_on_one_stream() -> None: + """End to end: a BufferedSocketReader over-reads, then a direct + read_exact picks up exactly where it left off.""" + sock = _socket_with(b"", on_wire=b"\x00\x2aREST-OF-THE-STREAM") + reader = BufferedSocketReader(sock) + assert reader.read_short() == 42 + assert len(sock._recv_buf) - sock._recv_pos > 0, ( + "precondition: the reader must have over-read for this to mean " + "anything" + ) + assert sock.read_exact(18) == b"REST-OF-THE-STREAM" + + +# --------------------------------------------------------------------------- +# fill_recv_buf must not believe an arbitrary length +# --------------------------------------------------------------------------- + + +def test_absurd_length_is_refused_not_allocated() -> None: + sock = _socket_with(b"", on_wire=b"") + with pytest.raises(ProtocolError, match="refusing to read"): + sock.fill_recv_buf(MAX_READ_BYTES + 1) + assert sock._sock.recv_calls == [], "must refuse before any recv" + + +def test_refusal_names_the_knob() -> None: + """The error has to be actionable in both directions: framing is lost, + or the value genuinely is that big and the limit needs raising.""" + sock = _socket_with(b"", on_wire=b"") + with pytest.raises(ProtocolError) as exc: + sock.fill_recv_buf(2**31 - 1) + message = str(exc.value) + assert "2147483647" in message + assert "IFX_MAX_READ_BYTES" in message + + +def test_a_normal_length_is_unaffected() -> None: + sock = _socket_with(b"", on_wire=b"x" * 100) + sock.fill_recv_buf(100) + assert len(sock._recv_buf) - sock._recv_pos >= 100 + + +# --------------------------------------------------------------------------- +# skip +# --------------------------------------------------------------------------- + + +def test_buffered_skip_does_not_rewind_on_a_negative_count() -> None: + """The base reader's skip delegates to read_exact, which guards. This + one advances the cursor arithmetically, so an unguarded negative count + re-decodes bytes already consumed as if they were the next field.""" + sock = _socket_with(b"ABCDEFGH") + sock._recv_pos = 4 + BufferedSocketReader(sock).skip(-4) + assert sock._recv_pos == 4, "skip must never move the cursor backwards" + + +def test_buffered_skip_advances_normally() -> None: + sock = _socket_with(b"ABCDEFGH") + reader = BufferedSocketReader(sock) + reader.skip(4) + assert reader.read_exact(4) == b"EFGH" diff --git a/tests/test_transaction_state.py b/tests/test_transaction_state.py new file mode 100644 index 0000000..d99e551 --- /dev/null +++ b/tests/test_transaction_state.py @@ -0,0 +1,191 @@ +"""Transaction control run as SQL, and the flag that didn't notice. + +``Connection._in_transaction`` decides whether ``commit()`` and +``rollback()`` send anything at all, and the pool reads it to decide +whether a returned connection needs cleaning up. It was maintained +solely by the driver's own implicit ``SQ_BEGIN``, so a caller who wrote +``cursor.execute("BEGIN WORK")`` — an entirely reasonable thing to +write — walked straight past it. + +With autocommit on, nothing stopped that statement reaching the server. +A transaction opened, the flag stayed False, and ``rollback()`` returned +successfully having sent nothing. The rows it was asked to discard were +still there. The connection then went back to the pool holding an open +transaction and its locks, because the pool's cleanup is guarded by the +same flag. + +With autocommit off it failed instead, and for a sillier reason: the +driver's implicit ``SQ_BEGIN`` fired first, so the caller's ``BEGIN +WORK`` got ``-535``, "already in transaction". The driver and the user +competing to open the same transaction, and the user losing. + +The server labels these statements: type 34 for BEGIN, 35 for COMMIT, 36 +for ROLLBACK, with and without the ``WORK`` keyword. JDBC reads the same +three values off the describe and calls ``setTxBeginState`` / +``setTxEndState``. It also calls ``initiateTransaction`` *after* the +describe rather than before, which is what makes the skip possible — +until the describe lands you can't know the statement is transaction +control. +""" + +from __future__ import annotations + +import contextlib + +import pytest + +import informix_db +from informix_db.cursors import _TX_CONTROL_TYPES +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=10.0, + read_timeout=25.0, + **kw, + ) + + +def test_transaction_control_types_are_what_the_server_says( + logged_db_params: ConnParams, +) -> None: + """Pin the three constants against a live server rather than trusting + the decompiled source. Both spellings must map to the same type.""" + with _connect(logged_db_params, autocommit=True) as conn: + cur = conn.cursor() + seen = {} + for sql in ("BEGIN WORK", "COMMIT WORK", "ROLLBACK WORK", + "BEGIN", "COMMIT", "ROLLBACK"): + with conn._wire_lock: + conn._send_pdu( + cur._build_prepare_pdu(sql, num_qmarks=0), + statement_boundary=True, + ) + cur._read_describe_response() + cur._release_after_failure() + seen[sql] = cur._statement_type + assert seen["BEGIN WORK"] == seen["BEGIN"] == 34 + assert seen["COMMIT WORK"] == seen["COMMIT"] == 35 + assert seen["ROLLBACK WORK"] == seen["ROLLBACK"] == 36 + assert set(seen.values()) == _TX_CONTROL_TYPES + + +def test_rollback_after_sql_begin_actually_rolls_back( + logged_db_params: ConnParams, +) -> None: + """The data-loss case. rollback() reported success and sent nothing, + so the row it was asked to discard survived.""" + with _connect(logged_db_params, autocommit=True) as conn: + cur = conn.cursor() + with contextlib.suppress(Exception): + cur.execute("DROP TABLE t_txstate") + cur.execute("CREATE TABLE t_txstate (k INT)") + try: + cur.execute("BEGIN WORK") + assert conn._in_transaction, "SQL BEGIN must set the flag" + cur.execute("INSERT INTO t_txstate VALUES (1)") + conn.rollback() + cur.execute("SELECT COUNT(*) FROM t_txstate") + assert cur.fetchone() == (0,), "rollback() was a silent no-op" + assert not conn._in_transaction + finally: + with contextlib.suppress(Exception): + cur.execute("DROP TABLE t_txstate") + + +def test_sql_commit_clears_the_flag(logged_db_params: ConnParams) -> None: + """The mirror image: with the flag stuck True after a SQL COMMIT, the + next rollback() would send SQ_RBWORK with no transaction open and + draw -255.""" + with _connect(logged_db_params, autocommit=True) as conn: + cur = conn.cursor() + with contextlib.suppress(Exception): + cur.execute("DROP TABLE t_txstate2") + cur.execute("CREATE TABLE t_txstate2 (k INT)") + try: + cur.execute("BEGIN WORK") + cur.execute("INSERT INTO t_txstate2 VALUES (1)") + cur.execute("COMMIT WORK") + assert not conn._in_transaction, "SQL COMMIT must clear the flag" + conn.rollback() # must be a no-op, not a -255 + cur.execute("SELECT COUNT(*) FROM t_txstate2") + assert cur.fetchone() == (1,), "the committed row must survive" + finally: + with contextlib.suppress(Exception): + cur.execute("DROP TABLE t_txstate2") + + +def test_sql_begin_does_not_collide_with_the_implicit_one( + logged_db_params: ConnParams, +) -> None: + """Non-autocommit. The driver's implicit SQ_BEGIN used to fire first + and the caller's BEGIN WORK then got -535.""" + with _connect(logged_db_params, autocommit=False) as conn: + cur = conn.cursor() + with contextlib.suppress(Exception): + cur.execute("DROP TABLE t_txstate3") + conn.commit() + cur.execute("CREATE TABLE t_txstate3 (k INT)") + conn.commit() + try: + cur.execute("BEGIN WORK") + cur.execute("INSERT INTO t_txstate3 VALUES (1)") + conn.rollback() + cur.execute("SELECT COUNT(*) FROM t_txstate3") + assert cur.fetchone() == (0,) + conn.commit() + finally: + with contextlib.suppress(Exception): + cur.execute("DROP TABLE t_txstate3") + conn.commit() + + +def test_ordinary_dml_still_opens_a_transaction( + logged_db_params: ConnParams, +) -> None: + """_ensure_transaction moved from before the PREPARE to after the + describe. It still has to fire for everything that isn't transaction + control, or non-autocommit DML runs outside a transaction.""" + with _connect(logged_db_params, autocommit=False) as conn: + cur = conn.cursor() + with contextlib.suppress(Exception): + cur.execute("DROP TABLE t_txstate4") + conn.commit() + cur.execute("CREATE TABLE t_txstate4 (k INT)") + conn.commit() + try: + assert not conn._in_transaction + cur.execute("INSERT INTO t_txstate4 VALUES (1)") + assert conn._in_transaction, "DML must open a transaction" + conn.rollback() + cur.execute("SELECT COUNT(*) FROM t_txstate4") + assert cur.fetchone() == (0,) + conn.commit() + finally: + with contextlib.suppress(Exception): + cur.execute("DROP TABLE t_txstate4") + conn.commit() + + +def test_a_failed_transaction_statement_does_not_move_the_flag( + logged_db_params: ConnParams, +) -> None: + """The sync runs on the success path only. A COMMIT that the server + rejects has not ended anything.""" + with _connect(logged_db_params, autocommit=True) as conn: + cur = conn.cursor() + assert not conn._in_transaction + with pytest.raises(informix_db.Error): + cur.execute("COMMIT WORK") # -255, nothing to commit + assert not conn._in_transaction + cur.execute("SELECT FIRST 1 tabid FROM systables") + assert cur.fetchone() is not None