"""Phase 6.b integration tests — DATETIME decoding for all qualifier ranges. DATETIME is BCD-packed with a qualifier embedded in encoded_length: ``(digit_count << 8) | (start_TU << 4) | end_TU`` where TU codes are YEAR=0, MONTH=2, DAY=4, HOUR=6, MIN=8, SEC=10, FRAC1=11..FRAC5=15. The decoder picks the appropriate Python type: - datetime.date for qualifiers ending at YEAR/MONTH/DAY - datetime.time for qualifiers starting at HOUR or later - datetime.datetime for spans crossing the date/time boundary """ from __future__ import annotations import datetime 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_datetime_year_to_second(conn_params: ConnParams) -> None: """Full date+time → datetime.datetime.""" with _connect(conn_params) as conn: cur = conn.cursor() cur.execute( "SELECT DATETIME(2026-05-04 12:34:56) YEAR TO SECOND " "FROM systables WHERE tabid = 1" ) assert cur.fetchone() == (datetime.datetime(2026, 5, 4, 12, 34, 56),) def test_datetime_year_to_day(conn_params: ConnParams) -> None: """Date-only qualifier → datetime.date.""" with _connect(conn_params) as conn: cur = conn.cursor() cur.execute( "SELECT DATETIME(2026-05-04) YEAR TO DAY " "FROM systables WHERE tabid = 1" ) assert cur.fetchone() == (datetime.date(2026, 5, 4),) def test_datetime_hour_to_second(conn_params: ConnParams) -> None: """Time-only qualifier → datetime.time.""" with _connect(conn_params) as conn: cur = conn.cursor() cur.execute( "SELECT DATETIME(12:34:56) HOUR TO SECOND " "FROM systables WHERE tabid = 1" ) assert cur.fetchone() == (datetime.time(12, 34, 56),) def test_datetime_current_with_fraction(conn_params: ConnParams) -> None: """CURRENT YEAR TO FRACTION returns a datetime.datetime.""" with _connect(conn_params) as conn: cur = conn.cursor() cur.execute( "SELECT CURRENT YEAR TO FRACTION(3) FROM systables WHERE tabid = 1" ) (val,) = cur.fetchone() assert isinstance(val, datetime.datetime) # Sanity: should be within a reasonable timeframe of "now" now = datetime.datetime.now() assert abs((val - now).total_seconds()) < 86400 # within a day def test_datetime_multiple_columns_in_one_row(conn_params: ConnParams) -> None: """Multiple DATETIME qualifiers in one row — proves per-column slicing.""" with _connect(conn_params) as conn: cur = conn.cursor() cur.execute( "SELECT " " DATETIME(2026-05-04 12:34:56) YEAR TO SECOND, " " DATETIME(2026-05-04) YEAR TO DAY, " " DATETIME(12:34:56) HOUR TO SECOND " "FROM systables WHERE tabid = 1" ) ts, d, t = cur.fetchone() assert ts == datetime.datetime(2026, 5, 4, 12, 34, 56) assert d == datetime.date(2026, 5, 4) assert t == datetime.time(12, 34, 56) def test_datetime_column_in_table(conn_params: ConnParams) -> None: """DATETIME stored in a table column round-trips correctly.""" with _connect(conn_params) as conn: cur = conn.cursor() cur.execute( "CREATE TEMP TABLE t_dt (id INTEGER, ts DATETIME YEAR TO SECOND)" ) cur.execute( "INSERT INTO t_dt VALUES (1, DATETIME(2026-01-15 09:30:00) YEAR TO SECOND)" ) cur.execute("SELECT ts FROM t_dt") assert cur.fetchone() == (datetime.datetime(2026, 1, 15, 9, 30, 0),) def test_datetime_null(conn_params: ConnParams) -> None: """NULL DATETIME decodes to Python None.""" with _connect(conn_params) as conn: cur = conn.cursor() cur.execute("CREATE TEMP TABLE t_dt2 (ts DATETIME YEAR TO SECOND)") cur.execute("INSERT INTO t_dt2 VALUES (NULL)") cur.execute("SELECT ts FROM t_dt2") assert cur.fetchone() == (None,)