diff --git a/services/ingestion/app/epcis_export.py b/services/ingestion/app/epcis_export.py index 695f515e3..a0c2a27fe 100644 --- a/services/ingestion/app/epcis_export.py +++ b/services/ingestion/app/epcis_export.py @@ -98,18 +98,32 @@ def _query_tenant_events( k_to.kde_value AS ship_to, k_to_gln.kde_value AS ship_to_gln, k_carrier.kde_value AS carrier, - k_temp.kde_value AS temperature_c + k_temp.kde_value AS temperature_c, + k_ref.kde_value AS reference_document, + k_tlc_src.kde_value AS tlc_source_reference, + k_prev.kde_value AS immediate_previous_source, + k_field.kde_value AS field_id, + k_harvester.kde_value AS harvester_name, + k_harvester_biz.kde_value AS harvester_business_name, + k_harvester_phone.kde_value AS harvester_business_phone FROM fsma.cte_events e - LEFT JOIN tenant_kdes k_from ON k_from.cte_event_id = e.id AND k_from.kde_key = 'ship_from_location' + LEFT JOIN tenant_kdes k_from ON k_from.cte_event_id = e.id AND k_from.tenant_id = e.tenant_id AND k_from.kde_key = 'ship_from_location' LEFT JOIN tenant_kdes k_epcis ON k_epcis.cte_event_id = e.id AND k_epcis.tenant_id = e.tenant_id AND k_epcis.kde_key = 'epcis_document_json' LEFT JOIN tenant_kdes k_gtin ON k_gtin.cte_event_id = e.id AND k_gtin.tenant_id = e.tenant_id AND k_gtin.kde_key = 'gtin' LEFT JOIN tenant_kdes k_product ON k_product.cte_event_id = e.id AND k_product.tenant_id = e.tenant_id AND k_product.kde_key = 'product_id' LEFT JOIN tenant_kdes k_epc_class ON k_epc_class.cte_event_id = e.id AND k_epc_class.tenant_id = e.tenant_id AND k_epc_class.kde_key = 'epc_class' - LEFT JOIN tenant_kdes k_from_gln ON k_from_gln.cte_event_id = e.id AND k_from_gln.kde_key = 'ship_from_gln' - LEFT JOIN tenant_kdes k_to ON k_to.cte_event_id = e.id AND k_to.kde_key = 'ship_to_location' - LEFT JOIN tenant_kdes k_to_gln ON k_to_gln.cte_event_id = e.id AND k_to_gln.kde_key = 'ship_to_gln' - LEFT JOIN tenant_kdes k_carrier ON k_carrier.cte_event_id = e.id AND k_carrier.kde_key = 'carrier_name' - LEFT JOIN tenant_kdes k_temp ON k_temp.cte_event_id = e.id AND k_temp.kde_key = 'temperature_celsius' + LEFT JOIN tenant_kdes k_from_gln ON k_from_gln.cte_event_id = e.id AND k_from_gln.tenant_id = e.tenant_id AND k_from_gln.kde_key = 'ship_from_gln' + LEFT JOIN tenant_kdes k_to ON k_to.cte_event_id = e.id AND k_to.tenant_id = e.tenant_id AND k_to.kde_key = 'ship_to_location' + LEFT JOIN tenant_kdes k_to_gln ON k_to_gln.cte_event_id = e.id AND k_to_gln.tenant_id = e.tenant_id AND k_to_gln.kde_key = 'ship_to_gln' + LEFT JOIN tenant_kdes k_carrier ON k_carrier.cte_event_id = e.id AND k_carrier.tenant_id = e.tenant_id AND k_carrier.kde_key = 'carrier_name' + LEFT JOIN tenant_kdes k_temp ON k_temp.cte_event_id = e.id AND k_temp.tenant_id = e.tenant_id AND k_temp.kde_key = 'temperature_celsius' + LEFT JOIN tenant_kdes k_ref ON k_ref.cte_event_id = e.id AND k_ref.tenant_id = e.tenant_id AND k_ref.kde_key = 'reference_document' + LEFT JOIN tenant_kdes k_tlc_src ON k_tlc_src.cte_event_id = e.id AND k_tlc_src.tenant_id = e.tenant_id AND k_tlc_src.kde_key = 'tlc_source_reference' + LEFT JOIN tenant_kdes k_prev ON k_prev.cte_event_id = e.id AND k_prev.tenant_id = e.tenant_id AND k_prev.kde_key = 'immediate_previous_source' + LEFT JOIN tenant_kdes k_field ON k_field.cte_event_id = e.id AND k_field.tenant_id = e.tenant_id AND k_field.kde_key = 'field_id' + LEFT JOIN tenant_kdes k_harvester ON k_harvester.cte_event_id = e.id AND k_harvester.tenant_id = e.tenant_id AND k_harvester.kde_key = 'harvester_name' + LEFT JOIN tenant_kdes k_harvester_biz ON k_harvester_biz.cte_event_id = e.id AND k_harvester_biz.tenant_id = e.tenant_id AND k_harvester_biz.kde_key = 'harvester_business_name' + LEFT JOIN tenant_kdes k_harvester_phone ON k_harvester_phone.cte_event_id = e.id AND k_harvester_phone.tenant_id = e.tenant_id AND k_harvester_phone.kde_key = 'harvester_business_phone' WHERE e.tenant_id = :tenant_id {"AND " + " AND ".join(where) if where else ""} ORDER BY e.event_timestamp DESC diff --git a/services/ingestion/tests/test_epcis_export_kde_joins.py b/services/ingestion/tests/test_epcis_export_kde_joins.py new file mode 100644 index 000000000..e83ec2bc0 --- /dev/null +++ b/services/ingestion/tests/test_epcis_export_kde_joins.py @@ -0,0 +1,129 @@ +"""Shared _query_tenant_events KDE join expansion (Phase B prerequisite). + +Synthetic fixtures only. Ensures reference_document, tlc_source_reference, +immediate_previous_source, and harvester KDEs are selected with tenant_id on +every cte_kdes join. +""" + +from __future__ import annotations + +import sys +from pathlib import Path +from uuid import uuid4 + +import pytest + +_ingestion_dir = Path(__file__).resolve().parent.parent +if str(_ingestion_dir) not in sys.path: + sys.path.insert(0, str(_ingestion_dir)) +_services_dir = _ingestion_dir.parent +if str(_services_dir) not in sys.path: + sys.path.insert(0, str(_services_dir)) + +pytest.importorskip("fastapi") + +import app.epcis_export as epcis_export +from app.epcis_export import _query_tenant_events + +REQUIRED_SELECT_ALIASES = ( + "reference_document", + "tlc_source_reference", + "immediate_previous_source", + "field_id", + "harvester_name", + "harvester_business_name", + "harvester_business_phone", + "ship_from", + "ship_to", + "ship_from_gln", + "ship_to_gln", +) + + +class _FakeResult: + def __init__(self, rows): + self._rows = rows + + def fetchall(self): + return list(self._rows) + + +class _FakeSession: + def __init__(self): + self.sql = "" + self.params = None + self.closed = False + + def execute(self, stmt, params=None): + self.sql = str(stmt) + self.params = params + return _FakeResult([]) + + def close(self): + self.closed = True + + +def test_query_selects_shipping_and_harvest_kde_aliases(monkeypatch: pytest.MonkeyPatch) -> None: + session = _FakeSession() + monkeypatch.setattr(epcis_export, "get_db_safe", lambda: session) + assert _query_tenant_events(str(uuid4()), None, None, None) == [] + sql = " ".join(session.sql.split()) + for alias in REQUIRED_SELECT_ALIASES: + assert f"AS {alias}" in sql or f"as {alias}" in sql + assert "kde_key = 'reference_document'" in sql + assert "kde_key = 'tlc_source_reference'" in sql + assert "kde_key = 'immediate_previous_source'" in sql + assert "kde_key = 'harvester_name'" in sql + assert "kde_key = 'field_id'" in sql + + +def test_every_tenant_kdes_join_has_tenant_predicate(monkeypatch: pytest.MonkeyPatch) -> None: + session = _FakeSession() + monkeypatch.setattr(epcis_export, "get_db_safe", lambda: session) + _query_tenant_events(str(uuid4()), None, None, None) + sql = session.sql + # Each LEFT JOIN tenant_kdes alias must also filter tenant_id = e.tenant_id + import re + + joins = re.findall(r"LEFT JOIN tenant_kdes (\w+) ON ([^\n]+)", sql) + assert len(joins) >= 17 + for alias, cond in joins: + assert f"{alias}.tenant_id = e.tenant_id" in cond, alias + assert session.closed + + +def test_joined_row_mapping_exposes_new_keys(monkeypatch: pytest.MonkeyPatch) -> None: + class _Row: + def __init__(self, mapping): + self._mapping = mapping + + class _Session: + def execute(self, *_a, **_k): + return _FakeResult( + [ + _Row( + { + "event_type": "shipping", + "traceability_lot_code": "LOT-ROM-2026-0312A", + "reference_document": "BM:BOL-9001", + "tlc_source_reference": "https://id.gs1.org/414/6141410000014", + "immediate_previous_source": None, + "field_id": None, + "harvester_name": None, + "harvester_business_name": None, + "harvester_business_phone": None, + "ship_from_gln": "6141410000014", + "ship_to_gln": "6141410000069", + } + ) + ] + ) + + def close(self): + pass + + monkeypatch.setattr(epcis_export, "get_db_safe", lambda: _Session()) + rows = _query_tenant_events(str(uuid4()), None, None, None) + assert rows[0]["reference_document"] == "BM:BOL-9001" + assert rows[0]["tlc_source_reference"].startswith("https://id.gs1.org/414/") + assert rows[0]["ship_from_gln"] == "6141410000014" diff --git a/services/ingestion/tests/test_second_format_kde_map.py b/services/ingestion/tests/test_second_format_kde_map.py new file mode 100644 index 000000000..1dfc18ce1 --- /dev/null +++ b/services/ingestion/tests/test_second_format_kde_map.py @@ -0,0 +1,187 @@ +"""Second format shipping KDE map cases T1 to T5 (synthetic fixtures only). + +Locks the map in /workspace/rnd/second-format-kde-map.md against +services/ingestion/tests/fixtures. No partner file. SEL-549 hold stays. +""" + +from __future__ import annotations + +import csv +import json +import sys +from pathlib import Path + +import pytest + +_ingestion_dir = Path(__file__).resolve().parent.parent +if str(_ingestion_dir) not in sys.path: + sys.path.insert(0, str(_ingestion_dir)) +_services_dir = _ingestion_dir.parent +if str(_services_dir) not in sys.path: + sys.path.insert(0, str(_services_dir)) + +from app.webhook_models import REQUIRED_KDES_BY_CTE, WebhookCTEType +from shared.fsma_validation import validate_tlc + +FIXTURES = Path(__file__).resolve().parent / "fixtures" +ASN_EXPECTED = FIXTURES / "edi_corpus" / "asn_856.expected.json" +HARVEST_CSV = FIXTURES / "01_harvesting.csv" + +SHIPPING_REQUIRED = list(REQUIRED_KDES_BY_CTE[WebhookCTEType.SHIPPING]) + + +def _harvest_lot() -> str: + with HARVEST_CSV.open(newline="") as fh: + row = next(csv.DictReader(fh)) + return row["traceability_lot_code"] + + +def _asn_to_shipping_kdes(asn: dict, *, lot=None, tlc_source=None, extras=None) -> dict: + """Map asn_856.expected.json shape onto REQUIRED shipping KDEs (map B).""" + item = asn["line_items"][0] + raw_date = asn.get("ship_date_raw") or "" + ship_date = None + if len(raw_date) == 8 and raw_date.isdigit(): + ship_date = f"{raw_date[0:4]}-{raw_date[4:6]}-{raw_date[6:8]}" + kdes = { + "traceability_lot_code": lot if lot is not None else item.get("lot_number"), + "product_description": item.get("product_description"), + "quantity": item.get("quantity"), + "unit_of_measure": item.get("unit_of_measure"), + "ship_date": ship_date, + "ship_from_location": asn.get("ship_from_name"), + "ship_to_location": asn.get("ship_to_name"), + "reference_document": ( + f"BM:{asn['reference_document_number']}" + if asn.get("reference_document_number") + else None + ), + "tlc_source_reference": tlc_source, + } + unmapped = { + "ship_from_gln": asn.get("ship_from_gln"), + "ship_to_gln": asn.get("ship_to_gln"), + "asn_number": asn.get("asn_number"), + "carrier": asn.get("carrier"), + } + if extras: + unmapped.update(extras) + return {"kdes": kdes, "unmapped_fields": unmapped} + + +def _gaps(kdes: dict) -> list[str]: + missing = [] + for key in SHIPPING_REQUIRED: + val = kdes.get(key) + if val is None or val == "": + missing.append(key) + return missing + + +def test_t1_asn_856_expected_gaps() -> None: + """T1: current asn_856 fixture is gap (missing TLC and TLC source).""" + asn = json.loads(ASN_EXPECTED.read_text()) + assert asn["line_items"][0]["lot_number"] is None + mapped = _asn_to_shipping_kdes(asn) + gaps = _gaps(mapped["kdes"]) + assert "traceability_lot_code" in gaps + assert "tlc_source_reference" in gaps + # Present fields from fixture must not be dropped + assert mapped["kdes"]["product_description"] == "ROMAINE-12CT" + assert mapped["kdes"]["quantity"] == 200.0 + assert mapped["kdes"]["reference_document"] == "BM:BOL-9001" + assert mapped["unmapped_fields"]["ship_from_gln"] == "6141410000014" + + +def test_t2_asn_plus_synthetic_lot_and_tlc_source_can_pass() -> None: + """T2: asn_856 + harvest lot + synthetic TLC source fills required nine.""" + asn = json.loads(ASN_EXPECTED.read_text()) + lot = _harvest_lot() + mapped = _asn_to_shipping_kdes( + asn, + lot=lot, + tlc_source="https://id.gs1.org/414/6141410000014", + ) + assert _gaps(mapped["kdes"]) == [] + lenient = validate_tlc(lot, strict=False) + assert lenient.is_valid + strict = validate_tlc(lot, strict=True) + # Harvest style lot: a GS1 pattern mismatch is a warning, not a failure (#3719) + assert strict.is_valid + assert any("production pattern" in w for w in strict.warnings) + + +def test_t3_strict_style_tlc_no_format_warning() -> None: + """T3: production style TLC passes lenient and strict format checks.""" + asn = json.loads(ASN_EXPECTED.read_text()) + lot = "00012345678901-LotA" + mapped = _asn_to_shipping_kdes( + asn, + lot=lot, + tlc_source="https://id.gs1.org/414/6141410000014", + ) + assert _gaps(mapped["kdes"]) == [] + assert validate_tlc(lot, strict=False).is_valid + assert validate_tlc(lot, strict=True).is_valid + + +def test_t4_fda_spreadsheet_shaped_shipping_row() -> None: + """T4: FDA Shipping sheet shaped synthetic row maps to the same nine KDEs.""" + lot = _harvest_lot() + row = { + "Traceability Lot Code (assigned by Initial Packer / First Land Based Receiver / person who Transformed the Food)": lot, + "Quantity": "200", + "Unit of Measure": "cases", + "Product Description - Product Name": "Organic Romaine Hearts", + "Product Description - Brand Name (if applicable)": "Valley Fresh", + "Shipping Date (mm/dd/yyyy)": "03/12/2026", + "Location Description for the Immediate Subsequent Recipient - Business Name": "Metro Distribution Center", + "Location Description for the Location from which you Shipped the Food - Business Name": "Valley Fresh Farms", + "Traceability Lot Code Source Reference (may be provided instead of the TLC Source location description)": "https://id.gs1.org/414/6141410000014", + "Reference Document Type": "BOL", + "Reference Document Number": "BOL-9001", + "PartnerInternalScore": "42", + } + kdes = { + "traceability_lot_code": row[ + "Traceability Lot Code (assigned by Initial Packer / First Land Based Receiver / person who Transformed the Food)" + ], + "product_description": row["Product Description - Product Name"], + "quantity": float(row["Quantity"]), + "unit_of_measure": row["Unit of Measure"], + "ship_date": "2026-03-12", + "ship_from_location": row[ + "Location Description for the Location from which you Shipped the Food - Business Name" + ], + "ship_to_location": row[ + "Location Description for the Immediate Subsequent Recipient - Business Name" + ], + "reference_document": f"{row['Reference Document Type']}:{row['Reference Document Number']}", + "tlc_source_reference": row[ + "Traceability Lot Code Source Reference (may be provided instead of the TLC Source location description)" + ], + } + extras = { + "Product Description - Brand Name (if applicable)": row[ + "Product Description - Brand Name (if applicable)" + ], + "PartnerInternalScore": row["PartnerInternalScore"], + } + assert _gaps(kdes) == [] + assert extras["PartnerInternalScore"] == "42" + assert "Valley Fresh" in extras["Product Description - Brand Name (if applicable)"] + + +def test_t5_unknown_column_preserved() -> None: + """T5: deliberate extra column is flagged and retained, never dropped.""" + asn = json.loads(ASN_EXPECTED.read_text()) + mapped = _asn_to_shipping_kdes( + asn, + lot=_harvest_lot(), + tlc_source="https://id.gs1.org/414/6141410000014", + extras={"PartnerInternalScore": "42", "REF*ZZ*CUSTOM": "KEEP-ME"}, + ) + assert "PartnerInternalScore" in mapped["unmapped_fields"] + assert mapped["unmapped_fields"]["REF*ZZ*CUSTOM"] == "KEEP-ME" + assert mapped["unmapped_fields"]["PartnerInternalScore"] == "42" + assert _gaps(mapped["kdes"]) == []