diff --git a/services/ingestion/app/edi_ingestion/rejection_log.py b/services/ingestion/app/edi_ingestion/rejection_log.py index c10eb5d02..13d3c9bab 100644 --- a/services/ingestion/app/edi_ingestion/rejection_log.py +++ b/services/ingestion/app/edi_ingestion/rejection_log.py @@ -55,6 +55,8 @@ def record_edi_rejection( extracted: dict[str, Any] | None = None, partner_id: str | None = None, source: str | None = None, + reason: str = "fsma_validation_failed", + needs_review: bool = False, ) -> dict[str, Any]: """Record an EDI FSMA validation rejection. @@ -78,7 +80,8 @@ def record_edi_rejection( # Snapshot of the extracted fields for forensic review. Drops # any non-serializable values defensively. "extracted": _scrub(extracted or {}), - "reason": "fsma_validation_failed", + "reason": reason, + "needs_review": needs_review, } store = _rejection_store_for(tenant_id) diff --git a/services/ingestion/app/edi_ingestion/routes.py b/services/ingestion/app/edi_ingestion/routes.py index f90831eeb..bb2c94ad2 100644 --- a/services/ingestion/app/edi_ingestion/routes.py +++ b/services/ingestion/app/edi_ingestion/routes.py @@ -26,6 +26,7 @@ import importlib import logging import os +from collections.abc import Callable from typing import Any, Optional from fastapi import APIRouter, Depends, File, Form, Header, HTTPException, Query, UploadFile @@ -280,6 +281,69 @@ def _reject_legacy_856_placeholders( ) +# Event-date field (and optional time field) per transaction set. +_EVENT_DATE_FIELDS: dict[str, tuple[str | None, str | None]] = { + "856": ("ship_date_raw", "ship_time_raw"), + "850": ("po_date_raw", None), + "810": ("invoice_date_raw", None), + "861": ("receive_date_raw", None), + "944": ("receive_date_raw", None), + "945": ("ship_date_raw", None), +} + + +def _preflight_event_date( + *, + transaction_set: str, + extracted: dict[str, Any], + sender_tenant_id: str, + traceability_lot_code: str, + partner_id: str | None, + source: str, +) -> None: + """Reject a bad EDI event date BEFORE the ISA13 is reserved. + + ``_enforce_envelope_integrity`` records the interchange control number, + so a date rejection raised after it would make the corrected resend a + 409 replay. Run this first. Every rejection is written to the EDI + rejection log with ``needs_review=True`` and its id is returned to the + caller. + """ + date_field, time_field = _EVENT_DATE_FIELDS.get(transaction_set, (None, None)) + if transaction_set in ("944", "945"): + time_field = None + try: + _ship_timestamp_and_date( + extracted.get(date_field) if date_field else None, + extracted.get(time_field) if time_field else None, + ) + except HTTPException as exc: + if exc.status_code != 400 or not isinstance(exc.detail, dict): + raise + detail = dict(exc.detail) + record = _record_edi_rejection( + tenant_id=sender_tenant_id, + transaction_set=transaction_set, + traceability_lot_code=traceability_lot_code, + errors=[ + { + "loc": [date_field or "event_date"], + "type": detail.get("error"), + "msg": detail.get("message"), + "input": detail.get("raw"), + } + ], + extracted=extracted, + partner_id=partner_id, + source=source, + reason=str(detail.get("error") or "invalid_edi_date"), + needs_review=True, + ) + detail["needs_review"] = True + detail["rejection_id"] = record["rejection_id"] + raise HTTPException(status_code=400, detail=detail) from exc + + router = APIRouter(prefix="/api/v1/ingest/edi", tags=["EDI Import"]) _resolve_tenant_id = resolve_tenant_id @@ -290,9 +354,15 @@ def _enforce_envelope_integrity( extracted: dict[str, Any], sender_tenant_id: str, transaction_set: str, + before_reserve: Callable[[], None] | None = None, ) -> None: """Apply #1160 + #1165 envelope validation. + ``before_reserve`` runs after the envelope and allowlist checks and + before the ISA13 is recorded. Content rejections (event dates) go + there so a disallowed sender still gets 403 and a rejected document + does not burn its interchange number. + - Reject on ISA/GS sender/receiver mismatch (X12 tenant-smuggling). - Reject on GS sender not in tenant's allowed trading-partner list. - Check ISA13 against the interchange dedup cache and return 409 @@ -342,6 +412,9 @@ def _enforce_envelope_integrity( }, ) + if before_reserve is not None: + before_reserve() + # #1165: retransmission dedup on ISA13. is_duplicate, _prev = check_and_record_interchange( sender_id=extracted.get("sender_id"), @@ -537,8 +610,21 @@ async def ingest_edi_856( extracted = _extract_856_fields(segments) - # #1160 + #1165: envelope mismatch / allowlist / ISA13 dedup. - _enforce_envelope_integrity(extracted, sender_tenant_id, "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( + transaction_set="856", + extracted=extracted, + sender_tenant_id=sender_tenant_id, + traceability_lot_code=traceability_lot_code, + partner_id=x_partner_id, + source=source, + ), + ) lines = extracted.get("line_items") or [] if len(lines) > 1: @@ -774,15 +860,7 @@ def _build_ingest_event_for_set( location_gln = _valid_gln_or_none(extracted.get("receiving_location_gln")) # Resolve timestamp - date_field_map = { - "856": ("ship_date_raw", "ship_time_raw"), - "850": ("po_date_raw", None), - "810": ("invoice_date_raw", None), - "861": ("receive_date_raw", None), - "944": ("receive_date_raw", None), - "945": ("ship_date_raw", None), - } - date_field, time_field = date_field_map.get(transaction_set, (None, None)) + date_field, time_field = _EVENT_DATE_FIELDS.get(transaction_set, (None, None)) if transaction_set in ("944", "945") and extracted.get("_event_timestamp"): timestamp, event_date = extracted["_event_timestamp"], extracted["_event_date"] else: @@ -961,7 +1039,9 @@ 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"), + tlc_source_reference: Optional[str] = Form( + None, description="Optional TLC source reference copied to 944/945 KDEs" + ), tenant_id: Optional[str] = Form(None, description="Optional tenant override"), source: str = Form("edi_inbound", description="Source system identifier"), product_description: Optional[str] = Form(None, description="Optional product description override"), @@ -1084,7 +1164,20 @@ async def ingest_edi_document( # #1160 + #1165: validate and reserve the shared envelope once per # document, before validating any transaction set or warehouse line. - _enforce_envelope_integrity(extracted, sender_tenant_id, transaction_set) + def _preflight_all_sets() -> None: + for slice_set, slice_extracted in zip(slice_sets, extractions): + _preflight_event_date( + transaction_set=slice_set, + extracted=slice_extracted, + sender_tenant_id=sender_tenant_id, + traceability_lot_code=traceability_lot_code, + 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. + _enforce_envelope_integrity(extracted, sender_tenant_id, transaction_set, before_reserve=_preflight_all_sets) # 856 SN1, 861 RCD, and warehouse W07/W12 lines are each one CTE. # ST/SE extraction and envelope handling stay at document scope, then diff --git a/services/ingestion/app/edi_ingestion/utils.py b/services/ingestion/app/edi_ingestion/utils.py index 4c816d92c..d5ce3f8c0 100644 --- a/services/ingestion/app/edi_ingestion/utils.py +++ b/services/ingestion/app/edi_ingestion/utils.py @@ -125,17 +125,30 @@ def _valid_gln_or_none(value: Optional[str]) -> Optional[str]: def _ship_timestamp_and_date(ship_date_raw: Optional[str], ship_time_raw: Optional[str]) -> tuple[str, str]: """Normalize an EDI shipment date/time into an ISO timestamp and date. - #1167: unparseable dates now raise HTTP 400 rather than silently - falling back to ``datetime.now()`` — retransmitted archival EDI was - being re-stamped as today's shipment. Stale dates (> 90 days old) - still snap to "now" for the ingest timestamp to keep webhook model - bounds valid, but the original parsed date is preserved separately - so downstream FSMA 204 lookback sees the real event date. + The returned timestamp is always the real event time from the EDI + source, never the ingest time. Backfilled EDI older than 90 days keeps + its real date (the webhook model accepts historical events for the + FSMA 204 2-year retention backfill). Previously the timestamp snapped + to ``now`` for such events, which gave each event two clocks and + corrupted EPCIS eventTime and R1 time to fix. + + Missing, unparseable, or > 24h-future dates raise HTTP 400 so the + event is rejected for review instead of being re-stamped (#1167). """ now = datetime.now(timezone.utc) - if not ship_date_raw: - return now.isoformat(), now.date().isoformat() + if not ship_date_raw or not str(ship_date_raw).strip(): + raise HTTPException( + status_code=400, + detail={ + "error": "missing_edi_date", + "raw": ship_date_raw, + "message": ( + "EDI event date is missing. The ingest time is not " + "substituted for the event time; supply the source date." + ), + }, + ) clean_date = re.sub(r"\D", "", ship_date_raw) @@ -170,12 +183,22 @@ def _ship_timestamp_and_date(ship_date_raw: Optional[str], ship_time_raw: Option }, ) from exc - # Keep webhook model bounds to avoid stale/future rejection — the - # ingest timestamp snaps to "now" if the EDI claims a shipment - # older than 90 days or > 24h in the future. The parsed event_date - # is returned unchanged so FSMA 204 traceability sees the real date. - if parsed < now - timedelta(days=90) or parsed > now + timedelta(hours=24): - return now.isoformat(), parsed.date().isoformat() + # Never replace the real event time with the ingest time. Historical + # dates are kept as-is; a date more than 24h ahead is rejected for + # review rather than overwritten. + if parsed > now + timedelta(hours=24): + raise HTTPException( + status_code=400, + detail={ + "error": "edi_date_in_future", + "raw": ship_date_raw, + "needs_review": True, + "message": ( + "EDI event date is more than 24 hours in the future; " + "flagged for review instead of being replaced with now()." + ), + }, + ) return parsed.isoformat(), parsed.date().isoformat() diff --git a/services/ingestion/tests/test_edi_date_reject_resend.py b/services/ingestion/tests/test_edi_date_reject_resend.py new file mode 100644 index 000000000..2e3af20a5 --- /dev/null +++ b/services/ingestion/tests/test_edi_date_reject_resend.py @@ -0,0 +1,173 @@ +"""A date-rejected EDI interchange must not burn its ISA13. + +Uses the repo's synthetic fixture ``tests/fixtures/edi_corpus/asn_856.edi``. +The rejected variant only moves the BSN ship date to 5 days after now; the +corrected resend is the unmodified fixture. +""" + +from __future__ import annotations + +import sys +from datetime import datetime, timedelta, timezone +from pathlib import Path + +import pytest + +service_dir = Path(__file__).parent.parent +sys.path.insert(0, str(service_dir)) + +pytest.importorskip("fastapi") + +import app.edi_ingestion.routes as edi_routes # noqa: E402 +from app.authz import IngestionPrincipal, get_ingestion_principal # noqa: E402 +from app.edi_ingestion import ( + dedup as dedup_mod, # noqa: E402 + rejection_log, # noqa: E402 + router as edi_router, # noqa: E402 +) +from app.webhook_models import EventResult, IngestResponse # noqa: E402 +from fastapi import FastAPI # noqa: E402 +from fastapi.testclient import TestClient # noqa: E402 + +FIXTURE = service_dir / "tests" / "fixtures" / "edi_corpus" / "asn_856.edi" +FIXTURE_BSN = "BSN*00*ASN-12345*20260310*1200~" +TENANT = "00000000-0000-0000-0000-000000000123" +# GTIN-14 + lot suffix, the FSMA 204 TLC format strict validation accepts +# (same value test_edi_wire_format_2974_2976_2978 uses). +VALID_TLC = "00012345678901-LOT001" +ALLOWLIST_ENV = "EDI_PARTNER_ALLOWLIST_" + TENANT.upper().replace("-", "_") + + +class _NxStore: + """Minimal SET NX semantics, matching how dedup uses Redis.""" + + def __init__(self) -> None: + self.keys: dict[str, str] = {} + + def set(self, key, value, nx=False, ex=None): + if nx and key in self.keys: + return None + self.keys[key] = value + return True + + def get(self, key): + return self.keys.get(key) + + +@pytest.fixture() +def store(monkeypatch: pytest.MonkeyPatch) -> _NxStore: + fake = _NxStore() + monkeypatch.setattr(dedup_mod, "_redis_client", lambda: fake) + return fake + + +@pytest.fixture() +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): + calls.append(payload) + return IngestResponse( + accepted=len(payload.events), + rejected=0, + total=len(payload.events), + events=[ + EventResult( + traceability_lot_code=event.traceability_lot_code, + cte_type=event.cte_type.value, + status="accepted", + event_id=f"evt-{idx}", + sha256_hash="hash", + chain_hash="chain", + ) + for idx, event in enumerate(payload.events) + ], + ) + + monkeypatch.setattr(edi_routes, "ingest_events", _fake_ingest_events) + return calls + + +@pytest.fixture() +def client(monkeypatch: pytest.MonkeyPatch, ingested: list) -> TestClient: + monkeypatch.delenv("EDI_REQUIRE_PARTNER_ID", raising=False) + monkeypatch.delenv(ALLOWLIST_ENV, raising=False) + monkeypatch.delenv("EDI_PARTNER_ALLOWLIST", raising=False) + rejection_log.reset_edi_rejections() + app = FastAPI() + app.include_router(edi_router) + app.dependency_overrides[get_ingestion_principal] = lambda: IngestionPrincipal( + key_id="test-key", scopes=["*"], auth_mode="test" + ) + with TestClient(app) as test_client: + yield test_client + + +def _future_variant() -> bytes: + raw = FIXTURE.read_text(encoding="utf-8") + assert FIXTURE_BSN in raw + future = (datetime.now(timezone.utc) + timedelta(days=5)).strftime("%Y%m%d") + return raw.replace(FIXTURE_BSN, f"BSN*00*ASN-12345*{future}*1200~").encode() + + +def _post(client: TestClient, raw: bytes, route: str): + return client.post( + f"/api/v1/ingest/edi{route}", + data={"traceability_lot_code": VALID_TLC, "tenant_id": TENANT}, + files={"file": ("asn_856.edi", raw, "application/edi-x12")}, + ) + + +@pytest.mark.parametrize("route", ["", "/document"]) +def test_reject_then_corrected_resend_is_accepted(client, store, ingested, route): + rejected = _post(client, _future_variant(), route) + assert rejected.status_code == 400 + detail = rejected.json()["detail"] + assert detail["error"] == "edi_date_in_future" + assert detail["needs_review"] is True + assert store.keys == {}, "ISA13 must not be reserved by a date rejection" + + assert ingested == [] + records = rejection_log.list_edi_rejections(TENANT) + assert [r["rejection_id"] for r in records] == [detail["rejection_id"]] + assert records[0]["needs_review"] is True + assert records[0]["reason"] == "edi_date_in_future" + + resend = _post(client, FIXTURE.read_bytes(), route) + assert resend.status_code == 201, resend.json() + assert resend.json()["ingestion_result"]["accepted"] == 1 + assert len(ingested) == 1 + event = ingested[0].events[0] + assert event.traceability_lot_code == VALID_TLC + assert event.timestamp.startswith("2026-03-10") + assert len(store.keys) == 1, "corrected resend reserves the ISA13" + + replay = _post(client, FIXTURE.read_bytes(), route) + assert replay.status_code == 409 + assert replay.json()["detail"]["error"] == "duplicate_interchange" + assert len(ingested) == 1 + + +def test_missing_date_rejection_is_logged_for_review(client, store): + raw = FIXTURE.read_text(encoding="utf-8").replace(FIXTURE_BSN, "BSN*00*ASN-12345**1200~") + response = _post(client, raw.encode(), "/document") + assert response.status_code == 400 + detail = response.json()["detail"] + assert detail["error"] == "missing_edi_date" + assert detail["needs_review"] is True + record = rejection_log.list_edi_rejections(TENANT)[0] + assert record["rejection_id"] == detail["rejection_id"] + assert record["needs_review"] is True + assert store.keys == {} + + +@pytest.mark.parametrize("route", ["", "/document"]) +def test_disallowed_sender_with_bad_date_gets_403_and_no_rejection(client, store, ingested, monkeypatch, route): + monkeypatch.setenv(ALLOWLIST_ENV, "SOMEONEELSE") + response = _post(client, _future_variant(), route) + assert response.status_code == 403, response.json() + assert response.json()["detail"]["error"] == "trading_partner_not_allowed" + assert rejection_log.list_edi_rejections(TENANT) == [] + assert store.keys == {} + assert ingested == [] diff --git a/services/ingestion/tests/test_edi_preserve_event_time.py b/services/ingestion/tests/test_edi_preserve_event_time.py new file mode 100644 index 000000000..e052489a0 --- /dev/null +++ b/services/ingestion/tests/test_edi_preserve_event_time.py @@ -0,0 +1,63 @@ +"""Data-free checks that EDI event time is never replaced by ingest time. + +Dates are constructed relative to now; no fixtures or records are used. +""" + +from __future__ import annotations + +import sys +from datetime import datetime, timedelta, timezone +from pathlib import Path + +import pytest + +service_dir = Path(__file__).parent.parent +sys.path.insert(0, str(service_dir)) + +pytest.importorskip("fastapi") + +from app.edi_ingestion.utils import _ship_timestamp_and_date # noqa: E402 +from fastapi import HTTPException # noqa: E402 + + +def _ccyymmdd(dt: datetime) -> str: + return dt.strftime("%Y%m%d") + + +def test_backfilled_date_keeps_real_event_time(): + real = datetime.now(timezone.utc) - timedelta(days=200) + timestamp, event_date = _ship_timestamp_and_date(_ccyymmdd(real), "1430") + parsed = datetime.fromisoformat(timestamp) + assert parsed.date() == real.date() + assert (parsed.hour, parsed.minute) == (14, 30) + assert event_date == real.date().isoformat() + assert timestamp[:10] == event_date + + +def test_recent_date_unchanged(): + real = datetime.now(timezone.utc) - timedelta(days=3) + timestamp, event_date = _ship_timestamp_and_date(_ccyymmdd(real), None) + assert timestamp[:10] == event_date == real.date().isoformat() + + +def test_future_date_is_flagged_not_replaced(): + future = datetime.now(timezone.utc) + timedelta(days=5) + with pytest.raises(HTTPException) as exc: + _ship_timestamp_and_date(_ccyymmdd(future), None) + assert exc.value.status_code == 400 + assert exc.value.detail["error"] == "edi_date_in_future" + assert exc.value.detail["needs_review"] is True + + +@pytest.mark.parametrize("raw", [None, "", " "]) +def test_missing_date_is_rejected(raw): + with pytest.raises(HTTPException) as exc: + _ship_timestamp_and_date(raw, None) + assert exc.value.status_code == 400 + assert exc.value.detail["error"] == "missing_edi_date" + + +def test_invalid_date_is_rejected(): + with pytest.raises(HTTPException) as exc: + _ship_timestamp_and_date("20261345", None) + assert exc.value.detail["error"] == "invalid_edi_date" 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 cfd73035d..7799bb865 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 @@ -23,6 +23,7 @@ import logging import sys +from datetime import datetime, timedelta, timezone from pathlib import Path import pytest @@ -50,8 +51,7 @@ # 106-char ISA header (index 105 is the segment terminator the parser # sniffs). Shared by every envelope below; only ISA13 varies. _ISA_TEMPLATE = ( - "ISA*00* *00* *ZZ*SENDERID *ZZ*RECEIVERID " - "*260310*1200*U*00401*{isa13}*0*P*>~" + "ISA*00* *00* *ZZ*SENDERID *ZZ*RECEIVERID *260310*1200*U*00401*{isa13}*0*P*>~" ) _GS_856 = "GS*SH*SENDERID*RECEIVERID*20260310*1200*1*X*004010~" @@ -374,9 +374,7 @@ def test_2978_unmapped_unit_warns_and_fails_closed_on_document_route( # --------------------------------------------------------------------------- -def test_2974_multi_sn1_856_quantity_is_summed_not_last_wins( - client: TestClient, captured_payload: dict -) -> None: +def test_2974_multi_sn1_856_quantity_is_summed_not_last_wins(client: TestClient, captured_payload: dict) -> None: """Repeated SN1 lines each become a CTE. Last-line-wins and summing both drop a lot from the recall record. """ @@ -417,9 +415,7 @@ def test_2974_multi_sn1_856_quantity_is_summed_not_last_wins( isa13="000029742", ) zero_extracted = _extract_fields_for_set("856", _parse_x12_segments(raw_zero.decode("utf-8"))) - assert zero_extracted["line_items"][0]["quantity"] is None, ( - "#2974: SN1**0*CA must not fabricate a quantity of 1.0" - ) + assert zero_extracted["line_items"][0]["quantity"] is None, "#2974: SN1**0*CA must not fabricate a quantity of 1.0" def test_blank_quantity_element_is_present_and_not_synthesized() -> None: @@ -447,14 +443,16 @@ def test_856_blank_sn1_quantity_is_rejected(client: TestClient, captured_payload The ship-from, ASN, BOL and other text elements contain no names, with SN1 quantity blank. The document must be refused with 422 - edi_non_positive_quantity and nothing persisted. + edi_non_positive_quantity and nothing persisted. The BSN date is + yesterday so the earlier event-date gate passes. """ + ship_date = (datetime.now(timezone.utc) - timedelta(days=1)).strftime("%Y%m%d").encode() raw = ( b"ISA*00* *00* *ZZ* *ZZ* " b"* * *U*00401* *0*P*>~" b"GS*SH*****1*X*004010~" b"ST*856*1~" - b"BSN*00**~" + b"BSN*00**" + ship_date + b"~" b"HL*1**S~" b"N1*SF*~" b"SN1***CA~" @@ -481,8 +479,7 @@ def test_856_blank_sn1_quantity_is_rejected(client: TestClient, captured_payload def test_sn1_lines_with_different_units_stay_separate_events(client: TestClient, captured_payload: dict) -> None: - """120 cases and 80 pounds are two CTEs. They are not 200 of either unit. - """ + """120 cases and 80 pounds are two CTEs. They are not 200 of either unit.""" raw = _interchange( [ _856_set( @@ -505,9 +502,7 @@ def test_sn1_lines_with_different_units_stay_separate_events(client: TestClient, ] -def test_2976_legacy_856_endpoint_refuses_a_batched_interchange( - client: TestClient, captured_payload: dict -) -> None: +def test_2976_legacy_856_endpoint_refuses_a_batched_interchange(client: TestClient, captured_payload: dict) -> None: """The legacy single-ASN endpoint extracts from the flat segment list, so a batched interchange would merge every ST/SE pair into its one event. It must refuse and point the caller at ``/document``. @@ -588,8 +583,7 @@ def test_2976_batched_interchange_never_half_lands(client: TestClient, captured_ def test_sn1_equivalent_unit_codes_stay_on_their_own_lines(): - """CA and CS both mean cases, and each SN1 still keeps its own quantity. - """ + """CA and CS both mean cases, and each SN1 still keeps its own quantity.""" from app.edi_ingestion.extractors import _extract_856_fields from app.edi_ingestion.parser import _parse_x12_segments from app.edi_ingestion.utils import _resolve_unit_code @@ -629,9 +623,7 @@ def test_two_unmappable_unit_codes_stay_on_their_lines(): assert [line["unit_of_measure"] for line in data["line_items"]] == ["QQ", "ZZ"] -def test_scalar_overrides_are_refused_for_a_batched_interchange( - client: TestClient, captured_payload: dict -): +def test_scalar_overrides_are_refused_for_a_batched_interchange(client: TestClient, captured_payload: dict): """A scalar form override describes ONE shipment. Applying it across a batch would overwrite each transaction set's own @@ -662,9 +654,7 @@ def test_scalar_overrides_are_refused_for_a_batched_interchange( assert "payload" not in captured_payload, "nothing may land when overrides are refused" -def test_scalar_overrides_still_work_for_a_single_transaction_set( - client: TestClient, captured_payload: dict -): +def test_scalar_overrides_still_work_for_a_single_transaction_set(client: TestClient, captured_payload: dict): """The guard must not break the ordinary single-shipment override path.""" raw = _interchange( [_856_set(control="0001", asn="ASN-SOLO", ship_from="Valley Fresh Farms", bol="BOL-SOLO")],