From 3a517781c75da09b637d9e5be1bf97058d41a3a3 Mon Sep 17 00:00:00 2001 From: Christopher <131207452+PetrefiedThunder@users.noreply.github.com> Date: Sat, 31 Oct 2026 01:48:42 -0700 Subject: [PATCH] fix(ingestion): give EDI events the shipping KDEs validation requires Every EDI 856 was rejected in ingest_events. REQUIRED_KDES_BY_CTE[SHIPPING] requires reference_document, tlc_source_reference and ship_date, but: - both EDI routes stored the BOL only as reference_document_number; - tlc_source_reference was only copied for 944/945, and the legacy /api/v1/ingest/edi route had no way to accept it; - /document set ship_date/receive_date only inside the 944/945 branch, so 856/850/810/861 events had no CTE date KDE. Now both routes set reference_document from the extracted BOL (keeping reference_document_number), the legacy route accepts tlc_source_reference, /document copies it to every set, and the date KDE is chosen by CTE type for every set. An 856 without tlc_source_reference is still rejected, but for that one field only: X12 has no place for it. test_r3_restored_fixtures_e2e asserted the old behavior (ship_date as an open gap on the 856 fixture); it now expects only tlc_source_reference. Co-Authored-By: Claude Opus 5.5 --- .../ingestion/app/edi_ingestion/routes.py | 29 ++++-- .../tests/test_edi_856_required_kdes.py | 88 +++++++++++++++++++ .../tests/test_r3_restored_fixtures_e2e.py | 4 +- 3 files changed, 112 insertions(+), 9 deletions(-) create mode 100644 services/ingestion/tests/test_edi_856_required_kdes.py diff --git a/services/ingestion/app/edi_ingestion/routes.py b/services/ingestion/app/edi_ingestion/routes.py index 8d36158ff..d5abeec1c 100644 --- a/services/ingestion/app/edi_ingestion/routes.py +++ b/services/ingestion/app/edi_ingestion/routes.py @@ -518,6 +518,9 @@ def _enforce_interchange_structure( async def ingest_edi_856( file: UploadFile = File(..., description="X12 EDI file (856 ASN)"), traceability_lot_code: str = Form(..., description="FSMA Traceability Lot Code"), + tlc_source_reference: Optional[str] = Form( + None, description="Optional TLC source reference (where the TLC was assigned) copied to each event's KDEs" + ), tenant_id: Optional[str] = Form(None, description="Optional tenant override"), source: str = Form("edi_856_inbound", description="Source system identifier"), product_description: Optional[str] = Form(None, description="Optional product description override"), @@ -698,6 +701,9 @@ async def ingest_edi_856( "ship_from_gln": ship_from_gln or ship_from_gln_raw, "ship_to_gln": ship_to_gln or ship_to_gln_raw, "reference_document_number": extracted.get("reference_document_number"), + # Shipping validation requires these exact names (21 CFR 1.1340(a)(7)-(8)). + "reference_document": extracted.get("reference_document_number"), + "tlc_source_reference": tlc_source_reference.strip() if tlc_source_reference else None, "carrier": extracted.get("carrier"), "asn_number": extracted.get("asn_number"), "edi_transaction_set": "856", @@ -882,6 +888,10 @@ def _build_ingest_event_for_set( "edi_isa13": extracted.get("isa13"), "edi_gs_control_number": extracted.get("gs_control_number"), "reference_document_number": extracted.get("reference_document_number"), + # Shipping/receiving validation requires these exact names + # (21 CFR 1.1340(a)(7)-(8), 1.1345(a)(7)-(8)) for every set. + "reference_document": extracted.get("reference_document_number"), + "tlc_source_reference": extracted.get("tlc_source_reference"), "event_date": event_date, } @@ -995,18 +1005,19 @@ def _build_ingest_event_for_set( "g62_date_qualifier": extracted.get("g62_date_qualifier"), "receive_date_raw": extracted.get("receive_date_raw"), "ship_date_raw": extracted.get("ship_date_raw"), - "reference_document": extracted.get("reference_document_number"), - "tlc_source_reference": extracted.get("tlc_source_reference"), } ) if transaction_set == "944": - kdes["receive_date"] = event_date kdes["immediate_previous_source"] = extracted.get("ship_from_name") or extracted.get("ship_from_gln") - else: - kdes["ship_date"] = event_date if extracted.get("depositor_gln"): kdes["depositor_gln"] = extracted["depositor_gln"] + # Every set needs the CTE's date KDE, not only 944/945. + if cte_type == WebhookCTEType.RECEIVING: + kdes["receive_date"] = event_date + else: + kdes["ship_date"] = event_date + return IngestEvent( cte_type=cte_type, traceability_lot_code=traceability_lot_code, @@ -1040,7 +1051,7 @@ async def ingest_edi_document( description="FSMA Traceability Lot Code for 856/850/810/861; required but ignored for 944/945 event TLCs, which use each line's lot code", ), tlc_source_reference: Optional[str] = Form( - None, description="Optional TLC source reference copied to 944/945 KDEs" + None, description="Optional TLC source reference copied to every event's KDEs" ), tenant_id: Optional[str] = Form(None, description="Optional tenant override"), source: str = Form("edi_inbound", description="Source system identifier"), @@ -1184,12 +1195,13 @@ def _preflight_all_sets() -> None: # the shared header is copied onto each line. event_inputs: list[tuple[str, dict[str, Any], list[list[str]]]] = [] first_event_indexes: list[int] = [] + tlc_source_ref = tlc_source_reference.strip() if tlc_source_reference else None for slice_set, slice_extracted, slice_segments in zip(slice_sets, extractions, set_slices): first_event_indexes.append(len(event_inputs)) if slice_set in ("856", "861", "944", "945"): header = {key: value for key, value in slice_extracted.items() if key != "line_items"} + header["tlc_source_reference"] = tlc_source_ref if slice_set in ("944", "945"): - header["tlc_source_reference"] = tlc_source_reference.strip() if tlc_source_reference else None date_field = "receive_date_raw" if slice_set == "944" else "ship_date_raw" header["_event_timestamp"], header["_event_date"] = _ship_timestamp_and_date( header.get(date_field), None @@ -1213,7 +1225,8 @@ def _preflight_all_sets() -> None: } event_inputs.append((slice_set, merged, slice_segments)) else: - event_inputs.append((slice_set, slice_extracted, slice_segments)) + with_ref = {**slice_extracted, "tlc_source_reference": tlc_source_ref} + event_inputs.append((slice_set, with_ref, slice_segments)) # #1174: FSMA validation is fail-closed by default. A schema failure # returns HTTP 422 with per-field errors and nothing is persisted. diff --git a/services/ingestion/tests/test_edi_856_required_kdes.py b/services/ingestion/tests/test_edi_856_required_kdes.py new file mode 100644 index 000000000..d04a8f795 --- /dev/null +++ b/services/ingestion/tests/test_edi_856_required_kdes.py @@ -0,0 +1,88 @@ +"""EDI 856 events must carry the KDEs that shipping validation requires. + +Both EDI routes built 856 events with ``reference_document_number`` only and +never set ``tlc_source_reference``, while ``REQUIRED_KDES_BY_CTE[SHIPPING]`` +requires ``reference_document`` and ``tlc_source_reference``. Every 856 was +therefore rejected in ``ingest_events``. These checks post the recorded ASN +fixture through each route and run the captured events through the real +KDE validator. +""" + +from __future__ import annotations + +import sys +from pathlib import Path + +import pytest +from fastapi import FastAPI +from fastapi.testclient import TestClient + +service_dir = Path(__file__).parent.parent +sys.path.insert(0, str(service_dir)) + +pytest.importorskip("fastapi") + +import app.edi_ingestion.routes as edi_ingestion_routes # noqa: E402 +from app.authz import IngestionPrincipal, get_ingestion_principal # noqa: E402 +from app.edi_ingestion import router as edi_router # noqa: E402 +from app.webhook_models import IngestResponse, WebhookPayload # noqa: E402 +from app.webhook_router_v2 import _validate_event_kdes # noqa: E402 + +_ASN_856 = Path(__file__).parent / "fixtures" / "edi_corpus" / "asn_856.edi" +_TENANT = "11111111-1111-1111-1111-111111111111" + + +@pytest.fixture() +def captured(monkeypatch: pytest.MonkeyPatch) -> list[WebhookPayload]: + payloads: list[WebhookPayload] = [] + + async def _capture(payload: WebhookPayload, **_kwargs) -> IngestResponse: + payloads.append(payload) + return IngestResponse(accepted=0, rejected=0, total=len(payload.events), events=[]) + + monkeypatch.setattr(edi_ingestion_routes, "ingest_events", _capture) + monkeypatch.setattr(edi_ingestion_routes, "_resolve_principal_tenant_id", lambda *args: _TENANT) + monkeypatch.setattr(edi_ingestion_routes, "check_and_record_interchange", lambda *args, **kwargs: (False, None)) + monkeypatch.delenv("EDI_REQUIRE_PARTNER_ID", raising=False) + monkeypatch.delenv("EDI_PARTNER_ALLOWLIST", raising=False) + return payloads + + +def _post(route: str, data: dict[str, str]): + app = FastAPI() + app.include_router(edi_router) + app.dependency_overrides[get_ingestion_principal] = lambda: IngestionPrincipal( + key_id="test", scopes=["edi.ingest"], tenant_id=_TENANT + ) + with TestClient(app) as client: + return client.post( + f"/api/v1/ingest/edi{route}", + data=data, + files={"file": ("asn_856.edi", _ASN_856.read_bytes(), "application/edi-x12")}, + ) + + +@pytest.mark.parametrize("route", ["", "/document"]) +def test_856_with_tlc_source_reference_passes_kde_validation(route: str, captured) -> None: + response = _post(route, {"traceability_lot_code": "LOT-856-KDE", "tlc_source_reference": " PACK-0310 "}) + + assert response.status_code == 201, response.text + events = [event for payload in captured for event in payload.events] + assert events + for event in events: + assert event.kdes["reference_document"] == "BOL-9001" + assert event.kdes["tlc_source_reference"] == "PACK-0310" + assert _validate_event_kdes(event) == [] + + +@pytest.mark.parametrize("route", ["", "/document"]) +def test_856_without_tlc_source_reference_only_misses_that_kde(route: str, captured) -> None: + response = _post(route, {"traceability_lot_code": "LOT-856-KDE"}) + + assert response.status_code == 201, response.text + events = [event for payload in captured for event in payload.events] + assert events + for event in events: + assert _validate_event_kdes(event) == [ + "Missing required KDE 'tlc_source_reference' for shipping CTE" + ] diff --git a/services/ingestion/tests/test_r3_restored_fixtures_e2e.py b/services/ingestion/tests/test_r3_restored_fixtures_e2e.py index fb062a9f2..0bb311290 100644 --- a/services/ingestion/tests/test_r3_restored_fixtures_e2e.py +++ b/services/ingestion/tests/test_r3_restored_fixtures_e2e.py @@ -144,7 +144,9 @@ def test_edi_856_fixture_end_to_end(): transition, gaps = gaps_by_tlc[tlc] assert transition == "open" assert {g.gap_kind for g in gaps} == {GAP_KIND_MISSING_KDE} - assert {"ship_date", "tlc_source_reference"} <= {g.gap_key for g in gaps} + # The 856 carries its ship date (DTM) and BOL; only the TLC source + # reference has no X12 home, so it stays a gap unless the caller sends it. + assert {g.gap_key for g in gaps} == {"tlc_source_reference"} headers, rows = _parse_sheet(sheet) assert headers == FDA_COLUMNS