diff --git a/services/ingestion/app/epcis_export.py b/services/ingestion/app/epcis_export.py index 695f515e3..d512c67d8 100644 --- a/services/ingestion/app/epcis_export.py +++ b/services/ingestion/app/epcis_export.py @@ -8,6 +8,8 @@ from __future__ import annotations +import csv +import io import json import logging import re @@ -31,7 +33,8 @@ FDA_CSV_HEADER = ( "CTE_Type,Traceability_Lot_Code,Product_Description,Quantity,Unit_of_Measure," "Event_Date,Event_Time,Ship_From_Location,Ship_From_GLN,Ship_To_Location," - "Ship_To_GLN,Carrier,Temperature_C,SHA256_Hash" + "Ship_To_GLN,Carrier,Temperature_C,Reference_Document,TLC_Source_Reference," + "Immediate_Previous_Source,SHA256_Hash,Additional_KDEs_JSON" ) @@ -715,38 +718,53 @@ async def export_fda( rows = _load_export_rows(tenant_id, request) data_source = "tenant" if rows else "empty" - if rows: - # Build CSV from real data - lines = [FDA_CSV_HEADER] - for r in rows: - ts = r["event_timestamp"] - date_str = ts.strftime("%Y-%m-%d") if ts else "" - time_str = ts.strftime("%H:%M:%SZ") if ts else "" - lines.append( - ",".join( - [ - str(r["event_type"] or "").upper(), - str(r["traceability_lot_code"] or ""), - sanitize_cell(str(r["product_description"] or "")).replace(",", ";"), - str(r["quantity"] or ""), - str(r["unit_of_measure"] or ""), - date_str, - time_str, - sanitize_cell(str(r["ship_from"] or r["location_name"] or "")).replace(",", ";"), - str(r["ship_from_gln"] or r["location_gln"] or ""), - sanitize_cell(str(r["ship_to"] or "")).replace(",", ";"), - str(r["ship_to_gln"] or ""), - sanitize_cell(str(r["carrier"] or "")).replace(",", ";"), - str(r["temperature_c"] or ""), - str(r["sha256_hash"] or ""), - ] - ) + # Use csv.writer so fields with commas (esp. Additional_KDEs_JSON) stay + # valid JSON / text; the writer quotes cells as needed (Hermes #3722). + buf = io.StringIO() + writer = csv.writer(buf, lineterminator="\n") + writer.writerow(FDA_CSV_HEADER.split(",")) + for r in rows: + ts = r["event_timestamp"] + date_str = ts.strftime("%Y-%m-%d") if ts else "" + time_str = ts.strftime("%H:%M:%SZ") if ts else "" + additional = { + key: r.get(key) + for key in ( + "field_id", + "harvester_name", + "harvester_business_name", + "harvester_business_phone", ) - csv_content = "\n".join(lines) + "\n" - event_count = len(rows) - else: - csv_content = FDA_CSV_HEADER + "\n" - event_count = 0 + if r.get(key) + } + # Emit valid JSON unchanged — do not replace commas; csv.writer quotes. + additional_json = ( + sanitize_cell(json.dumps(additional, sort_keys=True)) if additional else "" + ) + writer.writerow( + [ + str(r.get("event_type") or "").upper(), + str(r.get("traceability_lot_code") or ""), + sanitize_cell(str(r.get("product_description") or "")), + str(r.get("quantity") or ""), + str(r.get("unit_of_measure") or ""), + date_str, + time_str, + sanitize_cell(str(r.get("ship_from") or r.get("location_name") or "")), + str(r.get("ship_from_gln") or r.get("location_gln") or ""), + sanitize_cell(str(r.get("ship_to") or "")), + str(r.get("ship_to_gln") or ""), + sanitize_cell(str(r.get("carrier") or "")), + str(r.get("temperature_c") or ""), + sanitize_cell(str(r.get("reference_document") or "")), + sanitize_cell(str(r.get("tlc_source_reference") or "")), + sanitize_cell(str(r.get("immediate_previous_source") or "")), + str(r.get("sha256_hash") or ""), + additional_json, + ] + ) + csv_content = buf.getvalue() + event_count = len(rows) # EPIC-L (#1655): route filename through the shared safe_filename # builder so tenant_id can't carry CRLF / quotes into the header value. diff --git a/services/ingestion/tests/test_epcis_export_fda_csv_c3.py b/services/ingestion/tests/test_epcis_export_fda_csv_c3.py new file mode 100644 index 000000000..a4070b3ed --- /dev/null +++ b/services/ingestion/tests/test_epcis_export_fda_csv_c3.py @@ -0,0 +1,187 @@ +"""C3: ingestion FDA CSV aligned with compliance sortable KDE columns. + +Synthetic fixtures only. Uses harvest CSV lot strings and asn_856 GLNs. +""" + +from __future__ import annotations + +import csv +import json +import sys +from datetime import datetime, timezone +from pathlib import Path +from uuid import uuid4 + +import pytest +from fastapi import FastAPI, Request +from fastapi.testclient import TestClient + +_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 FDA_CSV_HEADER, router as epcis_export_router +from app.webhook_compat import _verify_api_key, require_tenant_context +from shared.tenant_context import TenantContext + +FIXTURES = Path(__file__).resolve().parent / "fixtures" +HARVEST_CSV = FIXTURES / "01_harvesting.csv" +ASN_EXPECTED = FIXTURES / "edi_corpus" / "asn_856.expected.json" + +REQUIRED_NEW = ( + "Reference_Document", + "TLC_Source_Reference", + "Immediate_Previous_Source", + "Additional_KDEs_JSON", +) + + +@pytest.fixture() +def client() -> TestClient: + app = FastAPI() + app.include_router(epcis_export_router) + + async def _tenant_ctx(request: Request) -> TenantContext: + tenant_id = None + if request.method in {"POST", "PUT", "PATCH"}: + try: + body = await request.json() + except Exception: + body = {} + if isinstance(body, dict): + tenant_id = body.get("tenant_id") + return TenantContext( + tenant_id=tenant_id or str(uuid4()), + principal_kind="test", + principal_id=uuid4().hex, + ) + + app.dependency_overrides[_verify_api_key] = lambda: None + app.dependency_overrides[require_tenant_context] = _tenant_ctx + return TestClient(app) + + +def test_fda_header_includes_compliance_kde_columns() -> None: + cols = FDA_CSV_HEADER.split(",") + for name in REQUIRED_NEW: + assert name in cols + # SHA256 still present; new cols sit before it or adjacent + assert "SHA256_Hash" in cols + assert cols.index("Reference_Document") < cols.index("SHA256_Hash") + + +def test_fda_export_row_fills_reference_and_tlc_source( + client: TestClient, monkeypatch: pytest.MonkeyPatch +) -> None: + with HARVEST_CSV.open(newline="") as fh: + harvest = next(csv.DictReader(fh)) + asn = json.loads(ASN_EXPECTED.read_text()) + lot = harvest["traceability_lot_code"] + + def _rows(*_a, **_k): + return [ + { + "event_type": "shipping", + "traceability_lot_code": lot, + "product_description": asn["line_items"][0]["product_description"], + "quantity": asn["line_items"][0]["quantity"], + "unit_of_measure": asn["line_items"][0]["unit_of_measure"], + "event_timestamp": datetime(2026, 3, 10, 12, 0, tzinfo=timezone.utc), + "location_gln": None, + "location_name": None, + "ship_from": asn["ship_from_name"], + "ship_from_gln": asn["ship_from_gln"], + "ship_to": asn["ship_to_name"], + "ship_to_gln": asn["ship_to_gln"], + "carrier": "CarrierCo", + "temperature_c": None, + "reference_document": f"BM:{asn['reference_document_number']}", + "tlc_source_reference": "https://id.gs1.org/414/6141410000014", + "immediate_previous_source": "", + "field_id": harvest.get("field_id"), + "harvester_name": harvest.get("harvester_name"), + "sha256_hash": "abc", + } + ] + + monkeypatch.setattr(epcis_export, "_load_export_rows", _rows) + response = client.post("/api/v1/export/fda", json={"tenant_id": str(uuid4())}) + assert response.status_code == 200 + lines = [ln for ln in response.text.splitlines() if ln] + assert lines[0] == FDA_CSV_HEADER + reader = csv.DictReader(lines) + row = next(reader) + assert row["Traceability_Lot_Code"] == lot + assert row["Reference_Document"] == "BM:BOL-9001" + assert "414/6141410000014" in row["TLC_Source_Reference"] + assert "FIELD-SVF-7" in row["Additional_KDEs_JSON"] + assert row["Ship_From_GLN"] == "6141410000014" + + +def test_fda_empty_export_header_only( + client: TestClient, monkeypatch: pytest.MonkeyPatch +) -> None: + monkeypatch.setattr(epcis_export, "_load_export_rows", lambda *_a, **_k: []) + response = client.post("/api/v1/export/fda", json={"tenant_id": str(uuid4())}) + assert response.status_code == 200 + assert [ln for ln in response.text.splitlines() if ln] == [FDA_CSV_HEADER] + +def test_fda_additional_kdes_json_csv_round_trip( + client: TestClient, monkeypatch: pytest.MonkeyPatch +) -> None: + """Additional_KDEs_JSON with 2+ keys must stay valid JSON through CSV quoting.""" + lot = "LOT-ROM-2026-0312A" + expected_kdes = { + "field_id": "FIELD-SVF-7", + "harvester_business_name": "Salinas Valley Farms LLC", + "harvester_business_phone": "+1-831-555-0100", + "harvester_name": "Salinas Valley Farms LLC", + } + + def _rows(*_a, **_k): + return [ + { + "event_type": "harvesting", + "traceability_lot_code": lot, + "product_description": "Organic Romaine Hearts", + "quantity": 5000, + "unit_of_measure": "lbs", + "event_timestamp": datetime(2026, 3, 12, 8, 0, tzinfo=timezone.utc), + "location_gln": "0061414100014", + "location_name": "Salinas Valley Farms", + "ship_from": "Salinas Valley Farms", + "ship_from_gln": "0061414100014", + "ship_to": "", + "ship_to_gln": "", + "carrier": "", + "temperature_c": None, + "reference_document": "HARV-SVF-2026-0312-001", + "tlc_source_reference": "", + "immediate_previous_source": "", + "field_id": expected_kdes["field_id"], + "harvester_name": expected_kdes["harvester_name"], + "harvester_business_name": expected_kdes["harvester_business_name"], + "harvester_business_phone": expected_kdes["harvester_business_phone"], + "sha256_hash": "abc", + } + ] + + monkeypatch.setattr(epcis_export, "_load_export_rows", _rows) + response = client.post("/api/v1/export/fda", json={"tenant_id": str(uuid4())}) + assert response.status_code == 200 + + # Write CSV body, read back with the csv module (not naive split). + reader = csv.DictReader(response.text.splitlines()) + row = next(reader) + cell = row["Additional_KDEs_JSON"] + assert cell.count(",") >= 1 # multi-key JSON keeps commas + parsed = json.loads(cell) + assert len(parsed) >= 2 + assert parsed == expected_kdes +