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.
164 lines
5.8 KiB
Python
164 lines
5.8 KiB
Python
"""Unit tests for ``_protocol.py`` framing primitives.
|
|
|
|
These run without a database — they assert that the encoders produce
|
|
exactly the bytes IBM's JDBC reference would produce, and that the
|
|
decoders round-trip cleanly. The byte expectations come from the
|
|
captured login PDU in ``docs/CAPTURES/01-connect-only.socat.log``.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from io import BytesIO
|
|
|
|
import pytest
|
|
|
|
from informix_db._protocol import IfxStreamReader, IfxStreamWriter, ProtocolError
|
|
|
|
|
|
def make_writer() -> tuple[IfxStreamWriter, BytesIO]:
|
|
buf = BytesIO()
|
|
return IfxStreamWriter(buf), buf
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Endianness
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestEndianness:
|
|
"""All multi-byte integers MUST be big-endian (network byte order)."""
|
|
|
|
def test_short_is_big_endian(self) -> None:
|
|
w, buf = make_writer()
|
|
w.write_short(0x1234)
|
|
assert buf.getvalue() == b"\x12\x34"
|
|
|
|
def test_int_is_big_endian(self) -> None:
|
|
w, buf = make_writer()
|
|
w.write_int(0x12345678)
|
|
assert buf.getvalue() == b"\x12\x34\x56\x78"
|
|
|
|
def test_long_bigint_is_big_endian(self) -> None:
|
|
w, buf = make_writer()
|
|
w.write_long_bigint(0x123456789ABCDEF0)
|
|
assert buf.getvalue() == b"\x12\x34\x56\x78\x9a\xbc\xde\xf0"
|
|
|
|
def test_observed_capability_int_matches(self) -> None:
|
|
"""Cap_2 from the captured PDU is 0x3c000000 — verify our encoder hits it."""
|
|
w, buf = make_writer()
|
|
w.write_int(0x3C000000)
|
|
assert buf.getvalue() == b"\x3c\x00\x00\x00"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Padding
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestPadding:
|
|
"""Variable-length payloads must be padded to even byte alignment."""
|
|
|
|
def test_even_length_no_padding(self) -> None:
|
|
w, buf = make_writer()
|
|
w.write_padded(b"\x01\x02")
|
|
assert buf.getvalue() == b"\x01\x02"
|
|
|
|
def test_odd_length_pads_with_nul(self) -> None:
|
|
w, buf = make_writer()
|
|
w.write_padded(b"\x01\x02\x03")
|
|
assert buf.getvalue() == b"\x01\x02\x03\x00"
|
|
|
|
def test_empty_no_padding(self) -> None:
|
|
w, buf = make_writer()
|
|
w.write_padded(b"")
|
|
assert buf.getvalue() == b""
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Length-prefixed strings
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestStringEncoding:
|
|
"""``write_string_with_nul`` must produce ``[short len+1][bytes][nul]``."""
|
|
|
|
def test_simple_ascii(self) -> None:
|
|
"""The username 'informix' from the captured login: 9 bytes (8+nul)."""
|
|
w, buf = make_writer()
|
|
w.write_string_with_nul("informix")
|
|
assert buf.getvalue() == b"\x00\x09informix\x00"
|
|
|
|
def test_password_in4mix_matches_capture(self) -> None:
|
|
"""The password 'in4mix' from the captured login: 7 bytes (6+nul)."""
|
|
w, buf = make_writer()
|
|
w.write_string_with_nul("in4mix")
|
|
assert buf.getvalue() == b"\x00\x07in4mix\x00"
|
|
|
|
def test_empty_string(self) -> None:
|
|
w, buf = make_writer()
|
|
w.write_string_with_nul("")
|
|
# Empty string: length=1 (just the nul), content=just the nul.
|
|
assert buf.getvalue() == b"\x00\x01\x00"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Reader round-trips
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestRoundTrip:
|
|
"""Every encoder must round-trip through its matching decoder."""
|
|
|
|
@pytest.mark.parametrize("value", [0, 1, -1, 32767, -32768, 0x1234])
|
|
def test_short_round_trip(self, value: int) -> None:
|
|
w, buf = make_writer()
|
|
w.write_short(value)
|
|
r = IfxStreamReader(BytesIO(buf.getvalue()))
|
|
assert r.read_short() == value
|
|
|
|
@pytest.mark.parametrize("value", [0, 1, -1, 0x7FFFFFFF, -0x80000000, 0x3C000000])
|
|
def test_int_round_trip(self, value: int) -> None:
|
|
w, buf = make_writer()
|
|
w.write_int(value)
|
|
r = IfxStreamReader(BytesIO(buf.getvalue()))
|
|
assert r.read_int() == value
|
|
|
|
@pytest.mark.parametrize("value", [0, 1, -1, 0x7FFFFFFFFFFFFFFF, -0x8000000000000000])
|
|
def test_long_bigint_round_trip(self, value: int) -> None:
|
|
w, buf = make_writer()
|
|
w.write_long_bigint(value)
|
|
r = IfxStreamReader(BytesIO(buf.getvalue()))
|
|
assert r.read_long_bigint() == value
|
|
|
|
@pytest.mark.parametrize("text", ["", "informix", "in4mix", "hello world"])
|
|
def test_string_with_nul_round_trip(self, text: str) -> None:
|
|
w, buf = make_writer()
|
|
w.write_string_with_nul(text)
|
|
r = IfxStreamReader(BytesIO(buf.getvalue()))
|
|
assert r.read_string_with_nul() == text
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Failure modes
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestFailureModes:
|
|
"""Truncated or malformed streams must raise ``ProtocolError``, not silently misbehave."""
|
|
|
|
def test_short_eof_raises(self) -> None:
|
|
r = IfxStreamReader(BytesIO(b"\x12")) # 1 byte, want 2
|
|
with pytest.raises(ProtocolError, match="unexpected EOF"):
|
|
r.read_short()
|
|
|
|
def test_int_eof_raises(self) -> None:
|
|
r = IfxStreamReader(BytesIO(b"\x00\x00\x00")) # 3 bytes, want 4
|
|
with pytest.raises(ProtocolError, match="unexpected EOF"):
|
|
r.read_int()
|
|
|
|
def test_negative_string_length_rejected(self) -> None:
|
|
# A short with the high bit set parses as negative; should fail fast.
|
|
r = IfxStreamReader(BytesIO(b"\xff\xff"))
|
|
with pytest.raises(ProtocolError, match="negative string length"):
|
|
r.read_string_with_nul()
|