diff --git a/scripts/check_edi_event_time_mismatch.py b/scripts/check_edi_event_time_mismatch.py new file mode 100644 index 000000000..88e6fcaab --- /dev/null +++ b/scripts/check_edi_event_time_mismatch.py @@ -0,0 +1,259 @@ +#!/usr/bin/env python3 +"""Count EDI events whose stored timestamp and EDI event date disagree. + +Read-only data check for the EDI two-clocks bug (regengine-ds, sweep P2 5). +An event disagrees when ``fsma.cte_events.event_timestamp`` is more than one +day away from its ``ship_date`` / ``event_date`` KDE. Only rows whose +``source`` starts with ``edi`` are counted (EDI ingest paths). + +Safety: +* The query runs inside ``SET TRANSACTION READ ONLY`` and is rolled back. +* Fail-closed host allow-list: only localhost / loopback / local compose + hosts run by default. Pass ``--allow-host HOST`` for a remote scratch DB. + ``REGENGINE_ENV`` / ``ENVIRONMENT`` / ``ENV`` naming production always + refuses. Never prints the database URL. +* Bare ``postgresql://`` URLs are rewritten to ``postgresql+psycopg://`` + (psycopg 3; psycopg2 is not installed). +* Tenant scope: pass ``--tenant-id UUID`` (sets ``app.tenant_id``) or + ``--all-tenants`` (requires a role that bypasses RLS). Ordinary + ``DATABASE_URL`` (``regengine`` role) cannot scan all tenants. + +Usage: python scripts/check_edi_event_time_mismatch.py --database-url URL \\ + --tenant-id UUID + (or set DATABASE_URL). Prints one integer on stdout: the disagreeing + count. Calendar-invalid date KDEs are skipped and reported on stderr. +""" + +from __future__ import annotations + +import argparse +import os +import sys +from datetime import date, datetime, timedelta +from pathlib import Path +from typing import Iterable, Optional, Sequence +from urllib.parse import parse_qs, urlsplit + +# ``shared.*`` lives under services/ +_ROOT = Path(__file__).resolve().parents[1] +for _p in (str(_ROOT / "services"), str(_ROOT)): + if _p not in sys.path: + sys.path.insert(0, _p) + +from shared.db_url import sync_postgres_url # noqa: E402 + +TOLERANCE = timedelta(days=1) +DATE_KDE_KEYS = ("ship_date", "event_date") + +# Affirmative allow-list of hosts this diagnostic may touch by default. +LOOPBACK_HOSTS = frozenset({"localhost", "127.0.0.1", "::1"}) +DEV_COMPOSE_HOSTS = frozenset({"postgres", "host.docker.internal"}) +LOCAL_HOSTS = LOOPBACK_HOSTS | DEV_COMPOSE_HOSTS + +_PROD_ENV_VALUES = frozenset({"prod", "production"}) + +# Scalar text of a JSONB kde_value (fsma.cte_kdes.kde_value is JSONB since v078). +_KDE_TEXT = "(k.kde_value #>> '{}')" + +# Calendar-valid ISO date without calling to_date / ::date on unchecked input. +# PostgreSQL's to_date raises on out-of-range fields (e.g. 2026-02-30), so a +# round-trip filter cannot count those rows. CASE short-circuits; only a +# predicate-validated string is cast. Leap years: divisible by 4, except +# centuries not divisible by 400. Year must be >= 1 (year 0000 is out of range). +_CALENDAR_VALID_ISO_DATE = ( + "(CASE WHEN (k.kde_value #>> '{}') ~ '^\\d{4}-\\d{2}-\\d{2}$' THEN (" + "CASE WHEN SUBSTRING((k.kde_value #>> '{}') FROM 1 FOR 4)::int >= 1 " + "AND SUBSTRING((k.kde_value #>> '{}') FROM 6 FOR 2)::int BETWEEN 1 AND 12 " + "AND SUBSTRING((k.kde_value #>> '{}') FROM 9 FOR 2)::int BETWEEN 1 AND (" + "CASE SUBSTRING((k.kde_value #>> '{}') FROM 6 FOR 2) " + "WHEN '01' THEN 31 WHEN '03' THEN 31 WHEN '05' THEN 31 WHEN '07' THEN 31 " + "WHEN '08' THEN 31 WHEN '10' THEN 31 WHEN '12' THEN 31 " + "WHEN '04' THEN 30 WHEN '06' THEN 30 WHEN '09' THEN 30 WHEN '11' THEN 30 " + "ELSE CASE WHEN ((SUBSTRING((k.kde_value #>> '{}') FROM 1 FOR 4)::int % 4 = 0 " + "AND SUBSTRING((k.kde_value #>> '{}') FROM 1 FOR 4)::int % 100 <> 0) " + "OR SUBSTRING((k.kde_value #>> '{}') FROM 1 FOR 4)::int % 400 = 0) " + "THEN 29 ELSE 28 END END) " + "THEN true ELSE false END) ELSE false END)" +) + +# Safe cast: evaluated only when the calendar predicate holds (CASE). +_SAFE_DATE = ( + "(CASE WHEN (k.kde_value #>> '{}') ~ '^\\d{4}-\\d{2}-\\d{2}$' THEN (" + "CASE WHEN SUBSTRING((k.kde_value #>> '{}') FROM 1 FOR 4)::int >= 1 " + "AND SUBSTRING((k.kde_value #>> '{}') FROM 6 FOR 2)::int BETWEEN 1 AND 12 " + "AND SUBSTRING((k.kde_value #>> '{}') FROM 9 FOR 2)::int BETWEEN 1 AND (" + "CASE SUBSTRING((k.kde_value #>> '{}') FROM 6 FOR 2) " + "WHEN '01' THEN 31 WHEN '03' THEN 31 WHEN '05' THEN 31 WHEN '07' THEN 31 " + "WHEN '08' THEN 31 WHEN '10' THEN 31 WHEN '12' THEN 31 " + "WHEN '04' THEN 30 WHEN '06' THEN 30 WHEN '09' THEN 30 WHEN '11' THEN 30 " + "ELSE CASE WHEN ((SUBSTRING((k.kde_value #>> '{}') FROM 1 FOR 4)::int % 4 = 0 " + "AND SUBSTRING((k.kde_value #>> '{}') FROM 1 FOR 4)::int % 100 <> 0) " + "OR SUBSTRING((k.kde_value #>> '{}') FROM 1 FOR 4)::int % 400 = 0) " + "THEN 29 ELSE 28 END END) " + "THEN ((k.kde_value #>> '{}')::date) ELSE NULL END) ELSE NULL END)" +) + +MISMATCH_SQL = f""" +SELECT COUNT(DISTINCT e.id) + FROM fsma.cte_events e + JOIN fsma.cte_kdes k + ON k.cte_event_id = e.id + AND k.tenant_id = e.tenant_id + AND k.kde_key IN ('ship_date', 'event_date') + WHERE e.source LIKE 'edi%' + AND {_SAFE_DATE} IS NOT NULL + AND (e.event_timestamp < ({_SAFE_DATE}::timestamp AT TIME ZONE 'UTC') - INTERVAL '1 day' + OR e.event_timestamp > ({_SAFE_DATE}::timestamp AT TIME ZONE 'UTC') + INTERVAL '2 days') +""" +# #>> '{}' extracts the JSONB scalar as text before matching, as +# compliance_score.py does. Calendar-invalid ISO-looking strings (e.g. +# 2026-02-30) are excluded via the non-throwing predicate above and counted +# separately by INVALID_DATE_SQL — neither query raises on them. +# e.source LIKE 'edi%' keeps this EDI-only (edi_inbound, edi_856_inbound, …). +# COUNT(DISTINCT e.id) counts each event once even when it carries both a +# ship_date and an event_date KDE. + +INVALID_DATE_SQL = f""" +SELECT COUNT(*) + FROM fsma.cte_kdes k + JOIN fsma.cte_events e + ON e.id = k.cte_event_id + AND e.tenant_id = k.tenant_id + WHERE e.source LIKE 'edi%' + AND k.kde_key IN ('ship_date', 'event_date') + AND (CASE WHEN (k.kde_value #>> '{{}}') ~ '^\d{{4}}-\d{{2}}-\d{{2}}$' + THEN NOT ({_CALENDAR_VALID_ISO_DATE}) + ELSE false + END) +""" + + +def is_mismatch(event_timestamp: datetime, event_date: date, tolerance: timedelta = TOLERANCE) -> bool: + """True when the timestamp is more than ``tolerance`` outside ``event_date``.""" + day_start = datetime(event_date.year, event_date.month, event_date.day, tzinfo=event_timestamp.tzinfo) + day_end = day_start + timedelta(days=1) + return event_timestamp < day_start - tolerance or event_timestamp > day_end + tolerance + + +def count_mismatches(rows: Iterable[tuple[datetime, date]], tolerance: timedelta = TOLERANCE) -> int: + return sum(1 for ts, d in rows if is_mismatch(ts, d, tolerance)) + + +def _effective_db_hosts(database_url: str) -> list[str]: + """Authority host plus any libpq ``host`` / ``hostaddr`` query values.""" + hosts: list[str] = [] + try: + parts = urlsplit(database_url) + except ValueError: + return [""] + if parts.hostname: + hosts.append(parts.hostname.lower()) + try: + query = parse_qs(parts.query) + except ValueError: + return hosts or [""] + for key in ("host", "hostaddr"): + for value in query.get(key, []): + hosts.extend(h.strip().lower() for h in value.split(",") if h.strip()) + return hosts or [""] + + +def refuse_production( + database_url: str, + env: Optional[dict] = None, + allowed_hosts: Optional[Sequence[str]] = None, +) -> Optional[str]: + """Return a refusal reason, or None when the target is allowed. + + Fail-closed allow-list: every effective host must be in LOCAL_HOSTS or in + ``allowed_hosts`` (from ``--allow-host``). Env vars naming production + always refuse, even for localhost. + """ + env = os.environ if env is None else env + for name in ("REGENGINE_ENV", "ENVIRONMENT", "ENV"): + if str(env.get(name, "")).lower() in _PROD_ENV_VALUES: + return f"{name} names production" + permit = set(LOCAL_HOSTS) + if allowed_hosts: + permit.update(h.strip().lower() for h in allowed_hosts if h and h.strip()) + hosts = _effective_db_hosts(database_url) + if not hosts or any(not h for h in hosts): + return "database URL host could not be parsed" + bad = [h for h in hosts if h not in permit] + if bad: + shown = ", ".join(bad) + return f"host(s) not on the local allow-list: {shown} (pass --allow-host for a remote scratch database)" + return None + + +def main(argv: Optional[list[str]] = None) -> int: + parser = argparse.ArgumentParser(description=__doc__.splitlines()[0]) + parser.add_argument("--database-url", default=os.environ.get("DATABASE_URL", "")) + parser.add_argument( + "--allow-host", + action="append", + default=[], + metavar="HOST", + help="Permit a non-local host (repeatable). For remote scratch DBs only.", + ) + parser.add_argument( + "--tenant-id", + default=None, + metavar="UUID", + help="Scope the count to one tenant (sets app.tenant_id for RLS).", + ) + parser.add_argument( + "--all-tenants", + action="store_true", + help="Scan every tenant. Requires a role that bypasses RLS (not the ordinary DATABASE_URL / regengine role).", + ) + args = parser.parse_args(argv) + if not args.database_url: + print("error: --database-url or DATABASE_URL required", file=sys.stderr) + return 2 + if bool(args.tenant_id) == bool(args.all_tenants): + print( + "error: pass exactly one of --tenant-id UUID or --all-tenants", + file=sys.stderr, + ) + return 2 + reason = refuse_production(args.database_url, allowed_hosts=args.allow_host) + if reason: + print(f"refusing to run: {reason}", file=sys.stderr) + return 3 + + from sqlalchemy import create_engine, text + + engine = create_engine(sync_postgres_url(args.database_url)) + with engine.connect() as conn: + trans = conn.begin() + try: + conn.execute(text("SET TRANSACTION READ ONLY")) + if args.tenant_id: + from shared.tenant_context import set_tenant_guc + + set_tenant_guc(conn, args.tenant_id) + try: + count = conn.execute(text(MISMATCH_SQL)).scalar_one() + invalid = conn.execute(text(INVALID_DATE_SQL)).scalar_one() + except Exception as exc: # noqa: BLE001 - surface RLS hint, then re-raise path + message = str(exc) + if "app.tenant_id not set" in message or "insufficient_privilege" in message: + print( + "error: row-level security blocked the query. " + "Pass --tenant-id UUID, or connect with a role that " + "bypasses RLS and pass --all-tenants.", + file=sys.stderr, + ) + return 4 + raise + finally: + trans.rollback() + if int(invalid) > 0: + print(f"invalid_date_kdes: {int(invalid)}", file=sys.stderr) + print(int(count)) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/services/ingestion/tests/test_edi_ingest_time_field.py b/services/ingestion/tests/test_edi_ingest_time_field.py new file mode 100644 index 000000000..22cf8a89d --- /dev/null +++ b/services/ingestion/tests/test_edi_ingest_time_field.py @@ -0,0 +1,257 @@ +"""Checks that EDI ingest time stays out of hashed KDEs, and the +timestamp/event_date mismatch check. Dates are built relative to now; the +only input file is the repo's synthetic fixture edi_corpus/asn_856.edi. +No database.""" + +from __future__ import annotations + +import importlib.util +import sys +import time +from datetime import date, datetime, timedelta, timezone +from pathlib import Path + +import pytest + +service_dir = Path(__file__).parent.parent +repo_root = service_dir.parent.parent +sys.path.insert(0, str(service_dir)) + +pytest.importorskip("fastapi") + +import app.edi_ingestion.routes as edi_routes # noqa: E402 +from app.edi_ingestion.extractors import _extract_fields_for_set # noqa: E402 +from app.edi_ingestion.parser import _parse_x12_segments # noqa: E402 +from app.edi_ingestion.utils import _ship_timestamp_and_date # noqa: E402 +from shared.cte_persistence.hashing import compute_idempotency_key # noqa: E402 + +_spec = importlib.util.spec_from_file_location( + "check_edi_event_time_mismatch", repo_root / "scripts" / "check_edi_event_time_mismatch.py" +) +check = importlib.util.module_from_spec(_spec) +_spec.loader.exec_module(check) + +UTC = timezone.utc + + +FIXTURE = service_dir / "tests" / "fixtures" / "edi_corpus" / "asn_856.edi" + + +def _fixture_event(): + """Build the 856 event from the repo's synthetic fixture, as /document does.""" + segments = _parse_x12_segments(FIXTURE.read_text(encoding="utf-8")) + extracted = _extract_fields_for_set("856", segments) + header = {k: v for k, v in extracted.items() if k != "line_items"} + merged = {**header, **extracted["line_items"][0], "edi_line_index": 1} + return edi_routes._build_ingest_event_for_set( + "856", merged, extracted["asn_number"], None, None, None, None, segments=segments + ) + + +def _key(event) -> str: + return compute_idempotency_key( + event.cte_type.value, + event.traceability_lot_code, + event.timestamp, + "edi_inbound", + dict(event.kdes), + location_gln=event.location_gln, + location_name=event.location_name, + ) + + +def test_same_document_same_idempotency_key(): + """A resend must hash the same, so backup dedupe works when Redis is down. + + Ingest time lives in the ``fsma.cte_events.ingested_at`` column, never in + the hashed KDEs. + """ + first = _fixture_event() + time.sleep(0.01) + second = _fixture_event() + assert "edi_ingested_at" not in first.kdes + assert first.kdes == second.kdes + assert _key(first) == _key(second) + + +def test_routes_do_not_put_ingest_time_in_kdes(): + src = (service_dir / "app" / "edi_ingestion" / "routes.py").read_text() + assert "edi_ingested_at" not in src + + +def test_backfilled_event_time_is_kept(): + real = datetime.now(UTC) - timedelta(days=200) + timestamp, event_date = _ship_timestamp_and_date(real.strftime("%Y%m%d"), None) + assert timestamp[:10] == event_date == real.date().isoformat() + + +def _day(offset_days: int) -> date: + return (datetime.now(UTC) - timedelta(days=offset_days)).date() + + +def _at(d: date, days: int = 0, hours: int = 0, minutes: int = 0, seconds: int = 0) -> datetime: + start = datetime(d.year, d.month, d.day, tzinfo=UTC) + return start + timedelta(days=days, hours=hours, minutes=minutes, seconds=seconds) + + +@pytest.mark.parametrize( + "days, hours, minutes, seconds, expected", + [ + (0, 14, 30, 0, False), # same day + (1, 23, 0, 0, False), # within a day after the day ends + (-1, 1, 0, 0, False), # within a day before the day starts + (2, 0, 0, 1, True), # just over a day after the day ends + (-1, 0, -1, 0, True), # just over a day before the day starts + (200, 9, 0, 0, True), # snapped to an ingest time months later + ], +) +def test_is_mismatch(days, hours, minutes, seconds, expected): + d = _day(200) + assert check.is_mismatch(_at(d, days, hours, minutes, seconds), d) is expected + + +def test_count_mismatches_old_vs_new_behaviour(): + day = _day(200) + ingest_now = datetime.now(UTC) + real = _at(day, hours=14, minutes=30) + assert check.count_mismatches([(ingest_now, day)] * 3) == 3 # snapped to ingest time + assert check.count_mismatches([(real, day)] * 3) == 0 # real event time kept + + +def test_sql_is_read_only_count(): + sql = check.MISMATCH_SQL.upper() + assert sql.lstrip().startswith("SELECT COUNT(DISTINCT E.ID)") + for verb in ("INSERT", "UPDATE", "DELETE", "DROP", "ALTER", "TRUNCATE"): + assert verb not in sql + + +@pytest.mark.parametrize( + "url, env, allowed_hosts, refused", + [ + ("postgresql://u@localhost:5432/regengine_test", {}, None, False), + ("postgresql://u@127.0.0.1/regengine", {}, None, False), + ("postgresql://u@postgres:5432/regengine", {}, None, False), + # Fail-closed: remote host without --allow-host, even without "prod" in URL. + ("postgresql://u@db.internal/regengine", {}, None, True), + ("postgresql://u@db.prod.example/regengine", {}, None, True), + ("postgresql://u@db.internal/regengine", {}, ["db.internal"], False), + ("postgresql://u@localhost/regengine", {"REGENGINE_ENV": "production"}, None, True), + ("postgresql://u@localhost/regengine", {"ENVIRONMENT": "Prod"}, None, True), + ], +) +def test_refuses_production(url, env, allowed_hosts, refused): + assert (check.refuse_production(url, env, allowed_hosts=allowed_hosts) is not None) is refused + + +def test_main_requires_url(monkeypatch, capsys): + monkeypatch.delenv("DATABASE_URL", raising=False) + assert check.main([]) == 2 + + +def test_main_requires_tenant_scope(monkeypatch, capsys): + monkeypatch.delenv("DATABASE_URL", raising=False) + monkeypatch.delenv("REGENGINE_ENV", raising=False) + assert check.main(["--database-url", "postgresql://u@localhost/x"]) == 2 + err = capsys.readouterr().err + assert "--tenant-id" in err and "--all-tenants" in err + + +def test_main_refuses_production_without_connecting(monkeypatch): + monkeypatch.setenv("REGENGINE_ENV", "production") + assert check.main(["--database-url", "postgresql://u@localhost/x", "--all-tenants"]) == 3 + + +def test_main_refuses_remote_host_without_allow_host(monkeypatch, capsys): + monkeypatch.delenv("REGENGINE_ENV", raising=False) + monkeypatch.delenv("ENVIRONMENT", raising=False) + monkeypatch.delenv("ENV", raising=False) + assert check.main(["--database-url", "postgresql://u@db.internal/x", "--all-tenants"]) == 3 + assert "allow-list" in capsys.readouterr().err + + +def test_sync_postgres_url_pins_psycopg3(): + assert check.sync_postgres_url("postgresql://u@localhost/x").startswith("postgresql+psycopg://") + assert check.sync_postgres_url("postgres://u@localhost/x").startswith("postgresql+psycopg://") + + +def test_main_passes_psycopg_url_to_create_engine(monkeypatch): + """Bare postgresql:// must not reach create_engine (psycopg2 is absent).""" + seen = {} + + class _Conn: + def __enter__(self): + return self + + def __exit__(self, *a): + return False + + def begin(self): + return self + + def execute(self, *a, **k): + class _R: + def scalar_one(self_inner): + return 0 + + return _R() + + def rollback(self): + return None + + class _Engine: + def connect(self): + return _Conn() + + def fake_create_engine(url, *a, **k): + seen["url"] = url + return _Engine() + + import sqlalchemy + + monkeypatch.setattr(sqlalchemy, "create_engine", fake_create_engine) + # Also patch the late import site inside main via sqlalchemy module. + monkeypatch.delenv("REGENGINE_ENV", raising=False) + rc = check.main( + [ + "--database-url", + "postgresql://u@localhost:5432/regengine_test", + "--all-tenants", + ] + ) + assert rc == 0 + assert seen["url"].startswith("postgresql+psycopg://") + assert "psycopg2" not in seen["url"] + + +def test_sql_filters_edi_source_and_rejects_invalid_dates(): + sql = check.MISMATCH_SQL + assert "e.source LIKE 'edi%'" in sql + assert "to_date" not in sql + assert "to_char" not in sql + assert "CASE WHEN" in sql + assert "edi%" in check.INVALID_DATE_SQL + assert "to_date" not in check.INVALID_DATE_SQL + assert "NOT (" in check.INVALID_DATE_SQL + + +def test_date_validity_sql_guards_and_year_range(): + """Verify CASE WHEN guards around regex, SUBSTRING::int, and ::date casts, and year >= 1.""" + assert ">= 1" in check._CALENDAR_VALID_ISO_DATE + assert ">= 1" in check._SAFE_DATE + assert check._CALENDAR_VALID_ISO_DATE.strip().startswith( + "(CASE WHEN (k.kde_value #>> '{}') ~ '^\\d{4}-\\d{2}-\\d{2}$'" + ) + assert check._SAFE_DATE.strip().startswith("(CASE WHEN (k.kde_value #>> '{}') ~ '^\\d{4}-\\d{2}-\\d{2}$'") + assert not check._CALENDAR_VALID_ISO_DATE.startswith("(k.kde_value #>> '{}') ~") + assert "CASE WHEN (k.kde_value #>> '{}') ~ '^\\d{4}-\\d{2}-\\d{2}$'" in check.INVALID_DATE_SQL + + +def test_r1_time_to_fix_reads_ingest_clocks_not_event_time(): + src = (repo_root / "services" / "shared" / "data_gap_tracking.py").read_text() + sql_blocks = [b for b in src.split('text(\n """')[1:]] + assert sql_blocks + for block in sql_blocks: + body = block.split('"""')[0].lower() + assert "event_timestamp" not in body + assert "event_date" not in body + assert "closed_at = GREATEST(now(), opened_at)" in src diff --git a/services/shared/data_gap_tracking.py b/services/shared/data_gap_tracking.py index fe585ab48..f9a66f986 100644 --- a/services/shared/data_gap_tracking.py +++ b/services/shared/data_gap_tracking.py @@ -14,6 +14,14 @@ that lot and CTE, stamping the closing record id and its canonical SHA-256 (the persisted ledger hash) as the receipt. +Clock: time to fix is measured on ingest-time clocks only. ``opened_at`` +defaults to the database ``now()`` when the gap is flagged. ``closed_at`` +is ``GREATEST(now(), opened_at)`` when the fixing record lands, so a clock +skew that puts the database behind ``opened_at`` clamps to ``opened_at`` +rather than writing a negative duration. Neither reads the event +``timestamp``/``event_date``, so backfilled EDI events (whose event time is +months old) never inflate or shrink time to fix. Keep it that way. + Gap logging is observability. It must never block or roll back ingest, so the writer catches every error, logs it, and returns. """ diff --git a/tests/migrations/test_edi_event_time_mismatch_jsonb.py b/tests/migrations/test_edi_event_time_mismatch_jsonb.py new file mode 100644 index 000000000..108d3fa5e --- /dev/null +++ b/tests/migrations/test_edi_event_time_mismatch_jsonb.py @@ -0,0 +1,183 @@ +"""Run the EDI event-time mismatch query against a real JSONB ``kde_value``. + +``fsma.cte_kdes.kde_value`` has been JSONB since migration v078, so the +query in ``scripts/check_edi_event_time_mismatch.py`` must extract the scalar +text (``#>> '{}'``) before its regex match and date cast. This test builds a +minimal events/kdes pair with a JSONB column in a disposable Postgres +(``rls_postgres_or_skip``: the CI service container or a testcontainers +Postgres), runs ``MISMATCH_SQL`` / ``INVALID_DATE_SQL`` inside a transaction +and rolls it back. It never runs the script itself and never touches a real +database. + +Tables are created under a unique schema (not ``fsma``) so the assertions +still run when the connected database already has ``fsma.cte_kdes`` from a +prior migration step. The SQL under test is rewritten only for that schema +name; the predicate and cast logic are unchanged. +""" + +from __future__ import annotations + +import importlib.util +import json +import uuid +from datetime import datetime, timedelta, timezone +from pathlib import Path + +import pytest +from sqlalchemy import text + +from tests.migrations.rls_db import rls_postgres_or_skip + +_SCRIPT = Path(__file__).resolve().parents[2] / "scripts" / "check_edi_event_time_mismatch.py" +_spec = importlib.util.spec_from_file_location("check_edi_event_time_mismatch", _SCRIPT) +check = importlib.util.module_from_spec(_spec) +_spec.loader.exec_module(check) # type: ignore[union-attr] + + +def _sql_for_schema(sql: str, schema: str) -> str: + """Point the diagnostic SQL at a disposable schema instead of ``fsma``.""" + return sql.replace("fsma.", f"{schema}.") + + +def test_mismatch_sql_extracts_jsonb_scalar_text() -> None: + sql = check.MISMATCH_SQL + assert "#>> '{}'" in sql + assert "CAST(k.kde_value" not in sql + assert "k.kde_value ~" not in sql + assert "e.source LIKE 'edi%'" in sql + assert "to_date" not in sql + assert "CASE WHEN" in sql + assert "to_date" not in check.INVALID_DATE_SQL + + +def test_mismatch_sql_counts_against_real_jsonb_column() -> None: + now = datetime.now(timezone.utc).replace(microsecond=0) + day = (now - timedelta(days=10)).date() + day_start = datetime(day.year, day.month, day.day, tzinfo=timezone.utc) + tenant = uuid.uuid4() + schema = f"edi_mismatch_{uuid.uuid4().hex[:12]}" + mismatch_sql = _sql_for_schema(check.MISMATCH_SQL, schema) + invalid_sql = _sql_for_schema(check.INVALID_DATE_SQL, schema) + with rls_postgres_or_skip() as (_url, engine): + with engine.connect() as conn: + trans = conn.begin() + try: + conn.execute(text(f"CREATE SCHEMA {schema}")) + conn.execute( + text( + f"CREATE TABLE {schema}.cte_events (id uuid PRIMARY KEY, tenant_id uuid NOT NULL," + " event_timestamp timestamptz NOT NULL, source text NOT NULL)" + ) + ) + conn.execute( + text( + f"CREATE TABLE {schema}.cte_kdes (cte_event_id uuid NOT NULL, tenant_id uuid NOT NULL," + " kde_key text NOT NULL, kde_value jsonb)" + ) + ) + + def add(ts: datetime, kdes: dict, source: str = "edi_inbound") -> None: + eid = uuid.uuid4() + conn.execute( + text(f"INSERT INTO {schema}.cte_events VALUES (:i, :t, :ts, :s)"), + {"i": eid, "t": tenant, "ts": ts, "s": source}, + ) + for k, v in kdes.items(): + conn.execute( + text(f"INSERT INTO {schema}.cte_kdes VALUES (:i, :t, :k, CAST(:v AS jsonb))"), + {"i": eid, "t": tenant, "k": k, "v": json.dumps(v)}, + ) + + iso = day.isoformat() + # Agrees: timestamp inside the event day. + add(day_start + timedelta(hours=12), {"ship_date": iso}) + # Disagrees by ingest-time drift; both date KDEs -> counted once. + add(now, {"ship_date": iso, "event_date": iso}) + # Disagrees, event_date only. + add(day_start - timedelta(days=3), {"event_date": iso}) + # Non-EDI source with a mismatch must not inflate the EDI count. + add(now, {"ship_date": iso}, source="webhook") + # Calendar-invalid ISO-looking date must be counted as invalid, not abort. + add(now, {"ship_date": "2026-02-30"}) + add(now, {"event_date": "2024-02-30"}) # also invalid + add(now, {"ship_date": "1900-02-29"}) # century non-leap + add(now, {"ship_date": "0000-01-01"}) # year 0000 invalid + # Non-date JSONB values are ignored, not cast errors. + add(now, {"ship_date": "not-a-date"}) + add(now, {"ship_date": ""}) + add(now, {"ship_date": "20-02-2026x"}) + add(now, {"event_date": {"nested": iso}}) + add(now, {"ship_date": None}) + + count = conn.execute(text(mismatch_sql)).scalar() + assert count == 2 + invalid = conn.execute(text(invalid_sql)).scalar() + assert invalid == 4 + # Valid leap day must not be counted as invalid, and must not abort. + leap = uuid.uuid4() + conn.execute( + text(f"INSERT INTO {schema}.cte_events VALUES (:i, :t, :ts, :s)"), + {"i": leap, "t": tenant, "ts": now, "s": "edi_inbound"}, + ) + conn.execute( + text(f"INSERT INTO {schema}.cte_kdes VALUES (:i, :t, :k, CAST(:v AS jsonb))"), + {"i": leap, "t": tenant, "k": "ship_date", "v": json.dumps("2024-02-29")}, + ) + assert conn.execute(text(invalid_sql)).scalar() == 4 + # Python-side helper still agrees for a single valid ship_date row. + rows = conn.execute( + text( + f"SELECT e.event_timestamp, " + f"{check._SAFE_DATE} " + f"FROM {schema}.cte_events e" + f" JOIN {schema}.cte_kdes k ON k.cte_event_id = e.id AND k.kde_key = 'ship_date'" + f" WHERE e.source LIKE 'edi%'" + f" AND {check._CALENDAR_VALID_ISO_DATE}" + ) + ).all() + # One agreeing ship_date event + one disagreeing (ingest drift) + leap disagreeing. + assert check.count_mismatches(rows) >= 1 + finally: + trans.rollback() + try: + with engine.connect() as cleanup_conn: + cleanup_conn.execute(text(f"DROP SCHEMA IF EXISTS {schema} CASCADE")) + cleanup_conn.commit() + except Exception: + pass + + +@pytest.mark.parametrize( + "val, expected_valid, expected_safe_date, expected_invalid_date", + [ + ("not-a-date", False, None, False), + ("", False, None, False), + ("20-02-2026x", False, None, False), + ("0000-01-01", False, None, True), + ("2026-02-30", False, None, True), + ("2024-02-29", True, "2024-02-29", False), + ], +) +def test_date_validity_sql_handles_all_date_shapes( + val: str, expected_valid: bool, expected_safe_date: str | None, expected_invalid_date: bool +) -> None: + """Verifies non-date strings, year 0000, invalid dates, and valid leap days.""" + with rls_postgres_or_skip() as (_url, engine): + with engine.connect() as conn: + res = conn.execute( + text( + f"SELECT " + f"({check._CALENDAR_VALID_ISO_DATE}) AS is_valid, " + f"({check._SAFE_DATE}) AS safe_d, " + f"(CASE WHEN (k.kde_value #>> '{{}}') ~ '^\\d{{4}}-\\d{{2}}-\\d{{2}}$' " + f"THEN NOT ({check._CALENDAR_VALID_ISO_DATE}) ELSE false END) AS is_invalid " + f"FROM (VALUES (CAST(:v AS jsonb))) AS k(kde_value)" + ), + {"v": json.dumps(val)}, + ).one() + assert res.is_valid is expected_valid + if expected_safe_date is None: + assert res.safe_d is None + else: + assert res.safe_d.isoformat() == expected_safe_date + assert res.is_invalid is expected_invalid_date