diff --git a/services/ingestion/app/epcis_export.py b/services/ingestion/app/epcis_export.py index 695f515e3..cc042eaf5 100644 --- a/services/ingestion/app/epcis_export.py +++ b/services/ingestion/app/epcis_export.py @@ -19,6 +19,7 @@ from fastapi.responses import JSONResponse, StreamingResponse from pydantic import BaseModel, Field from shared.database import get_db_safe +from shared.epcis_lot_identity import is_full_class_uri, lot_token_only from shared.fsma_validation import normalize_gtin, verify_gs1_check_digit from shared.tenant_context import TenantContext from sqlalchemy import text @@ -46,6 +47,53 @@ class ExportStorageUnavailable(RuntimeError): """The event store could not be read; never report this as "no data".""" +def _lot_code_filter(lot_code: str | None) -> tuple[str | None, dict]: + """SQL predicate for an export lot filter that reads both TLC shapes (SEL-598). + + Rows may store the bare lot token (pre-H2, or ``LOT_IDENTITY_LOSSLESS`` + off) or the full LGTIN / GS1 Digital Link class URI (flag on). A bare + filter also matches URIs whose lot segment ends with that token; a URI + filter also matches the bare token. ``_row_matches_lot`` then keeps + only rows whose lot token is exactly the requested one. + """ + if not lot_code: + return None, {} + params: dict = {"lot_code": lot_code} + clauses = ["e.traceability_lot_code = :lot_code"] + token = lot_token_only(lot_code) + if token and token != lot_code: + # Full URI requested: also match rows stored with the bare token. + params["lot_token"] = token + clauses.append("e.traceability_lot_code = :lot_token") + elif token: + # Bare lot requested: also match full class URIs carrying it. + suffixes = sorted({f"/10/{token}", f"/10/{quote(token, safe='-._~')}", f".{token}", f".{quote(token, safe='-._~')}"}) + for index, suffix in enumerate(suffixes): + name = f"lot_suffix_{index}" + params[name] = suffix + clauses.append( + "((lower(e.traceability_lot_code) LIKE 'https://id.gs1.org/01/%'" + " OR lower(e.traceability_lot_code) LIKE 'urn:epc:class:lgtin:%')" + f" AND right(e.traceability_lot_code, char_length(:{name})) = :{name})" + ) + return "(" + " OR ".join(clauses) + ")", params + + +def _row_matches_lot(stored: object, lot_code: str) -> bool: + """Exact lot match across bare-token and full-class-URI storage.""" + if not isinstance(stored, str): + return False + if stored == lot_code: + return True + wanted = lot_token_only(lot_code) + if not wanted: + return False + if is_full_class_uri(lot_code): + # A URI filter matches its own URI (above) or a bare historical row. + return stored == wanted + return lot_token_only(stored) == wanted and (stored == wanted or is_full_class_uri(stored)) + + def _query_tenant_events( tenant_id: str, lot_code: str | None, date_from: str | None, date_to: str | None ) -> list[dict]: @@ -59,9 +107,10 @@ def _query_tenant_events( try: where = [] params: dict = {"tenant_id": tenant_id} - if lot_code: - where.append("e.traceability_lot_code = :lot_code") - params["lot_code"] = lot_code + lot_predicate, lot_params = _lot_code_filter(lot_code) + if lot_predicate: + where.append(lot_predicate) + params.update(lot_params) if date_from: where.append("e.event_timestamp >= :date_from") params["date_from"] = date_from @@ -117,12 +166,15 @@ def _query_tenant_events( """), {**params, "row_limit": _MAX_EXPORT_EVENTS + 1}, ).fetchall() - return [dict(row._mapping) for row in rows] + result = [dict(row._mapping) for row in rows] except Exception as exc: logger.error("db_query_failed_for_export", extra={"error": str(exc)}) raise ExportStorageUnavailable(str(exc)) from exc finally: db.close() + if lot_code: + result = [row for row in result if _row_matches_lot(row.get("traceability_lot_code"), lot_code)] + return result def _load_export_rows(tenant_id: str, request: "ExportRequest") -> list[dict]: @@ -332,13 +384,20 @@ def _valid_class_uri(value: object) -> bool: def _epc_class(row: dict) -> str | None: """Use an existing class URI, product_id, or a GTIN Digital Link. + SEL-598 / H2: if ``traceability_lot_code`` (or ``epc_class`` / + ``product_id``) already holds a full LGTIN / Digital Link, emit it + unchanged. Never pass a full class URI as the ``{lot}`` argument when + rebuilding from GTIN + lot token. + GTIN/GLN digits do not reveal the company-prefix split required by EPC - URNs. A lot code alone cannot identify an EPC class — callers must + URNs. A lot code alone cannot identify an EPC class: callers must persist ``epc_class``, ``product_id`` (class URI), ``gtin``, or an embedded ``epcis_document_json`` quantity class. """ - for key in ("epc_class", "product_id"): + for key in ("epc_class", "product_id", "traceability_lot_code", "canonical_epc_class"): value = _kde_scalar(row.get(key)) + # Strict export gate only — do not emit URIs the lenient parser accepts + # but `_valid_class_uri` rejects (Codex P2 / CodeRabbit). if _valid_class_uri(value): return value raw = row.get("epcis_document_json") @@ -354,7 +413,16 @@ def _epc_class(row: dict) -> str | None: except (TypeError, ValueError): pass gtin_raw = _kde_scalar(row.get("gtin")) or _kde_scalar(row.get("product_id")) - lot = _kde_scalar(row.get("traceability_lot_code")) + raw_lot = _kde_scalar(row.get("lot_token")) or _kde_scalar(row.get("traceability_lot_code")) + # Never pass a full URI as the lot argument (SEL-598). + lot = lot_token_only(raw_lot) or ( + raw_lot + if raw_lot + and not is_full_class_uri(raw_lot) + and "://" not in raw_lot + and not str(raw_lot).lower().startswith("urn:epc:") + else None + ) digits = re.sub(r"\D", "", gtin_raw) if digits and lot: try: diff --git a/services/ingestion/tests/test_epcis_export_1342.py b/services/ingestion/tests/test_epcis_export_1342.py index 9437fffac..588adefc5 100644 --- a/services/ingestion/tests/test_epcis_export_1342.py +++ b/services/ingestion/tests/test_epcis_export_1342.py @@ -117,8 +117,12 @@ def test_optional_filters_preserve_tenant_scope( sql, params = session.calls[0][1] expected_where = "WHERE e.tenant_id = :tenant_id" expected_params = {"tenant_id": tenant_id, "row_limit": epcis_export._MAX_EXPORT_EVENTS + 1} + if lot_code is not None: + # SEL-598 dual-read: bare and full-class-URI TLCs both match. + lot_predicate, lot_params = epcis_export._lot_code_filter(lot_code) + expected_where += f" AND {lot_predicate}" + expected_params.update(lot_params) for name, value, predicate in ( - ("lot_code", lot_code, "e.traceability_lot_code = :lot_code"), ("date_from", date_from, "e.event_timestamp >= :date_from"), ("date_to", date_to, "e.event_timestamp < :date_to"), ): diff --git a/services/shared/canonical_event.py b/services/shared/canonical_event.py index 193a8bca7..a2083cf8d 100644 --- a/services/shared/canonical_event.py +++ b/services/shared/canonical_event.py @@ -32,6 +32,12 @@ from uuid import UUID, uuid4 from pydantic import BaseModel, Field, field_validator +from shared.epcis_lot_identity import ( + lossless_enabled, + lot_token_only, + normalize_epcis_lot, + parse_class_uri, +) logger = logging.getLogger("canonical-event") @@ -1038,8 +1044,7 @@ def _resolve_bizstep_to_cte_type(biz_step: Any) -> CTEType: raw = "" if biz_step is None else str(biz_step).strip() if not raw: raise UnmappedBizStepError( - "EPCIS event has no bizStep — cannot derive an FSMA 204 CTE type. " - "Supply a CBV or FSMA bizStep URI. (#2815)" + "EPCIS event has no bizStep — cannot derive an FSMA 204 CTE type. Supply a CBV or FSMA bizStep URI. (#2815)" ) lowered = raw.lower() @@ -1097,8 +1102,13 @@ def _epcis_list(epcis_data: Dict[str, Any], key: str) -> List[Any]: return list(value) if isinstance(value, list) else [] -def _epcis_traceability_lot_code(epcis_data: Dict[str, Any]) -> Tuple[str, Optional[str]]: - """Resolve ``(traceability_lot_code, gtin)`` for an EPCIS event. +def _legacy_epcis_traceability_lot_code(epcis_data: Dict[str, Any]) -> Tuple[str, Optional[str]]: + """Pre-SEL-598 resolver, used while ``LOT_IDENTITY_LOSSLESS`` is off. + + Kept byte-for-byte so flag-off writes keep the same TLC, GTIN KDE, event + hash and idempotency key as before H2. + + Resolve ``(traceability_lot_code, gtin)`` for an EPCIS event. Precedence: ILMD ``fsma:traceabilityLotCode`` / ``cbvmda:lotNumber`` (the same order as ``epcis/extraction._extract_lot_data``), then the lot of an @@ -1148,7 +1158,7 @@ def _epcis_traceability_lot_code(epcis_data: Dict[str, Any]) -> Tuple[str, Optio ) -def _epcis_transformation_inputs(epcis_data: Dict[str, Any]) -> List[Dict[str, Any]]: +def _legacy_epcis_transformation_inputs(epcis_data: Dict[str, Any]) -> List[Dict[str, Any]]: """Per-input lot, quantity and unit for a TransformationEvent. ``inputQuantityList`` carries class-level (LGTIN) inputs with an explicit @@ -1187,6 +1197,157 @@ def _epcis_transformation_inputs(epcis_data: Dict[str, Any]) -> List[Dict[str, A return inputs +def _epcis_traceability_lot_code(epcis_data: Dict[str, Any]) -> Tuple[str, Optional[str]]: + """Resolve ``(traceability_lot_code, gtin)`` for an EPCIS event. + + SEL-598 / H2: when ``LOT_IDENTITY_LOSSLESS`` is enabled (opt-in), the + stored ``traceability_lot_code`` is the full LGTIN or GS1 Digital Link + class URI when present. The bare lot token remains available via + ``lot_token_only`` / ``normalize_epcis_lot(...).lot_token``. + + Precedence (H2 / SEL-598, intentional vs legacy extraction): a full class + URI among quantity classes / EPCs first (canonical stored identity), then + ILMD ``fsma:traceabilityLotCode`` / ``cbvmda:lotNumber`` as the bare partner + TLC when no class URI is present (preserve bare TLC — do not rebuild a + Digital Link from an SGTIN-derived GTIN at this boundary), then SGTIN + serial, then flat top-level keys. Legacy ``epcis/extraction`` prefers ILMD + first; canonical + export agree on class-URI-first after H2. Raises + ``ValueError`` when nothing identifies the lot. + + With the flag off (default) this is the pre-H2 resolver unchanged: the + stored TLC, GTIN KDE, event hash and idempotency key do not move. + """ + if not lossless_enabled(): + return _legacy_epcis_traceability_lot_code(epcis_data) + ilmd = epcis_data.get("ilmd") + if not isinstance(ilmd, dict): + ilmd = {} + epcs = _epcis_list(epcis_data, "epcList") + _epcis_list(epcis_data, "outputEPCList") + extension = epcis_data.get("extension") + extension = extension if isinstance(extension, dict) else {} + classes = [ + q.get("epcClass") + for q in ( + _epcis_list(epcis_data, "outputQuantityList") + + _epcis_list(epcis_data, "quantityList") + + (list(extension.get("quantityList") or []) if isinstance(extension.get("quantityList"), list) else []) + ) + if isinstance(q, dict) and q.get("epcClass") + ] + # Real GTIN-14 only (never the EPC company.item fragment from _epc_gtin). + gtin = None + for e in list(classes) + list(epcs): + parsed = parse_class_uri(e) + if parsed and parsed.gtin: + gtin = parsed.gtin + break + identity = normalize_epcis_lot(e, prefer_full=False) + if identity and identity.gtin: + gtin = identity.gtin + break + + # Prefer full class URI from EPC / quantity class (SEL-598 canonical). + for epc in list(classes) + list(epcs): + parsed = parse_class_uri(epc) + if parsed: + if lossless_enabled(): + return parsed.canonical_id, parsed.gtin or gtin + return parsed.lot_token, parsed.gtin or gtin + + for key in ("fsma:traceabilityLotCode", "traceabilityLotCode", "cbvmda:lotNumber", "lotNumber"): + value = ilmd.get(key) + if value is not None and str(value).strip(): + # Preserve bare partner TLC; do not rebuild Digital Link here. + identity = normalize_epcis_lot(value) + if identity: + return identity.canonical_id, identity.gtin or gtin + return str(value).strip(), gtin + + for epc in epcs: + if "sgtin" in str(epc).lower(): + identity = normalize_epcis_lot(epc) + if identity: + return identity.canonical_id, identity.gtin or gtin + segment = _epc_last_segment(epc) + if segment: + return segment, gtin + + for key in ("traceability_lot_code", "lotNumber"): + value = epcis_data.get(key) + if value is not None and str(value).strip(): + # Preserve bare flat TLC (same boundary rule as ILMD). + identity = normalize_epcis_lot(value) + if identity: + return identity.canonical_id, identity.gtin or gtin + return str(value).strip(), gtin + + raise ValueError( + "EPCIS event carries no lot identifier (ILMD traceabilityLotCode or lotNumber, an LGTIN or SGTIN " + "EPC, or traceability_lot_code); refusing to fabricate one" + ) + + +def _epcis_transformation_inputs(epcis_data: Dict[str, Any]) -> List[Dict[str, Any]]: + """Per-input lot, quantity and unit for a TransformationEvent. + + ``inputQuantityList`` carries class-level (LGTIN) inputs with an explicit + quantity; a quantityElement without ``uom`` is a count of instances in + EPCIS, recorded as ``each``. ``inputEPCList`` carries instance EPCs, each + of which is exactly one item. Lots are not merged or summed. + """ + if not lossless_enabled(): + return _legacy_epcis_transformation_inputs(epcis_data) + inputs: List[Dict[str, Any]] = [] + seen: set = set() + for element in _epcis_list(epcis_data, "inputQuantityList"): + if not isinstance(element, dict): + continue + epc_class = element.get("epcClass") + identity = normalize_epcis_lot(epc_class) + if identity is not None: + lot_key = identity.canonical_id if lossless_enabled() else identity.lot_token + lot_token = identity.lot_token + else: + lot_token = _epc_last_segment(epc_class) + lot_key = lot_token + dedup_key = ("class", lot_key) + if not lot_token or dedup_key in seen: + continue + raw_quantity = element.get("quantity") + try: + quantity = float(raw_quantity) if raw_quantity is not None else None + except (TypeError, ValueError): + quantity = None + inputs.append( + { + "traceability_lot_code": lot_key, + "lot_token": lot_token, + "quantity": quantity, + "unit_of_measure": (element.get("uom") or "each") if quantity is not None else None, + "epc_class": epc_class, + } + ) + seen.add(dedup_key) + for epc in _epcis_list(epcis_data, "inputEPCList"): + identity = normalize_epcis_lot(epc) + segment = identity.lot_token if identity else _epc_last_segment(epc) + dedup_key = ("epc", segment) + if not segment or dedup_key in seen: + continue + lot_key = identity.canonical_id if (identity and lossless_enabled()) else segment + inputs.append( + { + "traceability_lot_code": lot_key, + "lot_token": segment, + "quantity": 1.0, + "unit_of_measure": "each", + "epc": str(epc), + } + ) + seen.add(dedup_key) + return inputs + + def normalize_epcis_event( epcis_data: Dict[str, Any], tenant_id: str, diff --git a/services/shared/epcis_lot_identity.py b/services/shared/epcis_lot_identity.py new file mode 100644 index 000000000..f26d0feb5 --- /dev/null +++ b/services/shared/epcis_lot_identity.py @@ -0,0 +1,254 @@ +"""SEL-598 lossless EPCIS lot identity (H2). + +Canonical stored identity is the full LGTIN or GS1 Digital Link (AI 01 + 10). +The bare lot token is derived for UI, gap reports, and export rebuild paths. +Round-trip acceptance is exact equality on the canonical full ID. + +Feature flag: ``LOT_IDENTITY_LOSSLESS`` (default disabled). Set to ``1`` / +``true`` to store the full class URI as ``traceability_lot_code``. + +Keep it off until a dual-key or migration path exists for the event hash and +idempotency key. Both include the TLC, so switching a lot from its bare token +to the full URI changes them, and re-ingesting events stored before the switch +would create duplicates. With the flag off, canonical writes are the same as +before H2. +""" + +from __future__ import annotations + +import os +import re +from dataclasses import dataclass +from typing import Any, Optional +from urllib.parse import unquote + +_LGTIN_RE = re.compile(r"^urn:epc:class:lgtin:(\d+)\.(\d+)\.(.+)$", re.IGNORECASE) +_SGTIN_RE = re.compile(r"^urn:epc:id:sgtin:(\d+)\.(\d+)\.(.+)$", re.IGNORECASE) +_DL_CLASS_RE = re.compile( + r"^https://id\.gs1\.org/01/(\d{8,14})/10/([^/?#]+)$", + re.IGNORECASE, +) + + +def lossless_enabled() -> bool: + raw = (os.environ.get("LOT_IDENTITY_LOSSLESS") or "0").strip().lower() + return raw in {"1", "true", "yes", "on"} + + +@dataclass(frozen=True) +class LotIdentity: + """Normalized lot identity for ingest, canonical store, and export.""" + + canonical_id: str + lot_token: str + gtin: Optional[str] + is_full_class_uri: bool + + +def _company_item_to_gtin14(company: str, item: str) -> Optional[str]: + """Rebuild GTIN-14 from EPC Pure Identity company.item (SGTIN/LGTIN). + + Company prefix + item reference must sum to 13 digits. The first digit of + the item reference is the Indicator; the remaining item digits follow the + company prefix; then append the GS1 check digit. + Example: ``0614141.123452`` → ``10614141234520`` (not ``00614141123452``). + """ + if not company.isdigit() or not item.isdigit() or not item: + return None + if len(company) + len(item) != 13: + return None + indicator = item[0] + item_without_indicator = item[1:] + digits13 = indicator + company + item_without_indicator + if len(digits13) != 13: + return None + return digits13 + _gs1_check_digit(digits13) + + +def _gs1_check_digit(digits13: str) -> str: + total = 0 + for i, ch in enumerate(digits13): + n = int(ch) + total += n * (3 if (i % 2 == 0) else 1) + return str((10 - (total % 10)) % 10) + + +def normalize_gtin14(raw: str) -> Optional[str]: + digits = re.sub(r"\D", "", raw or "") + if not digits: + return None + if len(digits) < 8 or len(digits) > 14: + return None + digits = digits.zfill(14) + body, check = digits[:-1], digits[-1] + if _gs1_check_digit(body) != check: + # Still return zero-padded digits for dual-read rebuild; caller may validate + return digits + return digits + + +def parse_class_uri(value: object) -> Optional[LotIdentity]: + """Parse a full LGTIN or Digital Link class URI into LotIdentity.""" + if not isinstance(value, str): + return None + text = value.strip() + if not text or "0000000.000000" in text: + return None + + m = _LGTIN_RE.match(text) + if m: + company, item, lot_raw = m.group(1), m.group(2), m.group(3) + lot = unquote(lot_raw) + if not lot: + return None + gtin = _company_item_to_gtin14(company, item) + return LotIdentity( + canonical_id=text, + lot_token=lot, + gtin=gtin, + is_full_class_uri=True, + ) + + m = _DL_CLASS_RE.match(text) + if m: + gtin_raw, lot_raw = m.group(1), m.group(2) + lot = unquote(lot_raw).strip() + if not lot: + return None + gtin = normalize_gtin14(gtin_raw) or gtin_raw.zfill(14) + return LotIdentity( + canonical_id=text, + lot_token=lot, + gtin=gtin, + is_full_class_uri=True, + ) + + return None + + +def is_full_class_uri(value: object) -> bool: + parsed = parse_class_uri(value) + return bool(parsed and parsed.is_full_class_uri) + + +def lot_token_only(value: object) -> Optional[str]: + """Derived bare lot segment; never treat a full class URI as the token.""" + parsed = parse_class_uri(value) + if parsed: + return parsed.lot_token + if not isinstance(value, str): + return None + text = value.strip() + if not text: + return None + lowered = text.lower() + # Legacy LGTIN/SGTIN segment helper, only for EPC URNs the strict class + # parse rejected. A bare lot that merely contains "lgtin"/"sgtin" is a + # lot token, not a URI, and must not be dropped. + if lowered.startswith("urn:epc:") and ("lgtin" in lowered or "sgtin" in lowered): + parts = text.rsplit(".", 1) + if len(parts) == 2 and parts[1]: + return unquote(parts[1]) + return None + # Reject values that look like URIs but failed class parse + if "://" in text or lowered.startswith("urn:epc:"): + return None + return text + + +def normalize_epcis_lot( + value: object, + *, + gtin: Optional[str] = None, + prefer_full: Optional[bool] = None, +) -> Optional[LotIdentity]: + """Normalize any lot-ish value into LotIdentity. + + When ``prefer_full`` is None, uses ``lossless_enabled()``. + Dual-read: bare lot + GTIN rebuilds a Digital Link canonical ID. + """ + use_full = lossless_enabled() if prefer_full is None else prefer_full + + parsed = parse_class_uri(value) + if parsed: + if use_full: + return parsed + return LotIdentity( + canonical_id=parsed.lot_token, + lot_token=parsed.lot_token, + gtin=parsed.gtin or (normalize_gtin14(gtin) if gtin else None), + is_full_class_uri=False, + ) + + # SGTIN is instance-level; expose serial as token but not a class URI + if isinstance(value, str): + sm = _SGTIN_RE.match(value.strip()) + if sm: + serial = unquote(sm.group(3)) + g = _company_item_to_gtin14(sm.group(1), sm.group(2)) + return LotIdentity( + canonical_id=serial if use_full else serial, + lot_token=serial, + gtin=g or (normalize_gtin14(gtin) if gtin else None), + is_full_class_uri=False, + ) + + token = lot_token_only(value) + if not token: + return None + + gtin14 = normalize_gtin14(gtin) if gtin else None + if use_full and gtin14: + # Rebuild Digital Link; do not invent GTIN when absent + from urllib.parse import quote + + canonical = f"https://id.gs1.org/01/{gtin14}/10/{quote(token, safe='-._~')}" + return LotIdentity( + canonical_id=canonical, + lot_token=token, + gtin=gtin14, + is_full_class_uri=True, + ) + + return LotIdentity( + canonical_id=token, + lot_token=token, + gtin=gtin14, + is_full_class_uri=False, + ) + + +def resolve_stored_lot( + *, + epc_or_class: Any = None, + ilmd_tlc: Any = None, + flat_tlc: Any = None, + gtin: Optional[str] = None, +) -> LotIdentity: + """Resolve identity with SEL-598 precedence for canonical write. + + Prefer a full class URI from EPC/quantity class, then ILMD TLC when it is + already a full URI, then rebuild from GTIN + derived token, else bare token. + """ + for candidate in (epc_or_class, ilmd_tlc, flat_tlc): + parsed = parse_class_uri(candidate) + if parsed and lossless_enabled(): + return parsed + if parsed and not lossless_enabled(): + return LotIdentity( + canonical_id=parsed.lot_token, + lot_token=parsed.lot_token, + gtin=parsed.gtin or (normalize_gtin14(gtin) if gtin else None), + is_full_class_uri=False, + ) + + # ILMD / flat may carry bare lot while EPC carries the class URI (handled above) + for candidate in (ilmd_tlc, flat_tlc, epc_or_class): + identity = normalize_epcis_lot(candidate, gtin=gtin) + if identity: + return identity + + raise ValueError( + "EPCIS event carries no lot identifier (ILMD traceabilityLotCode or lotNumber, " + "an LGTIN or SGTIN EPC, or traceability_lot_code); refusing to fabricate one" + ) diff --git a/tests/shared/test_epcis_lot_identity_sel598.py b/tests/shared/test_epcis_lot_identity_sel598.py new file mode 100644 index 000000000..1add71773 --- /dev/null +++ b/tests/shared/test_epcis_lot_identity_sel598.py @@ -0,0 +1,284 @@ +"""SEL-598 / H2: lossless EPCIS lot identity (structural + synthetic URIs). + +Synthetic identifiers only (A-2165). No partner data. No compliance claims. +""" + +from __future__ import annotations + +import os + +import pytest +from shared.canonical_event import ( + _epcis_traceability_lot_code, + _epcis_transformation_inputs, + _legacy_epcis_traceability_lot_code, + _legacy_epcis_transformation_inputs, +) +from shared.epcis_lot_identity import ( + is_full_class_uri, + lossless_enabled, + lot_token_only, + normalize_epcis_lot, + parse_class_uri, +) + +# Synthetic GS1-shaped IDs (not partner data). GTIN 00614141123452 check digit +# is illustrative; normalize_epcis_lot rebuilds Digital Links from digits. +SYN_GTIN14 = "00614141123452" +# EPC Pure Identity 0614141.123452 → indicator 1 + check digit +SYN_GTIN_FROM_LGTIN = "10614141234520" +SYN_LOT = "LOT-ROM-2026-0312A" +SYN_LGTIN = f"urn:epc:class:lgtin:0614141.123452.{SYN_LOT}" +SYN_DL = f"https://id.gs1.org/01/{SYN_GTIN14}/10/{SYN_LOT}" +SYN_LGTIN_OTHER_GTIN = f"urn:epc:class:lgtin:0614141.999990.{SYN_LOT}" + + + +@pytest.fixture(autouse=True) +def _lossless_on(monkeypatch): + """The H2 behaviour under test is opt-in; tests that need it off say so.""" + monkeypatch.setenv("LOT_IDENTITY_LOSSLESS", "1") + + +def test_lgtin_canonical_not_truncated_on_parse(): + parsed = parse_class_uri(SYN_LGTIN) + assert parsed is not None + assert parsed.is_full_class_uri is True + assert parsed.canonical_id == SYN_LGTIN + assert parsed.lot_token == SYN_LOT + assert parsed.canonical_id != SYN_LOT + + +def test_digital_link_canonical_not_truncated_on_parse(): + parsed = parse_class_uri(SYN_DL) + assert parsed is not None + assert parsed.canonical_id == SYN_DL + assert parsed.lot_token == SYN_LOT + + +def test_full_uri_not_passed_as_lot_arg_to_epc_class(): + """lot_token_only must refuse to treat a class URI as the lot segment.""" + assert lot_token_only(SYN_DL) == SYN_LOT + assert lot_token_only(SYN_LGTIN) == SYN_LOT + assert is_full_class_uri(SYN_DL) + # Rebuild path: normalize with GTIN + accidental full URI still yields token + identity = normalize_epcis_lot(SYN_DL, gtin=SYN_GTIN14) + assert identity is not None + assert identity.lot_token == SYN_LOT + assert "://" not in identity.lot_token + assert not identity.lot_token.lower().startswith("urn:epc:") + + +def test_same_lot_different_gtin_are_distinct(): + a = parse_class_uri(SYN_LGTIN) + b = parse_class_uri(SYN_LGTIN_OTHER_GTIN) + assert a is not None and b is not None + assert a.lot_token == b.lot_token == SYN_LOT + assert a.canonical_id != b.canonical_id + + +def test_both_pass_fixture_round_trip_full_id(): + """Synthetic both-pass: full Digital Link survives normalize → token → rebuild.""" + first = normalize_epcis_lot(SYN_DL) + assert first is not None + assert first.canonical_id == SYN_DL + rebuilt = normalize_epcis_lot(first.lot_token, gtin=first.gtin or SYN_GTIN14) + assert rebuilt is not None + assert rebuilt.is_full_class_uri is True + # Exact equality on full Digital Link form + assert rebuilt.canonical_id == SYN_DL + + +def test_lenient_only_fixture_preserves_raw_tlc_and_gtin(): + """Synthetic lenient-only: bare TLC + GTIN rebuilds without inventing GTIN.""" + bare = normalize_epcis_lot(SYN_LOT, gtin=SYN_GTIN14) + assert bare is not None + assert bare.lot_token == SYN_LOT + assert bare.gtin is not None + assert SYN_LOT in bare.canonical_id + # Without GTIN, do not invent a class URI + no_gtin = normalize_epcis_lot(SYN_LOT, gtin=None) + assert no_gtin is not None + assert no_gtin.canonical_id == SYN_LOT + assert no_gtin.is_full_class_uri is False + + +def test_canonical_path_preserves_full_lgtin_when_lossless(monkeypatch): + monkeypatch.setenv("LOT_IDENTITY_LOSSLESS", "1") + assert lossless_enabled() is True + event = { + "quantityList": [{"epcClass": SYN_LGTIN, "quantity": 1, "uom": "EA"}], + "ilmd": {}, + } + tlc, gtin = _epcis_traceability_lot_code(event) + assert tlc == SYN_LGTIN + assert tlc != SYN_LOT + + +def test_canonical_path_segment_only_when_flag_off(monkeypatch): + monkeypatch.setenv("LOT_IDENTITY_LOSSLESS", "0") + assert lossless_enabled() is False + event = { + "quantityList": [{"epcClass": SYN_LGTIN, "quantity": 1, "uom": "EA"}], + "ilmd": {}, + } + tlc, _gtin = _epcis_traceability_lot_code(event) + assert tlc == SYN_LOT + + +def test_transformation_inputs_keep_full_class_uri(monkeypatch): + monkeypatch.setenv("LOT_IDENTITY_LOSSLESS", "1") + event = { + "inputQuantityList": [ + {"epcClass": SYN_LGTIN, "quantity": 2.5, "uom": "KGM"}, + ], + } + inputs = _epcis_transformation_inputs(event) + assert len(inputs) == 1 + assert inputs[0]["traceability_lot_code"] == SYN_LGTIN + assert inputs[0]["lot_token"] == SYN_LOT + assert inputs[0]["epc_class"] == SYN_LGTIN + + +def test_export_epc_class_emits_stored_full_uri_unchanged(): + from ingestion.app.epcis_export import _epc_class + + row = { + "traceability_lot_code": SYN_DL, + "gtin": SYN_GTIN14, + } + assert _epc_class(row) == SYN_DL + + +def test_export_epc_class_does_not_embed_full_uri_as_lot(): + from ingestion.app.epcis_export import _epc_class + + # Stored TLC accidentally holds a full URI while epc_class is empty; + # rebuild must not produce .../10/https%3A%2F%2F... + row = { + "traceability_lot_code": SYN_DL, + "gtin": SYN_GTIN14, + "epc_class": None, + "product_id": None, + } + out = _epc_class(row) + assert out == SYN_DL + assert out is not None + assert "%3A" not in out + assert out.count("https://") == 1 + + +def test_lgtin_company_item_rebuilds_indicator_and_check_digit(): + parsed = parse_class_uri(SYN_LGTIN) + assert parsed is not None + assert parsed.gtin == SYN_GTIN_FROM_LGTIN + # Must not use the naive 0+company+item concatenation + assert parsed.gtin != "00614141123452" + + +def test_bare_ilmd_tlc_preserved_with_sgtin_epc(monkeypatch): + """SGTIN + bare ILMD lot must not fabricate a Digital Link TLC.""" + monkeypatch.setenv("LOT_IDENTITY_LOSSLESS", "1") + sgtin = "urn:epc:id:sgtin:0614141.123452.SERIAL9" + event = { + "epcList": [sgtin], + "ilmd": {"cbvmda:lotNumber": SYN_LOT}, + } + tlc, gtin = _epcis_traceability_lot_code(event) + assert tlc == SYN_LOT + assert "://" not in tlc + assert gtin == SYN_GTIN_FROM_LGTIN + + +def test_transformation_inputs_same_lot_different_gtin_kept(monkeypatch): + monkeypatch.setenv("LOT_IDENTITY_LOSSLESS", "1") + event = { + "inputQuantityList": [ + {"epcClass": SYN_LGTIN, "quantity": 1, "uom": "EA"}, + {"epcClass": SYN_LGTIN_OTHER_GTIN, "quantity": 2, "uom": "EA"}, + ], + } + inputs = _epcis_transformation_inputs(event) + assert len(inputs) == 2 + codes = {i["traceability_lot_code"] for i in inputs} + assert codes == {SYN_LGTIN, SYN_LGTIN_OTHER_GTIN} + + +def test_export_rejects_invalid_class_uri_from_lenient_parse(): + from ingestion.app.epcis_export import _epc_class + + # Bad company+item length — lenient parser may accept; export must not emit. + bad = "urn:epc:class:lgtin:1.2.LOT1" + row = {"traceability_lot_code": bad, "gtin": None, "epc_class": None, "product_id": None} + assert _epc_class(row) is None + + # Digital Link with invalid check digit + bad_dl = "https://id.gs1.org/01/09506000134351/10/LOT1" + row2 = {"traceability_lot_code": bad_dl, "gtin": None, "epc_class": None, "product_id": None} + assert _epc_class(row2) is None + + +def test_lossless_is_opt_in_by_default(monkeypatch): + monkeypatch.delenv("LOT_IDENTITY_LOSSLESS", raising=False) + assert lossless_enabled() is False + monkeypatch.setenv("LOT_IDENTITY_LOSSLESS", "true") + assert lossless_enabled() is True + + +@pytest.mark.parametrize( + "event", + [ + {"quantityList": [{"epcClass": SYN_LGTIN, "quantity": 1, "uom": "EA"}], "ilmd": {}}, + {"epcList": ["urn:epc:id:sgtin:0614141.123452.SERIAL9"], "ilmd": {"cbvmda:lotNumber": SYN_LOT}}, + {"quantityList": [{"epcClass": SYN_LGTIN_OTHER_GTIN, "quantity": 1}], "ilmd": {"fsma:traceabilityLotCode": "TLC-ILMD-1"}}, + {"traceability_lot_code": " FLAT-LOT-9 "}, + ], +) +def test_flag_off_resolver_is_pre_h2_unchanged(monkeypatch, event): + """Flag off must keep TLC and GTIN KDE (and so hash/idempotency) as before H2.""" + monkeypatch.setenv("LOT_IDENTITY_LOSSLESS", "0") + assert _epcis_traceability_lot_code(event) == _legacy_epcis_traceability_lot_code(event) + + +def test_flag_off_transformation_inputs_are_pre_h2_unchanged(monkeypatch): + monkeypatch.setenv("LOT_IDENTITY_LOSSLESS", "0") + event = { + "inputQuantityList": [ + {"epcClass": SYN_LGTIN, "quantity": 1, "uom": "EA"}, + {"epcClass": SYN_LGTIN_OTHER_GTIN, "quantity": 2, "uom": "EA"}, + ], + "inputEPCList": ["urn:epc:id:sgtin:0614141.123452.SERIAL9"], + } + assert _epcis_transformation_inputs(event) == _legacy_epcis_transformation_inputs(event) + + +@pytest.mark.parametrize("bare", ["SGTINLOT-7", "PLGTIN.A", "lot-lgtin"]) +def test_lot_token_only_keeps_bare_lot_containing_gtin_words(bare): + assert lot_token_only(bare) == bare + + +def test_export_lot_filter_bare_lot_reads_both_shapes(): + from ingestion.app.epcis_export import _lot_code_filter, _row_matches_lot + + predicate, params = _lot_code_filter(SYN_LOT) + assert "e.traceability_lot_code = :lot_code" in predicate + assert f"/10/{SYN_LOT}" in params.values() + assert f".{SYN_LOT}" in params.values() + assert _row_matches_lot(SYN_LOT, SYN_LOT) + assert _row_matches_lot(SYN_DL, SYN_LOT) + assert _row_matches_lot(SYN_LGTIN, SYN_LOT) + # Same suffix, different lot: excluded after the coarse SQL match. + assert not _row_matches_lot(f"urn:epc:class:lgtin:0614141.123452.X.{SYN_LOT}", SYN_LOT) + assert not _row_matches_lot("OTHER-LOT", SYN_LOT) + + +def test_export_lot_filter_full_uri_also_reads_bare_rows(): + from ingestion.app.epcis_export import _lot_code_filter, _row_matches_lot + + predicate, params = _lot_code_filter(SYN_DL) + assert params == {"lot_code": SYN_DL, "lot_token": SYN_LOT} + assert "e.traceability_lot_code = :lot_token" in predicate + assert _row_matches_lot(SYN_DL, SYN_DL) + assert _row_matches_lot(SYN_LOT, SYN_DL) + # A different GTIN's URI with the same lot is not the requested class. + assert not _row_matches_lot(SYN_LGTIN_OTHER_GTIN, SYN_DL)