Ryan Malloy
d04000dfc3
Before:
ProgrammingError: server returned SQ_ERR sqlcode=-201 isamcode=0
After:
ProgrammingError: [-201] A syntax error has occurred (offset 1)
ProgrammingError: [-206] The specified table is not in the database
(near 'no_such_table') [ISAM -111] (offset 27)
IntegrityError: [-268] Cannot insert duplicate value -
violates UNIQUE constraint (near 'on table u')
[ISAM -100] (offset 23)
IntegrityError: [-391] Cannot insert NULL value into a NOT NULL column
(near 't.id') (offset 23)
OperationalError: [-255] Not in transaction
PEP 249 exception classes mapped from sqlcode:
-201, -206, -217, -286, -310, ... → ProgrammingError
-239, -268, -291, -292, -391, -703 → IntegrityError
-255, -256, -407, -440, -908, ... → OperationalError
-329, -349, -510 → NotSupportedError
others → DatabaseError (safe fallback)
SQ_ERR wire decode (per IfxSqli.receiveError line 2717):
[short sqlcode][short isamcode][int offset]
[short near_token_len][bytes name][optional pad][short SQ_EOT]
The "near" token is the object name where the error occurred (table or
column name for "not found" errors); empty for most syntax errors.
Structured fields attached to every Informix error for programmatic
inspection:
e.sqlcode — Informix error code (e.g. -206)
e.isamcode — ISAM/RSAM-level error (e.g. -111 = "table not found")
e.offset — character offset in the SQL where the error occurred
e.near — object name in the "near 'XYZ'" clause (or "")
Connection state survives errors: a failed query doesn't poison the
session — subsequent execute() calls work normally. Verified by
test_connection_survives_query_error.
Built-in error catalog of ~50 most common Informix sqlcodes shipped
in src/informix_db/_errcodes.py. Users can extend at runtime with
register_error_text(code, text). Unknown codes get a generic
"Informix error <N>" with structured fields still populated.
Module changes:
src/informix_db/_errcodes.py (new) — error catalog + exception
classification + register_error_text()
src/informix_db/cursors.py — _raise_sq_err now uses the catalog
src/informix_db/connections.py — same upgrade for the connection-side
SQ_ERR path (catches commit/rollback errors etc.)
Tests: 40 unit + 63 integration (8 new error tests) = 103 total, all
green, ruff clean. Tests cover:
- syntax error → ProgrammingError(-201)
- table not found → ProgrammingError(-206) with near='no_such_table'
- column not found → ProgrammingError(-217)
- UNIQUE violation → IntegrityError(-268)
- NOT NULL violation → IntegrityError(-391)
- commit on unlogged DB → OperationalError(-255)
- connection survives errors (subsequent queries work)
- all errors carry structured sqlcode/isamcode/offset/near attrs
102 lines
3.7 KiB
Python
102 lines
3.7 KiB
Python
"""Phase 6.a integration tests — DECIMAL/MONEY row decoding.
|
|
|
|
Decode-only for now. Encoding (Decimal as a parameter) is Phase 6.x —
|
|
the encoder is implemented in ``converters.py`` but the server rejects
|
|
the bytes (precision packing not quite right). Workaround: cast to
|
|
float at the call site or pass via SQL literal.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from decimal import Decimal
|
|
|
|
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_count_returns_decimal(conn_params: ConnParams) -> None:
|
|
"""COUNT(*) returns DECIMAL — must decode to a Python int-like Decimal."""
|
|
with _connect(conn_params) as conn:
|
|
cur = conn.cursor()
|
|
cur.execute("SELECT COUNT(*) FROM systables")
|
|
(n,) = cur.fetchone()
|
|
assert isinstance(n, Decimal)
|
|
assert n > 0 # systables has at least some rows
|
|
|
|
|
|
def test_sum_returns_decimal(conn_params: ConnParams) -> None:
|
|
"""SUM returns DECIMAL — verify exact integer arithmetic."""
|
|
with _connect(conn_params) as conn:
|
|
cur = conn.cursor()
|
|
cur.execute("SELECT SUM(tabid) FROM systables WHERE tabid <= 10")
|
|
# 1+2+...+10 = 55
|
|
assert cur.fetchone() == (Decimal("55"),)
|
|
|
|
|
|
def test_avg_returns_decimal_with_fraction(conn_params: ConnParams) -> None:
|
|
"""AVG returns DECIMAL with fractional part."""
|
|
with _connect(conn_params) as conn:
|
|
cur = conn.cursor()
|
|
cur.execute("SELECT AVG(tabid) FROM systables WHERE tabid <= 10")
|
|
# avg(1..10) = 5.5
|
|
assert cur.fetchone() == (Decimal("5.5"),)
|
|
|
|
|
|
def test_decimal_literal_positive(conn_params: ConnParams) -> None:
|
|
"""Literal DECIMAL value with explicit precision/scale."""
|
|
with _connect(conn_params) as conn:
|
|
cur = conn.cursor()
|
|
cur.execute("SELECT 1234.56::DECIMAL(10,2) FROM systables WHERE tabid = 1")
|
|
assert cur.fetchone() == (Decimal("1234.56"),)
|
|
|
|
|
|
def test_decimal_literal_negative(conn_params: ConnParams) -> None:
|
|
"""Negative DECIMAL — exercises the asymmetric base-100 complement decode."""
|
|
with _connect(conn_params) as conn:
|
|
cur = conn.cursor()
|
|
cur.execute("SELECT -1234.56::DECIMAL(10,2) FROM systables WHERE tabid = 1")
|
|
assert cur.fetchone() == (Decimal("-1234.56"),)
|
|
|
|
|
|
def test_decimal_small_fraction(conn_params: ConnParams) -> None:
|
|
with _connect(conn_params) as conn:
|
|
cur = conn.cursor()
|
|
cur.execute("SELECT 0.5::DECIMAL(10,2), -0.5::DECIMAL(10,2) FROM systables WHERE tabid = 1")
|
|
assert cur.fetchone() == (Decimal("0.5"), Decimal("-0.5"))
|
|
|
|
|
|
def test_decimal_null(conn_params: ConnParams) -> None:
|
|
"""NULL DECIMAL decodes as Python None."""
|
|
with _connect(conn_params) as conn:
|
|
cur = conn.cursor()
|
|
cur.execute("CREATE TEMP TABLE td (n DECIMAL(10,2))")
|
|
cur.execute("INSERT INTO td VALUES (NULL)")
|
|
cur.execute("SELECT n FROM td")
|
|
assert cur.fetchone() == (None,)
|
|
|
|
|
|
def test_decimal_param_binding_raises_for_now(conn_params: ConnParams) -> None:
|
|
"""Decimal as a bind parameter is Phase 6.x — currently raises."""
|
|
with _connect(conn_params) as conn:
|
|
cur = conn.cursor()
|
|
cur.execute("CREATE TEMP TABLE td2 (n DECIMAL(10,2))")
|
|
with pytest.raises(NotImplementedError, match="Decimal"):
|
|
cur.execute("INSERT INTO td2 VALUES (?)", (Decimal("3.14"),))
|