informix-db/tests/test_statement_release.py

"""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()