informix-db/tests/test_capabilities.py

"""Integration tests for live SQ_PROTOCOLS negotiation.

These run against whatever server ``IFX_PORT`` points at, so `make
test-matrix` exercises them on 12.10, 14.10, and 15 in turn.

The important one is ``test_no_violated_assumptions``: this driver
hardcodes several wire-framing choices that SQLI actually negotiates, and
a mismatch corrupts rows silently. That test turns "we assume this" into
"we check this on every supported server".
"""

from __future__ import annotations

import pytest

import informix_db
from informix_db._capabilities import ENHANCED_PROTOCOL_CAP, ServerCapabilities
from tests.conftest import ConnParams

pytestmark = 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=10.0,
        read_timeout=10.0,
    )


def test_capabilities_are_decoded(conn_params: ConnParams) -> None:
    with _connect(conn_params) as conn:
        caps = conn.server_capabilities
        assert isinstance(caps, ServerCapabilities)
        assert caps.raw_mask, "server sent an empty protocols reply"
        assert caps.bits


def test_no_violated_assumptions(conn_params: ConnParams) -> None:
    """Every wire-framing shape we hardcode is one the server negotiated.

    If this fails, rows are being decoded against the wrong framing and
    the fix is to branch on the capability rather than assume it.
    """
    with _connect(conn_params) as conn:
        caps = conn.server_capabilities
        assert caps is not None
        assert caps.violated_assumptions() == []


def test_enhanced_protocol_negotiated(conn_params: ConnParams) -> None:
    """Cap_1 is the client's declared protocol level echoed back. Every
    supported server accepts 316."""
    with _connect(conn_params) as conn:
        caps = conn.server_capabilities
        assert caps is not None
        assert caps.cap_1 == ENHANCED_PROTOCOL_CAP
        assert caps.enhanced_protocol


def test_framing_capabilities_present(conn_params: ConnParams) -> None:
    with _connect(conn_params) as conn:
        caps = conn.server_capabilities
        assert caps is not None
        assert caps.four_byte_offset
        assert caps.varchar_var_len
        assert caps.remove_64k_limit
        assert caps.usver


def test_server_version_is_exposed(conn_params: ConnParams) -> None:
    """Assert the shape rather than a value so this holds across the matrix."""
    with _connect(conn_params) as conn:
        version = conn.server_version
        assert "Informix" in version
        assert "Version" in version


def test_server_version_reports_release_not_protocol_version(
    conn_params: ConnParams,
) -> None:
    """A field report flagged ``server_version`` as looking like a
    client-SDK version: the login response announces 12.10 servers as 9.56
    and 14.10 as 9.59. ``server_version`` now answers with the release; the
    raw login string moved to ``server_version_internal``."""
    with _connect(conn_params) as conn:
        release = conn.server_version
        internal = conn.server_version_internal
        assert "Informix" in internal
        # On 12.10/14.10 these genuinely differ; on 15 they agree.
        if "9.5" in internal:
            assert release != internal
            assert "9.5" not in release, (
                f"server_version still reports the protocol version: {release!r}"
            )


def test_server_version_is_cached(conn_params: ConnParams) -> None:
    """It costs a round-trip, so repeated access must not repeat it."""
    with _connect(conn_params) as conn:
        assert conn.server_version == conn.server_version
        assert conn._server_version_full is not None


def test_server_version_degrades_without_a_database(
    conn_params: ConnParams,
) -> None:
    """The DBINFO lookup needs an open database. With none, the property
    must fall back rather than raise — a version lookup should never be
    able to break a working connection."""
    conn = informix_db.connect(
        host=conn_params.host,
        port=conn_params.port,
        user=conn_params.user,
        password=conn_params.password,
        database=None,
        server=conn_params.server,
        connect_timeout=10.0,
        read_timeout=10.0,
    )
    try:
        assert isinstance(conn.server_version, str)
    finally:
        conn.close()


def test_capabilities_survive_multiple_connections(
    conn_params: ConnParams,
) -> None:
    """Negotiation happens per-connection; two connections to the same
    server must agree."""
    with _connect(conn_params) as first, _connect(conn_params) as second:
        assert first.server_capabilities is not None
        assert second.server_capabilities is not None
        assert (
            first.server_capabilities.raw_mask
            == second.server_capabilities.raw_mask
        )
        assert first.server_capabilities.bits == second.server_capabilities.bits