diff --git a/docs/runbooks/FDA_RECALL_RESPONSE.md b/docs/runbooks/FDA_RECALL_RESPONSE.md index 1280b2f33..9f8e518d3 100644 --- a/docs/runbooks/FDA_RECALL_RESPONSE.md +++ b/docs/runbooks/FDA_RECALL_RESPONSE.md @@ -176,7 +176,7 @@ section included, fails both the browser verifier and `services/admin/scripts/ve ### 3.3 FDA CSV Column Specification -The export uses 34 columns defined in `FDA_COLUMNS` (`fda_export_service.py:29-65`): +The export uses 34 columns defined in `FDA_COLUMNS` (`fda_export_service.py`): **Core fields:** TLC, Product Description, Quantity, UOM, Event Type (CTE), Event Date/Time, Location GLN/Name, Ship From/To, Immediate Previous Source, TLC Source, Source Document @@ -186,6 +186,8 @@ The export uses 34 columns defined in `FDA_COLUMNS` (`fda_export_service.py:29-6 **FSMA 204 required:** System Entry Timestamp (when the record was entered into the system, maps to `ingested_at` column) +**Open Data Gaps:** always the last column. It names the data gaps open on that row's lot and CTE when the export was generated, for example `missing_kde: tlc_source_reference`. Blank means none was open. If the gaps cannot be read, the export stops with a 503. + ### 3.4 Check Response Headers The export endpoint returns compliance metadata in response headers: diff --git a/services/ingestion/app/fda_export/queries.py b/services/ingestion/app/fda_export/queries.py index 83d9584d5..89eade52d 100644 --- a/services/ingestion/app/fda_export/queries.py +++ b/services/ingestion/app/fda_export/queries.py @@ -11,7 +11,11 @@ from datetime import date as _date, datetime, time as _time, timedelta, timezone from typing import Any, Optional +from fastapi import HTTPException +from shared.data_gap_tracking import fetch_gaps_open_at +from shared.tenant_context import set_tenant_guc_if_uuid from sqlalchemy import text +from sqlalchemy.exc import SQLAlchemyError logger = logging.getLogger("fda-export") @@ -325,6 +329,50 @@ def fetch_recall_events_canonical(db_session, where_clause: str, params: dict, l return rows +def attach_open_gaps( + db_session, + tenant_id: str, + events: list[dict], + *, + as_of: Optional[datetime] = None, +) -> datetime: + """Set ``open_data_gaps`` on each event and return the moment it describes. + + A gap belongs to a lot and CTE, so every row for that pair carries it. + ``as_of`` defaults to now. Exporters record the returned moment as the + audit row's ``generated_at`` and ``/export/verify`` passes that value back, + so a gap that closes after the export does not change the regenerated + bytes. + + A blank cell states that no gap was open, so a failed lookup stops the + export with a 503 rather than emit rows it cannot vouch for. Scoped twice: + the ``tenant_id`` predicate here and forced RLS on the table. + """ + as_of = as_of or datetime.now(timezone.utc) + lot_codes = {str(event.get("traceability_lot_code") or "") for event in events} + try: + set_tenant_guc_if_uuid(db_session, tenant_id) + gaps = fetch_gaps_open_at(db_session, tenant_id, lot_codes, as_of) + except SQLAlchemyError as exc: + logger.error( + "fda_export_open_gap_lookup_failed", + extra={"tenant_id": tenant_id, "error_type": type(exc).__name__}, + exc_info=True, + ) + raise HTTPException( + status_code=503, + detail=( + "FDA export halted: open data gaps could not be read, so the " + "export cannot state which rows have an outstanding item." + ), + ) from exc + for event in events: + lot_code = str(event.get("traceability_lot_code") or "").strip() + cte_type = str(event.get("event_type") or "").strip().lower() + event["open_data_gaps"] = list(gaps.get((lot_code, cte_type), ())) + return as_of + + def _verify_chain_for_export(persistence, db_session, tenant_id: str): """Chain verification for export handlers, honoring the WS-1b flag. @@ -491,6 +539,7 @@ def log_recall_export( location: Optional[str] = None, event_type: Optional[str] = None, include_pii: bool = False, + generated_at: Optional[datetime] = None, ) -> None: """Write the recall-export audit log entry. @@ -525,9 +574,10 @@ def log_recall_export( INSERT INTO fsma.fda_export_log (tenant_id, export_type, record_count, export_hash, generated_by, query_tlc, query_start_date, query_end_date, - include_pii) + include_pii, generated_at) VALUES (:tid, :etype, :cnt, :hash, :generated_by, - :tlc, :sd, :ed, :pii) + :tlc, :sd, :ed, :pii, + COALESCE(CAST(:generated_at AS timestamptz), now())) """), { "tid": tenant_id, @@ -539,6 +589,7 @@ def log_recall_export( "sd": start_date, "ed": end_date, "pii": bool(include_pii), + "generated_at": generated_at, }, ) db_session.commit() @@ -769,6 +820,7 @@ def log_v2_export( source_ip: Optional[str] = None, include_pii: bool = False, event_type: Optional[str] = None, + generated_at: Optional[datetime] = None, ) -> None: """Write the v2 export audit log entry. @@ -781,9 +833,9 @@ def log_v2_export( INSERT INTO fsma.fda_export_log (tenant_id, export_type, record_count, export_hash, generated_by, query_tlc, query_start_date, query_end_date, - include_pii, query_event_type) + include_pii, query_event_type, generated_at) VALUES (:tid, :etype, :cnt, :hash, :generated_by, :tlc, :sd, :ed, - :pii, :qetype) + :pii, :qetype, COALESCE(CAST(:generated_at AS timestamptz), now())) """), { "tid": tenant_id, @@ -796,6 +848,7 @@ def log_v2_export( "ed": end_date, "pii": bool(include_pii), "qetype": event_type, + "generated_at": generated_at, }, ) db_session.commit() diff --git a/services/ingestion/app/fda_export/recall.py b/services/ingestion/app/fda_export/recall.py index 97b290744..8179727ab 100644 --- a/services/ingestion/app/fda_export/recall.py +++ b/services/ingestion/app/fda_export/recall.py @@ -30,6 +30,7 @@ from .queries import ( AuditLogWriteError, _verify_chain_for_export, + attach_open_gaps, build_recall_where_clause, build_recall_where_clause_canonical, export_v1_reads_canonical, @@ -176,6 +177,7 @@ async def export_recall_filtered_handler( # Convert to FDA export format events = rows_to_event_dicts(rows) + generated_at = attach_open_gaps(db_session, tenant_id, events) csv_content, export_hash = generate_csv_and_hash(events, include_pii=include_pii) chain_verification = _verify_chain_for_export(persistence, db_session, tenant_id) completeness_summary = _build_completeness_summary(events) @@ -212,6 +214,7 @@ async def export_recall_filtered_handler( location=location, event_type=event_type, include_pii=include_pii, + generated_at=generated_at, ) timestamp = make_timestamp() diff --git a/services/ingestion/app/fda_export/router.py b/services/ingestion/app/fda_export/router.py index 34947985a..b68ac6d58 100644 --- a/services/ingestion/app/fda_export/router.py +++ b/services/ingestion/app/fda_export/router.py @@ -68,7 +68,9 @@ from .merkle import get_merkle_proof_handler, get_merkle_root_handler from .queries import ( AuditLogWriteError, + ExportOverflowError, _verify_chain_for_export, + attach_open_gaps, build_recall_where_clause, build_v2_where_clause, fetch_export_log_history, @@ -78,7 +80,6 @@ format_export_log_rows, log_recall_export, log_v2_export, - ExportOverflowError, reconstruct_full_export_events, rows_to_event_dicts, v2_rows_to_event_dicts, @@ -305,6 +306,7 @@ def export_fda_spreadsheet( # flag flows from the query string → authorization gate → CSV # generator; the generated hash covers the redacted (or full) # content exactly, so ``verify_export`` can reproduce it. + generated_at = attach_open_gaps(db_session, tenant_id, events) csv_content = _generate_csv(events, include_pii=include_pii) export_hash = hashlib.sha256(csv_content.encode("utf-8")).hexdigest() chain_verification = _verify_chain_for_export(persistence, db_session, tenant_id) @@ -339,6 +341,7 @@ def export_fda_spreadsheet( generated_by=generated_by, export_type="fda_spreadsheet_package" if format == "package" else "fda_spreadsheet", include_pii=include_pii, + generated_at=generated_at, ) db_session.commit() except Exception as exc: @@ -611,6 +614,7 @@ def export_all_events( ), ) + generated_at = attach_open_gaps(db_session, tenant_id, deduped) csv_content = _generate_csv(deduped, include_pii=include_pii) export_hash = hashlib.sha256(csv_content.encode("utf-8")).hexdigest() chain_verification = _verify_chain_for_export(persistence, db_session, tenant_id) @@ -639,6 +643,7 @@ def export_all_events( export_type="fda_export_all_package" if format == "package" else "fda_export_all", include_pii=include_pii, query_event_type=event_type, + generated_at=generated_at, ) db_session.commit() except Exception as exc: @@ -1061,6 +1066,7 @@ def export_fda_spreadsheet_v2( events = v2_rows_to_event_dicts(rows) # ----- Generate CSV with compliance columns ----- + generated_at = attach_open_gaps(db_session, tenant_id, events) csv_content, export_hash = generate_csv_v2_and_hash(events, include_pii=include_pii) chain_verification = _verify_chain_for_export(persistence, db_session, tenant_id) completeness_summary = _build_completeness_summary(events) @@ -1111,6 +1117,7 @@ def export_fda_spreadsheet_v2( source_ip=identity["source_ip"], include_pii=include_pii, event_type=event_type, + generated_at=generated_at, ) timestamp = make_timestamp() diff --git a/services/ingestion/app/fda_export/verification.py b/services/ingestion/app/fda_export/verification.py index 494513127..29a448f83 100644 --- a/services/ingestion/app/fda_export/verification.py +++ b/services/ingestion/app/fda_export/verification.py @@ -14,11 +14,12 @@ from ..authz import require_permission from ..export_models import ExportVerifyResponse -from ..fda_export_service import _generate_csv, _generate_csv_v2 +from ..fda_export_service import _generate_csv, _generate_csv_before_open_gaps, _generate_csv_v2 from ..subscription_gate import require_active_subscription from .queries import ( FULL_EXPORT_TYPES, ExportOverflowError, + attach_open_gaps, build_v2_where_clause, fetch_v2_events, reconstruct_full_export_events, @@ -140,8 +141,6 @@ async def verify_export_handler( ), ) events = v2_rows_to_event_dicts(v2_rows) - csv_content = _generate_csv_v2(events, include_pii=include_pii) - regenerated_hash = hashlib.sha256(csv_content.encode("utf-8")).hexdigest() elif reconstruction == "single_tlc": events = persistence.query_events_by_tlc( tenant_id=tenant_id, @@ -172,9 +171,20 @@ async def verify_export_handler( ), ) from exc - if reconstruction != "v2": - csv_content = _generate_csv(events, include_pii=include_pii) - regenerated_hash = hashlib.sha256(csv_content.encode("utf-8")).hexdigest() + # Gaps as of the export, not now: one that has closed since must not + # read as tampering. + attach_open_gaps(db_session, tenant_id, events, as_of=generated_at) + generate_csv = _generate_csv_v2 if reconstruction == "v2" else _generate_csv + csv_content = generate_csv(events, include_pii=include_pii) + regenerated_hash = hashlib.sha256(csv_content.encode("utf-8")).hexdigest() + + # An export logged before the Open Data Gaps column existed has the + # older layout. Without this, every such export would read as tampered. + if regenerated_hash != original_hash: + older_layout = _generate_csv_before_open_gaps(events, include_pii=include_pii, v2=reconstruction == "v2") + older_hash = hashlib.sha256(older_layout.encode("utf-8")).hexdigest() + if older_hash == original_hash: + regenerated_hash = older_hash match = regenerated_hash == original_hash diff --git a/services/ingestion/app/fda_export_service.py b/services/ingestion/app/fda_export_service.py index 113283029..8f4cad36c 100644 --- a/services/ingestion/app/fda_export_service.py +++ b/services/ingestion/app/fda_export_service.py @@ -69,8 +69,17 @@ "Additional KDEs (JSON)", # FSMA 204 §1.1455(c): date/time event entered into system "System Entry Timestamp", + # Names the gaps open on this row's lot and CTE when the export was + # generated, so a blank required KDE reads as a known outstanding item. + # Always the last column, in v1 and v2, so no earlier column moves. + "Open Data Gaps", ] +OPEN_GAPS_COLUMN = FDA_COLUMNS[-1] + +# Exports logged before the column existed are verified against this layout. +FDA_COLUMNS_WITHOUT_OPEN_GAPS = FDA_COLUMNS[:-1] + # KDE keys that get their own named columns in the FDA export _NAMED_KDE_COLUMNS = { @@ -162,7 +171,7 @@ def _redact_location_value(value: str, column_name: str, include_pii: bool) -> s # Extended column spec: original FDA columns + compliance + traceability graph columns -FDA_COLUMNS_V2 = FDA_COLUMNS + [ +FDA_COLUMNS_V2_WITHOUT_OPEN_GAPS = FDA_COLUMNS_WITHOUT_OPEN_GAPS + [ "Compliance Status", "Rule Failures", # Transformation traceability (added Phase 2) @@ -170,6 +179,8 @@ def _redact_location_value(value: str, column_name: str, include_pii: bool) -> s "Trace Seed TLC", # The TLC originally queried (useful when this row is a linked lot) ] +FDA_COLUMNS_V2 = FDA_COLUMNS_V2_WITHOUT_OPEN_GAPS + [OPEN_GAPS_COLUMN] + def _event_to_fda_row(event: dict, *, include_pii: bool = False) -> dict: """Convert a persisted CTE event to an FDA spreadsheet row. @@ -284,6 +295,10 @@ def _event_to_fda_row(event: dict, *, include_pii: bool = False) -> dict: # This maps to the ingested_at column which defaults to NOW() on INSERT. row["System Entry Timestamp"] = sanitize_cell(event.get("ingested_at", "") or "") + # Set by ``attach_open_gaps``. Blank means no gap was open at export time, + # so an export whose gap lookup failed must not reach this point. + row[OPEN_GAPS_COLUMN] = sanitize_cell("; ".join(event.get("open_data_gaps") or ())) + return row @@ -697,6 +712,7 @@ def _build_known_gaps( chain_payload: dict, coverage_gate_bypassed: bool, rule_versions: Optional[dict], + events_with_open_data_gaps: int = 0, ) -> dict: """Collect everything this package knows it does NOT prove. @@ -711,6 +727,7 @@ def _build_known_gaps( rules = rule_versions or {} gaps: dict[str, Any] = { "required_kde_coverage_ratio": completeness_summary.get("required_kde_coverage_ratio"), + "events_with_open_data_gaps": events_with_open_data_gaps, "events_with_missing_required_fields": missing_total, "missing_required_by_field": completeness_summary.get("missing_required_by_field") or {}, # The per-event list in ``completeness`` is capped (250); say so @@ -785,6 +802,7 @@ def _build_fda_package( chain_payload=chain_payload, coverage_gate_bypassed=coverage_gate_bypassed, rule_versions=rule_versions, + events_with_open_data_gaps=sum(1 for event in events if event.get("open_data_gaps")), ) evidence_context = { "version": "1.0", @@ -818,6 +836,10 @@ def _build_fda_package( + "\n" + pii_note + "\n" + "Open Data Gaps: the last spreadsheet column names the data gaps open on that row's " + "traceability lot and CTE when this export was generated, for example a required KDE " + "still outstanding. Blank means none was open.\n" + "\n" "Verification:\n" "- Recompute each manifest file hash and the Merkle root to confirm " "the package bytes are internally consistent.\n" @@ -1017,3 +1039,22 @@ def _generate_csv_v2(events: list[dict], *, include_pii: bool = False) -> str: writer.writerow(_event_to_fda_row_v2(event, include_pii=include_pii)) return output.getvalue() + + +def _generate_csv_before_open_gaps(events: list[dict], *, include_pii: bool, v2: bool) -> str: + """Write the layout from before the Open Data Gaps column existed. + + Only ``/export/verify`` uses this, to reproduce exports logged back then. + """ + columns = FDA_COLUMNS_V2_WITHOUT_OPEN_GAPS if v2 else FDA_COLUMNS_WITHOUT_OPEN_GAPS + output = io.StringIO() + writer = csv.DictWriter(output, fieldnames=columns, quoting=csv.QUOTE_ALL, extrasaction="ignore") + writer.writeheader() + + for event in events: + if v2: + writer.writerow(_event_to_fda_row_v2(event, include_pii=include_pii)) + else: + writer.writerow(_event_to_fda_row(event, include_pii=include_pii)) + + return output.getvalue() diff --git a/services/ingestion/tests/test_fda_export_open_gaps.py b/services/ingestion/tests/test_fda_export_open_gaps.py new file mode 100644 index 000000000..3815fd596 --- /dev/null +++ b/services/ingestion/tests/test_fda_export_open_gaps.py @@ -0,0 +1,340 @@ +"""The FDA spreadsheet marks rows whose lot and CTE have an open data gap. + +Locks: +- Open Data Gaps is the last column in the v1 and v2 sheets, and no earlier + column moves. +- A row with a gap names it; a row without one is unchanged and its cell blank. +- Gaps match on lot and CTE, so another CTE of the same lot stays blank. +- A failed lookup stops the export: a blank cell states no gap was open. +- /export/verify reads gaps as of the export, so a gap that closed later is not + reported as tampering, and an export from before the column still verifies. +""" + +from __future__ import annotations + +import ast +import asyncio +import csv +import hashlib +import io +import json +import sys +import zipfile +from datetime import datetime, timezone +from pathlib import Path +from types import ModuleType, SimpleNamespace + +import pytest +from app import fda_export_service as svc +from app.fda_export import verification as v +from app.fda_export.queries import attach_open_gaps +from fastapi import HTTPException +from shared.data_gap_tracking import fetch_gaps_open_at +from sqlalchemy.exc import ProgrammingError + +TENANT = "7b6f7f0e-6f0a-4a5e-9d0b-1f2f7c1f0a11" +EXPORTED_AT = datetime(2026, 11, 1, 12, 0, tzinfo=timezone.utc) +GAP_LABEL = "missing_kde: tlc_source_reference" + + +def _event(tlc="LOT-1", event_type="shipping") -> dict: + return { + "id": f"{tlc}-{event_type}", + "traceability_lot_code": tlc, + "product_description": "Romaine", + "quantity": 10, + "unit_of_measure": "cases", + "event_type": event_type, + "event_timestamp": "2026-10-30T10:00:00+00:00", + "sha256_hash": "a" * 64, + "chain_hash": "b" * 64, + "kdes": {}, + } + + +def _rows(csv_content: str) -> list[dict]: + return list(csv.DictReader(io.StringIO(csv_content))) + + +class _Result: + def __init__(self, rows=(), row=None): + self._rows = list(rows) + self._row = row + + def fetchall(self): + return self._rows + + def fetchone(self): + return self._row + + +class _GapSession: + """Answers the gap query from ``gap_rows`` and records its parameters.""" + + def __init__(self, gap_rows=(), audit_row=None): + self.gap_rows = gap_rows + self.audit_row = audit_row + self.gap_params: list[dict] = [] + self.gap_sql: list[str] = [] + + def execute(self, stmt, params=None): + sql = str(stmt) + if "public.data_gap_events" in sql: + self.gap_params.append(dict(params)) + self.gap_sql.append(sql) + return _Result(self.gap_rows) + if "fda_export_log" in sql: + return _Result(row=self.audit_row) + return _Result() + + def close(self): + pass + + +# --------------------------------------------------------------------------- +# Spreadsheet +# --------------------------------------------------------------------------- + + +def test_column_is_last_and_no_earlier_column_moves() -> None: + assert svc.FDA_COLUMNS[-1] == svc.OPEN_GAPS_COLUMN + assert svc.FDA_COLUMNS_V2[-1] == svc.OPEN_GAPS_COLUMN + assert svc.FDA_COLUMNS[:-1] == svc.FDA_COLUMNS_WITHOUT_OPEN_GAPS + assert svc.FDA_COLUMNS_V2[:-1] == svc.FDA_COLUMNS_V2_WITHOUT_OPEN_GAPS + assert svc.FDA_COLUMNS_V2_WITHOUT_OPEN_GAPS[: len(svc.FDA_COLUMNS_WITHOUT_OPEN_GAPS)] == ( + svc.FDA_COLUMNS_WITHOUT_OPEN_GAPS + ) + + +@pytest.mark.parametrize("generate", [svc._generate_csv, svc._generate_csv_v2]) +def test_row_with_a_gap_is_marked_and_row_without_is_blank(generate) -> None: + marked = {**_event("LOT-1"), "open_data_gaps": [GAP_LABEL, "missing_kde: other"]} + clean = {**_event("LOT-2"), "open_data_gaps": []} + + rows = _rows(generate([marked, clean])) + + assert rows[0][svc.OPEN_GAPS_COLUMN] == f"{GAP_LABEL}; missing_kde: other" + assert rows[1][svc.OPEN_GAPS_COLUMN] == "" + + +@pytest.mark.parametrize( + ("generate", "columns", "v2"), + [ + (svc._generate_csv, svc.FDA_COLUMNS_WITHOUT_OPEN_GAPS, False), + (svc._generate_csv_v2, svc.FDA_COLUMNS_V2_WITHOUT_OPEN_GAPS, True), + ], +) +def test_no_gap_row_keeps_every_other_cell(generate, columns, v2) -> None: + """Only the new cell differs from the layout before the column.""" + event = {**_event(), "open_data_gaps": []} + + current = _rows(generate([event]))[0] + older = _rows(svc._generate_csv_before_open_gaps([event], include_pii=False, v2=v2))[0] + + assert list(older) == columns + assert {column: current[column] for column in columns} == older + + +def test_cell_is_sanitized() -> None: + """A gap key can echo a validator message, so it gets the same formula guard.""" + event = {**_event(), "open_data_gaps": ['=HYPERLINK("http://x")']} + + assert not _rows(svc._generate_csv([event]))[0][svc.OPEN_GAPS_COLUMN].startswith("=") + + +def test_pii_redaction_leaves_the_marker() -> None: + event = {**_event(), "open_data_gaps": [GAP_LABEL]} + + assert _rows(svc._generate_csv([event], include_pii=False))[0][svc.OPEN_GAPS_COLUMN] == GAP_LABEL + + +# --------------------------------------------------------------------------- +# Lookup +# --------------------------------------------------------------------------- + + +def test_lookup_is_tenant_scoped_and_as_of_the_export() -> None: + session = _GapSession() + + fetch_gaps_open_at(session, TENANT, ["LOT-2", " LOT-1 ", "LOT-2", ""], EXPORTED_AT) + + assert session.gap_params == [{"tid": TENANT, "lots": ["LOT-1", "LOT-2"], "as_of": EXPORTED_AT}] + assert "tenant_id = CAST(:tid AS uuid)" in session.gap_sql[0] + + +def test_lookup_labels_are_sorted_so_bytes_are_stable() -> None: + session = _GapSession( + [ + ("LOT-1", "shipping", "missing_kde", "z_field"), + ("LOT-1", "shipping", "cte_validation_failed", "rule-7"), + ("LOT-1", "shipping", "missing_kde", "a_field"), + ] + ) + + gaps = fetch_gaps_open_at(session, TENANT, ["LOT-1"], EXPORTED_AT) + + assert gaps == { + ("LOT-1", "shipping"): ( + "cte_validation_failed: rule-7", + "missing_kde: a_field", + "missing_kde: z_field", + ) + } + + +def test_no_lots_runs_no_query() -> None: + session = _GapSession() + + assert fetch_gaps_open_at(session, TENANT, ["", " "], EXPORTED_AT) == {} + assert session.gap_params == [] + + +def test_attach_matches_lot_and_cte() -> None: + session = _GapSession([("LOT-1", "shipping", "missing_kde", "tlc_source_reference")]) + shipped = _event("LOT-1", "SHIPPING") + received = _event("LOT-1", "receiving") + other_lot = _event("LOT-2", "shipping") + + returned = attach_open_gaps(session, TENANT, [shipped, received, other_lot], as_of=EXPORTED_AT) + + assert returned == EXPORTED_AT + assert shipped["open_data_gaps"] == [GAP_LABEL] + assert received["open_data_gaps"] == [] + assert other_lot["open_data_gaps"] == [] + + +def test_attach_binds_the_tenant_for_rls() -> None: + calls: list[dict] = [] + + class Session(_GapSession): + def execute(self, stmt, params=None): + if "set_config" in str(stmt): + calls.append(dict(params)) + return super().execute(stmt, params) + + attach_open_gaps(Session(), TENANT, [_event()], as_of=EXPORTED_AT) + + assert any(TENANT in call.values() for call in calls) + + +def test_failed_lookup_stops_the_export() -> None: + class Denied(_GapSession): + def execute(self, stmt, params=None): + if "public.data_gap_events" in str(stmt): + raise ProgrammingError("SELECT", {}, Exception("permission denied")) + return super().execute(stmt, params) + + event = _event() + + with pytest.raises(HTTPException) as exc: + attach_open_gaps(Denied(), TENANT, [event], as_of=EXPORTED_AT) + + assert exc.value.status_code == 503 + assert "open_data_gaps" not in event + + +# --------------------------------------------------------------------------- +# Package +# --------------------------------------------------------------------------- + + +def test_package_counts_marked_rows_in_the_hashed_evidence() -> None: + events = [{**_event("LOT-1"), "open_data_gaps": [GAP_LABEL]}, {**_event("LOT-2"), "open_data_gaps": []}] + csv_content = svc._generate_csv(events) + csv_hash = hashlib.sha256(csv_content.encode("utf-8")).hexdigest() + chain = SimpleNamespace(valid=True, chain_length=2, errors=[], checked_at="now") + completeness = svc._build_completeness_summary(events) + chain_payload = svc._build_chain_verification_payload( + tenant_id=TENANT, + tlc=None, + events=events, + csv_hash=csv_hash, + chain_verification=chain, + completeness_summary=completeness, + ) + + package_bytes, meta = svc._build_fda_package( + events=events, + csv_content=csv_content, + csv_hash=csv_hash, + chain_payload=chain_payload, + completeness_summary=completeness, + tenant_id=TENANT, + tlc=None, + query_start_date=None, + query_end_date=None, + ) + + with zipfile.ZipFile(io.BytesIO(package_bytes)) as archive: + evidence = json.loads(archive.read(meta["evidence_context_name"])) + readme = archive.read("README.txt").decode("utf-8") + assert evidence["known_gaps"]["events_with_open_data_gaps"] == 1 + assert svc.OPEN_GAPS_COLUMN in readme + + +# --------------------------------------------------------------------------- +# /export/verify +# --------------------------------------------------------------------------- + + +@pytest.mark.parametrize("module", ["router.py", "recall.py"]) +def test_every_export_stores_the_moment_its_gaps_describe(module) -> None: + """Verify reads gaps as of the audit row, so each audit write must carry that moment.""" + tree = ast.parse((Path(v.__file__).parent / module).read_text(encoding="utf-8")) + writers = [ + node + for node in ast.walk(tree) + if isinstance(node, ast.Call) + and (getattr(node.func, "attr", None) == "log_export" or getattr(node.func, "id", "").endswith("_export")) + and getattr(node.func, "id", "log_export").startswith("log_") + ] + + assert writers + for call in writers: + assert "generated_at" in {keyword.arg for keyword in call.keywords}, f"line {call.lineno}" + + + +def _audit_row(export_hash: str) -> tuple: + return ("fda_spreadsheet", "LOT-1", None, None, 1, export_hash, EXPORTED_AT, False, None) + + +def _verify(monkeypatch, session: _GapSession) -> dict: + db_module = ModuleType("shared.database") + db_module.SessionLocal = lambda: session + cte_module = ModuleType("shared.cte_persistence") + cte_module.CTEPersistence = lambda db: SimpleNamespace(query_events_by_tlc=lambda **_: [_event("LOT-1")]) + monkeypatch.setitem(sys.modules, "shared.database", db_module) + monkeypatch.setitem(sys.modules, "shared.cte_persistence", cte_module) + return asyncio.run(v.verify_export_handler("exp-1", TENANT)) + + +def test_verify_reads_gaps_as_of_the_export(monkeypatch) -> None: + """The stored moment is what a gap closed since then is judged against.""" + marked = svc._generate_csv([{**_event("LOT-1"), "open_data_gaps": [GAP_LABEL]}]) + session = _GapSession( + gap_rows=[("LOT-1", "shipping", "missing_kde", "tlc_source_reference")], + audit_row=_audit_row(hashlib.sha256(marked.encode("utf-8")).hexdigest()), + ) + + out = _verify(monkeypatch, session) + + assert session.gap_params[0]["as_of"] == EXPORTED_AT + assert out["data_integrity"] == "VERIFIED" + + +def test_verify_accepts_an_export_from_before_the_column(monkeypatch) -> None: + older = svc._generate_csv_before_open_gaps([_event("LOT-1")], include_pii=False, v2=False) + original_hash = hashlib.sha256(older.encode("utf-8")).hexdigest() + + out = _verify(monkeypatch, _GapSession(audit_row=_audit_row(original_hash))) + + assert out["hashes_match"] is True + assert out["regenerated_hash"] == original_hash + + +def test_verify_still_reports_a_real_mismatch(monkeypatch) -> None: + out = _verify(monkeypatch, _GapSession(audit_row=_audit_row("0" * 64))) + + assert out["data_integrity"] == "MISMATCH_DETECTED" + assert out["regenerated_hash"] != "0" * 64 diff --git a/services/ingestion/tests/test_fda_export_open_gaps_pg.py b/services/ingestion/tests/test_fda_export_open_gaps_pg.py new file mode 100644 index 000000000..79a9df15d --- /dev/null +++ b/services/ingestion/tests/test_fda_export_open_gaps_pg.py @@ -0,0 +1,99 @@ +"""The FDA export's gap lookup against real ``public.data_gap_events`` rows. + +The unit tests in ``test_fda_export_open_gaps.py`` run on a fake session that +never parses the SQL. This module runs the real statement at alembic head. + +Skips cleanly when no migrated Postgres is reachable, except under +``REGENGINE_REQUIRE_DB=1`` where an unreachable database fails rather than +skipping (#2530). +""" + +from __future__ import annotations + +import os +import uuid +from datetime import datetime, timedelta, timezone + +import pytest +from shared.data_gap_tracking import fetch_gaps_open_at +from shared.db_url import sync_postgres_url +from sqlalchemy import create_engine, text + +DSN = os.environ.get( + "REMEDIATION_TEST_SUPERUSER_DSN", + "postgresql://regengine:regengine@localhost:55432/regengine", # pragma: allowlist secret — throwaway scratch-DB role +) +RECEIPT = "a" * 64 + + +def _schema_ready(dsn: str) -> bool: + try: + engine = create_engine(sync_postgres_url(dsn), pool_pre_ping=True) + with engine.connect() as conn: + found = conn.execute(text("SELECT to_regclass('public.data_gap_events') IS NOT NULL")).scalar() + engine.dispose() + return bool(found) + except Exception: + return False + + +_DB_READY = _schema_ready(DSN) + +if not _DB_READY and os.environ.get("REGENGINE_REQUIRE_DB") == "1": + raise RuntimeError(f"REGENGINE_REQUIRE_DB=1 but no migrated Postgres is reachable at {DSN!r} — refusing to skip.") + +pytestmark = [ + pytest.mark.integration, + pytest.mark.skipif(not _DB_READY, reason="migrated Postgres with fsma schema not available"), +] + + +@pytest.fixture() +def conn(): + """A connection whose transaction is rolled back, so seeds are never committed.""" + engine = create_engine(sync_postgres_url(DSN)) + with engine.connect() as connection: + outer = connection.begin() + try: + yield connection + finally: + outer.rollback() + engine.dispose() + + +def _seed(conn, tenant_id: str, gap_key: str, opened_at: datetime, closed_at: datetime | None = None) -> None: + conn.execute( + text( + """ + INSERT INTO public.data_gap_events + (tenant_id, supplier_key, lot_code, cte_type, gap_kind, gap_key, + opened_at, closed_at, closing_record_id, closing_record_hash) + VALUES (:tid, 'unassigned', 'LOT-1', 'shipping', 'missing_kde', :key, + :opened, :closed, :rid, :rhash) + """ + ), + { + "tid": tenant_id, + "key": gap_key, + "opened": opened_at, + "closed": closed_at, + "rid": str(uuid.uuid4()) if closed_at else None, + "rhash": RECEIPT if closed_at else None, + }, + ) + + +def test_gap_state_is_read_as_of_a_moment_for_one_tenant(conn) -> None: + now = datetime.now(timezone.utc) + tenant_id, other_tenant_id = str(uuid.uuid4()), str(uuid.uuid4()) + as_of = now - timedelta(days=1) + + _seed(conn, tenant_id, "open_then", now - timedelta(days=2)) + _seed(conn, tenant_id, "closed_since", now - timedelta(days=2), now - timedelta(hours=1)) + _seed(conn, tenant_id, "closed_before", now - timedelta(days=3), now - timedelta(days=2)) + _seed(conn, tenant_id, "opened_after", now - timedelta(hours=1)) + _seed(conn, other_tenant_id, "other_tenant", now - timedelta(days=2)) + + gaps = fetch_gaps_open_at(conn, tenant_id, ["LOT-1"], as_of) + + assert gaps == {("LOT-1", "shipping"): ("missing_kde: closed_since", "missing_kde: open_then")} diff --git a/services/ingestion/tests/test_fda_export_v2_reproducibility.py b/services/ingestion/tests/test_fda_export_v2_reproducibility.py index a7652d7e3..5093ba5ee 100644 --- a/services/ingestion/tests/test_fda_export_v2_reproducibility.py +++ b/services/ingestion/tests/test_fda_export_v2_reproducibility.py @@ -243,6 +243,7 @@ def test_v2_tlc_export_verifies_with_v2_csv(self, monkeypatch): [ _Result(row=_audit_row(export_hash=original_hash)), _Result(rows=rows), + _Result(), # open-gap lookup ] ) _install(monkeypatch, session) @@ -273,6 +274,7 @@ def test_v2_full_export_applies_stored_event_type_filter(self, monkeypatch): ) ), _Result(rows=rows), + _Result(), # open-gap lookup ] ) _install(monkeypatch, session) diff --git a/services/ingestion/tests/test_fda_export_verification_1342.py b/services/ingestion/tests/test_fda_export_verification_1342.py index 592f5459e..8d4a6413e 100644 --- a/services/ingestion/tests/test_fda_export_verification_1342.py +++ b/services/ingestion/tests/test_fda_export_verification_1342.py @@ -53,6 +53,10 @@ def __init__(self, row): def fetchone(self): return self._row + def fetchall(self): + # The open-gap lookup: these fakes model an export with no gaps. + return [] + class _FakeSession: def __init__(self, *, row=None, execute_raises=None): diff --git a/services/shared/cte_persistence/exports.py b/services/shared/cte_persistence/exports.py index 15ee81e9e..5562c1da5 100644 --- a/services/shared/cte_persistence/exports.py +++ b/services/shared/cte_persistence/exports.py @@ -7,6 +7,7 @@ from __future__ import annotations import hashlib +from datetime import datetime from typing import Optional from uuid import uuid4 @@ -66,6 +67,7 @@ def log_export( export_type: str = "fda_spreadsheet", include_pii: bool = False, query_event_type: Optional[str] = None, + generated_at: Optional[datetime] = None, ) -> str: """Log an FDA export event and return the export log ID. @@ -86,6 +88,10 @@ def log_export( tampered — so both are recorded on the audit row. A schema that predates the v113 migration falls back to the legacy insert below, leaving them NULL, which the verifier reads as "not reproducible". + + ``generated_at`` is the moment the export's Open Data Gaps column + describes, so ``/export/verify`` can read gaps as of the same moment. + Omitted, the column default applies. """ export_id = str(uuid4()) fingerprint = self._compute_export_fingerprint( @@ -105,10 +111,11 @@ def log_export( id, tenant_id, export_type, query_tlc, query_start_date, query_end_date, record_count, export_hash, generated_by, export_fingerprint, - include_pii, query_event_type + include_pii, query_event_type, generated_at ) VALUES ( :id, :tid, :etype, :tlc, :start, :end, :count, - :hash, :by, :fp, :pii, :qetype + :hash, :by, :fp, :pii, :qetype, + COALESCE(CAST(:generated_at AS timestamptz), now()) ) ON CONFLICT (tenant_id, export_fingerprint) WHERE export_fingerprint IS NOT NULL @@ -129,6 +136,7 @@ def log_export( "fp": fingerprint, "pii": bool(include_pii), "qetype": query_event_type, + "generated_at": generated_at, }, ).fetchone() except Exception: diff --git a/services/shared/data_gap_tracking.py b/services/shared/data_gap_tracking.py index f9a66f986..311a79eee 100644 --- a/services/shared/data_gap_tracking.py +++ b/services/shared/data_gap_tracking.py @@ -320,6 +320,43 @@ def record_outcomes(db_session: Any, tenant_id: str, outcomes: Sequence[Validati ) +# Open as of a moment, not as of now. The same predicate is in is_open_at. An +# FDA export is regenerated later by /export/verify, and a gap that has closed +# since must still read as open at the original moment or the bytes differ. +_GAPS_OPEN_AT_SQL = text( + """ + SELECT lot_code, cte_type, gap_kind, gap_key + FROM public.data_gap_events + WHERE tenant_id = CAST(:tid AS uuid) + AND lot_code = ANY(:lots) + AND opened_at <= CAST(:as_of AS timestamptz) + AND (closed_at IS NULL OR closed_at > CAST(:as_of AS timestamptz)) + """ +) + + +def fetch_gaps_open_at( + db_session: Any, + tenant_id: str, + lot_codes: Iterable[str], + as_of: datetime, +) -> dict[tuple[str, str], tuple[str, ...]]: + """Gaps open at ``as_of``, by (lot code, lower-case CTE type). + + Values are sorted ``": "`` labels, so a caller writing + them into an export gets the same bytes on every run. Raises on a database + error: unlike the writer, a reader's caller decides what failure means. + """ + lots = sorted({lot.strip() for lot in lot_codes if lot and lot.strip()}) + if not lots: + return {} + rows = db_session.execute(_GAPS_OPEN_AT_SQL, {"tid": tenant_id, "lots": lots, "as_of": as_of}).fetchall() + labels: dict[tuple[str, str], set[str]] = {} + for lot_code, cte_type, gap_kind, gap_key in rows: + labels.setdefault((lot_code, cte_type), set()).add(f"{gap_kind}: {gap_key}") + return {key: tuple(sorted(values)) for key, values in labels.items()} + + def fetch_supplier_summaries( db_session: Any, tenant_id: str,