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
import os
import shutil
import subprocess
from collections.abc import Iterator
from pathlib import Path

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,
    )


def _java_available() -> bool:
    """JDBC reference client requires java + the IfxJdbc jar."""
    if not shutil.which("java"):
        return False
    return Path("build/ifxjdbc.jar").exists() and Path("build/tests").exists()


@pytest.fixture
def blob_table_with_data(
    logged_db_params: ConnParams,
) -> Iterator[str]:
    """Create a BLOB table and seed it via the JDBC reference client.

    Smart-LOB writes require the SQ_FPROUTINE + SQ_LODATA protocols
    that our driver doesn't implement yet (Phase 10). We use the
    JDBC reference client (``RefBlob``) to seed test data.
    """
    if not _java_available():
        pytest.skip(
            "JDBC reference client unavailable (need java + build/ifxjdbc.jar)"
        )

    table = "t_blob_test"
    # Drop if exists
    with _connect(logged_db_params) as conn:
        cur = conn.cursor()
        with contextlib.suppress(Exception):
            cur.execute(f"DROP TABLE {table}")

    # Use Java helper to populate (compile RefBlob inline if needed)
    helper_dir = Path("build/tests/reference")
    helper_dir.mkdir(parents=True, exist_ok=True)
    helper_src = Path("tests/reference/RefBlobTest.java")
    if not helper_src.exists():
        helper_src.write_text(
            'package tests.reference;\n'
            'import java.sql.*;\n'
            'public class RefBlobTest {\n'
            '    public static void main(String[] args) throws Exception {\n'
            '        String table = args[0], payload = args[1];\n'
            '        Class.forName("com.informix.jdbc.IfxDriver");\n'
            '        try (Connection c = DriverManager.getConnection(\n'
            '             "jdbc:informix-sqli://127.0.0.1:9088/testdb:INFORMIXSERVER=informix",\n'
            '             "informix", "in4mix")) {\n'
            '            c.setAutoCommit(true);\n'
            '            try (Statement s = c.createStatement()) {\n'
            '                s.execute("CREATE TABLE " + table + " (id INT, data BLOB)");\n'
            '            }\n'
            '            try (PreparedStatement ps = c.prepareStatement(\n'
            '                    "INSERT INTO " + table + " VALUES (?, ?)")) {\n'
            '                ps.setInt(1, 1);\n'
            '                ps.setBytes(2, payload.getBytes());\n'
            '                ps.executeUpdate();\n'
            '            }\n'
            '        }\n'
            '    }\n'
            '}\n'
        )
    subprocess.run(
        [
            "javac", "-cp", "build/ifxjdbc.jar",
            "-d", "build/", str(helper_src),
        ],
        check=True, capture_output=True,
    )
    subprocess.run(
        [
            "java", "-cp", "build/ifxjdbc.jar:build/",
            "tests.reference.RefBlobTest", table, "hello smart-lob bytes",
        ],
        check=True, capture_output=True,
        env={**os.environ, "IFX_DATABASE": "testdb"},
    )

    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