"""Every door out of a statement has to release it on the way through. A statement that fails is still allocated server-side. Skipping the RELEASE bricks the connection: the next PREPARE collides with the leaked one, and every subsequent call returns a nonsense error whose offset points back at the *failed* SQL rather than the new statement. There are six exits from the execute paths, and the guard was added to them one at a time, each after a user hit it: 1. ``_execute_dml`` drain 2. ``_execute_dml_with_params`` build 3. ``_execute_dml_with_params`` drain 4. ``executemany`` pipeline build/send 5. ``_execute_select_with_params`` build 6. ``_execute_select`` fetch loop Two were still open, and both are ordinary to reach: * **The parameterized-SELECT bind drain.** The build was guarded and the drain was not. Passing a string where the column is an INT gets a clean encode, so the rejection comes from the *server* (-1213, -415) during the bind drain — past the guard. A wrong-typed parameter is about as common as application mistakes get. * **The scrollable-cursor open.** No guard at all, and the worst place to lack one: the GC-time finalizer is armed on the line *after* the drain, so a failure there left the statement allocated with no fallback of any kind. The guard is now one helper rather than six hand-rolled copies, which also fixed a defect in the copies that had a cursor to close: CLOSE and RELEASE shared a single ``contextlib.suppress`` block, so a CLOSE that raised skipped the RELEASE — losing the half that actually matters. """ from __future__ import annotations import pytest import informix_db from informix_db.cursors import _RELEASE_PDU, Cursor from tests.conftest import ConnParams # --------------------------------------------------------------------------- # The helper — no server needed # --------------------------------------------------------------------------- class _FakeConn: def __init__(self) -> None: self.sent: list[bytes] = [] def _send_pdu(self, pdu: bytes) -> None: self.sent.append(pdu) class _FakeCursor: """Duck-types the four attributes ``_release_after_failure`` touches.""" def __init__(self, *, close_raises: bool = False) -> None: self._conn = _FakeConn() self._close_raises = close_raises def _build_close_pdu(self) -> bytes: return b"CLOSE" def _build_release_pdu(self) -> bytes: return _RELEASE_PDU def _drain_to_eot(self) -> None: if self._close_raises and self._conn.sent[-1] == b"CLOSE": raise OSError("wire went away mid-close") def test_release_is_sent_even_when_close_fails() -> None: """The regression in the hand-rolled copies. CLOSE and RELEASE shared one suppress block, so a failing CLOSE swallowed the RELEASE — and a lost cursor handle is a nuisance while a lost statement breaks the next call.""" cur = _FakeCursor(close_raises=True) Cursor._release_after_failure(cur, close_cursor=True) assert _RELEASE_PDU in cur._conn.sent, ( "a CLOSE that raises must not prevent the RELEASE" ) def test_close_is_skipped_when_no_cursor_was_opened() -> None: cur = _FakeCursor() Cursor._release_after_failure(cur) assert cur._conn.sent == [_RELEASE_PDU] def test_cleanup_never_propagates() -> None: """Cleanup runs inside an ``except``. If it raises, it replaces the real SQL error with a secondary failure from the cleanup path, which is strictly worse for the caller.""" class _Hostile(_FakeCursor): def _drain_to_eot(self) -> None: raise OSError("connection reset") Cursor._release_after_failure(_Hostile(), close_cursor=True) # --------------------------------------------------------------------------- # Against a real server # --------------------------------------------------------------------------- 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, # Bounded: a leaked statement used to manifest as a hang. read_timeout=25.0, autocommit=True, ) @pytest.mark.integration @pytest.mark.parametrize( ("sql", "params"), [ pytest.param( "SELECT tabid FROM systables WHERE tabid = ?", ("not-an-int",), id="string-for-int", ), pytest.param( "SELECT ?::INT FROM systables WHERE tabid = 1", (2**40,), id="out-of-range-int", ), ], ) def test_server_rejected_bind_releases_statement( conn_params: ConnParams, sql: str, params: tuple ) -> None: """A parameterized SELECT whose bind the *server* rejects. The value encodes cleanly, so this lands past the build guard and in the drain that had none. Repeated because a leak only shows on the call after it — the first failure looks fine on its own.""" with _connect(conn_params) as conn: cur = conn.cursor() for i in range(4): with pytest.raises(informix_db.Error): cur.execute(sql, params) cur.execute("SELECT FIRST 1 tabid FROM systables") assert cur.fetchone() is not None, f"broken after {i + 1} binds" @pytest.mark.integration def test_failed_scroll_open_releases_statement(conn_params: ConnParams) -> None: """FOR UPDATE prepares cleanly and fails at OPEN with -526, which is exactly the branch that had no guard.""" with _connect(conn_params) as conn: scroll = conn.cursor(scrollable=True) for i in range(4): with pytest.raises(informix_db.Error): scroll.execute("SELECT tabid FROM systables FOR UPDATE") assert not scroll._server_cursor_open, ( "a failed open must not leave the cursor marked live" ) other = conn.cursor() other.execute("SELECT FIRST 1 tabid FROM systables") assert other.fetchone() is not None, f"broken after {i + 1} opens" other.close() scroll.close() @pytest.mark.integration def test_scrollable_cursor_still_works_after_a_failed_open( conn_params: ConnParams, ) -> None: """The same cursor object must be reusable — a failed open is an ordinary error, not a terminal state for the cursor.""" with _connect(conn_params) as conn: scroll = conn.cursor(scrollable=True) with pytest.raises(informix_db.Error): scroll.execute("SELECT tabid FROM systables FOR UPDATE") scroll.execute("SELECT tabid FROM systables ORDER BY tabid") assert scroll.fetch_first() is not None scroll.close()