Ryan Malloy
1c19c71cb6
Introduces driver-managed transactions that work seamlessly across logged and unlogged databases. The user calls commit() and rollback() without needing to know which kind they're hitting — the connection tracks transaction state internally. Three protocol facts came out of integration testing: 1. Logged DBs in non-ANSI mode require an explicit SQ_BEGIN before the first DML — the server doesn't auto-open a transaction. Connection._ensure_transaction() sends SQ_BEGIN lazily and is idempotent within an open txn. After commit/rollback, the next DML triggers a fresh BEGIN. 2. SQ_RBWORK has a [short savepoint=0] payload before the SQ_EOT framing tag — sending SQ_RBWORK alone causes the server to hang silently (waiting for the missing 2 bytes). SQ_CMMTWORK has no payload. This is the same pattern as the SHORT-vs-INT bug from Phase 4.x and the 2-byte length prefix from Phase 6.c — when the server hangs, it's an incomplete PDU body. 3. SQ_XACTSTAT (tag 99) is a logged-DB-only message that's interleaved with normal responses. Now drained in all four response-reading paths: cursor _drain_to_eot, _read_describe_ response, _read_fetch_response, and connection _drain_to_eot. For unlogged DBs (e.g., sysmaster), SQ_BEGIN returns -201 and we cache that result so subsequent DML doesn't re-probe. commit() and rollback() are silent no-ops in that case — same client code works across both DB modes. Tests: * New tests/test_transactions.py — 10 integration tests covering commit visibility, rollback isolation, multi-row rollback, partial commit-then-rollback, autocommit behavior, cross-connection durability, UPDATE/DELETE rollback, implicit per-statement txn. * conftest.py auto-creates testdb (logged) for the suite. * Two old tests rewritten to assert new no-op behavior on unlogged DBs (test_commit_rollback_in_unlogged_db_is_noop, test_commit_in_unlogged_db_is_noop). Total: 53 unit + 98 integration = 151 tests. The Phase 3 "gate test" (test_rollback_hides_insert) — a rolled-back INSERT must be invisible to subsequent SELECTs in the same session — now passes against a real logged database for the first time.
130 lines
5.0 KiB
Python
130 lines
5.0 KiB
Python
"""Phase 5.a integration tests — error message decoding + PEP 249 classification.
|
|
|
|
Server SQ_ERR responses get decoded to:
|
|
- the appropriate PEP 249 exception class (Programming/Integrity/...)
|
|
- a human-readable message with sqlcode, near-token, ISAM code, offset
|
|
- structured ``sqlcode``, ``isamcode``, ``offset``, ``near`` attributes
|
|
|
|
The connection survives errors — subsequent queries work normally.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import pytest
|
|
|
|
import informix_db
|
|
from tests.conftest import ConnParams
|
|
|
|
pytestmark = pytest.mark.integration
|
|
|
|
|
|
def _connect(conn_params: ConnParams) -> informix_db.Connection:
|
|
return 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,
|
|
)
|
|
|
|
|
|
def test_syntax_error_is_programming_error(conn_params: ConnParams) -> None:
|
|
"""sqlcode -201 (syntax error) → ProgrammingError."""
|
|
with _connect(conn_params) as conn:
|
|
cur = conn.cursor()
|
|
with pytest.raises(informix_db.ProgrammingError) as excinfo:
|
|
cur.execute("SELEKT * FROM systables")
|
|
e = excinfo.value
|
|
assert e.sqlcode == -201
|
|
assert "syntax error" in str(e).lower()
|
|
|
|
|
|
def test_table_not_found_is_programming_error(conn_params: ConnParams) -> None:
|
|
"""sqlcode -206 (table not found) → ProgrammingError; near-token has table name."""
|
|
with _connect(conn_params) as conn:
|
|
cur = conn.cursor()
|
|
with pytest.raises(informix_db.ProgrammingError) as excinfo:
|
|
cur.execute("SELECT * FROM no_such_table_xyz")
|
|
e = excinfo.value
|
|
assert e.sqlcode == -206
|
|
assert e.near == "no_such_table_xyz"
|
|
assert "not in the database" in str(e)
|
|
|
|
|
|
def test_column_not_found_is_programming_error(conn_params: ConnParams) -> None:
|
|
"""sqlcode -217 (column not found) → ProgrammingError; near-token has column name."""
|
|
with _connect(conn_params) as conn:
|
|
cur = conn.cursor()
|
|
with pytest.raises(informix_db.ProgrammingError) as excinfo:
|
|
cur.execute("SELECT no_such_column FROM systables")
|
|
e = excinfo.value
|
|
assert e.sqlcode == -217
|
|
assert e.near == "no_such_column"
|
|
|
|
|
|
def test_unique_violation_is_integrity_error(conn_params: ConnParams) -> None:
|
|
"""sqlcode -268 (UNIQUE violation) → IntegrityError."""
|
|
with _connect(conn_params) as conn:
|
|
cur = conn.cursor()
|
|
cur.execute("CREATE TEMP TABLE t_uniq (id INTEGER PRIMARY KEY)")
|
|
cur.execute("INSERT INTO t_uniq VALUES (?)", (1,))
|
|
with pytest.raises(informix_db.IntegrityError) as excinfo:
|
|
cur.execute("INSERT INTO t_uniq VALUES (?)", (1,))
|
|
e = excinfo.value
|
|
assert e.sqlcode == -268
|
|
assert "duplicate" in str(e).lower()
|
|
|
|
|
|
def test_not_null_violation_is_integrity_error(conn_params: ConnParams) -> None:
|
|
"""sqlcode -391 (NOT NULL violation) → IntegrityError."""
|
|
with _connect(conn_params) as conn:
|
|
cur = conn.cursor()
|
|
cur.execute("CREATE TEMP TABLE t_nn (id INTEGER NOT NULL)")
|
|
with pytest.raises(informix_db.IntegrityError) as excinfo:
|
|
cur.execute("INSERT INTO t_nn VALUES (?)", (None,))
|
|
assert excinfo.value.sqlcode == -391
|
|
|
|
|
|
def test_commit_in_unlogged_db_is_noop(conn_params: ConnParams) -> None:
|
|
"""commit() on an unlogged DB does NOT raise — it's a silent no-op.
|
|
|
|
Phase 7 made this driver-side smart. Calling commit() when no
|
|
server-side transaction is open is a no-op (same as autocommit-mode
|
|
commit). This matches what most DB-API drivers do for graceful
|
|
degradation. The PEP 249 spec is silent on this case.
|
|
"""
|
|
with _connect(conn_params) as conn:
|
|
conn.commit() # must not raise
|
|
conn.rollback() # must not raise either
|
|
|
|
|
|
def test_connection_survives_query_error(conn_params: ConnParams) -> None:
|
|
"""Critical: a failed query must not poison the connection — subsequent queries must work."""
|
|
with _connect(conn_params) as conn:
|
|
cur = conn.cursor()
|
|
# First query fails
|
|
with pytest.raises(informix_db.ProgrammingError):
|
|
cur.execute("SELECT * FROM no_such_table")
|
|
# Second query works
|
|
cur.execute("SELECT 1 FROM systables WHERE tabid = 1")
|
|
assert cur.fetchone() == (1,)
|
|
|
|
|
|
def test_error_has_structured_fields(conn_params: ConnParams) -> None:
|
|
"""``sqlcode``, ``isamcode``, ``offset``, ``near`` are attached for inspection."""
|
|
with _connect(conn_params) as conn:
|
|
cur = conn.cursor()
|
|
with pytest.raises(informix_db.Error) as excinfo:
|
|
cur.execute("SELECT * FROM no_such_table")
|
|
e = excinfo.value
|
|
# All four structured attributes present
|
|
assert hasattr(e, "sqlcode")
|
|
assert hasattr(e, "isamcode")
|
|
assert hasattr(e, "offset")
|
|
assert hasattr(e, "near")
|
|
assert e.sqlcode == -206
|
|
assert e.near == "no_such_table"
|