diff --git a/docs/architecture/RULES_ENFORCEMENT.md b/docs/architecture/RULES_ENFORCEMENT.md index cd88f2ac5..7a03b97e3 100644 --- a/docs/architecture/RULES_ENFORCEMENT.md +++ b/docs/architecture/RULES_ENFORCEMENT.md @@ -30,6 +30,8 @@ Set via `RULES_ENGINE_ENFORCE` (read at call time, no restart needed — flip + | `cte_only` | Reject when `summary.critical_failures` is non-empty. Warnings/info still accept. | | `all` | Reject on any `summary.compliant is False` (including warning-only failures). | +`ingest_events` passes `deferred_kdes` for EDI shipments stored with an open data gap. Under `all`, a warning that fails only on those KDEs does not reject. The verdict stays `fail`, and any other failure on the event still rejects. + Two invariants hold in every mode: 1. **No-verdict never rejects.** When `summary.compliant is None` (no rules seeded for the tenant, non-FTL product, evaluator skipped), the event accepts — "we didn't evaluate" is not the same as "the event violates a rule." diff --git a/services/ingestion/app/edi_ingestion/routes.py b/services/ingestion/app/edi_ingestion/routes.py index ce2aa63ba..c18cf4173 100644 --- a/services/ingestion/app/edi_ingestion/routes.py +++ b/services/ingestion/app/edi_ingestion/routes.py @@ -29,7 +29,7 @@ import logging import os import uuid -from collections.abc import Callable +from collections.abc import Callable, Iterable from typing import Any, Optional from fastapi import APIRouter, Depends, File, Form, Header, HTTPException, Query, UploadFile @@ -73,10 +73,16 @@ _valid_gln_or_none, _verify_partner_id, ) -from .validation import _validate_edi_as_fsma_event +from .validation import _validate_edi_as_fsma_event, _warehouse_line_tlc logger = logging.getLogger("edi-ingestion") +# X12 has no element for the TLC source reference (21 CFR 1.1340(a)(7)), so a +# shipment arrives without it unless the uploader sends the form field. Store +# the shipment with an open data gap rather than lose the rest of the record. +# Receiving keeps the strict gate, as its rule for this KDE is critical. +_EDI_DEFERRED_KDES = {WebhookCTEType.SHIPPING: frozenset({"tlc_source_reference"})} + def _record_edi_rejection(**kwargs): """Resolve the live rejection-log module before writing. @@ -93,6 +99,70 @@ def _record_edi_rejection(**kwargs): return rejection_log.record_edi_rejection(**kwargs) +def _reject_shared_tlc_source_for_distinct_lots( + *, + tlc_source_reference: str | None, + transaction_sets: Iterable[tuple[str, dict[str, Any]]], + traceability_lot_code: str, + sender_tenant_id: str, + partner_id: str | None, + source: str, +) -> None: + """Refuse one TLC source reference across several shipped lots. + + The form field is a single value copied to every event. On a document with + several lots it could link a lot to a source that never assigned its code, + which is worse than the open gap the lot gets without the field. Receiving + has no open gap to fall back on, as the field is required there, so + refusing it would leave a multi-lot receipt no way in. + """ + if not (tlc_source_reference and tlc_source_reference.strip()): + return + shipping_sets = [ + (transaction_set, extracted) + for transaction_set, extracted in transaction_sets + if _CTE_TYPE_BY_SET.get(transaction_set, WebhookCTEType.SHIPPING) == WebhookCTEType.SHIPPING + ] + lots: set[str] = set() + for transaction_set, extracted in shipping_sets: + for line in extracted.get("line_items") or [{}]: + lot = line.get("lot_number") + lot = lot.strip() if isinstance(lot, str) else "" + if not lot: + lots.add(traceability_lot_code) + elif transaction_set == "945": + lots.add(_warehouse_line_tlc(lot, line)) + else: + lots.add(lot) + if len(lots) <= 1: + return + error = "edi_tlc_source_reference_with_multiple_lots" + message = ( + "tlc_source_reference applies to one lot, but this document has several. " + "Send each lot separately, or resubmit without the field." + ) + first_set, first_extracted = shipping_sets[0] + record = _record_edi_rejection( + tenant_id=sender_tenant_id, + transaction_set=first_set, + traceability_lot_code=traceability_lot_code, + errors=[{"loc": ["tlc_source_reference"], "type": error, "msg": message}], + extracted=first_extracted, + partner_id=partner_id, + source=source, + reason=error, + ) + raise HTTPException( + status_code=422, + detail={ + "error": error, + "message": message, + "lot_count": len(lots), + "rejection_id": record["rejection_id"], + }, + ) + + # ISA13 claims taken by ``_enforce_envelope_integrity`` during the current # request, as ``release_interchange`` arguments. ``_release_isa13_unless_stored`` # installs the list. @@ -576,6 +646,13 @@ 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=( + "TLC source reference (where the TLC was assigned), copied to each event's KDEs. " + "X12 cannot carry it: a shipment without it is stored with an open data gap." + ), + ), 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"), @@ -670,19 +747,25 @@ async def ingest_edi_856( # Event-date rejections run after the allowlist and before the ISA13 # is reserved, so a corrected resend is not refused as a 409 replay. - _enforce_envelope_integrity( - extracted, - sender_tenant_id, - "856", - before_reserve=lambda: _preflight_event_date( + def _preflight() -> None: + _preflight_event_date( transaction_set="856", extracted=extracted, sender_tenant_id=sender_tenant_id, traceability_lot_code=traceability_lot_code, partner_id=x_partner_id, source=source, - ), - ) + ) + _reject_shared_tlc_source_for_distinct_lots( + tlc_source_reference=tlc_source_reference, + transaction_sets=[("856", extracted)], + traceability_lot_code=traceability_lot_code, + sender_tenant_id=sender_tenant_id, + partner_id=x_partner_id, + source=source, + ) + + _enforce_envelope_integrity(extracted, sender_tenant_id, "856", before_reserve=_preflight) lines = extracted.get("line_items") or [] if len(lines) > 1: @@ -756,6 +839,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", @@ -796,6 +882,7 @@ async def ingest_edi_856( payload, principal_tenant_id=sender_tenant_id, x_regengine_api_key=x_regengine_api_key, + deferred_kdes=_EDI_DEFERRED_KDES, ) return EDIIngestResponse( @@ -940,6 +1027,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, } @@ -1053,18 +1144,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, @@ -1099,7 +1191,11 @@ 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=( + "TLC source reference, copied to every event's KDEs. X12 cannot carry it: " + "a shipment without it is stored with an open data gap." + ), ), tenant_id: Optional[str] = Form(None, description="Optional tenant override"), source: str = Form("edi_inbound", description="Source system identifier"), @@ -1233,6 +1329,14 @@ def _preflight_all_sets() -> None: partner_id=x_partner_id, source=source, ) + _reject_shared_tlc_source_for_distinct_lots( + tlc_source_reference=tlc_source_reference, + transaction_sets=list(zip(slice_sets, extractions)), + traceability_lot_code=traceability_lot_code, + sender_tenant_id=sender_tenant_id, + partner_id=x_partner_id, + source=source, + ) # Event-date rejections run after the allowlist and before the ISA13 # is reserved, so a corrected resend is not refused as a 409 replay. @@ -1243,12 +1347,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 @@ -1272,7 +1377,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. @@ -1510,6 +1616,7 @@ def _summarize(slice_set: str, slice_extracted: dict[str, Any]) -> dict[str, Any payload, principal_tenant_id=sender_tenant_id, x_regengine_api_key=x_regengine_api_key, + deferred_kdes=_EDI_DEFERRED_KDES, ) return EDIIngestResponse( diff --git a/services/ingestion/app/edi_ingestion/validation.py b/services/ingestion/app/edi_ingestion/validation.py index c2a848235..198b6ef9c 100644 --- a/services/ingestion/app/edi_ingestion/validation.py +++ b/services/ingestion/app/edi_ingestion/validation.py @@ -23,6 +23,32 @@ class _WarehouseLineRequiredKDEs(BaseModel): quantity: float = Field(gt=0) +def _warehouse_line_tlc(lot: str, line: dict[str, Any]) -> str: + """The TLC stored for a warehouse line with a non-blank lot. + + A lot that isn't production-shaped gets the line's GTIN or UPC as a prefix, + so one short lot on two items is two TLCs. + """ + from shared.schemas import _TLC_PRODUCTION_RE, FSMAEvent + + tlc = lot + # Pattern misses are advisory, so check explicitly to retain the + # warehouse GTIN/UPC rebuild for both accepted and short lots. + try: + tlc = FSMAEvent.validate_tlc_format(tlc) + rebuild = not _TLC_PRODUCTION_RE.fullmatch(tlc) + except ValueError: + rebuild = True + if rebuild: + for item_id, lengths in ( + (line.get("gtin"), (8, 12, 13, 14)), + (line.get("upc"), (12, 13)), + ): + if isinstance(item_id, str) and item_id.isascii() and item_id.isdigit() and len(item_id) in lengths: + return f"{item_id.zfill(14)}-{tlc}" + return tlc + + def _validate_edi_as_fsma_event( extracted: dict[str, Any], transaction_set: str, @@ -39,7 +65,7 @@ def _validate_edi_as_fsma_event( forced every caller to treat ``None`` as "maybe-failed, maybe- unavailable" and masked schema errors in the audit trail. """ - from shared.schemas import _TLC_PRODUCTION_RE, FSMAEvent, FSMAEventType + from shared.schemas import FSMAEvent, FSMAEventType type_map = { "856": FSMAEventType.SHIPPING, @@ -62,21 +88,7 @@ def _validate_edi_as_fsma_event( # TLC that FSMAEvent would accept, collapsing blank lots onto one id. if not tlc: _WarehouseLineRequiredKDEs(lot_number=tlc, quantity=line.quantity) - # Pattern misses are advisory, so check explicitly to retain the - # warehouse GTIN/UPC rebuild for both accepted and short lots. - try: - tlc = FSMAEvent.validate_tlc_format(tlc) - rebuild = not _TLC_PRODUCTION_RE.fullmatch(tlc) - except ValueError: - rebuild = True - if rebuild: - for item_id, lengths in ( - (extracted.get("gtin"), (8, 12, 13, 14)), - (extracted.get("upc"), (12, 13)), - ): - if isinstance(item_id, str) and item_id.isascii() and item_id.isdigit() and len(item_id) in lengths: - tlc = f"{item_id.zfill(14)}-{tlc}" - break + tlc = _warehouse_line_tlc(tlc, extracted) elif transaction_set in ("856", "861"): # The line lot is the TLC verbatim. Do not prefix a GTIN (#1104). # A line with no lot keeps the caller-supplied form field. diff --git a/services/ingestion/app/webhook_compat.py b/services/ingestion/app/webhook_compat.py index 74ced5e92..131105545 100644 --- a/services/ingestion/app/webhook_compat.py +++ b/services/ingestion/app/webhook_compat.py @@ -20,7 +20,7 @@ import logging import uuid from dataclasses import dataclass, field -from typing import Any, Dict, List, Optional, Union +from typing import Any, Dict, List, Mapping, Optional, Union from fastapi import Depends, Header, HTTPException, Request from shared.api_key_store import APIKeyResponse @@ -29,14 +29,16 @@ ValidationOutcome, gap_from_rule_reason, gaps_from_kde_errors, + open_gaps_for_stored_record, record_outcomes, ) from shared.database import get_db_session as _get_db_session from shared.permissions import has_permission from shared.tenant_context import TenantContext, resolve_tenant_context +from sqlalchemy.exc import SQLAlchemyError from .authz import IngestionPrincipal, get_ingestion_principal -from .webhook_models import EventResult, IngestEvent, IngestResponse, WebhookPayload +from .webhook_models import EventResult, IngestEvent, IngestResponse, WebhookCTEType, WebhookPayload from .webhook_router_v2 import ( _check_rate_limit, _generate_alerts, @@ -237,6 +239,7 @@ async def ingest_events( x_regengine_api_key: Optional[str] = None, principal: Optional[IngestionPrincipal] = None, source_file: Optional[SourceFileReceipt] = None, + deferred_kdes: Optional[Mapping[WebhookCTEType, frozenset[str]]] = None, ) -> IngestResponse: """Internal ingestion helper — not an HTTP route. @@ -256,6 +259,12 @@ async def ingest_events( row is processed, retains the file verbatim, stamps ``ingestion_run_id`` on every canonical event, and closes the run with counts covering every row of the file. The run id is returned on the response. + + ``deferred_kdes`` names, per CTE, required KDEs the source format cannot + carry. An event missing only those is stored, with an open data gap and + ``open_gaps`` on its result, instead of being rejected. Only server-side + callers set this. Never derive it from the payload, which the sender + controls: that would let a caller switch off the KDE gate. """ if not principal_tenant_id: raise PermissionError("ingest_events requires principal_tenant_id from authenticated context") @@ -307,6 +316,7 @@ async def ingest_events( seen_in_batch: set[str] = set() # Data gap time to fix: written after the ingest commit, fail-open. gap_outcomes: list[ValidationOutcome] = [] + closes_queued: set[tuple[str, str]] = set() try: if source_file is not None: @@ -329,7 +339,9 @@ async def ingest_events( seen_in_batch.add(dedup_key) errors = _validate_event_kdes(event) - if errors: + open_gaps = gaps_from_kde_errors(errors) + deferrable = deferred_kdes.get(event.cte_type, frozenset()) if errors and deferred_kdes else frozenset() + if errors and not (open_gaps and all(gap.gap_key in deferrable for gap in open_gaps)): results.append( EventResult( traceability_lot_code=event.traceability_lot_code, @@ -344,7 +356,7 @@ async def ingest_events( order=event_index, lot_code=event.traceability_lot_code, cte_type=event.cte_type.value, - gaps=gaps_from_kde_errors(errors), + gaps=open_gaps, ) ) continue @@ -429,7 +441,13 @@ async def ingest_events( if summary is not None: from shared.rules.enforcement import current_mode, should_reject - should, reject_reason = should_reject(summary) + if open_gaps: + should, reject_reason = should_reject( + summary, + deferred_kdes=frozenset(gap.gap_key for gap in open_gaps), + ) + else: + should, reject_reason = should_reject(summary) if should: # Finding B — durable rejection audit row, # written via its own session BEFORE the @@ -485,6 +503,20 @@ async def ingest_events( ) continue + # An incomplete record and its gap commit together: if the + # gap can't be written, the event is not accepted. + if open_gaps: + try: + open_gaps_for_stored_record( + db_session, + tenant_id, + lot_code=event.traceability_lot_code, + cte_type=event.cte_type.value, + gaps=open_gaps, + ) + except SQLAlchemyError as gap_err: + raise RuntimeError("deferred gap write failed") from gap_err + # Accepted path. Savepoint commits the primary CTE write. savepoint.commit() results.append( @@ -495,18 +527,35 @@ async def ingest_events( event_id=store_result.event_id, sha256_hash=store_result.sha256_hash, chain_hash=store_result.chain_hash, + open_gaps=[gap.gap_key for gap in open_gaps], ) ) accepted += 1 - gap_outcomes.append( - ValidationOutcome( - order=event_index, - lot_code=event.traceability_lot_code, - cte_type=event.cte_type.value, - closing_record_id=store_result.event_id, - closing_record_hash=store_result.sha256_hash, + # A record with deferred KDEs is stored but incomplete, so it + # must not close the lot's existing gaps. Its gaps are already + # written; requeue them only to follow a close queued earlier + # in this batch, as a replay could reopen another request's. + gap_scope = (event.traceability_lot_code.strip(), event.cte_type.value) + if not open_gaps: + closes_queued.add(gap_scope) + gap_outcomes.append( + ValidationOutcome( + order=event_index, + lot_code=event.traceability_lot_code, + cte_type=event.cte_type.value, + closing_record_id=store_result.event_id, + closing_record_hash=store_result.sha256_hash, + ) + ) + elif gap_scope in closes_queued: + gap_outcomes.append( + ValidationOutcome( + order=event_index, + lot_code=event.traceability_lot_code, + cte_type=event.cte_type.value, + gaps=open_gaps, + ) ) - ) _publish_graph_sync(store_result.event_id, event, tenant_id) # Non-blocking failures → exception queue. Separate diff --git a/services/ingestion/app/webhook_models.py b/services/ingestion/app/webhook_models.py index 08017e288..9b17d4101 100644 --- a/services/ingestion/app/webhook_models.py +++ b/services/ingestion/app/webhook_models.py @@ -329,6 +329,10 @@ class EventResult(BaseModel): sha256_hash: Optional[str] = None chain_hash: Optional[str] = None errors: List[str] = Field(default_factory=list) + open_gaps: List[str] = Field( + default_factory=list, + description="Required KDEs this accepted event still lacks; each has an open data gap", + ) class IngestResponse(BaseModel): 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..4a5222667 --- /dev/null +++ b/services/ingestion/tests/test_edi_856_required_kdes.py @@ -0,0 +1,203 @@ +"""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, HTTPException +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.rejection_log as rejection_log # noqa: E402 +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, WebhookCTEType, 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" + ] + + +@pytest.mark.parametrize("route", ["", "/document"]) +def test_both_routes_defer_only_shipping_tlc_source_reference(route: str, monkeypatch: pytest.MonkeyPatch, captured) -> None: + seen: list[dict] = [] + + async def _capture_kwargs(payload: WebhookPayload, **kwargs) -> IngestResponse: + seen.append(kwargs) + return IngestResponse(accepted=0, rejected=0, total=len(payload.events), events=[]) + + monkeypatch.setattr(edi_ingestion_routes, "ingest_events", _capture_kwargs) + + assert _post(route, {"traceability_lot_code": "LOT-856-KDE"}).status_code == 201 + assert seen + for kwargs in seen: + assert kwargs["deferred_kdes"] == {WebhookCTEType.SHIPPING: frozenset({"tlc_source_reference"})} + + +def _two_lot_asn() -> bytes: + raw = _ASN_856.read_text(encoding="utf-8") + line = "LIN**SK*ROMAINE-12CT~SN1**200*CA~" + assert line in raw + two_lots = raw.replace( + line, + "LIN**SK*ROMAINE-12CT~SN1**100*CA~REF*LT*LOT-A~LIN**SK*ROMAINE-12CT~SN1**100*CA~REF*LT*LOT-B~", + ) + return two_lots.replace("SE*10*0001", "SE*14*0001").encode() + + +def _post_two_lots(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", _two_lot_asn(), "application/edi-x12")}, + ) + + +@pytest.mark.parametrize("route", ["", "/document"]) +def test_one_reference_cannot_cover_distinct_lots(route: str, monkeypatch: pytest.MonkeyPatch, captured) -> None: + rejection_log.reset_edi_rejections() + claims: list[tuple] = [] + + def _claim(*args, **kwargs): + claims.append(args) + return False, None + + monkeypatch.setattr(edi_ingestion_routes, "check_and_record_interchange", _claim) + + response = _post_two_lots(route, {"traceability_lot_code": "LOT-856-KDE", "tlc_source_reference": "PACK-0310"}) + + assert response.status_code == 422, response.text + assert response.json()["detail"]["error"] == "edi_tlc_source_reference_with_multiple_lots" + assert captured == [] + assert claims == [], "a refused document must not reserve its ISA13" + recorded = rejection_log.list_edi_rejections(_TENANT) + assert [record["reason"] for record in recorded] == ["edi_tlc_source_reference_with_multiple_lots"] + assert response.json()["detail"]["rejection_id"] == recorded[0]["rejection_id"] + + +@pytest.mark.parametrize("route", ["", "/document"]) +def test_distinct_lots_without_a_reference_are_ingested(route: str, captured) -> None: + response = _post_two_lots(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 {event.traceability_lot_code for event in events} == {"LOT-A", "LOT-B"} + assert all(event.kdes.get("tlc_source_reference") is None for event in events) + + +@pytest.mark.parametrize("transaction_set", ["861", "944"]) +def test_multi_lot_receipt_may_share_a_reference(transaction_set: str) -> None: + """Receiving requires the KDE, so refusing it would leave no way in.""" + edi_ingestion_routes._reject_shared_tlc_source_for_distinct_lots( + tlc_source_reference="PACK-0310", + transaction_sets=[(transaction_set, {"line_items": [{"lot_number": "LOT-A"}, {"lot_number": "LOT-B"}]})], + traceability_lot_code="LOT-FORM", + sender_tenant_id=_TENANT, + partner_id=None, + source="edi_inbound", + ) + + +def _check_945_lines(*lines: dict) -> None: + edi_ingestion_routes._reject_shared_tlc_source_for_distinct_lots( + tlc_source_reference="PACK-0310", + transaction_sets=[("945", {"line_items": list(lines)})], + traceability_lot_code="LOT-FORM", + sender_tenant_id=_TENANT, + partner_id=None, + source="edi_inbound", + ) + + +def test_one_short_lot_on_two_gtins_is_two_tlcs() -> None: + with pytest.raises(HTTPException) as refused: + _check_945_lines( + {"lot_number": "A1", "gtin": "00012345678901"}, + {"lot_number": "A1", "gtin": "00012345678902"}, + ) + + assert refused.value.detail["error"] == "edi_tlc_source_reference_with_multiple_lots" + + +def test_one_short_lot_on_one_gtin_is_one_tlc() -> None: + _check_945_lines( + {"lot_number": "A1", "gtin": "00012345678901"}, + {"lot_number": "A1", "gtin": "00012345678901"}, + ) diff --git a/services/ingestion/tests/test_edi_date_reject_resend.py b/services/ingestion/tests/test_edi_date_reject_resend.py index e641961c4..7f184f00a 100644 --- a/services/ingestion/tests/test_edi_date_reject_resend.py +++ b/services/ingestion/tests/test_edi_date_reject_resend.py @@ -73,7 +73,7 @@ def ingested(monkeypatch: pytest.MonkeyPatch) -> list: """Stub the canonical write; capture each accepted payload.""" calls: list = [] - async def _fake_ingest_events(payload, *, principal_tenant_id=None, x_regengine_api_key=None): + async def _fake_ingest_events(payload, *, principal_tenant_id=None, x_regengine_api_key=None, **_kwargs): calls.append(payload) return IngestResponse( accepted=len(payload.events), @@ -222,7 +222,7 @@ async def _broken(*args, **kwargs): def test_all_events_rejected_frees_isa13(client, store, monkeypatch, route): """201 with accepted=0 stored nothing, so the corrected resend must not 409.""" - async def _rejecting(payload, *, principal_tenant_id=None, x_regengine_api_key=None): + async def _rejecting(payload, *, principal_tenant_id=None, x_regengine_api_key=None, **_kwargs): return IngestResponse( accepted=0, rejected=len(payload.events), diff --git a/services/ingestion/tests/test_edi_deferred_kdes.py b/services/ingestion/tests/test_edi_deferred_kdes.py new file mode 100644 index 000000000..942582cbd --- /dev/null +++ b/services/ingestion/tests/test_edi_deferred_kdes.py @@ -0,0 +1,243 @@ +"""EDI shipments without a TLC source reference are stored with an open gap. + +X12 has no element for ``tlc_source_reference``. The EDI routes defer that one +KDE on shipping events: the event is stored, its result names the gap, and a +``missing_kde`` gap stays open. Everything else keeps the strict gate, and the +exemption must be unreachable from anything the sender controls. +""" + +from __future__ import annotations + +import asyncio +from types import SimpleNamespace +from typing import Any + +import pytest +from sqlalchemy.exc import ProgrammingError + +from services.ingestion.app import webhook_compat as wc +from services.ingestion.app.edi_ingestion import routes as edi_routes +from services.ingestion.app.webhook_models import IngestEvent, WebhookCTEType, WebhookPayload +from shared.data_gap_tracking import GAP_KIND_MISSING_KDE, Gap + +TENANT = "11111111-1111-1111-1111-111111111111" +TLC = "00012345678901-LOT001" +EVENT_HASH = "a" * 64 +TLC_SOURCE_GAP = Gap(GAP_KIND_MISSING_KDE, "tlc_source_reference") + + +class _Savepoint: + def __init__(self, state: SimpleNamespace) -> None: + self.state = state + self.stored_before = len(state.stored) + + def commit(self) -> None: + return None + + def rollback(self) -> None: + del self.state.stored[self.stored_before :] + + +class _Session: + def __init__(self, state: SimpleNamespace) -> None: + self.state = state + + def begin_nested(self) -> "_Savepoint": + return _Savepoint(self.state) + + def execute(self, _statement: Any, params: dict[str, Any]) -> None: + if self.state.gap_write_error is not None: + raise self.state.gap_write_error + self.state.inline_gaps.append(params["key"]) + + def commit(self) -> None: + return None + + def rollback(self) -> None: + return None + + def close(self) -> None: + return None + + +@pytest.fixture() +def harness(monkeypatch: pytest.MonkeyPatch) -> SimpleNamespace: + state = SimpleNamespace(stored=[], outcomes=[], inline_gaps=[], gap_write_error=None) + + class _Persistence: + def __init__(self, db: Any) -> None: + return None + + def store_event(self, **kwargs: Any) -> SimpleNamespace: + state.stored.append(kwargs) + return SimpleNamespace(event_id="ev-1", sha256_hash=EVENT_HASH, chain_hash="c" * 64) + + def _no_canonical(*_args: Any, **_kwargs: Any) -> None: + raise ValueError("canonical write not under test") + + def _sessions(): + yield _Session(state) + + import shared.canonical_event as canonical_event + import shared.cte_persistence as cte_persistence + + monkeypatch.setattr(cte_persistence, "CTEPersistence", _Persistence) + monkeypatch.setattr(canonical_event, "normalize_webhook_event", _no_canonical) + monkeypatch.setattr(wc, "_get_db_session", _sessions) + monkeypatch.setattr(wc, "_check_rate_limit", lambda _tenant: None) + monkeypatch.setattr(wc, "_verify_api_key_sync", lambda **_kwargs: None) + monkeypatch.setattr(wc, "_generate_alerts", lambda _event: []) + monkeypatch.setattr(wc, "_publish_graph_sync", lambda *_args: None) + monkeypatch.setattr(wc, "record_outcomes", lambda _db, _tenant, outcomes: state.outcomes.extend(outcomes)) + return state + + +def _shipment(**kde_overrides: Any) -> IngestEvent: + kdes = { + "ship_date": "2026-03-10", + "ship_from_location": "Valley Fresh Farms", + "ship_to_location": "Metro Distribution Center", + "reference_document": "BOL-9001", + **kde_overrides, + } + return IngestEvent( + cte_type=WebhookCTEType.SHIPPING, + traceability_lot_code=TLC, + product_description="Romaine 12CT", + quantity=200, + unit_of_measure="cases", + timestamp="2026-03-10T12:00:00Z", + location_name="Valley Fresh Farms", + kdes={key: value for key, value in kdes.items() if value is not None}, + ) + + +def _receipt() -> IngestEvent: + return IngestEvent( + cte_type=WebhookCTEType.RECEIVING, + traceability_lot_code=TLC, + product_description="Romaine 12CT", + quantity=200, + unit_of_measure="cases", + timestamp="2026-03-10T12:00:00Z", + location_name="Metro Distribution Center", + kdes={ + "receive_date": "2026-03-10", + "receiving_location": "Metro Distribution Center", + "immediate_previous_source": "Valley Fresh Farms", + "reference_document": "BOL-9001", + }, + ) + + +def _ingest(event: IngestEvent, *, source: str = "edi_856_inbound", **kwargs: Any): + return _ingest_batch([event], source=source, **kwargs) + + +def _ingest_batch(events: list[IngestEvent], *, source: str = "edi_856_inbound", **kwargs: Any): + payload = WebhookPayload(tenant_id=TENANT, source=source, events=events) + return asyncio.run( + wc.ingest_events( + payload, + principal_tenant_id=TENANT, + x_regengine_api_key="test-key", # pragma: allowlist secret — synthetic test fixture only + **kwargs, + ) + ) + + +def test_deferred_kde_stores_event_and_keeps_gap_open(harness) -> None: + response = _ingest(_shipment(), deferred_kdes=edi_routes._EDI_DEFERRED_KDES) + + assert (response.accepted, response.rejected) == (1, 0) + assert response.events[0].open_gaps == ["tlc_source_reference"] + assert len(harness.stored) == 1 + assert harness.inline_gaps == ["tlc_source_reference"] + + # Nothing queued: no receipt to close the lot's gaps, and no replay that + # could reopen a gap another request closes after this commit. + assert harness.outcomes == [] + + +def test_batch_keeps_arrival_order_for_a_lot(harness) -> None: + """A later incomplete record keeps its gap; a later complete one closes it.""" + complete = _shipment(tlc_source_reference="PACK-0310") + incomplete = _shipment().model_copy(update={"timestamp": "2026-03-10T13:00:00Z"}) + + _ingest_batch([complete, incomplete], deferred_kdes=edi_routes._EDI_DEFERRED_KDES) + + assert [outcome.gaps for outcome in harness.outcomes] == [(), (TLC_SOURCE_GAP,)] + assert harness.outcomes[0].closing_record_id == "ev-1" + assert harness.outcomes[1].closing_record_id is None + + +def test_incomplete_then_complete_in_a_batch_closes_the_gap(harness) -> None: + incomplete = _shipment() + complete = _shipment(tlc_source_reference="PACK-0310").model_copy(update={"timestamp": "2026-03-10T13:00:00Z"}) + + _ingest_batch([incomplete, complete], deferred_kdes=edi_routes._EDI_DEFERRED_KDES) + + assert harness.inline_gaps == ["tlc_source_reference"] + assert [outcome.closing_record_id for outcome in harness.outcomes] == ["ev-1"] + + +def test_event_is_not_accepted_when_its_gap_cannot_be_written(harness) -> None: + harness.gap_write_error = ProgrammingError("INSERT", {}, Exception("permission denied")) + + response = _ingest(_shipment(), deferred_kdes=edi_routes._EDI_DEFERRED_KDES) + + assert (response.accepted, response.rejected) == (0, 1) + assert response.events[0].open_gaps == [] + assert harness.stored == [], "the stored event must roll back with its gap" + + +def test_supplied_reference_stores_event_and_closes_gaps(harness) -> None: + response = _ingest( + _shipment(tlc_source_reference="PACK-0310"), + deferred_kdes=edi_routes._EDI_DEFERRED_KDES, + ) + + assert response.accepted == 1 + assert response.events[0].open_gaps == [] + outcome = harness.outcomes[0] + assert outcome.gaps == () + assert (outcome.closing_record_id, outcome.closing_record_hash) == ("ev-1", EVENT_HASH) + + +def test_another_missing_kde_still_rejects(harness) -> None: + response = _ingest(_shipment(reference_document=None), deferred_kdes=edi_routes._EDI_DEFERRED_KDES) + + assert (response.accepted, response.rejected) == (0, 1) + assert harness.stored == [] + assert set(harness.outcomes[0].gaps) == { + TLC_SOURCE_GAP, + Gap(GAP_KIND_MISSING_KDE, "reference_document"), + } + + +def test_receiving_is_not_deferred(harness) -> None: + response = _ingest(_receipt(), deferred_kdes=edi_routes._EDI_DEFERRED_KDES) + + assert (response.accepted, response.rejected) == (0, 1) + assert harness.stored == [] + + +def test_error_with_no_gap_key_still_rejects(harness, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setattr(wc, "_validate_event_kdes", lambda _event: [" "]) + + response = _ingest(_shipment(), deferred_kdes=edi_routes._EDI_DEFERRED_KDES) + + assert (response.accepted, response.rejected) == (0, 1) + assert harness.stored == [] + + +def test_sender_cannot_claim_the_exemption(harness) -> None: + """EDI-looking source and KDEs change nothing without the server-side argument.""" + response = _ingest( + _shipment(edi_transaction_set="856", deferred_kdes=["tlc_source_reference"]), + source="edi_856_inbound", + ) + + assert (response.accepted, response.rejected) == (0, 1) + assert harness.stored == [] + assert response.events[0].open_gaps == [] diff --git a/services/ingestion/tests/test_edi_wire_format_2974_2976_2978.py b/services/ingestion/tests/test_edi_wire_format_2974_2976_2978.py index a69e42f86..f959a5847 100644 --- a/services/ingestion/tests/test_edi_wire_format_2974_2976_2978.py +++ b/services/ingestion/tests/test_edi_wire_format_2974_2976_2978.py @@ -140,7 +140,7 @@ def client(monkeypatch: pytest.MonkeyPatch, captured_payload: dict) -> TestClien auth_mode="test", ) - async def _fake_ingest_events(payload, *, principal_tenant_id=None, x_regengine_api_key=None): + async def _fake_ingest_events(payload, *, principal_tenant_id=None, x_regengine_api_key=None, **_kwargs): captured_payload["payload"] = payload return IngestResponse( accepted=len(payload.events), 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 diff --git a/services/shared/data_gap_tracking.py b/services/shared/data_gap_tracking.py index f9a66f986..b78b81bf0 100644 --- a/services/shared/data_gap_tracking.py +++ b/services/shared/data_gap_tracking.py @@ -252,6 +252,27 @@ def _apply_outcome(db_session: Any, tenant_id: str, outcome: ValidationOutcome) ) +def open_gaps_for_stored_record( + db_session: Any, + tenant_id: str, + *, + lot_code: str, + cte_type: str, + gaps: tuple[Gap, ...], +) -> None: + """Open gaps inside the record's own transaction. Raises on failure. + + Gap logging is normally fail-open, because a failed record is not stored + anyway. A record stored with a known gap is different: if the gap is lost, + nothing shows the record is incomplete. Commit the two together. + """ + _apply_outcome( + db_session, + tenant_id, + ValidationOutcome(order=0, lot_code=lot_code, cte_type=cte_type, gaps=gaps), + ) + + def record_outcomes(db_session: Any, tenant_id: str, outcomes: Sequence[ValidationOutcome]) -> None: """Apply queued outcomes in arrival order, then commit. Never raises. diff --git a/services/shared/rules/enforcement.py b/services/shared/rules/enforcement.py index 8212857f5..33da05b8f 100644 --- a/services/shared/rules/enforcement.py +++ b/services/shared/rules/enforcement.py @@ -27,9 +27,13 @@ from __future__ import annotations import os -from typing import Optional, Tuple +from typing import FrozenSet, Optional, Sequence, Tuple -from .types import EvaluationSummary +from .types import EvaluationSummary, RuleEvaluationResult + +# Other fields a rule accepts for the same KDE. A rule that fails on a deferred +# KDE reports these as missing too, so they don't make the failure a new one. +_KDE_ALTERNATE_FIELDS = {"tlc_source_reference": ("kdes.tlc_source_gln",)} class EnforcementMode: @@ -155,7 +159,22 @@ def current_mode() -> str: return raw -def should_reject(summary: EvaluationSummary) -> Tuple[bool, Optional[str]]: +def _is_deferred_failure(result: RuleEvaluationResult, deferred_kdes: FrozenSet[str]) -> bool: + """True when a non-critical failure comes only from deferred KDEs.""" + if result.result != "fail" or result.severity == "critical": + return False + allowed = {f"kdes.{kde}" for kde in deferred_kdes} + for kde in deferred_kdes: + allowed.update(_KDE_ALTERNATE_FIELDS.get(kde, ())) + missing = {entry.get("field") for entry in result.evidence_fields_inspected if not entry.get("present")} + return bool(missing) and missing <= allowed + + +def should_reject( + summary: EvaluationSummary, + *, + deferred_kdes: FrozenSet[str] = frozenset(), +) -> Tuple[bool, Optional[str]]: """Return ``(reject, reason)`` for a rules evaluation. ``reject``: True means the caller should reject the event (typically @@ -163,6 +182,12 @@ def should_reject(summary: EvaluationSummary) -> Tuple[bool, Optional[str]]: ``reason``: a short human-readable string suitable for the HTTP response body when rejecting. ``None`` when not rejecting. + + ``deferred_kdes`` names KDEs the caller stores with an open data gap. + Under ``all``, a warning that fails only because of them does not + reject, or the gap could never be stored. Only server-side callers may + set this. The verdict stays ``fail``, so the exception queue still sees + it, and every other failure on the event still rejects. """ mode = current_mode() if mode == EnforcementMode.OFF: @@ -181,9 +206,13 @@ def should_reject(summary: EvaluationSummary) -> Tuple[bool, Optional[str]]: return True, _first_failure_reason(summary) return False, None if mode == EnforcementMode.ALL: - if summary.compliant is False: - return True, _first_failure_reason(summary) - return False, None + if summary.compliant is not False: + return False, None + deferred = [result for result in summary.results if _is_deferred_failure(result, deferred_kdes)] + # Fail closed: accept only if these results explain every failure. + if deferred and summary.errored == 0 and len(deferred) == summary.failed: + return False, None + return True, _first_failure_reason(summary, skip=deferred) return False, None @@ -262,9 +291,13 @@ def handle_preeval_exception(err: BaseException, *, where: str) -> None: ) from err -def _first_failure_reason(summary: EvaluationSummary) -> str: +def _first_failure_reason( + summary: EvaluationSummary, + *, + skip: Sequence[RuleEvaluationResult] = (), +) -> str: for result in summary.critical_failures or summary.results: - if result.result in ("fail", "error"): + if result.result in ("fail", "error") and not any(result is skipped for skipped in skip): rule_id = result.rule_id or "unknown_rule" why = result.why_failed or "failed" return f"{rule_id}: {why}" diff --git a/services/shared/tests/test_rules_enforcement_deferred_kdes.py b/services/shared/tests/test_rules_enforcement_deferred_kdes.py new file mode 100644 index 000000000..ecbf22ba7 --- /dev/null +++ b/services/shared/tests/test_rules_enforcement_deferred_kdes.py @@ -0,0 +1,96 @@ +"""Under ``all``, a warning caused only by a deferred KDE does not reject. + +EDI shipments are stored with an open data gap when X12 can't carry the TLC +source reference. The seeded rule for that KDE is a warning, and ``all`` +rejects on warnings, so without this the gap could never be stored. +""" + +from __future__ import annotations + +import pytest + +from shared.rules.enforcement import should_reject +from shared.rules.types import EvaluationSummary, RuleEvaluationResult + +DEFERRED = frozenset({"tlc_source_reference"}) + + +def _field(name: str, present: bool = False) -> dict: + return {"field": name, "value": None, "present": present} + + +def _tlc_source_warning(severity: str = "warning") -> RuleEvaluationResult: + return RuleEvaluationResult( + rule_id="tlc-source", + severity=severity, + result="fail", + why_failed="missing TLC source", + evidence_fields_inspected=[_field("kdes.tlc_source_reference"), _field("kdes.tlc_source_gln")], + ) + + +def _other_warning() -> RuleEvaluationResult: + return RuleEvaluationResult( + rule_id="other", + severity="warning", + result="fail", + why_failed="other failure", + evidence_fields_inspected=[_field("kdes.carrier")], + ) + + +def _summary(*results: RuleEvaluationResult, errored: int = 0) -> EvaluationSummary: + failed = sum(1 for result in results if result.result == "fail") + return EvaluationSummary( + total_rules=len(results) + 1, + passed=1, + failed=failed, + errored=errored, + results=list(results), + critical_failures=[result for result in results if result.severity == "critical"], + ) + + +@pytest.fixture(autouse=True) +def _all_mode(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv("RULES_ENGINE_ENFORCE", "all") + + +def test_deferred_failure_does_not_reject_and_stays_noncompliant() -> None: + summary = _summary(_tlc_source_warning()) + + assert should_reject(summary, deferred_kdes=DEFERRED) == (False, None) + assert summary.compliant is False + + +def test_without_deferred_kdes_the_warning_still_rejects() -> None: + assert should_reject(_summary(_tlc_source_warning()))[0] is True + + +def test_another_failure_still_rejects_and_is_the_reason() -> None: + reject, reason = should_reject(_summary(_tlc_source_warning(), _other_warning()), deferred_kdes=DEFERRED) + + assert reject is True + assert reason == "other: other failure" + + +def test_critical_rule_on_the_same_kde_still_rejects() -> None: + assert should_reject(_summary(_tlc_source_warning("critical")), deferred_kdes=DEFERRED)[0] is True + + +def test_rule_needing_another_missing_field_still_rejects() -> None: + result = _tlc_source_warning() + result.evidence_fields_inspected.append(_field("kdes.carrier")) + + assert should_reject(_summary(result), deferred_kdes=DEFERRED)[0] is True + + +def test_evaluator_error_still_rejects() -> None: + assert should_reject(_summary(_tlc_source_warning(), errored=1), deferred_kdes=DEFERRED)[0] is True + + +def test_unexplained_failure_count_fails_closed() -> None: + summary = _summary(_tlc_source_warning()) + summary.failed = 2 + + assert should_reject(summary, deferred_kdes=DEFERRED)[0] is True