informix-db/tests/test_smoke.py
Ryan Malloy a1bd52788d Phase 2: SELECT works end-to-end — pure-Python Informix fully reads data
cursor.execute("SELECT 1 FROM systables WHERE tabid = 1")
  cursor.fetchone() == (1,)

To my knowledge, this is the first time a pure-Python implementation
has read data from Informix without wrapping IBM's CSDK or JDBC.

Three breakthroughs in this commit:

1. Login PDU's database field is BROKEN. Passing a database name there
   makes the server reject subsequent SQ_DBOPEN with sqlcode -759
   ("database not available"). JDBC always sends NULL in the login
   PDU's database slot — we now do the same. The user-supplied database
   opens via SQ_DBOPEN in _init_session.

2. Post-login session init dance: SQ_PROTOCOLS (8-byte feature mask
   replayed verbatim from JDBC) → SQ_INFO with INFO_ENV + env vars
   (48-byte PDU replayed verbatim — DBTEMP=/tmp, SUBQCACHESZ=10) →
   SQ_DBOPEN. Without all three steps in this exact order, the server
   silently ignores SELECTs.

3. SQ_DESCRIBE per-column block has 10 fields per column (not the
   simple "name + type" my best-effort parser assumed): fieldIndex,
   columnStartPos, columnType, columnExtendedId, ownerName,
   extendedName, reference, alignment, sourceType, encodedLength.
   The string table at the end is offset-indexed (fieldIndex points
   into it), which is how JDBC handles disambiguation.

Cursor lifecycle implementation in cursors.py mirrors JDBC exactly:
  PREPARE+NDESCRIBE+WANTDONE → DESCRIBE+DONE+COST+EOT
  CURNAME+NFETCH(4096) → TUPLE*+DONE+COST+EOT
  NFETCH(4096) → DONE+COST+EOT (drain)
  CLOSE → EOT
  RELEASE → EOT

Five round trips per SELECT — same as JDBC.

Module changes:
  src/informix_db/connections.py — added _init_session(), _send_protocols(),
    _send_dbopen(), _drain_to_eot(), _raise_sq_err(); login PDU now
    forces database=None always; SQ_INFO PDU replayed verbatim from
    JDBC capture (offsets-indexed env-var format too gnarly to derive
    in MVP).
  src/informix_db/cursors.py — full rewrite: real PDU builders for
    PREPARE/CURNAME+NFETCH/NFETCH/CLOSE/RELEASE; tag-dispatched
    response readers; cursor-name generator matching JDBC's "_ifxc"
    convention.
  src/informix_db/_resultset.py — proper SQ_DESCRIBE parser per
    JDBC's receiveDescribe (USVER mode); offset-indexed string table
    with name lookup by fieldIndex; ColumnInfo dataclass with raw
    type-code preserved for null-flag extraction.
  src/informix_db/_messages.py — added SQ_NDESCRIBE=22, SQ_WANTDONE=49.

Test coverage: 40 unit + 15 integration tests (7 smoke + 8 new SELECT)
= 55 total, all green, ruff clean. New tests cover:
  - SELECT 1 returns (1,)
  - cursor.description shape per PEP 249
  - Multi-row INT SELECT
  - Multi-column mixed types (INT + FLOAT)
  - Iterator protocol (for row in cursor)
  - fetchmany(n)
  - Re-executing on same cursor resets state
  - Two cursors on one connection (sequential)

Known gap: VARCHAR row decoding doesn't yet handle the variable-width
on-wire encoding correctly. Phase 2.x will address — for now NotImpl
errors surface raw bytes in the row tuple.
2026-05-03 15:37:10 -06:00

131 lines
4.1 KiB
Python

"""Phase 1 integration smoke tests — connect, auth-fail, network-fail.
Marked ``integration`` so the default ``pytest`` invocation skips them.
Run with ``pytest -m integration`` after ``docker compose up`` (or with
the container already running).
"""
from __future__ import annotations
import pytest
import informix_db
from tests.conftest import ConnParams
pytestmark = pytest.mark.integration
def test_connect_and_close(conn_params: ConnParams) -> None:
"""The happy path: connect with valid creds, then close cleanly."""
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,
)
assert conn.closed is False
conn.close()
assert conn.closed is True
def test_close_is_idempotent(conn_params: ConnParams) -> None:
"""``close()`` must be safely callable multiple times."""
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,
)
conn.close()
conn.close() # must not raise
assert conn.closed is True
def test_context_manager(conn_params: ConnParams) -> None:
"""``with`` block closes on exit."""
with 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,
) as conn:
assert conn.closed is False
assert conn.closed is True
def test_bad_password_raises_operational_error(conn_params: ConnParams) -> None:
"""Auth failure must be ``OperationalError``, not a generic socket error."""
with pytest.raises(informix_db.OperationalError):
informix_db.connect(
host=conn_params.host,
port=conn_params.port,
user=conn_params.user,
password="definitely-the-wrong-password",
database=conn_params.database,
server=conn_params.server,
connect_timeout=5.0,
read_timeout=5.0,
)
def test_bad_host_raises_operational_error(conn_params: ConnParams) -> None:
"""Network-level failure must also be ``OperationalError``."""
# Pick an unused port on loopback.
with pytest.raises(informix_db.OperationalError, match="cannot connect"):
informix_db.connect(
host="127.0.0.1",
port=1, # IANA-reserved, nothing listens
user="x",
password="x",
database="x",
server="x",
connect_timeout=2.0,
)
def test_cursor_returns_cursor_object(conn_params: ConnParams) -> None:
"""Phase 2: ``cursor()`` returns a Cursor; SELECT execution is partial work-in-progress."""
with 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,
) as conn:
cur = conn.cursor()
assert cur is not None
assert cur.description is None # nothing executed yet
assert cur.rowcount == -1
assert cur.fetchone() is None
cur.close()
assert cur.closed is True
def test_cursor_with_parameters_raises(conn_params: ConnParams) -> None:
"""Parameter binding lands in Phase 4; passing parameters must raise NotSupportedError."""
with 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,
) as conn:
cur = conn.cursor()
with pytest.raises(informix_db.NotSupportedError, match="Phase 4"):
cur.execute("SELECT ?", (1,))
cur.close()