"""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_round_trip(conn_params: ConnParams) -> None: """Decimal as a bind parameter round-trips through INSERT + SELECT.""" with _connect(conn_params) as conn: cur = conn.cursor() cur.execute("CREATE TEMP TABLE td2 (id INTEGER, n DECIMAL(12,2))") for i, d in enumerate([Decimal("1234.56"), Decimal("-99.99"), Decimal("0.5")]): cur.execute("INSERT INTO td2 VALUES (?, ?)", (i, d)) cur.execute("SELECT n FROM td2 ORDER BY id") rows = cur.fetchall() assert rows == [(Decimal("1234.56"),), (Decimal("-99.99"),), (Decimal("0.5"),)]