informix-db/tests/conftest.py

"""Shared pytest fixtures.

The integration tests need a running Informix server on ``localhost:9088``
with the default Developer Edition credentials (``informix`` / ``in4mix``
on the ``sysmaster`` database). Run::

    docker compose -f tests/docker-compose.yml up -d

before the integration suite. See ``docs/DECISION_LOG.md`` for the
pinned image digest and credential rationale.

Connection parameters are overridable via env vars
(``IFX_HOST`` / ``IFX_PORT`` / ``IFX_USER`` / ``IFX_PASSWORD`` /
``IFX_DATABASE`` / ``IFX_SERVER``) so the same suite runs against a
non-default instance if you have one.
"""

from __future__ import annotations

import os
import socket
from collections.abc import Iterator
from typing import NamedTuple

import pytest


class ConnParams(NamedTuple):
    host: str
    port: int
    user: str
    password: str
    database: str
    server: str


@pytest.fixture(scope="session")
def conn_params() -> ConnParams:
    """Connection parameters for the test Informix instance."""
    return ConnParams(
        host=os.environ.get("IFX_HOST", "127.0.0.1"),
        port=int(os.environ.get("IFX_PORT", "9088")),
        user=os.environ.get("IFX_USER", "informix"),
        password=os.environ.get("IFX_PASSWORD", "in4mix"),
        database=os.environ.get("IFX_DATABASE", "sysmaster"),
        server=os.environ.get("IFX_SERVER", "informix"),
    )


@pytest.fixture(scope="session", autouse=True)
def _check_integration_server(request: pytest.FixtureRequest, conn_params: ConnParams) -> None:
    """Skip integration tests cleanly when the server isn't up.

    Avoids a sea of ``ConnectionRefusedError`` failures masking real bugs
    when the user runs the full suite without ``docker compose up``.
    """
    if "integration" not in request.config.getoption("-m", default=""):
        return
    try:
        with socket.create_connection((conn_params.host, conn_params.port), timeout=2.0):
            return
    except OSError as e:
        pytest.skip(
            f"Informix server not reachable at {conn_params.host}:{conn_params.port} ({e}). "
            f"Start it with: docker compose -f tests/docker-compose.yml up -d"
        )


@pytest.fixture
def ifx_connection(conn_params: ConnParams) -> Iterator[object]:
    """Open a fresh Informix connection for one test, then close it."""
    import informix_db

    conn = 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,
    )
    try:
        yield conn
    finally:
        conn.close()


@pytest.fixture(scope="session")
def logged_db_params(conn_params: ConnParams) -> ConnParams:
    """Connection parameters for a LOGGED database — required for real
    transaction tests (commit/rollback against an unlogged DB is a no-op).

    Defaults to ``testdb`` (created by Phase 7 setup with
    ``CREATE DATABASE testdb WITH LOG``). Override via
    ``IFX_LOGGED_DATABASE``. The database must exist; if missing, the
    test will fail with a clear "database not found" error.
    """
    return conn_params._replace(
        database=os.environ.get("IFX_LOGGED_DATABASE", "testdb"),
    )


@pytest.fixture(scope="session", autouse=True)
def _ensure_testdb(
    request: pytest.FixtureRequest, conn_params: ConnParams
) -> None:
    """Lazily ensure ``testdb`` exists if integration tests are running.

    If the user has set ``IFX_LOGGED_DATABASE`` to something else, we
    don't try to create it (assume they manage that DB themselves).
    """
    if "integration" not in request.config.getoption("-m", default=""):
        return
    if os.environ.get("IFX_LOGGED_DATABASE"):
        return  # user-managed DB; don't auto-create
    try:
        import informix_db

        conn = informix_db.connect(
            host=conn_params.host,
            port=conn_params.port,
            user=conn_params.user,
            password=conn_params.password,
            database="sysmaster",
            server=conn_params.server,
            connect_timeout=5.0,
            autocommit=True,
        )
        cur = conn.cursor()
        cur.execute(
            "SELECT name FROM sysdatabases WHERE name = 'testdb'"
        )
        if not cur.fetchone():
            cur.execute("CREATE DATABASE testdb WITH LOG")
        conn.close()
    except Exception:
        # If creation fails, downstream tests will surface a clearer error.
        pass