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.
122 lines
4.6 KiB
Python
122 lines
4.6 KiB
Python
"""Phase 3 integration tests — DDL + DML.
|
|
|
|
Tests CREATE TEMP TABLE, INSERT, UPDATE, DELETE end-to-end with row counts
|
|
verified via subsequent SELECTs.
|
|
"""
|
|
|
|
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_create_temp_table(conn_params: ConnParams) -> None:
|
|
"""DDL: CREATE TEMP TABLE returns rowcount=0 and no description."""
|
|
with _connect(conn_params) as conn:
|
|
cur = conn.cursor()
|
|
cur.execute("CREATE TEMP TABLE t_phase3_a (id INTEGER, name VARCHAR(50))")
|
|
assert cur.description is None
|
|
assert cur.rowcount == 0
|
|
|
|
|
|
def test_insert_then_select_persists(conn_params: ConnParams) -> None:
|
|
"""DML: INSERT actually persists; SELECT returns inserted rows."""
|
|
with _connect(conn_params) as conn:
|
|
cur = conn.cursor()
|
|
cur.execute("CREATE TEMP TABLE t_phase3_b (id INTEGER, name VARCHAR(50))")
|
|
cur.execute("INSERT INTO t_phase3_b VALUES (1, 'alpha')")
|
|
assert cur.rowcount == 1
|
|
cur.execute("INSERT INTO t_phase3_b VALUES (2, 'beta')")
|
|
assert cur.rowcount == 1
|
|
cur.execute("INSERT INTO t_phase3_b VALUES (3, 'gamma')")
|
|
assert cur.rowcount == 1
|
|
|
|
cur.execute("SELECT id, name FROM t_phase3_b ORDER BY id")
|
|
rows = cur.fetchall()
|
|
assert rows == [(1, "alpha"), (2, "beta"), (3, "gamma")]
|
|
|
|
|
|
def test_update_with_where(conn_params: ConnParams) -> None:
|
|
"""UPDATE with WHERE clause changes rowcount-many rows."""
|
|
with _connect(conn_params) as conn:
|
|
cur = conn.cursor()
|
|
cur.execute("CREATE TEMP TABLE t_phase3_c (id INTEGER, name VARCHAR(50))")
|
|
cur.execute("INSERT INTO t_phase3_c VALUES (1, 'old')")
|
|
cur.execute("INSERT INTO t_phase3_c VALUES (2, 'old')")
|
|
cur.execute("INSERT INTO t_phase3_c VALUES (3, 'old')")
|
|
|
|
cur.execute("UPDATE t_phase3_c SET name = 'new' WHERE id = 2")
|
|
# rowcount semantics: at least the affected row count
|
|
# (Phase 3.x will refine — for now we check the SELECT)
|
|
|
|
cur.execute("SELECT id, name FROM t_phase3_c ORDER BY id")
|
|
rows = cur.fetchall()
|
|
assert rows == [(1, "old"), (2, "new"), (3, "old")]
|
|
|
|
|
|
def test_delete_with_where(conn_params: ConnParams) -> None:
|
|
"""DELETE with WHERE removes the matched row."""
|
|
with _connect(conn_params) as conn:
|
|
cur = conn.cursor()
|
|
cur.execute("CREATE TEMP TABLE t_phase3_d (id INTEGER, name VARCHAR(50))")
|
|
cur.execute("INSERT INTO t_phase3_d VALUES (1, 'keep')")
|
|
cur.execute("INSERT INTO t_phase3_d VALUES (2, 'delete')")
|
|
cur.execute("INSERT INTO t_phase3_d VALUES (3, 'keep')")
|
|
|
|
cur.execute("DELETE FROM t_phase3_d WHERE id = 2")
|
|
|
|
cur.execute("SELECT id, name FROM t_phase3_d ORDER BY id")
|
|
rows = cur.fetchall()
|
|
assert rows == [(1, "keep"), (3, "keep")]
|
|
|
|
|
|
def test_full_dml_cycle_in_one_connection(conn_params: ConnParams) -> None:
|
|
"""Mix DDL, multiple DML, and SELECT on one connection — proves session state."""
|
|
with _connect(conn_params) as conn:
|
|
cur = conn.cursor()
|
|
cur.execute("CREATE TEMP TABLE t_phase3_e (id INTEGER, val INTEGER)")
|
|
for i in range(1, 6):
|
|
cur.execute(f"INSERT INTO t_phase3_e VALUES ({i}, {i * 10})")
|
|
|
|
cur.execute("SELECT val FROM t_phase3_e WHERE id > 2 ORDER BY id")
|
|
rows = cur.fetchall()
|
|
assert rows == [(30,), (40,), (50,)]
|
|
|
|
cur.execute("DELETE FROM t_phase3_e WHERE id <= 2")
|
|
cur.execute("SELECT id FROM t_phase3_e ORDER BY id")
|
|
rows = cur.fetchall()
|
|
assert rows == [(3,), (4,), (5,)]
|
|
|
|
|
|
def test_commit_rollback_in_unlogged_db_is_noop(conn_params: ConnParams) -> None:
|
|
"""commit() and rollback() are no-ops in sysmaster (unlogged).
|
|
|
|
Phase 7 made these driver-side smart: the connection tracks whether
|
|
a transaction is open. On unlogged DBs, ``_ensure_transaction`` is
|
|
cached as "unsupported" after the first -201 from SQ_BEGIN, so
|
|
subsequent commit/rollback calls skip the wire send entirely.
|
|
Real transaction semantics are tested in test_transactions.py
|
|
against the logged ``testdb`` database.
|
|
"""
|
|
with _connect(conn_params) as conn:
|
|
# No transaction is open; both should silently no-op.
|
|
conn.commit()
|
|
conn.rollback()
|