informix-db/tests/test_interval.py

"""Phase 6.d integration tests — INTERVAL decoding for both qualifier families.

INTERVAL splits into two families that decode to different Python types:
  - DAY-FRACTION (start TU >= DAY=4)  → ``datetime.timedelta``
  - YEAR-MONTH   (start TU <= MONTH=2) → :class:`informix_db.IntervalYM`

The wire format is identical to DECIMAL/DATETIME (BCD with sign+exp head
byte); the qualifier short tells us how to interpret the digits as
time fields.
"""

from __future__ import annotations

import datetime

import pytest

import informix_db
from informix_db import IntervalYM
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_interval_day_to_second(conn_params: ConnParams) -> None:
    """INTERVAL DAY TO SECOND: 3 days, 4 hours, 5 minutes, 6 seconds."""
    with _connect(conn_params) as conn:
        cur = conn.cursor()
        cur.execute(
            "SELECT INTERVAL(3 04:05:06) DAY TO SECOND "
            "FROM systables WHERE tabid = 1"
        )
        (val,) = cur.fetchone()
        assert val == datetime.timedelta(
            days=3, hours=4, minutes=5, seconds=6
        )


def test_interval_hour_to_second(conn_params: ConnParams) -> None:
    """INTERVAL HOUR TO SECOND: 12:34:56."""
    with _connect(conn_params) as conn:
        cur = conn.cursor()
        cur.execute(
            "SELECT INTERVAL(12:34:56) HOUR TO SECOND "
            "FROM systables WHERE tabid = 1"
        )
        (val,) = cur.fetchone()
        assert val == datetime.timedelta(hours=12, minutes=34, seconds=56)


def test_interval_minute_to_second(conn_params: ConnParams) -> None:
    """INTERVAL MINUTE TO SECOND: 23:45."""
    with _connect(conn_params) as conn:
        cur = conn.cursor()
        cur.execute(
            "SELECT INTERVAL(23:45) MINUTE TO SECOND "
            "FROM systables WHERE tabid = 1"
        )
        (val,) = cur.fetchone()
        assert val == datetime.timedelta(minutes=23, seconds=45)


def test_interval_year_to_month(conn_params: ConnParams) -> None:
    """INTERVAL YEAR TO MONTH: 5 years, 3 months."""
    with _connect(conn_params) as conn:
        cur = conn.cursor()
        cur.execute(
            "SELECT INTERVAL(5-3) YEAR TO MONTH "
            "FROM systables WHERE tabid = 1"
        )
        (val,) = cur.fetchone()
        assert val == IntervalYM(months=5 * 12 + 3)


def test_interval_year_only(conn_params: ConnParams) -> None:
    """INTERVAL YEAR — exercises start==end, year-only qualifier."""
    with _connect(conn_params) as conn:
        cur = conn.cursor()
        cur.execute(
            "SELECT INTERVAL(2026) YEAR(4) TO YEAR "
            "FROM systables WHERE tabid = 1"
        )
        (val,) = cur.fetchone()
        assert val == IntervalYM(months=2026 * 12)


def test_interval_negative(conn_params: ConnParams) -> None:
    """Negative INTERVAL — exercises 9's-complement decode for intervals."""
    with _connect(conn_params) as conn:
        cur = conn.cursor()
        cur.execute(
            "SELECT INTERVAL(-3 04:05:06) DAY TO SECOND "
            "FROM systables WHERE tabid = 1"
        )
        (val,) = cur.fetchone()
        # Note: Informix encodes the whole interval as negative — all
        # fields share the sign.
        assert val == datetime.timedelta(
            days=-3, hours=-4, minutes=-5, seconds=-6
        )


def test_interval_in_table_column(conn_params: ConnParams) -> None:
    """INTERVAL stored in a table column round-trips correctly."""
    with _connect(conn_params) as conn:
        cur = conn.cursor()
        cur.execute(
            "CREATE TEMP TABLE t_iv "
            "(id INTEGER, "
            " span INTERVAL DAY(2) TO SECOND, "
            " age INTERVAL YEAR(4) TO MONTH)"
        )
        cur.execute(
            "INSERT INTO t_iv VALUES "
            "(1, INTERVAL(7 12:00:00) DAY TO SECOND, "
            "INTERVAL(2-6) YEAR TO MONTH)"
        )
        cur.execute("SELECT span, age FROM t_iv")
        span, age = cur.fetchone()
        assert span == datetime.timedelta(days=7, hours=12)
        assert age == IntervalYM(months=2 * 12 + 6)


def test_interval_null(conn_params: ConnParams) -> None:
    """NULL INTERVAL decodes to Python None."""
    with _connect(conn_params) as conn:
        cur = conn.cursor()
        cur.execute("CREATE TEMP TABLE t_iv2 (span INTERVAL DAY TO SECOND)")
        cur.execute("INSERT INTO t_iv2 VALUES (NULL)")
        cur.execute("SELECT span FROM t_iv2")
        assert cur.fetchone() == (None,)


def test_interval_multiple_columns(conn_params: ConnParams) -> None:
    """Multiple INTERVAL qualifiers in one row — proves per-column slicing."""
    with _connect(conn_params) as conn:
        cur = conn.cursor()
        cur.execute(
            "SELECT "
            "  INTERVAL(3 04:05:06) DAY TO SECOND, "
            "  INTERVAL(1-6) YEAR TO MONTH, "
            "  INTERVAL(99:30) HOUR(2) TO MINUTE "
            "FROM systables WHERE tabid = 1"
        )
        df, ym, hm = cur.fetchone()
        assert df == datetime.timedelta(days=3, hours=4, minutes=5, seconds=6)
        assert ym == IntervalYM(months=18)
        assert hm == datetime.timedelta(hours=99, minutes=30)