diff --git a/services/ingestion/app/epcis_export.py b/services/ingestion/app/epcis_export.py index 88bf999b0..695f515e3 100644 --- a/services/ingestion/app/epcis_export.py +++ b/services/ingestion/app/epcis_export.py @@ -1,9 +1,9 @@ """ EPCIS Export Router. -Generates GS1 EPCIS 2.0 JSON-LD exports for retailer compliance -(Walmart, Kroger, Costco). Also provides FDA 21 CFR 1.1455 sortable -spreadsheet export format. +Generates GS1 EPCIS 2.0 JSON-LD exports (Preview) and FDA 21 CFR 1.1455 +sortable spreadsheet export. EPCIS output is labeled Preview; validate with +your trading partner before relying on it for compliance. """ from __future__ import annotations @@ -152,7 +152,7 @@ def _load_export_rows(tenant_id: str, request: "ExportRequest") -> list[dict]: "shipping": "urn:epcglobal:cbv:bizstep:shipping", "transformation": "urn:epcglobal:cbv:bizstep:commissioning", "initial_packing": "urn:epcglobal:cbv:bizstep:packing", - "harvesting": "urn:epcglobal:cbv:bizstep:collecting", + "harvesting": "urn:epcglobal:cbv:bizstep:creating_class_instance", "cooling": "urn:epcglobal:cbv:bizstep:storing", "first_land_based_receiving": "urn:epcglobal:cbv:bizstep:receiving", } @@ -483,9 +483,10 @@ class ExportRequest(BaseModel): "/epcis", summary="Export in GS1 EPCIS 2.0 JSON-LD format", description=( - "Generates a GS1 EPCIS 2.0 compliant JSON-LD document containing all " - "traceability events. Compatible with Walmart, Kroger, and Costco portals. " - "Stored event hashes are included without cryptographic verification." + "Generates a GS1 EPCIS 2.0 JSON-LD document containing traceability " + "events (Preview). Validate with your trading partner before relying " + "on it for compliance. Stored event hashes are included without " + "cryptographic verification." ), ) async def export_epcis( @@ -623,6 +624,8 @@ def _skip(row: dict, reason: str) -> None: "valid", "errors", "checkedAt", + "preview", + "previewNotice", ) }, }, @@ -645,6 +648,11 @@ def _skip(row: dict, reason: str) -> None: # Loading stored hashes does not verify event content or chain integrity. "integrityVerified": False, "chainHashVerified": False, + "preview": True, + "previewNotice": ( + "EPCIS export is in Preview. Validate with your trading partner " + "before relying on it for compliance." + ), **({"regengine:message": "No data yet"} if not rows else {}), }, } @@ -680,6 +688,7 @@ def _skip(row: dict, reason: str) -> None: "X-RegEngine-Skipped-Events-Count": str(skipped), "X-RegEngine-Integrity-Verified": "false", "X-RegEngine-Data-Source": data_source, + "X-RegEngine-Preview": "true", } if validate: response_headers["X-RegEngine-Schema-Valid"] = "true" if not validation_errors else "false" @@ -756,6 +765,7 @@ async def export_fda( "X-RegEngine-Events-Count": str(event_count), "X-RegEngine-Format": "FDA_21CFR1.1455", "X-RegEngine-Data-Source": data_source, + "X-RegEngine-Preview": "true", } if data_source == "empty": fda_headers["X-RegEngine-Message"] = "No data yet" @@ -779,15 +789,17 @@ async def list_export_formats(): { "id": "epcis", "name": "GS1 EPCIS 2.0 JSON-LD", - "description": "Industry standard for supply chain event data. Compatible with Walmart, Kroger, Costco.", + "description": "EPCIS 2.0 JSON-LD export (Preview).", "content_type": "application/ld+json", - "retailers": ["Walmart", "Kroger", "Costco", "Target"], + "status": "preview", + "retailers": [], }, { "id": "fda", "name": "FDA Sortable Spreadsheet (CSV)", - "description": "Sortable electronic format required by 21 CFR 1.1455 for FDA records requests.", + "description": "Sortable electronic format for FDA records requests (Preview). Align with 21 CFR 1.1455 after partner validation.", "content_type": "text/csv", + "status": "preview", "retailers": ["FDA"], }, ] diff --git a/services/ingestion/tests/test_epcis_export_gs1_2.py b/services/ingestion/tests/test_epcis_export_gs1_2.py index 874d9de6b..790bc3833 100644 --- a/services/ingestion/tests/test_epcis_export_gs1_2.py +++ b/services/ingestion/tests/test_epcis_export_gs1_2.py @@ -34,6 +34,45 @@ def class_uri(): return f"urn:epc:class:lgtin:{digits[:7]}.{digits[7:]}.{uuid4().hex}" +def _independent_gs1_mod10_check_digit(data_digits: str) -> int: + """GS1 General Specifications mod-10, written out for the test. + + From the rightmost data digit: weights alternate 3, 1, 3, 1, ... + Check digit is (10 - (sum % 10)) % 10. Independent of + shared.fsma_validation / exporter._gs1_check_digit_valid. + """ + if not data_digits.isdigit(): + raise ValueError("data digits must be numeric") + total = 0 + for i, ch in enumerate(reversed(data_digits)): + total += (3 if i % 2 == 0 else 1) * int(ch) + return (10 - (total % 10)) % 10 + + +def _gs1_with_check(data_digits: str) -> str: + """Append check digit from the independent mod-10 above.""" + return data_digits + str(_independent_gs1_mod10_check_digit(data_digits)) + + +def generated_gtin14() -> str: + base = str(uuid4().int % (10**13)).zfill(13) + return _gs1_with_check(base) + + +def generated_gln13() -> str: + base = str(uuid4().int % (10**12)).zfill(12) + return _gs1_with_check(base) + + +def generated_sscc18() -> str: + base = str(uuid4().int % (10**17)).zfill(17) + return _gs1_with_check(base) + + +def generated_lot() -> str: + return f"SYN-{uuid4().hex[:12]}" + + def assert_conformant(document): jsonschema = pytest.importorskip("jsonschema") jsonschema.Draft7Validator( @@ -84,6 +123,7 @@ def test_empty_export_has_no_fabricated_events(): def test_stored_event_export_preserves_metadata_namespace(): + """Namespace + @context terms without pyld (pyld is not in requirements.lock).""" event = { "type": "ObjectEvent", "eventTime": datetime.now(timezone.utc).isoformat(), @@ -101,44 +141,77 @@ def test_stored_event_export_preserves_metadata_namespace(): assert_conformant(document) assert "regengine:metadata" in document assert "metadata" not in document - jsonld = pytest.importorskip("pyld").jsonld - context = fixture("epcis-context.jsonld") - dropped = [] + assert document["@context"][0] == CONTEXT_URL + assert isinstance(document["@context"][1], dict) + assert document["@context"][1].get("regengine") == "https://regengine.co/ns/" + assert "regengine:source" in document["regengine:metadata"] + assert "regengine:event_count" in document["regengine:metadata"] - def loader(url, options=None): - if url != CONTEXT_URL: - raise ValueError(f"Unexpected context URL: {url}") - return {"contextUrl": None, "documentUrl": url, "document": context} - expanded = jsonld.expand(document, options={"documentLoader": loader}, on_property_dropped=dropped.append) - assert not dropped - assert "https://regengine.co/ns/metadata" in expanded[0] +def test_gs1_check_digit_independent_mod10_and_published_examples(): + """Independent mod-10 + known published vectors; production must agree. + Published examples are the independent guard (not minted via production). + Generated IDs are minted with the test-local mod-10, then checked by the + exporter so a reversed weight in production would fail. + """ + # Published GS1 examples (data + expected check digit), each cited from a + # GS1 document on ref.gs1.org / gs1.org — no third-party pages. + published = ( + # GTIN-14 09506000134352 — GS1 General Specifications Release 26.0 + # §1.2.4.4 (JSON-LD example gs1:gtin); also QR Code powered by GS1 + # best practices (2023), Best practice 1. + ("0950600013435", "2", 14), + # GLN-13 0614141000005 — GS1 Verifiable Credentials White Paper on + # Data Model and Validations 1.0.1 §5.1.2 (partyGLN for GS1 US). + ("061414100000", "5", 13), + # SSCC-18 376104250021234569 — GS1 General Specifications Release 26.0 + # §7.9.1 Table 7-9 (18-digit check digit calculation example). + ("37610425002123456", "9", 18), + ) + for data, expected_cd, length in published: + assert _independent_gs1_mod10_check_digit(data) == int(expected_cd) + full = data + expected_cd + assert len(full) == length + assert exporter._gs1_check_digit_valid(full) + wrong_cd = "0" if expected_cd != "0" else "1" + bad = data + wrong_cd + assert _independent_gs1_mod10_check_digit(data) != int(wrong_cd) + assert not exporter._gs1_check_digit_valid(bad) -def test_gs1_check_digit_weights_match_known_valid_examples(): - """Rightmost data digit must receive weight 3 (GS1 mod-10).""" - # GTIN-14, GLN-13, and EAN-13 textbook examples - assert exporter._gs1_check_digit_valid("09506000134352") - assert exporter._gs1_check_digit_valid("0614141000005") - assert exporter._gs1_check_digit_valid("4006381333931") - # SSCC-18: base 10614141123456789 → check digit 7 - assert exporter._gs1_check_digit_valid("106141411234567897") - assert not exporter._gs1_check_digit_valid("09506000134351") + # Generated structure + production agreement with independent mint. + gtin = generated_gtin14() + gln = generated_gln13() + sscc = generated_sscc18() + assert len(gtin) == 14 and gtin.isdigit() + assert len(gln) == 13 and gln.isdigit() + assert len(sscc) == 18 and sscc.isdigit() + assert _independent_gs1_mod10_check_digit(gtin[:-1]) == int(gtin[-1]) + assert _independent_gs1_mod10_check_digit(gln[:-1]) == int(gln[-1]) + assert _independent_gs1_mod10_check_digit(sscc[:-1]) == int(sscc[-1]) + assert exporter._gs1_check_digit_valid(gtin) + assert exporter._gs1_check_digit_valid(gln) + assert exporter._gs1_check_digit_valid(sscc) + bad_gtin = gtin[:-1] + ("0" if gtin[-1] != "0" else "1") + assert not exporter._gs1_check_digit_valid(bad_gtin) def test_export_builds_class_from_query_gtin_and_lot(): """Regression: query returns gtin (not a test-only epc_class column).""" + gtin = generated_gtin14() + gln = generated_gln13() + lot = generated_lot() row = { "event_type": "receiving", "epcis_event_type": "ObjectEvent", "epcis_action": "ADD", "epcis_biz_step": "receiving", "event_timestamp": datetime.now(timezone.utc), - "gtin": "09506000134352", - "traceability_lot_code": "SYN-LOT-001", + "gtin": gtin, + "traceability_lot_code": lot, "quantity": 10, "unit_of_measure": "lbs", - "location_gln": "0614141000005", + "location_gln": gln, } ctx = type("Tenant", (), {"tenant_id": str(uuid4())})() with patch.object(exporter, "_query_tenant_events", return_value=[row]): @@ -147,10 +220,10 @@ def test_export_builds_class_from_query_gtin_and_lot(): ) document = json.loads(response.body) event = document["epcisBody"]["eventList"][0] - assert event["quantityList"][0]["epcClass"] == "https://id.gs1.org/01/09506000134352/10/SYN-LOT-001" + assert event["quantityList"][0]["epcClass"] == f"https://id.gs1.org/01/{gtin}/10/{lot}" assert event["quantityList"][0]["uom"] == "LBR" assert event["regengine:unitOfMeasure"] == "lbs" - assert event["readPoint"]["id"] == "https://id.gs1.org/414/0614141000005" + assert event["readPoint"]["id"] == f"https://id.gs1.org/414/{gln}" assert document["regengine:exportMetadata"]["skippedEventsCount"] == 0 assert "regengine:message" not in document["regengine:exportMetadata"] @@ -164,7 +237,7 @@ def test_export_builds_class_from_product_id_kde(): "epcis_biz_step": None, "event_timestamp": datetime.now(timezone.utc), "product_id": epc_class, - "traceability_lot_code": "SYN-LOT-002", + "traceability_lot_code": generated_lot(), "quantity": None, "unit_of_measure": None, "location_gln": None, @@ -188,7 +261,7 @@ def test_transformation_cte_emitted_as_transformation_event(): "epcis_biz_step": "commissioning", "event_timestamp": datetime.now(timezone.utc), "epc_class": epc_class, - "traceability_lot_code": "OUT-LOT", + "traceability_lot_code": generated_lot(), "quantity": 5, "unit_of_measure": "kg", "sha256_hash": "a" * 64, @@ -211,8 +284,8 @@ def test_transformation_cte_emitted_as_transformation_event(): assert document["regengine:exportMetadata"]["skippedEventsCount"] == 0 -def test_harvesting_maps_to_collecting_bizstep(): - """Phase A keeps harvesting → collecting.""" +def test_harvesting_maps_to_creating_class_instance_bizstep(): + """Harvesting commissions a class instance (CBV creating_class_instance).""" epc_class = class_uri() row = { "event_type": "harvesting", @@ -221,7 +294,7 @@ def test_harvesting_maps_to_collecting_bizstep(): "epcis_biz_step": None, "event_timestamp": datetime.now(timezone.utc), "epc_class": epc_class, - "traceability_lot_code": "HARVEST-1", + "traceability_lot_code": generated_lot(), "quantity": 1, "unit_of_measure": "EA", "location_gln": None, @@ -232,4 +305,4 @@ def test_harvesting_maps_to_collecting_bizstep(): exporter.export_epcis(exporter.ExportRequest(tenant_id=ctx.tenant_id), validate=False, ctx=ctx) ) document = json.loads(response.body) - assert document["epcisBody"]["eventList"][0]["bizStep"] == "collecting" + assert document["epcisBody"]["eventList"][0]["bizStep"] == "creating_class_instance" diff --git a/services/ingestion/tests/test_epcis_export_preview.py b/services/ingestion/tests/test_epcis_export_preview.py new file mode 100644 index 000000000..9194f6e5a --- /dev/null +++ b/services/ingestion/tests/test_epcis_export_preview.py @@ -0,0 +1,180 @@ +"""H5 Preview labels for EPCIS export (spec section 6 / section 7 test 10). + +Synthetic fixtures only. No partner data. EPCIS stays Preview. +""" + +from __future__ import annotations + +import json +import sys +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 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" +PREVIEW_NOTICE = ( + "EPCIS export is in Preview. Validate with your trading partner " + "before relying on it for compliance." +) +NAMED_RETAILERS = ("Walmart", "Kroger", "Costco", "Target") + + +def _independent_gs1_mod10_check_digit(data_digits: str) -> int: + total = 0 + for i, ch in enumerate(reversed(data_digits)): + total += (3 if i % 2 == 0 else 1) * int(ch) + return (10 - (total % 10)) % 10 + + +def generated_gtin14() -> str: + base = str(uuid4().int % (10**13)).zfill(13) + return base + str(_independent_gs1_mod10_check_digit(base)) + + + +@pytest.fixture() +def client() -> TestClient: + app = FastAPI() + app.include_router(epcis_export_router) + + async def _tenant_ctx(request: Request) -> TenantContext: + tenant_id = request.path_params.get("tenant_id") + if not tenant_id and request.method in {"POST", "PUT", "PATCH"}: + try: + body = await request.json() + except Exception: + body = {} + if isinstance(body, dict): + tenant_id = body.get("tenant_id") or body.get("tenantId") + 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 _empty_rows(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setattr(epcis_export, "_load_export_rows", lambda *_a, **_k: []) + + +class TestPreviewHeaderAndMetadata: + """Spec section 7 test 10: Preview header and metadata on every response.""" + + def test_preview_header_on_empty_export( + self, client: TestClient, monkeypatch: pytest.MonkeyPatch + ) -> None: + _empty_rows(monkeypatch) + response = client.post( + "/api/v1/export/epcis", + json={"tenant_id": str(uuid4())}, + ) + assert response.status_code == 200 + assert response.headers.get("X-RegEngine-Preview") == "true" + meta = response.json()["regengine:exportMetadata"] + assert meta["preview"] is True + assert meta["previewNotice"] == PREVIEW_NOTICE + assert meta["eventsCount"] == 0 + + def test_preview_header_on_success( + self, client: TestClient, monkeypatch: pytest.MonkeyPatch + ) -> None: + # Synthetic harvest lot string only (fixture path under tests/fixtures). + assert HARVEST_CSV.is_file() + lot = HARVEST_CSV.read_text().splitlines()[1].split(",")[0] + assert lot.startswith("LOT-") + + gtin = generated_gtin14() + + def _one_row(*_a: object, **_k: object) -> list[dict]: + return [ + { + "event_type": "harvesting", + "epc_class": f"https://id.gs1.org/01/{gtin}/10/" + lot, + "traceability_lot_code": lot, + "quantity": 5000, + "unit_of_measure": "lbs", + "event_timestamp": "2026-03-12T12:00:00+00:00", + "location_gln": "0061414100014", # from fixtures/01_harvesting.csv + "sha256_hash": None, + "epcis_event_type": "ObjectEvent", + "epcis_action": "ADD", + "epcis_biz_step": None, + } + ] + + monkeypatch.setattr(epcis_export, "_load_export_rows", _one_row) + response = client.post( + "/api/v1/export/epcis", + json={"tenant_id": str(uuid4())}, + ) + assert response.status_code == 200 + assert response.headers.get("X-RegEngine-Preview") == "true" + body = response.json() + meta = body["regengine:exportMetadata"] + assert meta["preview"] is True + assert meta["previewNotice"] == PREVIEW_NOTICE + assert meta["eventsCount"] >= 1 + # preview keys registered in @context + ctx = body["@context"] + assert isinstance(ctx, list) and len(ctx) >= 2 + assert ctx[1].get("preview") == "regengine:preview" + assert ctx[1].get("previewNotice") == "regengine:previewNotice" + + def test_preview_header_on_fda_empty( + self, client: TestClient, monkeypatch: pytest.MonkeyPatch + ) -> None: + _empty_rows(monkeypatch) + response = client.post( + "/api/v1/export/fda", + json={"tenant_id": str(uuid4())}, + ) + assert response.status_code == 200 + assert response.headers.get("X-RegEngine-Preview") == "true" + + +class TestFormatsPreviewStatus: + def test_formats_epcis_status_preview(self, client: TestClient) -> None: + response = client.get("/api/v1/export/formats") + assert response.status_code == 200 + formats = {f["id"]: f for f in response.json()["formats"]} + epcis = formats["epcis"] + assert epcis["status"] == "preview" + assert "Preview" in epcis["description"] + assert epcis["retailers"] == [] + + def test_formats_fda_status_preview(self, client: TestClient) -> None: + response = client.get("/api/v1/export/formats") + assert response.status_code == 200 + formats = {f["id"]: f for f in response.json()["formats"]} + fda = formats["fda"] + assert fda["status"] == "preview" + assert "Preview" in fda["description"] + + def test_formats_has_no_named_retailers(self, client: TestClient) -> None: + response = client.get("/api/v1/export/formats") + blob = json.dumps(response.json()) + for name in NAMED_RETAILERS: + assert name not in blob