informix-db/tests/test_statement_classification.py

"""Deciding whether a statement needs a cursor, by asking rather than guessing.

The driver chose between "open a cursor and fetch" and "execute and
release" by checking whether the first word of the SQL was ``SELECT``.
That gets five ordinary forms wrong — a leading comment in any of the
three Informix flavours, a parenthesized select, a parenthesized UNION,
and a CTE. All five are perfectly good queries, and all five failed with
``-260 Cursor name already in use``, an error that describes neither the
cause nor anything the caller did. It says "cursor" because the driver
sent SQ_EXECUTE where the server was waiting to open one.

The server had been telling us the answer the whole time.
``statement_type`` is the first field of the DESCRIBE response, and
``parse_describe`` has always parsed it into the metadata dict, where
nothing read it. Every SELECT form above reports 2.

The comment that justified the heuristic said ``nfields`` couldn't
distinguish these cases, because ``INSERT INTO t VALUES (?)`` also
describes a column. That much was true — and it argued for the wrong
conclusion, because ``statement_type`` distinguishes them exactly. That
INSERT reports 6.

The predicate is now JDBC's ``IfxSqli.isResultSet``: type 2, or type 56
(``EXECUTE PROCEDURE``/``FUNCTION``) with at least one column, since a
routine may or may not return rows. That last clause fixes
``EXECUTE FUNCTION`` as a side effect — it used to run down the DML path
and discard its return value.
"""

from __future__ import annotations

import contextlib

import pytest

import informix_db
from informix_db.cursors import _produces_result_set
from tests.conftest import ConnParams

# ---------------------------------------------------------------------------
# The predicate
# ---------------------------------------------------------------------------


@pytest.mark.parametrize(
    ("statement_type", "ncolumns", "expected", "why"),
    [
        (2, 1, True, "SELECT"),
        (2, 0, True, "SELECT describing no columns is still a SELECT"),
        (6, 1, False, "INSERT ... VALUES (?) describes a column but is DML"),
        (6, 0, False, "INSERT with literals"),
        (32, 0, False, "DELETE"),
        (33, 0, False, "UPDATE"),
        (45, 0, False, "CREATE"),
        (56, 0, False, "EXECUTE PROCEDURE returning nothing"),
        (56, 2, True, "EXECUTE FUNCTION returning rows"),
        (0, 0, False, "unknown type defaults to the non-cursor path"),
    ],
)
def test_result_set_predicate(
    statement_type: int, ncolumns: int, expected: bool, why: str
) -> None:
    assert _produces_result_set(statement_type, ncolumns) is expected, why


# ---------------------------------------------------------------------------
# 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,
        read_timeout=25.0,
        autocommit=True,
    )


_ONE = "SELECT FIRST 1 tabid FROM systables"


@pytest.mark.integration
@pytest.mark.parametrize(
    ("label", "sql"),
    [
        pytest.param("plain", _ONE, id="plain"),
        pytest.param("lowercase", _ONE.lower(), id="lowercase"),
        pytest.param("leading-whitespace", f"  \n\t {_ONE}", id="whitespace"),
        pytest.param("line-comment", f"-- pick one\n{_ONE}", id="line-comment"),
        pytest.param("block-comment", f"/* pick one */ {_ONE}", id="block-comment"),
        pytest.param("brace-comment", f"{{ pick one }} {_ONE}", id="brace-comment"),
        pytest.param(
            "cte",
            "WITH c AS (SELECT tabid FROM systables) "
            "SELECT FIRST 1 tabid FROM c",
            id="cte",
        ),
        pytest.param("parenthesized", f"({_ONE})", id="parenthesized"),
        pytest.param(
            "union", f"{_ONE} UNION SELECT 99 FROM systables", id="union"
        ),
        pytest.param(
            "parenthesized-union",
            f"({_ONE}) UNION (SELECT 99 FROM systables)",
            id="paren-union",
        ),
    ],
)
def test_every_select_form_opens_a_cursor(
    conn_params: ConnParams, label: str, sql: str
) -> None:
    """The five non-``plain`` forms below the whitespace case all failed
    with -260 under the first-word heuristic."""
    with _connect(conn_params) as conn:
        cur = conn.cursor()
        try:
            cur.execute(sql)
        except informix_db.ProgrammingError as exc:
            # Informix 12.10 has no CTEs and rejects WITH at offset 1.
            # A syntax error is the server declining the grammar, which
            # is a different thing from the driver routing it wrongly —
            # that produced -260, not -201.
            if getattr(exc, "sqlcode", None) == -201:
                pytest.skip(f"server does not support this syntax: {label}")
            raise
        rows = cur.fetchall()
        assert rows, f"{label}: expected rows, got none"
        assert cur.description is not None


@pytest.mark.integration
def test_parameterized_insert_is_not_mistaken_for_a_query(
    conn_params: ConnParams,
) -> None:
    """The case the old comment worried about, and the reason it kept the
    heuristic: this DESCRIBEs a column. ``statement_type`` says 6."""
    with _connect(conn_params) as conn:
        cur = conn.cursor()
        cur.execute("CREATE TEMP TABLE t_cls (k INT)")
        cur.execute("INSERT INTO t_cls VALUES (?)", (7,))
        assert cur.rowcount == 1
        cur.execute("SELECT k FROM t_cls")
        assert cur.fetchall() == [(7,)]


@pytest.mark.integration
def test_execute_function_returns_its_value(conn_params: ConnParams) -> None:
    """Type 56 with columns. Under the first-word heuristic this ran down
    the DML path and the return value was discarded."""
    with _connect(conn_params) as conn:
        cur = conn.cursor()
        with contextlib.suppress(Exception):
            cur.execute("DROP FUNCTION ifxdrv_dbl")
        cur.execute(
            "CREATE FUNCTION ifxdrv_dbl(n INT) RETURNING INT; "
            "RETURN n * 2; END FUNCTION"
        )
        try:
            cur.execute("EXECUTE FUNCTION ifxdrv_dbl(21)")
            assert cur.fetchall() == [(42,)]
        finally:
            with contextlib.suppress(Exception):
                cur.execute("DROP FUNCTION ifxdrv_dbl")


@pytest.mark.integration
def test_dml_still_takes_the_non_cursor_path(conn_params: ConnParams) -> None:
    with _connect(conn_params) as conn:
        cur = conn.cursor()
        cur.execute("CREATE TEMP TABLE t_cls2 (k INT)")
        cur.execute("INSERT INTO t_cls2 VALUES (1)")
        cur.execute("UPDATE t_cls2 SET k = 2")
        assert cur.rowcount == 1
        cur.execute("DELETE FROM t_cls2")
        assert cur.rowcount == 1
        assert cur.description is None


@pytest.mark.integration
def test_executemany_refuses_a_query_the_first_word_missed(
    conn_params: ConnParams,
) -> None:
    """The pre-flight check shares the heuristic's blind spots, so a
    comment-prefixed SELECT reaches PREPARE. The post-DESCRIBE check
    catches it, and the connection stays usable — the refusal releases
    the statement. (A leading comment rather than a CTE, so this also
    runs on 12.10, which has no CTEs.)"""
    with _connect(conn_params) as conn:
        cur = conn.cursor()
        with pytest.raises(informix_db.NotSupportedError):
            cur.executemany(
                "/* batched? no */ SELECT FIRST 1 tabid FROM systables "
                "WHERE tabid <> ?",
                [(1,), (2,)],
            )
        cur.execute("SELECT FIRST 1 tabid FROM systables")
        assert cur.fetchone() is not None, "refusal leaked the statement"


@pytest.mark.integration
def test_executemany_still_refuses_a_plain_select(
    conn_params: ConnParams,
) -> None:
    with _connect(conn_params) as conn:
        cur = conn.cursor()
        with pytest.raises(informix_db.NotSupportedError):
            cur.executemany(
                "SELECT FIRST 1 tabid FROM systables WHERE tabid <> ?",
                [(1,), (2,)],
            )