diff --git a/services/ingestion/app/edi_ingestion/routes.py b/services/ingestion/app/edi_ingestion/routes.py index bb2c94ad2..8d36158ff 100644 --- a/services/ingestion/app/edi_ingestion/routes.py +++ b/services/ingestion/app/edi_ingestion/routes.py @@ -1240,6 +1240,8 @@ def _preflight_all_sets() -> None: ) if slice_set in ("856", "861", "944", "945"): slice_extracted["traceability_lot_code"] = validated["tlc"] + if validated.get("warnings"): + slice_extracted["fsma_validation_warnings"] = validated["warnings"] slice_extracted["fsma_validation_status"] = "passed" except ValidationError as ve: errors = ve.errors(include_context=False) if slice_set in ("856", "861", "944", "945") else ve.errors() @@ -1312,6 +1314,7 @@ def _summarize(slice_set: str, slice_extracted: dict[str, Any]) -> dict[str, Any "product_description": slice_extracted.get("product_description"), "reference_document_number": slice_extracted.get("reference_document_number"), "fsma_validation_status": slice_extracted.get("fsma_validation_status"), + "fsma_validation_warnings": slice_extracted.get("fsma_validation_warnings", []), } if slice_set in ("856", "861", "944", "945"): summary.update( diff --git a/services/ingestion/app/edi_ingestion/validation.py b/services/ingestion/app/edi_ingestion/validation.py index 8b0d263ca..c2a848235 100644 --- a/services/ingestion/app/edi_ingestion/validation.py +++ b/services/ingestion/app/edi_ingestion/validation.py @@ -39,7 +39,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 FSMAEvent, FSMAEventType + from shared.schemas import _TLC_PRODUCTION_RE, FSMAEvent, FSMAEventType type_map = { "856": FSMAEventType.SHIPPING, @@ -56,18 +56,26 @@ def _validate_edi_as_fsma_event( lot_number=extracted.get("lot_number"), quantity=extracted.get("quantity"), ) - tlc = line.lot_number - # Follow the shared EDI schema TLC rule; originator lot versus - # GTIN-prefixed TLC remains a founder decision (#1104). + tlc = line.lot_number.strip() + # A-2170: a whitespace-only lot strips empty. Reject before the + # GTIN/UPC rebuild so we never emit a production-shaped "GTIN-" + # 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: - FSMAEvent.validate_tlc_format(tlc) + 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)}-{line.lot_number}" + tlc = f"{item_id.zfill(14)}-{tlc}" break elif transaction_set in ("856", "861"): # The line lot is the TLC verbatim. Do not prefix a GTIN (#1104). 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 7799bb865..a69e42f86 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 @@ -197,8 +197,9 @@ def _post( # --------------------------------------------------------------------------- +@pytest.mark.parametrize("tlc", (_VALID_FSMA_TLC, "A" * 3)) def test_2976_multi_st_interchange_emits_one_event_per_transaction_set( - client: TestClient, captured_payload: dict + client: TestClient, captured_payload: dict, tlc: str ) -> None: """A batched interchange must produce one CTE per ST/SE pair with no cross-shipment field bleed. @@ -236,7 +237,7 @@ def test_2976_multi_st_interchange_emits_one_event_per_transaction_set( isa13="000029761", ) - response = _post(client, raw) + response = _post(client, raw, extra_data={"traceability_lot_code": tlc}) assert response.status_code == 201, response.text payload = captured_payload["payload"] @@ -262,6 +263,10 @@ def test_2976_multi_st_interchange_emits_one_event_per_transaction_set( body = response.json() assert body["extracted"]["transaction_set_count"] == 3 + if tlc == "A" * 3: + for transaction in body["extracted"]["transaction_sets"]: + assert transaction["fsma_validation_status"] == "passed" + assert any("production pattern" in warning for warning in transaction["fsma_validation_warnings"]) def test_2976_ge01_disagreeing_with_st_count_is_rejected(client: TestClient, captured_payload: dict) -> None: @@ -558,8 +563,8 @@ def test_2976_batched_interchange_never_half_lands(client: TestClient, captured_ "/api/v1/ingest/edi/document", params={"strict": "false"}, data={ - # Fails FSMAEvent.validate_tlc_format for every set. - "traceability_lot_code": "not-a-gtin-tlc", + # Below the shared TLC minimum for every set; pattern misses only warn. + "traceability_lot_code": "A" * 2, "tenant_id": TEST_TENANT_ID, }, files={"file": ("batch.edi", raw, "application/edi-x12")}, @@ -670,3 +675,59 @@ def test_scalar_overrides_still_work_for_a_single_transaction_set(client: TestCl assert response.status_code in (200, 201), response.text assert captured_payload["payload"].events[0].quantity == 42.0 + + +@pytest.mark.parametrize("transaction_set", ("856", "861", "944", "945")) +def test_tlc_pattern_miss_is_accepted_with_persisted_warning(transaction_set): + validated = edi_ingestion_routes._validate_edi_as_fsma_event( + {"lot_number": " \t" + "A" * 3 + "\n", "quantity": 1}, + transaction_set, + "A" * 3, + ) + assert validated["tlc"] == "A" * 3 + assert any("production pattern" in warning for warning in validated["warnings"]) + + +@pytest.mark.parametrize("transaction_set", ("944", "945")) +@pytest.mark.parametrize("lot", ("A" * 3, "0" * 14 + "A", "A")) +def test_warehouse_tlc_prefix_rebuild_is_reachable(transaction_set, lot): + validated = edi_ingestion_routes._validate_edi_as_fsma_event( + {"lot_number": lot, "quantity": 1, "gtin": "0" * 14}, + transaction_set, + lot, + ) + expected = lot if lot.startswith("0" * 14) else "0" * 14 + "-" + lot.strip() + assert validated["tlc"] == expected + assert validated["warnings"] == [] + + +@pytest.mark.parametrize("transaction_set", ("944", "945")) +@pytest.mark.parametrize("item_id", ("A" * 14, "0" * 3)) +def test_invalid_warehouse_prefix_keeps_lot_and_warning(transaction_set, item_id): + validated = edi_ingestion_routes._validate_edi_as_fsma_event( + {"lot_number": "A" * 3, "quantity": 1, "gtin": item_id, "upc": item_id}, + transaction_set, + "A" * 3, + ) + assert validated["tlc"] == "A" * 3 + assert any("production pattern" in warning for warning in validated["warnings"]) + + +@pytest.mark.parametrize("transaction_set", ("944", "945")) +@pytest.mark.parametrize("lot", ("", " ", " ", "\t")) +def test_warehouse_stripped_empty_lot_rejected_before_rebuild(transaction_set, lot): + """A-2170: empty or whitespace-only 944/945 lots reject like a missing lot. + + Must fail before the GTIN/UPC rebuild; otherwise a blank lot becomes + ``{gtin}-`` and FSMAEvent accepts it with no warning. + """ + from pydantic import ValidationError + + with pytest.raises(ValidationError) as excinfo: + edi_ingestion_routes._validate_edi_as_fsma_event( + {"lot_number": lot, "quantity": 1, "gtin": "0" * 14}, + transaction_set, + "caller-tlc", + ) + assert any(err["loc"][-1] == "lot_number" for err in excinfo.value.errors()) + diff --git a/services/ingestion/tests/test_epcis_cleartext_logging.py b/services/ingestion/tests/test_epcis_cleartext_logging.py index 96de0d3db..0fae0d152 100644 --- a/services/ingestion/tests/test_epcis_cleartext_logging.py +++ b/services/ingestion/tests/test_epcis_cleartext_logging.py @@ -117,9 +117,12 @@ def test_fsma_rejection_logs_omit_private_values(ingest, warning, monkeypatch, c monkeypatch.setattr(pmod, "_fsma_strict_mode", lambda: True) monkeypatch.setattr(pmod, "_epcis_store", {}) monkeypatch.setattr(pmod, "_epcis_idempotency_index", {}) + # TLC production-pattern misses are warnings now, so a valid-length TLC alone + # no longer triggers fsma_validation_failed. Blank eventTime still hard-fails + # FSMAEvent while keeping TLC in the audited payload for cleartext checks. event = { "type": "ObjectEvent", - "eventTime": datetime.now(timezone.utc).isoformat(), + "eventTime": " ", "action": "OBSERVE", "bizStep": "receiving", "ilmd": {"fsma:traceabilityLotCode": TLC}, @@ -133,7 +136,7 @@ def test_fsma_rejection_logs_omit_private_values(ingest, warning, monkeypatch, c assert warning in [record.getMessage() for record in caplog.records] assert audited[0][1] == TENANT assert audited[0][2]["tlc"] == TLC - assert any(error["loc"] == ("tlc",) for error in audited[0][0]) + assert any(error["loc"] == ("event_time",) for error in audited[0][0]) _assert_clean(caplog) @@ -161,10 +164,12 @@ def _raise(db_session, tenant_id, prepared): def test_pydantic_validation_failure_log_omits_private_values(monkeypatch, caplog): audited = [] monkeypatch.setattr(vmod, "_audit_log_validation_failure", lambda e, t, n: audited.append((e, t, n))) + # Valid-length TLC is accepted with a production-pattern warning; blank + # event_time still hard-fails while keeping TLC in the audited payload. normalized = { "event_type": "receiving", "tlc": TLC, - "event_time": datetime.now(timezone.utc).isoformat(), + "event_time": " ", } with caplog.at_level(logging.WARNING): result = vmod._validate_as_fsma_event(normalized, tenant_id=TENANT) @@ -172,5 +177,7 @@ def test_pydantic_validation_failure_log_omits_private_values(monkeypatch, caplo assert result is None assert "fsma_event_validation_failed" in _log_text(caplog) assert audited[0][1:] == (TENANT, normalized) - assert any(error["loc"] == ("tlc",) and error["type"] == "value_error" for error in audited[0][0]) + assert any( + error["loc"] == ("event_time",) and error["type"] == "value_error" for error in audited[0][0] + ) _assert_clean(caplog) diff --git a/services/ingestion/tests/test_r3_restored_fixtures_e2e.py b/services/ingestion/tests/test_r3_restored_fixtures_e2e.py index f5a30eb25..fb062a9f2 100644 --- a/services/ingestion/tests/test_r3_restored_fixtures_e2e.py +++ b/services/ingestion/tests/test_r3_restored_fixtures_e2e.py @@ -38,7 +38,6 @@ from app.edi_ingestion.validation import _validate_edi_as_fsma_event # noqa: E402 from app.fda_export_service import FDA_COLUMNS, _generate_csv # noqa: E402 from app.webhook_router_v2 import _validate_event_kdes # noqa: E402 -from pydantic import ValidationError # noqa: E402 from shared.data_gap_tracking import ( # noqa: E402 GAP_KIND_MISSING_KDE, decide_transition, @@ -102,9 +101,7 @@ def test_csv_harvesting_fixture_end_to_end(): headers, rows = _parse_sheet(sheet) assert headers == FDA_COLUMNS - assert [r["Traceability Lot Code (TLC)"] for r in rows] == [ - s["traceability_lot_code"] for s in source - ] + assert [r["Traceability Lot Code (TLC)"] for r in rows] == [s["traceability_lot_code"] for s in source] for row, src in zip(rows, source): assert row["Event Type (CTE)"].lower() == "harvesting" assert row["Product Description"] == src["product_description"] @@ -133,12 +130,11 @@ def test_edi_856_fixture_end_to_end(): merged = {**header, **lines[0], "edi_line_index": 1} tlc = extracted["asn_number"] - with pytest.raises(ValidationError): - _validate_edi_as_fsma_event(merged, "856", tlc, None) + validated = _validate_edi_as_fsma_event(merged, "856", tlc, None) + assert validated["tlc"] == tlc.strip() + assert any("production pattern" in warning for warning in validated["warnings"]) - event = edi_routes._build_ingest_event_for_set( - "856", merged, tlc, None, None, None, None, segments=segments - ) + event = edi_routes._build_ingest_event_for_set("856", merged, tlc, None, None, None, None, segments=segments) assert event.cte_type.value == "shipping" assert event.quantity == lines[0]["quantity"] assert event.kdes["ship_from_gln"] == expected["ship_from_gln"] diff --git a/services/shared/fsma_validation.py b/services/shared/fsma_validation.py index f2e2dd56c..2bdc15cd2 100644 --- a/services/shared/fsma_validation.py +++ b/services/shared/fsma_validation.py @@ -17,7 +17,7 @@ detect_identifier_type – auto-detect identifier kind from format validate_gln – GLN-13 with GS1 check digit validate_gtin – GTIN-8/12/13/14 with GS1 check digit -validate_tlc – Traceability Lot Code (GTIN-14 prefix in strict mode) +validate_tlc – Traceability Lot Code (GTIN-14 pattern warning in strict mode) validate_sscc – SSCC-18 with GS1 check digit validate_identifier – auto-detecting dispatcher validate_batch – batch validation helper @@ -187,14 +187,14 @@ def validate_tlc(tlc: str, *, strict: bool = False) -> ValidationResult: """ Validate Traceability Lot Code (TLC). - A production TLC must match the GTIN-14 prefix pattern: + Strict mode checks the production GTIN-14 prefix pattern: ``^\\d{14}[A-Za-z0-9\\-\\.]+$`` (14 numeric digits followed by an - alphanumeric lot suffix). In non-strict mode the format is more - permissive (minimum 3 chars, alphanumeric + common separators). + alphanumeric lot suffix). Strict mismatches are warnings, not failures. + Both modes require at least 3 characters and warn on unusual characters. Args: tlc: Raw TLC string to validate. - strict: When True, enforce the production GTIN-14 prefix pattern. + strict: When True, warn on a production GTIN-14 prefix pattern mismatch. """ raw = str(tlc) if tlc else "" cleaned = raw.strip() @@ -209,28 +209,14 @@ def validate_tlc(tlc: str, *, strict: bool = False) -> ValidationResult: warnings: List[str] = [] - if strict: - # Production mode: GTIN-14 prefix + alphanumeric lot suffix - if not _TLC_PRODUCTION_PATTERN.match(cleaned): - return ValidationResult( - is_valid=False, - identifier_type=IdentifierType.TLC, - original_value=raw, - errors=[ - ValidationError( - field="tlc", - message=( - "TLC does not match production pattern " - "(expected 14-digit GTIN prefix + alphanumeric lot suffix, " - "e.g. '00012345678901-LotA')" - ), - ) - ], - ) - else: - # Permissive mode: reject obviously invalid characters - if not re.match(r"^[A-Za-z0-9\-\._/]+$", cleaned): - warnings.append("TLC contains unusual characters; consider strict validation") + if strict and not _TLC_PRODUCTION_PATTERN.match(cleaned): + warnings.append( + "Warning: TLC does not match production pattern " + "(14-digit GTIN prefix + alphanumeric lot suffix); this is not a validation failure" + ) + + if not re.match(r"^[A-Za-z0-9\-\._/]+$", cleaned): + warnings.append("TLC contains unusual characters; consider strict validation") # Warn on very long TLCs if len(cleaned) > 50: diff --git a/services/shared/schemas.py b/services/shared/schemas.py index 517ec0ef0..f4f3fc8b2 100644 --- a/services/shared/schemas.py +++ b/services/shared/schemas.py @@ -11,7 +11,9 @@ from urllib.parse import unquote from uuid import UUID, uuid4 -from pydantic import BaseModel, Field, field_validator +from pydantic import BaseModel, Field, field_validator, model_validator + +from .fsma_validation import validate_tlc # Example constants to avoid duplication in schema examples EX_SUBJECT = "food facilities on the FTL" @@ -372,7 +374,7 @@ class FSMAEvent(BaseModel): model before being emitted to Kafka. Required KDEs (rejection triggers): - - ``tlc`` – GTIN-14 + lot, LGTIN class URN, or GS1 Digital Link + - ``tlc`` – at least 3 characters after stripping whitespace - ``event_time`` – ISO 8601 timestamp (required) """ @@ -386,11 +388,8 @@ class FSMAEvent(BaseModel): ) tlc: str = Field( ..., - description="Traceability Lot Code (GTIN-14 + lot, LGTIN class URN, or GS1 Digital Link)", - # 15 is the GTIN-14 form (14 digits + 1 lot character). An LGTIN class - # URN is at least 36 characters and a GS1 Digital Link at least 31, - # so this floor does not reject either lot-level form. - min_length=15, + description="Traceability Lot Code (production-pattern mismatches warn only)", + min_length=3, ) product_description: str = Field( default="Unknown Product", @@ -409,6 +408,8 @@ class FSMAEvent(BaseModel): reference_document_number: Optional[str] = Field(None, description="Source document reference number") tenant_id: Optional[str] = Field(None, description="Tenant identifier for multi-tenancy") + warnings: List[str] = Field(default_factory=list) + # Legacy fields retained for backward compatibility with document extraction date: Optional[str] = Field(None, description="Event date in YYYY-MM-DD format") cte_type: Optional[CTEType] = Field(None, description="Legacy CTE type (use event_type instead)") @@ -424,18 +425,23 @@ class FSMAEvent(BaseModel): @field_validator("tlc") @classmethod def validate_tlc_format(cls, v: str) -> str: - """Accept GTIN-14 + lot, an LGTIN class URN, or a GS1 Digital Link.""" - if _TLC_PRODUCTION_RE.fullmatch(v): - return v - for pattern in (_TLC_LGTIN_RE, _TLC_DIGITAL_LINK_RE): - matched = pattern.fullmatch(v) - if matched is not None and _decoded_ai10_lot_within_limit(matched.group("lot")): - return v - raise ValueError( - f"TLC '{v}' does not match FSMA 204 production pattern " - "(accepted formats: a fourteen-digit GTIN prefix followed by a lot " - "suffix, an LGTIN class URN, or a GS1 Digital Link)" - ) + """Strip surrounding whitespace; reject blank or short TLCs. + + Production-pattern advisories are collected on each model instance. + """ + cleaned = v.strip() + if not cleaned: + raise ValueError("TLC must not be blank") + if len(cleaned) < 3: + raise ValueError("TLC must be at least 3 characters") + return cleaned + + @model_validator(mode="after") + def collect_tlc_warnings(self) -> "FSMAEvent": + for warning in validate_tlc(self.tlc, strict=True).warnings: + if warning not in self.warnings: + self.warnings.append(warning) + return self @field_validator("event_time") @classmethod diff --git a/services/shared/tests/test_break_fsma_auth_guards.py b/services/shared/tests/test_break_fsma_auth_guards.py index 6301d57d6..235d0096d 100644 --- a/services/shared/tests/test_break_fsma_auth_guards.py +++ b/services/shared/tests/test_break_fsma_auth_guards.py @@ -169,8 +169,13 @@ def test_validate_gln_gtin_sscc_mistyped_fail_closed() -> None: def test_tlc_strict_boundary() -> None: from shared.fsma_validation import validate_tlc - assert validate_tlc("00012345678901", strict=True).is_valid is False - assert validate_tlc("00012345678901-LotA", strict=True).is_valid is True + result = validate_tlc("0" * 14, strict=True) + assert result.is_valid is True + assert result.errors == [] + assert any("production pattern" in w for w in result.warnings) + matched = validate_tlc("0" * 14 + "A", strict=True) + assert matched.is_valid is True + assert not any("production pattern" in w for w in matched.warnings) # --- auth key stores fail closed on non-string credentials ----------------- diff --git a/tests/test_fsma_validation.py b/tests/test_fsma_validation.py index 71d2d83c6..b0a0480ca 100644 --- a/tests/test_fsma_validation.py +++ b/tests/test_fsma_validation.py @@ -234,33 +234,49 @@ def test_tlc_normalization(self): assert result.is_valid assert result.normalized_value == "lot-abc-123" - def test_tlc_too_short(self): + @pytest.mark.parametrize("strict", [False, True]) + def test_tlc_too_short(self, strict): """Test TLC below minimum length.""" - result = validate_tlc("AB") + result = validate_tlc("A" * 2, strict=strict) assert not result.is_valid assert any("at least 3" in err.message for err in result.errors) def test_tlc_strict_mode_valid(self): """Test TLC validation in strict mode with valid GTIN-14 prefix.""" - result = validate_tlc("00012345678901-LotA", strict=True) - assert result.is_valid + result = validate_tlc("0" * 14 + "A", strict=True) + assert result.is_valid is True + assert result.errors == [] + assert not any("production pattern" in w for w in result.warnings) def test_tlc_strict_mode_invalid(self): - """Test TLC in strict mode rejects non-GTIN-14 prefix.""" - result = validate_tlc("L-2024-01-15-A", strict=True) - assert not result.is_valid - assert any("production pattern" in err.message for err in result.errors) + """A production-pattern miss is a warning, not a failure.""" + result = validate_tlc("A" * 3, strict=True) + assert result.is_valid is True + assert result.errors == [] + assert any("production pattern" in w and "Warning" in w for w in result.warnings) + + @pytest.mark.parametrize("tlc,pattern_warning", [("A" * 3, True), ("0" * 14 + "A", False)]) + def test_tlc_strict_pattern_shapes(self, tlc, pattern_warning): + """Structural padding represents format only, not a company prefix.""" + result = validate_tlc(tlc, strict=True) + assert result.is_valid is True + assert result.errors == [] + assert any("production pattern" in w for w in result.warnings) is pattern_warning def test_tlc_strict_mode_dots_valid(self): """Test TLC strict mode accepts dots in lot suffix.""" - result = validate_tlc("00012345678901-ABC.DEF", strict=True) + result = validate_tlc("0" * 14 + "A" * 3 + "." + "B" * 3, strict=True) assert result.is_valid - def test_tlc_unusual_chars_warning(self): - """Test TLC with unusual characters produces warning in permissive mode.""" - result = validate_tlc("LOT: ABC123") # colon + space + @pytest.mark.parametrize("strict", [False, True]) + def test_tlc_unusual_chars_warning(self, strict): + """Both modes warn on unusual characters without adding errors.""" + result = validate_tlc("A" * 3 + ": " + "B", strict=strict) # colon + space assert result.is_valid + assert result.errors == [] assert any("unusual characters" in w for w in result.warnings) + if strict: + assert any("production pattern" in w for w in result.warnings) def test_tlc_very_long_warning(self): """Test very long TLC produces warning.""" diff --git a/tests/test_shared/test_fsmaevent_lot_level_tlc.py b/tests/test_shared/test_fsmaevent_lot_level_tlc.py index a31397c0c..a438facbf 100644 --- a/tests/test_shared/test_fsmaevent_lot_level_tlc.py +++ b/tests/test_shared/test_fsmaevent_lot_level_tlc.py @@ -1,17 +1,18 @@ """Data-free checks for FSMAEvent lot-level TLC acceptance. -No identifier values, GTINs, lots, or records. These tests reject blank -input, bare scheme prefixes, a bare percent, and a trailing newline on -inputs that are already rejected. They also check pattern shape: ASCII -digits, an optional Digital Link stem before /01, and that the AI (10) -length limit is applied to the percent-decoded lot. +No identifier records. Blank and short input fail; production-pattern misses +warn and strip surrounding whitespace. Pattern-shape checks remain independent of +acceptance, including anchoring and the percent-decoded AI (10) limit. """ from __future__ import annotations import re +import warnings +from typing import Annotated import pytest +from pydantic import TypeAdapter from shared.schemas import ( _DL_CUSTOM_STEM, _DL_LOT_RAW_CLASS, @@ -30,31 +31,39 @@ "urn:epc:class:lgtin:..", "https:///01//10/", ) -_ALREADY_REJECTED = (*_BLANK, *_BARE_PREFIXES) -@pytest.mark.parametrize("value", _BLANK) +@pytest.mark.parametrize("value", (*_BLANK, *(value + "\n" for value in _BLANK))) def test_empty_and_whitespace_tlc_rejected(value: str) -> None: - with pytest.raises(ValueError, match="does not match FSMA 204 production pattern"): + with pytest.raises(ValueError, match="blank|at least 3 characters"): FSMAEvent.validate_tlc_format(value) -@pytest.mark.parametrize("value", _BARE_PREFIXES) -def test_bare_lot_level_prefixes_rejected(value: str) -> None: - with pytest.raises(ValueError, match="does not match FSMA 204 production pattern"): +@pytest.mark.parametrize("value", ("A", "A" * 2, " A ")) +def test_short_tlc_rejected(value: str) -> None: + with pytest.raises(ValueError, match="at least 3 characters"): FSMAEvent.validate_tlc_format(value) -@pytest.mark.parametrize("value", _ALREADY_REJECTED) -def test_trailing_newline_rejected_on_inputs_already_rejected(value: str) -> None: - with pytest.raises(ValueError, match="does not match FSMA 204 production pattern"): - FSMAEvent.validate_tlc_format(value + "\n") +@pytest.mark.parametrize( + "value", + ( + "A" * 3, + *_BARE_PREFIXES, + *(prefix + "\n" for prefix in _BARE_PREFIXES), + *(prefix + "%" for prefix in _BARE_PREFIXES), + ), +) +def test_production_pattern_miss_strips_value(value: str) -> None: + assert FSMAEvent.validate_tlc_format(value) == value.strip() -@pytest.mark.parametrize("prefix", _BARE_PREFIXES) -def test_bare_percent_rejected_on_prefixes(prefix: str) -> None: - with pytest.raises(ValueError, match="does not match FSMA 204 production pattern"): - FSMAEvent.validate_tlc_format(prefix + "%") +def test_production_pattern_match_has_no_warning() -> None: + value = "0" * 14 + "A" + with warnings.catch_warnings(record=True) as caught: + warnings.simplefilter("always") + assert FSMAEvent.validate_tlc_format(value) == value + assert not caught def test_lot_level_patterns_are_anchored() -> None: @@ -105,3 +114,31 @@ def test_ai10_length_limit_is_applied_after_percent_decode() -> None: assert not _decoded_ai10_lot_within_limit("%41" * 21) assert not _decoded_ai10_lot_within_limit("") assert not _decoded_ai10_lot_within_limit("%") + + +def test_tlc_field_minimum_does_not_reject_plain_three_character_value() -> None: + field = FSMAEvent.model_fields["tlc"] + adapter = TypeAdapter(Annotated[field.annotation, *field.metadata]) + assert adapter.validate_python("A" * 3) == "A" * 3 + with pytest.raises(ValueError): + adapter.validate_python("A" * 2) + + +@pytest.mark.parametrize("value", ("A" * 3, " \t" + "A" * 3 + "\n")) +def test_each_tlc_event_records_pattern_warning(value: str) -> None: + from datetime import datetime, timezone + + for _ in range(2): + event = FSMAEvent(event_type="SHIPPING", tlc=value, event_time=datetime.now(timezone.utc).isoformat()) + assert event.tlc == value.strip() + assert any("production pattern" in warning for warning in event.warnings) + assert event.model_dump()["warnings"] == event.warnings + restored = FSMAEvent.model_validate(event.model_dump()) + assert restored.warnings == event.warnings + + +def test_production_tlc_event_has_no_advisory() -> None: + from datetime import datetime, timezone + + event = FSMAEvent(event_type="SHIPPING", tlc="0" * 14 + "A", event_time=datetime.now(timezone.utc).isoformat()) + assert event.warnings == []