cyclone/backend/tests/test_migration_0020.py

"""Tests for migration 0020_add_batch_txn_set_control_number.sql.

SP37 adds a nullable ``batches.transaction_set_control_number`` column
populated from the parsed 837's ST02 (transaction set control number)
on every write. The 999 ack join (Pass 1) needs to resolve by ST02,
not ISA13, so this column is the join key. This migration is purely
additive: nullable, no default, backfills from
``raw_result_json.envelope.transaction_set_control_number`` where the
source JSON already carries it.

For the backfill-shape tests we point ``db_migrate.MIGRATIONS_DIR``
at the real migrations directory, apply all migrations once to bring
the fresh DB up to v20, insert representative rows, then replay the
exact UPDATE statement the migration uses. Replaying the UPDATE
proves the SQL works as intended even though the migration itself
already ran over an empty table.

The backfill UPDATE is loaded directly from the migration file (not
a hand-maintained copy) so the test cannot drift from production —
see :func:`_load_migration_0020_backfill_sql` and the regression
locks in ``test_migration_0020_no_drift.py``.
"""

from __future__ import annotations

import json
from pathlib import Path

import pytest
import sqlalchemy as sa

from cyclone import db_migrate


def _migration_0020_path() -> Path:
    return (
        Path(__file__).parent.parent / "src" / "cyclone" / "migrations"
        / "0020_add_batch_txn_set_control_number.sql"
    )


def _extract_update_statements(sql: str) -> list[str]:
    """Split a migration into statements; return the UPDATE ones.

    Mirrors db_migrate.run()'s splitter (strip ``--`` comments,
    split on ``;``) so the test extraction can never disagree with
    what the runner actually executes.
    """
    lines = [
        line for line in sql.splitlines()
        if not line.strip().startswith("--")
    ]
    cleaned = "\n".join(lines)
    return [
        stmt.strip() for stmt in cleaned.split(";")
        if stmt.strip()
    ]


def _load_migration_0020_backfill_sql() -> str:
    """Return migration 0020's UPDATE statement (the backfill).

    Reads the migration file at test time and extracts its UPDATE.
    Test code that needs to replay the backfill against rows that
    didn't exist when init_db ran uses this helper — guarantees the
    replayed SQL is byte-identical to what production will run.
    """
    sql = _migration_0020_path().read_text()
    updates = [
        s for s in _extract_update_statements(sql)
        if s.upper().startswith("UPDATE ")
    ]
    assert len(updates) == 1, (
        f"expected exactly 1 UPDATE in migration 0020, found {len(updates)}: {updates}"
    )
    return updates[0]


def _fresh_engine(path: Path) -> sa.Engine:
    return sa.create_engine(f"sqlite:///{path}", future=True)


def _table_info(engine: sa.Engine, table: str) -> list[tuple]:
    """Return PRAGMA table_info rows for ``table`` as plain tuples."""
    with engine.connect() as conn:
        return list(conn.exec_driver_sql(f"PRAGMA table_info({table});").tuples())


def _real_migrations_dir() -> Path:
    return Path(__file__).parent.parent / "src" / "cyclone" / "migrations"


@pytest.fixture
def migrated_engine(tmp_path: Path, monkeypatch: pytest.MonkeyPatch):
    """Yield an engine at v20 against which every real migration has run.

    Points ``db_migrate.MIGRATIONS_DIR`` at the real migrations
    directory so the test exercises the actual 0020 file, then runs
    the migration runner on a per-test fresh DB.
    """
    monkeypatch.setattr(db_migrate, "MIGRATIONS_DIR", _real_migrations_dir())
    engine = _fresh_engine(tmp_path / "mig0020.db")
    db_migrate.run(engine)

    # Confirm head is 21 (every migration applied, including SP39's 0021).
    with engine.connect() as conn:
        v = conn.exec_driver_sql("PRAGMA user_version").scalar() or 0
    assert v == 21, f"expected migration head=21, got {v}"

    yield engine
    engine.dispose()


def test_migration_0020_creates_column(migrated_engine) -> None:
    """Migration 0020 adds ``transaction_set_control_number`` to ``batches``."""
    cols = _table_info(migrated_engine, "batches")
    col_names = {row[1] for row in cols}
    assert "transaction_set_control_number" in col_names, (
        "batches.transaction_set_control_number missing — migration 0020 did not run. "
        f"Existing columns: {sorted(col_names)}"
    )


def test_migration_0020_column_is_nullable(migrated_engine) -> None:
    """The new column is nullable (no DEFAULT, no NOT NULL) so existing
    batches that don't yet carry the ST02 stay valid."""
    cols = {row[1]: row for row in _table_info(migrated_engine, "batches")}
    row = cols["transaction_set_control_number"]
    # PRAGMA table_info tuples: (cid, name, type, notnull, dflt_value, pk)
    assert row[3] == 0, f"notnull flag must be 0 (nullable), got {row[3]}"
    assert row[4] is None, f"dflt_value must be NULL, got {row[4]!r}"
    assert "TEXT" in (row[2] or "").upper(), f"expected TEXT column, got {row[2]!r}"


def test_migration_0020_backfills_when_key_present(
    migrated_engine: sa.Engine,
) -> None:
    """Replay the migration's backfill UPDATE against a row whose
    ``raw_result_json`` carries the key — must populate the new column.

    Replay (rather than waiting for ``db_migrate.run()`` to do it) is
    the only way to test the SQL against a row that didn't exist when
    init_db() ran; the migration's UPDATE naturally runs only over
    pre-existing rows.
    """
    st02_value = "ST0001"
    raw_json = {"envelope": {"transaction_set_control_number": st02_value}}

    with migrated_engine.begin() as conn:
        conn.exec_driver_sql(
            "INSERT INTO batches (id, kind, input_filename, parsed_at, raw_result_json) "
            "VALUES ('B-ST02-1', '837p', 'mig0020-st02.txt', '2026-07-07 00:00:00', ?)",
            (json.dumps(raw_json),),
        )
        conn.exec_driver_sql(_load_migration_0020_backfill_sql())

    with migrated_engine.connect() as conn:
        row = conn.exec_driver_sql(
            "SELECT transaction_set_control_number FROM batches WHERE id='B-ST02-1'"
        ).first()
    assert row is not None
    assert row[0] == st02_value, (
        f"backfill failed: expected {st02_value!r}, got {row[0]!r}"
    )


def test_migration_0020_backfill_conditional_on_key_present(
    migrated_engine: sa.Engine,
) -> None:
    """Replay the UPDATE against a row whose ``raw_result_json`` does NOT
    carry the key — must stay NULL. Proves the UPDATE is conditional
    (the ``json_extract(...) IS NOT NULL`` guard) rather than
    unconditionally overwriting with NULL."""
    raw_json = {"envelope": {"control_number": "ISA0001"}}  # no txn-set key
    with migrated_engine.begin() as conn:
        conn.exec_driver_sql(
            "INSERT INTO batches (id, kind, input_filename, parsed_at, raw_result_json) "
            "VALUES ('B-NOST-1', '837p', 'mig0020-nost.txt', '2026-07-07 00:00:00', ?)",
            (json.dumps(raw_json),),
        )
        conn.exec_driver_sql(_load_migration_0020_backfill_sql())

    with migrated_engine.connect() as conn:
        row = conn.exec_driver_sql(
            "SELECT transaction_set_control_number FROM batches WHERE id='B-NOST-1'"
        ).first()
    assert row is not None
    assert row[0] is None, (
        f"backfill must not overwrite when key is absent; got {row[0]!r}"
    )


def test_migration_0020_backfill_handles_null_raw_result_json(
    migrated_engine: sa.Engine,
) -> None:
    """A Batch row with ``raw_result_json IS NULL`` (the prior SP's
    unparsed state) must not crash the backfill and must stay NULL."""
    with migrated_engine.begin() as conn:
        conn.exec_driver_sql(
            "INSERT INTO batches (id, kind, input_filename, parsed_at) "
            "VALUES ('B-NULL-1', '837p', 'mig0020-null.txt', '2026-07-07 00:00:00')"
        )
        conn.exec_driver_sql(_load_migration_0020_backfill_sql())

    with migrated_engine.connect() as conn:
        row = conn.exec_driver_sql(
            "SELECT transaction_set_control_number FROM batches WHERE id='B-NULL-1'"
        ).first()
    assert row is not None
    assert row[0] is None