informix-db/tests/test_smart_lob.py

"""Phase 9 integration tests — smart-LOB BLOB/CLOB locator decoding.

Smart-LOB columns (BLOB type 102, CLOB type 101) are presented in the
SQ_DESCRIBE response as ``UDTFIXED`` (type 41) with extended_id 10 (BLOB)
or 11 (CLOB) and ``encoded_length=72`` (the locator size). The 72 bytes
in the SQ_TUPLE are an opaque server-side reference, NOT the actual data.

This phase surfaces the locator as a typed :class:`informix_db.BlobLocator`
or :class:`informix_db.ClobLocator` so users can recognize the column
type and not mistake the raw 72 bytes for actual content. Retrieving
the actual bytes requires the ``SQ_FPROUTINE`` + ``SQ_LODATA`` wire
protocols, deferred to Phase 10.

Test data is populated via the JDBC reference client (a Java helper)
since our driver doesn't yet support smart-LOB writes either.
"""

from __future__ import annotations

import contextlib
import dataclasses
from collections.abc import Iterator

import pytest

import informix_db
from tests.conftest import ConnParams

pytestmark = pytest.mark.integration


def _connect(params: ConnParams) -> informix_db.Connection:
    return informix_db.connect(
        host=params.host,
        port=params.port,
        user=params.user,
        password=params.password,
        database=params.database,
        server=params.server,
        connect_timeout=10.0,
        read_timeout=10.0,
        autocommit=True,
    )


@pytest.fixture
def blob_table_with_data(
    logged_db_params: ConnParams,
) -> Iterator[str]:
    """Create a BLOB table and seed it via Phase 11's write_blob_column.

    Originally (pre-Phase-11) this fixture used a JDBC helper to seed
    the row because our driver couldn't write smart-LOBs. Phase 11
    eliminated that dependency — pure Python end-to-end.
    """
    table = "t_blob_test"
    with _connect(logged_db_params) as conn:
        cur = conn.cursor()
        with contextlib.suppress(Exception):
            cur.execute(f"DROP TABLE {table}")
        try:
            cur.execute(f"CREATE TABLE {table} (id INT, data BLOB)")
        except informix_db.Error as e:
            pytest.skip(f"sbspace unavailable ({e!r})")
        cur.write_blob_column(
            f"INSERT INTO {table} VALUES (?, BLOB_PLACEHOLDER)",
            b"hello smart-lob bytes",
            (1,),
        )
    try:
        yield table
    finally:
        with _connect(logged_db_params) as conn:
            cur = conn.cursor()
            with contextlib.suppress(Exception):
                cur.execute(f"DROP TABLE {table}")


def test_blob_column_returns_blob_locator(
    logged_db_params: ConnParams, blob_table_with_data: str
) -> None:
    """SELECTing a BLOB column returns a :class:`BlobLocator`."""
    with _connect(logged_db_params) as conn:
        cur = conn.cursor()
        cur.execute(f"SELECT id, data FROM {blob_table_with_data}")
        rows = cur.fetchall()
        assert len(rows) == 1
        assert rows[0][0] == 1
        assert isinstance(rows[0][1], informix_db.BlobLocator)
        assert len(rows[0][1].raw) == 72


def test_blob_column_description_metadata(
    logged_db_params: ConnParams, blob_table_with_data: str
) -> None:
    """``cursor.description`` for BLOB column reports type=41 (UDTFIXED) size=72."""
    with _connect(logged_db_params) as conn:
        cur = conn.cursor()
        cur.execute(f"SELECT id, data FROM {blob_table_with_data} WHERE 1=0")
        # description is (name, type_code, display_size, internal_size,
        #                 precision, scale, null_ok)
        assert cur.description is not None
        data_col = cur.description[1]
        assert data_col[0] == "data"
        assert data_col[1] == 41  # UDTFIXED
        assert data_col[2] == 72  # display_size = locator size


def test_blob_locator_is_immutable(
    logged_db_params: ConnParams, blob_table_with_data: str
) -> None:
    """BlobLocator is frozen: the 72-byte ref can't be reassigned in place."""
    with _connect(logged_db_params) as conn:
        cur = conn.cursor()
        cur.execute(f"SELECT data FROM {blob_table_with_data}")
        (locator,) = cur.fetchone()
        with pytest.raises(dataclasses.FrozenInstanceError):
            locator.raw = b"x" * 72  # type: ignore[misc]


def test_blob_locator_repr_is_safe(
    logged_db_params: ConnParams, blob_table_with_data: str
) -> None:
    """``repr(locator)`` doesn't leak the raw bytes (which are opaque/internal)."""
    with _connect(logged_db_params) as conn:
        cur = conn.cursor()
        cur.execute(f"SELECT data FROM {blob_table_with_data}")
        (locator,) = cur.fetchone()
        r = repr(locator)
        assert "BlobLocator" in r
        # raw bytes must NOT leak into repr
        assert locator.raw.hex() not in r