Ryan Malloy
9b1fd8af2c
This commit takes informix-db from documentation-only (Phase 0 spike)
to a functional connect() / close() against a real Informix server.
To our knowledge, this is the first pure-socket Informix client in any
language — no CSDK, no JVM, no native libraries.
Layered architecture per the plan, mirroring PyMySQL's shape:
src/informix_db/
__init__.py — PEP 249 surface (connect, exceptions, paramstyle="numeric")
exceptions.py — full PEP 249 hierarchy declared up front
_socket.py — raw socket I/O (read_exact, write_all, timeouts)
_protocol.py — IfxStreamReader / IfxStreamWriter framing primitives
(big-endian, 16-bit-aligned variable payloads,
length-prefixed nul-terminated strings)
_messages.py — SQ_* tags from IfxMessageTypes + ASF/login markers
_auth.py — pluggable auth handlers; plain-password is the
only Phase-1 implementation
connections.py — Connection class: builds the binary login PDU
(SLheader + PFheader byte-for-byte per
PROTOCOL_NOTES.md §3), sends it, parses the
server response, wires up close()
Phase 1 design decisions locked in DECISION_LOG.md:
- paramstyle = "numeric" (matches Informix ESQL/C convention)
- Python >= 3.10
- autocommit defaults to off (PEP 249 implicit)
- License: MIT
- Distribution name: informix-db (verified PyPI-available)
Test coverage: 34 unit tests (codec round-trips against synthetic byte
streams; observed login-PDU values from the spike captures asserted as
exact byte literals) + 6 integration tests (connect, idempotent close,
context manager, bad-password → OperationalError, bad-host →
OperationalError, cursor() raises NotImplementedError).
pytest — runs 34 unit tests, no Docker needed
pytest -m integration — runs 6 integration tests against the
Developer Edition container (pinned by digest
in tests/docker-compose.yml)
pytest -m "" — runs everything
ruff is clean across src/ and tests/.
One bug found during smoke testing: threading.get_ident() can exceed
signed 32-bit on some processes, overflowing struct.pack("!i"). Fixed
the same way the JDBC reference does — clamp to signed 32-bit, fall
back to 0 if out of range. The field is diagnostic only.
One protocol-level observation that AMENDED the JDBC source reading:
the "capability section" in the login PDU is three independently
negotiated 4-byte ints (Cap_1=1, Cap_2=0x3c000000, Cap_3=0), not one
int + 8 reserved zero bytes as my CFR decompile read suggested. The
server echoes them back identically. Trust the wire over the
decompiler.
Phase 1 verification matrix (from PROTOCOL_NOTES.md §12):
- Login byte layout: confirmed (server accepts our pure-Python PDU)
- Disconnection: confirmed (SQ_EXIT round-trip works)
- Framing primitives: confirmed (34 unit tests)
- Error path: bad password → OperationalError, bad host → OperationalError
Phase 2 (Cursor / SELECT / basic types) is the next phase. The hard
unknowns there — exact column-descriptor layout, statement-time error
format — were called out as bounded gaps in Phase 0 and have existing
captures (02-select-1.socat.log, 02-dml-cycle.socat.log) to characterize
against.
87 lines
3.2 KiB
Python
87 lines
3.2 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_not_yet_implemented(conn_params: ConnParams) -> None:
|
|
"""Phase 1 declares ``cursor()`` as NotImplementedError; Phase 2 lands it."""
|
|
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, pytest.raises(NotImplementedError, match="Phase 2"):
|
|
conn.cursor()
|