From 8e5a985cdf127ab5084a64455df8441783b653af Mon Sep 17 00:00:00 2001 From: Christopher Sellers <1@christophersellers.com> Date: Tue, 27 Oct 2026 21:33:28 -0700 Subject: [PATCH 1/6] fix(ingest): clear missing-column errors and a per-row disposition log for partner CSVs SEL-549. Every data row is now accounted for as accepted, rejected, flagged or skipped, and a file missing required columns fails with one error naming them. Signed-off-by: Christopher Sellers <1@christophersellers.com> --- services/ingestion/app/csv_templates.py | 386 ++++++++++++++++-- services/ingestion/app/webhook_models.py | 47 +++ .../test_csv_partner_file_hardening_sel549.py | 116 ++++++ 3 files changed, 518 insertions(+), 31 deletions(-) create mode 100644 services/ingestion/tests/test_csv_partner_file_hardening_sel549.py diff --git a/services/ingestion/app/csv_templates.py b/services/ingestion/app/csv_templates.py index d67f089ed..aae543646 100644 --- a/services/ingestion/app/csv_templates.py +++ b/services/ingestion/app/csv_templates.py @@ -58,6 +58,7 @@ from .webhook_models import ( REQUIRED_KDES_BY_CTE, CsvIngestResponse, + CsvRowDispositionEntry, CteTypeSuggestionResult, HeaderMappingEngineResult, HeaderMappingSuggestionResult, @@ -715,10 +716,143 @@ def _parse_input_quantities( return inputs, None, None +# --------------------------------------------------------------------------- +# SEL-549: partner-file header checks + row disposition log +# --------------------------------------------------------------------------- +# Every physical CSV row the partner sent must be accounted for as accepted, +# rejected (with reason), flagged, or skipped — never silently dropped. +# Header guesses (Jev applied mappings) are logged as flagged field decisions +# without echoing cell values. + + +_REQUIRED_PARTNER_FIELD_GROUPS: tuple[tuple[str, tuple[str, ...]], ...] = ( + ("traceability_lot_code", _LOT_CODE_COLUMNS), + ("quantity", _QUANTITY_COLUMNS), + ("unit_of_measure", _UNIT_COLUMNS), + ("event_date", tuple(_DATE_FIELDS)), +) + + +def _normalized_header_names(header: Optional[Sequence[Optional[str]]]) -> list[str]: + if not header: + return [] + return [normalize_csv_header(name) for name in header if name is not None and str(name).strip()] + + +def _header_covers_field( + normalized_headers: set[str], + header_mapping: Optional[Mapping[str, str]], + aliases: Sequence[str], +) -> bool: + """True when the file (or an applied mapping) can supply one of ``aliases``.""" + alias_set = set(aliases) + if normalized_headers & alias_set: + return True + if not header_mapping: + return False + for source_key, canonical in header_mapping.items(): + if canonical in alias_set and (not source_key or source_key in normalized_headers): + return True + return False + + +def _missing_required_partner_columns( + header: Optional[Sequence[Optional[str]]], + header_mapping: Optional[Mapping[str, str]] = None, + fallback_cte: Optional[str] = None, +) -> list[str]: + """Canonical names of required partner columns the header cannot supply.""" + normalized = set(_normalized_header_names(header)) + missing: list[str] = [] + for canonical, aliases in _REQUIRED_PARTNER_FIELD_GROUPS: + if not _header_covers_field(normalized, header_mapping, aliases): + missing.append(canonical) + # cte_type is not checked here: a file without one keeps its per-row + # "No CTE type" rejection, which carries the Jev CTE suggestion. + return missing + + +def _unrecognized_partner_columns( + header: Optional[Sequence[Optional[str]]], + header_mapping: Optional[Mapping[str, str]] = None, +) -> list[str]: + """Partner columns outside the parser's recognised set (after applied mappings).""" + normalized = _normalized_header_names(header) + mapped_sources = set((header_mapping or {}).keys()) + unknown: list[str] = [] + seen: set[str] = set() + for name in normalized: + if name in seen: + continue + seen.add(name) + if name in _RECOGNIZED_COLUMNS: + continue + if name in mapped_sources: + continue + unknown.append(name) + return unknown + + +def _format_missing_columns_error( + missing: Sequence[str], + header: Optional[Sequence[Optional[str]]], +) -> str: + """Clear, actionable error naming each missing required column and aliases.""" + alias_by_canonical = {canonical: aliases for canonical, aliases in _REQUIRED_PARTNER_FIELD_GROUPS} + alias_by_canonical["cte_type"] = tuple(_CTE_TYPE_COLUMNS) + parts: list[str] = [] + for name in missing: + aliases = alias_by_canonical.get(name, (name,)) + alias_list = ", ".join(aliases) + parts.append(f"{name} (accepted column names: {alias_list})") + present = ", ".join(_normalized_header_names(header)) or "(none)" + return "Missing required columns for partner CSV ingest: " + "; ".join(parts) + f". Present columns: {present}" + + +def _is_data_record(record: Mapping[Optional[str], object]) -> bool: + """True for a record the row loop would read as data (not blank, not a comment).""" + values = [str(v or "").strip() for k, v in record.items() if k is not None] + if not any(values): + return False + return not values[0].startswith("#") + + +def _append_row_log( + row_log: Optional[list[CsvRowDispositionEntry]], + *, + row_number: int, + disposition: str, + reason: str, + field: Optional[str] = None, + detail: Optional[str] = None, + guessed_as: Optional[str] = None, +) -> None: + if row_log is None: + return + entry = CsvRowDispositionEntry( + row_number=row_number, + disposition=disposition, + reason=reason, + field=field, + detail=detail, + guessed_as=guessed_as, + ) + row_log.append(entry) + logger.info( + "csv_row_disposition row=%s disposition=%s reason=%s field=%s guessed_as=%s", + row_number, + disposition, + reason, + field or "", + guessed_as or "", + ) + + def parse_csv_rows( text: str, fallback_cte: Optional[str] = None, header_mapping: Optional[Mapping[str, str]] = None, + row_log: Optional[list[CsvRowDispositionEntry]] = None, ) -> tuple[list[IngestEvent], list[str], list[str]]: """Parse CSV text into ``IngestEvent`` objects. @@ -736,6 +870,10 @@ def parse_csv_rows( carries. The source column is left in place so the raw column still reaches ``kdes`` and nothing is lost. + ``row_log``, when provided, receives a ``CsvRowDispositionEntry`` for every + input row and for file-level header decisions (SEL-549). Cell values are + never written into the log. + Returns ``(events, parse_errors, kde_warnings)``. """ reader = csv.DictReader(io.StringIO(text)) @@ -750,6 +888,7 @@ def parse_csv_rows( # quote) is rejected whole with its line named, never partially ingested. records: list[dict] = [] skipped_overflow = 0 + header = None try: header = reader.fieldnames if header and any(name.strip() for name in header): @@ -757,14 +896,19 @@ def parse_csv_rows( for position, name in enumerate(header, start=1): normalized = normalize_csv_header(name) if normalized in seen_columns: - return ( - [], - [ - f"CSV has duplicate column {normalized!r} at positions " - f"{seen_columns[normalized]} and {position}; column names must be unique" - ], - [], + detail = ( + f"CSV has duplicate column {normalized!r} at positions " + f"{seen_columns[normalized]} and {position}; column names must be unique" ) + _append_row_log( + row_log, + row_number=1, + disposition="rejected", + reason="duplicate_column", + field=normalized, + detail=detail, + ) + return ([], [detail], []) seen_columns[normalized] = position for record in reader: records.append(record) @@ -783,25 +927,93 @@ def parse_csv_rows( except csv.Error as exc: line = reader.reader.line_num logger.warning("csv_unreadable line=%s error=%s", line, exc) - return ( - [], - [ - f"Line {line}: CSV could not be read. Check for carriage-return-only line endings, " - f"an unclosed quote, or a field longer than {csv.field_size_limit()} characters" - ], - [], + detail = ( + f"Line {line}: CSV could not be read. Check for carriage-return-only line endings, " + f"an unclosed quote, or a field longer than {csv.field_size_limit()} characters" ) + _append_row_log( + row_log, + row_number=line or 0, + disposition="rejected", + reason="unreadable_csv", + detail=detail, + ) + return ([], [detail], []) if header is not None and not any(name.strip() for name in header): - return [], ["CSV has no header row: the first line must list the column names"], [] + detail = "CSV has no header row: the first line must list the column names" + _append_row_log( + row_log, + row_number=1, + disposition="rejected", + reason="missing_header", + detail=detail, + ) + return ([], [detail], []) + + # SEL-549: fail fast with a clear error naming each missing required column + # (and the aliases we accept) rather than emitting one opaque per-row error. + # Only a file with data rows is checked; empty and header-only files keep + # their existing "no data rows" handling. + has_data = any(_is_data_record(r) for r in records) + missing_columns = ( + _missing_required_partner_columns(header, header_mapping, fallback_cte) if header and has_data else [] + ) + if missing_columns: + detail = _format_missing_columns_error(missing_columns, header) + _append_row_log( + row_log, + row_number=0, + disposition="rejected", + reason="missing_required_columns", + detail=detail, + ) + for name in missing_columns: + _append_row_log( + row_log, + row_number=0, + disposition="rejected", + reason="missing_column", + field=name, + detail=f"Required column {name!r} is absent from the partner file header", + ) + return ([], [detail], []) + + for bad in _unrecognized_partner_columns(header, header_mapping): + _append_row_log( + row_log, + row_number=0, + disposition="flagged", + reason="unrecognized_column", + field=bad, + detail=( + f"Column {bad!r} is not a recognised ingest field; values still pass through into KDEs when present" + ), + ) for row_num, row in enumerate(records, start=2): # _file_cte_type_for_jev reads exactly the rows before this cut-off. if row_num - 1 > MAX_CSV_ROWS: if skipped_overflow: - parse_errors.append( + overflow_detail = ( f"Row {row_num}: {skipped_overflow} more row(s) after this one were skipped and not ingested" ) - parse_errors.append(f"Row {row_num}: CSV exceeds maximum row limit of {MAX_CSV_ROWS}") + parse_errors.append(overflow_detail) + _append_row_log( + row_log, + row_number=row_num, + disposition="skipped", + reason="row_limit_overflow", + detail=overflow_detail, + ) + limit_detail = f"Row {row_num}: CSV exceeds maximum row limit of {MAX_CSV_ROWS}" + parse_errors.append(limit_detail) + _append_row_log( + row_log, + row_number=row_num, + disposition="rejected", + reason="row_limit_exceeded", + detail=limit_detail, + ) break # DictReader files values past the last header column under the None # restkey. A non-blank one rejects the row below rather than being @@ -813,22 +1025,79 @@ def parse_csv_rows( value = row.get(source_key) if value and not _row_provides(row, canonical_field): row[canonical_field] = value + _append_row_log( + row_log, + row_number=row_num, + disposition="flagged", + reason="header_guess_applied", + field=source_key, + guessed_as=canonical_field, + detail=( + f"Applied header mapping {source_key!r} -> {canonical_field!r} " + "(cell value omitted from log)" + ), + ) first_val = next(iter(row.values()), "") if first_val and str(first_val).startswith("#"): + _append_row_log( + row_log, + row_number=row_num, + disposition="skipped", + reason="comment_row", + detail="Comment row (first value starts with #) was not ingested", + ) continue if any(value.strip() for value in extra_values): - parse_errors.append( + detail = ( f"Row {row_num}: has more fields than the header " f"({len(header) + len(extra_values)} fields, header has {len(header)})" ) + parse_errors.append(detail) + _append_row_log( + row_log, + row_number=row_num, + disposition="rejected", + reason="extra_fields", + detail=detail, + ) continue if not any(row.values()): + _append_row_log( + row_log, + row_number=row_num, + disposition="skipped", + reason="blank_row", + detail="Blank data row was not ingested", + ) continue + def _reject(reason: str, message: str) -> None: + parse_errors.append(message) + _append_row_log( + row_log, + row_number=row_num, + disposition="rejected", + reason=reason, + detail=message, + ) + try: row_cte = _detect_row_cte_type(row, fallback_cte) + if row_cte and _detect_row_cte_type(row, None) is None: + _append_row_log( + row_log, + row_number=row_num, + disposition="flagged", + reason="cte_type_from_upload", + field="cte_type", + guessed_as=row_cte, + detail="Row has no CTE type; used the upload's cte_type", + ) if not row_cte: - parse_errors.append(f"Row {row_num}: No CTE type — add cte_type column or pass cte_type param") + _reject( + "missing_cte_type", + f"Row {row_num}: No CTE type — add cte_type column or pass cte_type param", + ) continue # ── Phase 1: Ingestion — flexible on format, strict on presence ── @@ -839,41 +1108,75 @@ def parse_csv_rows( event_time_raw = (row.get("event_time") or "").strip() ts, date_warning = _parse_date_flexible(raw_date, event_time_raw) if ts is None: - parse_errors.append(f"Row {row_num}: {date_warning}") + _reject("bad_date", f"Row {row_num}: {date_warning}") continue + if date_warning: + _append_row_log( + row_log, + row_number=row_num, + disposition="flagged", + reason="date_guessed", + field="date", + detail=date_warning, + ) # Location: strip whitespace and expand abbreviations loc_gln = _first_present(row, _LOCATION_GLN_COLUMNS) or None _loc_raw = _first_present(row, _LOCATION_NAME_COLUMNS) loc_name = _normalize_location(_loc_raw) if _loc_raw else None + if _loc_raw and loc_name != _loc_raw: + _append_row_log( + row_log, + row_number=row_num, + disposition="flagged", + reason="location_normalized", + field="location_name", + detail="Location name was normalised (abbreviations expanded)", + ) # Unit of measure: map abbreviations before Pydantic sees them. # No default: a unit the uploader did not record is not "units". uom_raw = _first_present(row, _UNIT_COLUMNS) if not uom_raw: - parse_errors.append(f"Row {row_num}: unit_of_measure is required") + _reject("missing_unit_of_measure", f"Row {row_num}: unit_of_measure is required") continue uom = _normalize_uom(uom_raw) + if uom != uom_raw: + _append_row_log( + row_log, + row_number=row_num, + disposition="flagged", + reason="unit_normalized", + field="unit_of_measure", + guessed_as=uom, + detail=f"Unit of measure mapped to {uom!r}", + ) # Quantity: required and numeric. The generic exception handler # below would reject the row too, but with an opaque message. quantity_raw = _first_present(row, _QUANTITY_COLUMNS) if not quantity_raw: - parse_errors.append(f"Row {row_num}: quantity is required") + _reject("missing_quantity", f"Row {row_num}: quantity is required") continue try: quantity = float(quantity_raw) except ValueError: - parse_errors.append(f"Row {row_num}: quantity '{quantity_raw}' is not a number") + _reject("bad_quantity", f"Row {row_num}: quantity '{quantity_raw}' is not a number") continue # IngestEvent requires a finite quantity greater than 0. Reject # here with a row-specific message instead of falling through to # the generic "could not be parsed" handler below. if not math.isfinite(quantity): - parse_errors.append(f"Row {row_num}: quantity '{quantity_raw}' must be a finite number") + _reject( + "bad_quantity", + f"Row {row_num}: quantity '{quantity_raw}' must be a finite number", + ) continue if quantity <= 0: - parse_errors.append(f"Row {row_num}: quantity '{quantity_raw}' must be greater than 0") + _reject( + "bad_quantity", + f"Row {row_num}: quantity '{quantity_raw}' must be greater than 0", + ) continue # Lot code: extract and run integrity checks @@ -917,7 +1220,7 @@ def parse_csv_rows( # warning for transformations. inputs, input_error, input_warning = _parse_input_quantities(row, input_lot_codes, row_cte) if input_error: - parse_errors.append(f"Row {row_num}: {input_error}") + _reject("bad_input_quantities", f"Row {row_num}: {input_error}") continue if inputs: kdes["inputs"] = inputs @@ -944,11 +1247,25 @@ def parse_csv_rows( missing_kdes = _validate_kde_completeness(row_cte, event, kdes) if missing_kdes: kde_warnings.append(f"Row {row_num}: Missing KDEs for {row_cte}: {', '.join(missing_kdes)}") + _append_row_log( + row_log, + row_number=row_num, + disposition="flagged", + reason="missing_kdes", + detail=f"Accepted with missing KDEs: {', '.join(missing_kdes)}", + ) + _append_row_log( + row_log, + row_number=row_num, + disposition="accepted", + reason="ingested", + detail=f"Row ingested as {row_cte}", + ) except Exception as e: # Log the detail server-side; return a generic per-row message so # raw exception text (library internals) isn't echoed to callers. logger.warning("csv_row_parse_failed row=%s error=%s", row_num, str(e)) - parse_errors.append(f"Row {row_num}: could not be parsed (see server logs)") + _reject("parse_exception", f"Row {row_num}: could not be parsed (see server logs)") return events, parse_errors, kde_warnings @@ -1459,13 +1776,19 @@ async def ingest_csv( len(header_mapping.applied_mapping), sorted(header_mapping.applied_mapping.items()), ) + row_log: list[CsvRowDispositionEntry] = [] events, parse_errors, kde_warnings = parse_csv_rows( - text, fallback_cte, header_mapping=header_mapping.applied_mapping + text, fallback_cte, header_mapping=header_mapping.applied_mapping, row_log=row_log ) kde_warnings.extend(xlsx_warnings) + missing_columns = [e.field for e in row_log if e.reason == "missing_column" and e.field] + unrecognized_columns = [e.field for e in row_log if e.reason == "unrecognized_column" and e.field] if not events and parse_errors: detail: dict = {"message": "No valid rows found", "errors": parse_errors} + if missing_columns: + # SEL-549: name the absent columns so the uploader can fix the header. + detail["missing_columns"] = missing_columns if xlsx_warnings: detail["kde_warnings"] = kde_warnings detail["kde_warning_count"] = len(kde_warnings) @@ -1544,9 +1867,7 @@ async def ingest_csv( response.total += 1 if kde_warnings: - logger.warning( - "csv_ingest_kde_warnings: %d warnings: %s", len(kde_warnings), "; ".join(kde_warnings[:5]) - ) + logger.warning("csv_ingest_kde_warnings: %d warnings: %s", len(kde_warnings), "; ".join(kde_warnings[:5])) logger.info( "csv_ingest_complete: total=%d accepted=%d rejected=%d kde_warnings=%d", @@ -1568,4 +1889,7 @@ async def ingest_csv( header_mapping_suggestions=_header_mapping_suggestion_results(header_mapping), header_mapping_engine=_header_mapping_engine_result(header_mapping, artifacts_recorded), cte_type_suggestion=_cte_type_suggestion_result(file_cte, file_cte_artifacts), + row_disposition_log=row_log, + missing_columns=missing_columns, + unrecognized_columns=unrecognized_columns, ) diff --git a/services/ingestion/app/webhook_models.py b/services/ingestion/app/webhook_models.py index c68f0b20e..e39ad308a 100644 --- a/services/ingestion/app/webhook_models.py +++ b/services/ingestion/app/webhook_models.py @@ -335,6 +335,38 @@ class IngestResponse(BaseModel): ) +class CsvRowDispositionEntry(BaseModel): + """One accounted-for CSV input row or header/field decision (SEL-549). + + Every physical data row the partner file presented must appear here as + accepted, rejected, flagged, or skipped — never silently dropped. + ``guessed_value`` is omitted on purpose: partner cell values must not + leave the ingest path via logs or API responses. + """ + + row_number: int = Field( + ..., + description="CSV line number (header is 1; data starts at 2). 0 means a file-level header decision.", + ) + disposition: str = Field( + ..., + description="accepted | rejected | flagged | skipped", + ) + reason: str = Field(..., description="Machine-readable reason code for this decision.") + field: Optional[str] = Field( + default=None, + description="Column or field name involved, when the decision is field-scoped.", + ) + detail: Optional[str] = Field( + default=None, + description="Human-readable detail. Never includes cell values or secrets.", + ) + guessed_as: Optional[str] = Field( + default=None, + description="Canonical field a partner column was mapped/guessed into, when applicable.", + ) + + class HeaderMappingSuggestionResult(BaseModel): """One Jev suggestion for a CSV column the parser did not recognise. @@ -439,6 +471,21 @@ class CsvIngestResponse(IngestResponse): description="Jev's suggested CTE for a file that declares none; null when not asked.", ) + # SEL-549: every input row accounted for. Defaults keep cached idempotent + # bodies valid against the schema after this ships. + row_disposition_log: List[CsvRowDispositionEntry] = Field( + default_factory=list, + description="Per-row (and file-level) accept/reject/flag/skip decisions; never silent drops.", + ) + missing_columns: List[str] = Field( + default_factory=list, + description="Required partner columns that were absent (canonical names); empty when the header was usable.", + ) + unrecognized_columns: List[str] = Field( + default_factory=list, + description="Partner columns the parser does not recognise (may still pass through into KDEs).", + ) + class RecentEventsResponse(BaseModel): """Response from the /recent events endpoint.""" diff --git a/services/ingestion/tests/test_csv_partner_file_hardening_sel549.py b/services/ingestion/tests/test_csv_partner_file_hardening_sel549.py new file mode 100644 index 000000000..62c1a04bf --- /dev/null +++ b/services/ingestion/tests/test_csv_partner_file_hardening_sel549.py @@ -0,0 +1,116 @@ +"""SEL-549: an unseen partner CSV fails clearly and never loses a row silently. + +Uses only synthetic data: the repo fixture ``tests/fixtures/01_harvesting.csv`` +and small inline CSVs built here (C.Lee's scoped exception to real-data-only). +""" + +from __future__ import annotations + +from pathlib import Path + +from app import csv_templates as ct + +FIXTURE = Path(__file__).parent / "fixtures" / "01_harvesting.csv" + +HEADER = "traceability_lot_code,product_description,quantity,unit_of_measure,harvest_date,location_name" + + +def _parse(text: str, cte: str | None = "harvesting", mapping=None): + log: list = [] + events, errors, warnings = ct.parse_csv_rows(text, cte, header_mapping=mapping, row_log=log) + return events, errors, warnings, log + + +def _data_rows(log): + return {e.row_number for e in log if e.row_number >= 2 and e.disposition in {"accepted", "rejected", "skipped"}} + + +def test_fixture_rows_are_all_accepted_and_logged(): + text = FIXTURE.read_text() + events, errors, _warnings, log = _parse(text) + data_lines = len([line for line in text.splitlines()[1:] if line.strip()]) + assert errors == [] + assert len(events) == data_lines + accepted = [e for e in log if e.disposition == "accepted"] + assert [e.row_number for e in accepted] == list(range(2, 2 + data_lines)) + + +def test_missing_required_columns_named_in_one_clear_error(): + text = "traceability_lot_code,product_description,harvest_date\nLOT-SYN-1,Synthetic Kale,2026-03-12\n" + events, errors, _warnings, log = _parse(text) + assert events == [] + assert len(errors) == 1 + assert "Missing required columns" in errors[0] + assert "quantity (accepted column names: quantity, qty)" in errors[0] + assert "unit_of_measure" in errors[0] + missing = {e.field for e in log if e.reason == "missing_column"} + assert missing == {"quantity", "unit_of_measure"} + + +def test_missing_cte_type_rejects_each_row_and_logs_it(): + text = f"{HEADER}\nLOT-SYN-1,Synthetic Kale,10,lbs,2026-03-12,Synthetic Farm\n" + _e, errors, _w, log = _parse(text, cte=None) + assert errors and "No CTE type" in errors[0] + assert [e.reason for e in log if e.disposition == "rejected"] == ["missing_cte_type"] + events, errors, _w, log = _parse(text, cte="harvesting") + assert errors == [] and len(events) == 1 + assert any(e.reason == "cte_type_from_upload" for e in log) + + +def test_every_row_accounted_for_none_silent(): + rows = [ + "LOT-SYN-1,Synthetic Kale,10,lbs,2026-03-12,Synthetic Farm", # 2 accepted + "# synthetic comment row,,,,,", # 3 skipped comment + ",,,,,", # 4 skipped blank + "LOT-SYN-2,Synthetic Kale,abc,lbs,2026-03-12,Synthetic Farm", # 5 bad quantity + "LOT-SYN-3,Synthetic Kale,10,,2026-03-12,Synthetic Farm", # 6 missing unit + "LOT-SYN-4,Synthetic Kale,10,lbs,not-a-date,Synthetic Farm", # 7 bad date + "LOT-SYN-5,Synthetic Kale,10,lbs,2026-03-12,Synthetic Farm,EXTRA", # 8 extra field + ] + text = HEADER + "\n" + "\n".join(rows) + "\n" + events, errors, _w, log = _parse(text) + assert len(events) == 1 + assert _data_rows(log) == set(range(2, 2 + len(rows))) + by_row = {e.row_number: e for e in log if e.disposition in {"accepted", "rejected", "skipped"}} + assert by_row[2].disposition == "accepted" + assert by_row[3].reason == "comment_row" + assert by_row[4].reason == "blank_row" + assert by_row[5].reason == "bad_quantity" + assert by_row[6].reason == "missing_unit_of_measure" + assert by_row[7].reason == "bad_date" + assert by_row[8].reason == "extra_fields" + rejected = [e for e in log if e.disposition == "rejected"] + assert len(rejected) == len(errors) == 4 + + +def test_unrecognized_columns_flagged_and_kept(): + text = f"{HEADER},partner_bin\nLOT-SYN-1,Synthetic Kale,10,lbs,2026-03-12,Synthetic Farm,BIN-7\n" + events, errors, _w, log = _parse(text) + assert errors == [] and len(events) == 1 + assert events[0].kdes.get("partner_bin") == "BIN-7" + flagged = [e for e in log if e.reason == "unrecognized_column"] + assert [e.field for e in flagged] == ["partner_bin"] + + +def test_header_mapping_guess_logged_without_cell_value(): + header = "traceability_lot_code,product_description,amount,unit_of_measure,harvest_date,location_name" + text = f"{header}\nLOT-SYN-1,Synthetic Kale,10,lbs,2026-03-12,Synthetic Farm\n" + events, errors, _w, log = _parse(text, mapping={"amount": "quantity"}) + assert errors == [] and len(events) == 1 + guesses = [e for e in log if e.reason == "header_guess_applied"] + assert len(guesses) == 1 + assert guesses[0].field == "amount" and guesses[0].guessed_as == "quantity" + assert "10" not in (guesses[0].detail or "") + + +def test_log_detail_never_echoes_guessed_cell_values(): + text = f"{HEADER}\nLOT-SYN-1,Synthetic Kale,10,lbs,2026-03-12,Synthetic Farm\n" + _e, _err, _w, log = _parse(text) + for entry in log: + if entry.disposition == "flagged": + assert "LOT-SYN-1" not in (entry.detail or "") + + +def test_header_only_file_is_not_a_missing_column_error(): + _e, errors, _w, _log = _parse("partner_col_a,partner_col_b\n") + assert errors == [] From 9e7b135d2191ed78734e83c39b2882f6c2bb47f3 Mon Sep 17 00:00:00 2001 From: Christopher Sellers <1@christophersellers.com> Date: Tue, 27 Oct 2026 21:36:18 -0700 Subject: [PATCH 2/6] test(ingest): SEL-549 acceptance tests; reject unknown CTE types and warn on unknown columns Signed-off-by: Christopher Sellers <1@christophersellers.com> --- services/ingestion/app/csv_templates.py | 44 +++++- .../test_csv_partner_file_hardening_sel549.py | 126 ++++++++++++++++++ 2 files changed, 169 insertions(+), 1 deletion(-) diff --git a/services/ingestion/app/csv_templates.py b/services/ingestion/app/csv_templates.py index aae543646..a34f719b6 100644 --- a/services/ingestion/app/csv_templates.py +++ b/services/ingestion/app/csv_templates.py @@ -809,6 +809,18 @@ def _format_missing_columns_error( return "Missing required columns for partner CSV ingest: " + "; ".join(parts) + f". Present columns: {present}" +def _unknown_row_cte_value(row: Mapping[str, Optional[str]]) -> tuple[Optional[str], Optional[str]]: + """``(value, column)`` when the row's first CTE value is not a known alias, else ``(None, None)``.""" + for col in _CTE_TYPE_COLUMNS: + raw = (row.get(col) or "").strip() + if not raw: + continue + if raw.lower().replace(" ", "_") in _CTE_TYPE_ALIASES: + return None, None + return raw, col + return None, None + + def _is_data_record(record: Mapping[Optional[str], object]) -> bool: """True for a record the row loop would read as data (not blank, not a comment).""" values = [str(v or "").strip() for k, v in record.items() if k is not None] @@ -978,7 +990,11 @@ def parse_csv_rows( ) return ([], [detail], []) - for bad in _unrecognized_partner_columns(header, header_mapping): + unrecognized = _unrecognized_partner_columns(header, header_mapping) + if unrecognized: + # Surface in the warning output too, not only the disposition log. + kde_warnings.append("Unrecognised column(s) kept as KDEs, not mapped to a field: " + ", ".join(unrecognized)) + for bad in unrecognized: _append_row_log( row_log, row_number=0, @@ -1020,11 +1036,13 @@ def parse_csv_rows( # dropped; blank ones (a trailing comma) carry nothing and are ignored. extra_values = row.pop(None, None) or [] row = {normalize_csv_header(k): (v or "").strip() for k, v in row.items()} + mapped_fields: set[str] = set() if header_mapping: for source_key, canonical_field in header_mapping.items(): value = row.get(source_key) if value and not _row_provides(row, canonical_field): row[canonical_field] = value + mapped_fields.add(canonical_field) _append_row_log( row_log, row_number=row_num, @@ -1082,6 +1100,30 @@ def _reject(reason: str, message: str) -> None: ) try: + unknown_cte, unknown_cte_col = _unknown_row_cte_value(row) + if unknown_cte is not None and unknown_cte_col in mapped_fields: + # A guessed (mapped) column whose value is not a CTE: keep the + # declared fallback, but flag it rather than accept quietly. + _append_row_log( + row_log, + row_number=row_num, + disposition="flagged", + reason="mapped_cte_not_recognized", + field=unknown_cte_col, + detail="Mapped column value is not a known CTE type; used the upload's cte_type", + ) + kde_warnings.append( + f"Row {row_num}: mapped {unknown_cte_col} value is not a known CTE type; used the upload's cte_type" + ) + elif unknown_cte is not None: + # SEL-549: an event type we don't recognise is never replaced + # by the upload's cte_type; the row is rejected and named. + _reject( + "unknown_cte_type", + f"Row {row_num}: unrecognised CTE type {unknown_cte!r}; " + f"use one of: {', '.join(sorted(set(_CTE_TYPE_ALIASES.values())))}", + ) + continue row_cte = _detect_row_cte_type(row, fallback_cte) if row_cte and _detect_row_cte_type(row, None) is None: _append_row_log( diff --git a/services/ingestion/tests/test_csv_partner_file_hardening_sel549.py b/services/ingestion/tests/test_csv_partner_file_hardening_sel549.py index 62c1a04bf..0bdca981a 100644 --- a/services/ingestion/tests/test_csv_partner_file_hardening_sel549.py +++ b/services/ingestion/tests/test_csv_partner_file_hardening_sel549.py @@ -114,3 +114,129 @@ def test_log_detail_never_echoes_guessed_cell_values(): def test_header_only_file_is_not_a_missing_column_error(): _e, errors, _w, _log = _parse("partner_col_a,partner_col_b\n") assert errors == [] + + +# --------------------------------------------------------------------------- +# Red Team acceptance bar: every unrecognised field, event type or format is +# flagged in the disposition log AND the error/warning output, and +# rows in == accepted + rejected + skipped (nothing silently dropped). +# Inputs are derived from the synthetic fixture 01_harvesting.csv. +# --------------------------------------------------------------------------- + +import csv as _csv +import io as _io + +import pytest + + +def _fixture_rows() -> tuple[list[str], list[dict]]: + reader = _csv.DictReader(_io.StringIO(FIXTURE.read_text())) + return list(reader.fieldnames or []), list(reader) + + +def _to_csv(header: list[str], rows: list[dict]) -> str: + out = _io.StringIO() + writer = _csv.DictWriter(out, fieldnames=header, lineterminator="\n", extrasaction="ignore") + writer.writeheader() + for row in rows: + writer.writerow(row) + return out.getvalue() + + +def _assert_counts_balance(text: str, events, log) -> None: + rows_in = len(text.splitlines()) - 1 + outcome = {e.row_number: e.disposition for e in log if e.row_number >= 2 and e.disposition != "flagged"} + assert len(outcome) == rows_in, "every input row needs exactly one outcome" + assert sum(1 for d in outcome.values() if d == "accepted") == len(events) + + +@pytest.fixture +def unknown_column_csv() -> str: + header, rows = _fixture_rows() + for i, row in enumerate(rows): + row["partner_pallet_code"] = f"PAL-SYN-{i}" + return _to_csv([*header, "partner_pallet_code"], rows) + + +@pytest.fixture +def unknown_cte_csv() -> str: + header, rows = _fixture_rows() + for row in rows: + row["cte_type"] = "harvesting" + rows[1]["cte_type"] = "teleporting" + return _to_csv([*header, "cte_type"], rows) + + +@pytest.fixture +def malformed_date_csv() -> str: + header, rows = _fixture_rows() + rows[0]["harvest_date"] = "32/13/2026" + return _to_csv(header, rows) + + +@pytest.fixture +def malformed_quantity_csv() -> str: + header, rows = _fixture_rows() + rows[0]["quantity"] = "5,000 lbs" + return _to_csv(header, rows) + + +@pytest.fixture +def missing_columns_csv() -> str: + header, rows = _fixture_rows() + kept = [h for h in header if h not in {"quantity", "unit_of_measure"}] + return _to_csv(kept, rows) + + +@pytest.fixture +def garbled_and_empty_rows_csv() -> str: + text = FIXTURE.read_text().rstrip("\n").split("\n") + header, good = text[0], text[1:] + garbled = "\x00\x07%%LOT??,,,,,,,,,,,,ZZZ" + empty = "," * (header.count(",")) + return "\n".join([header, good[0], garbled, empty, *good[1:]]) + "\n" + + +def test_acceptance_unknown_column_is_flagged(unknown_column_csv): + events, errors, warnings, log = _parse(unknown_column_csv) + assert any(e.reason == "unrecognized_column" and e.field == "partner_pallet_code" for e in log) + assert any("partner_pallet_code" in w for w in warnings) + assert errors == [] + _assert_counts_balance(unknown_column_csv, events, log) + + +def test_acceptance_unknown_cte_type_is_rejected_not_defaulted(unknown_cte_csv): + events, errors, _w, log = _parse(unknown_cte_csv, cte="harvesting") + assert [e.row_number for e in log if e.reason == "unknown_cte_type"] == [3] + assert any("Row 3" in e and "teleporting" in e for e in errors) + _assert_counts_balance(unknown_cte_csv, events, log) + + +def test_acceptance_malformed_date_is_flagged(malformed_date_csv): + events, errors, _w, log = _parse(malformed_date_csv) + assert [e.row_number for e in log if e.reason == "bad_date"] == [2] + assert any(e.startswith("Row 2:") for e in errors) + _assert_counts_balance(malformed_date_csv, events, log) + + +def test_acceptance_malformed_quantity_format_is_flagged(malformed_quantity_csv): + events, errors, _w, log = _parse(malformed_quantity_csv) + assert [e.row_number for e in log if e.reason == "bad_quantity"] == [2] + assert any(e.startswith("Row 2:") and "not a number" in e for e in errors) + _assert_counts_balance(malformed_quantity_csv, events, log) + + +def test_acceptance_missing_required_columns_flagged(missing_columns_csv): + events, errors, _w, log = _parse(missing_columns_csv) + assert events == [] + assert {e.field for e in log if e.reason == "missing_column"} == {"quantity", "unit_of_measure"} + assert len(errors) == 1 and "quantity" in errors[0] and "unit_of_measure" in errors[0] + + +def test_acceptance_garbled_and_empty_rows_flagged(garbled_and_empty_rows_csv): + events, errors, _w, log = _parse(garbled_and_empty_rows_csv) + outcome = {e.row_number: e for e in log if e.row_number >= 2 and e.disposition != "flagged"} + assert outcome[3].disposition == "rejected" # garbled + assert any(e.startswith("Row 3:") for e in errors) + assert outcome[4].reason == "blank_row" # empty + _assert_counts_balance(garbled_and_empty_rows_csv, events, log) From bf0265123dc312959c36a1d525b4f4308bc0e1b9 Mon Sep 17 00:00:00 2001 From: Christopher Sellers <1@christophersellers.com> Date: Wed, 28 Oct 2026 01:35:35 -0700 Subject: [PATCH 3/6] fix(ingest): finalize CSV row dispositions after ingest (SEL-549) Log accepted/rejected only after ingest_events so dedup and KDE failures are not reported as ingested. Balance counts for missing columns and row-limit overflow, and keep partner tests on fixture data. Signed-off-by: Christopher Sellers <1@christophersellers.com> --- services/ingestion/app/csv_templates.py | 131 +++++++++-- .../test_csv_partner_file_hardening_sel549.py | 219 +++++++++++++----- 2 files changed, 282 insertions(+), 68 deletions(-) diff --git a/services/ingestion/app/csv_templates.py b/services/ingestion/app/csv_templates.py index a34f719b6..8e1fe1c7a 100644 --- a/services/ingestion/app/csv_templates.py +++ b/services/ingestion/app/csv_templates.py @@ -860,11 +860,64 @@ def _append_row_log( ) +def _ingest_reject_reason(errors: Sequence[str]) -> str: + """Map ingest_events rejection text to a stable disposition reason code.""" + joined = " ".join(errors).lower() + if "duplicate event" in joined: + return "duplicate" + if joined.startswith("rule_violation") or "rule_violation" in joined: + return "rule_violation" + if errors: + return "ingest_rejected" + return "ingest_rejected" + + +def finalize_row_dispositions_from_ingest( + row_log: Optional[list[CsvRowDispositionEntry]], + event_source_rows: Sequence[int], + event_results: Sequence[object], +) -> None: + """Append the final per-row disposition after ``ingest_events`` returns. + + Parse-time logging never claims a successfully built event was ingested. + Each source row listed in ``event_source_rows`` (parallel to the parsed + events / the first ``len(event_source_rows)`` ingest results) gets exactly + one exclusive outcome here: accepted, or rejected (including duplicates). + """ + if row_log is None: + return + # ingest_csv may append synthetic EventResults for parse_errors after + # ingest_events; only the results that correspond to parsed events finalize. + for row_number, result in zip(event_source_rows, event_results): + status = getattr(result, "status", None) + errors = list(getattr(result, "errors", None) or []) + if status == "accepted": + _append_row_log( + row_log, + row_number=row_number, + disposition="accepted", + reason="ingested", + detail="Row ingested", + ) + continue + reason = _ingest_reject_reason(errors) + detail = errors[0] if errors else "Row rejected during ingest" + # Never echo partner cell values; EventResult errors are codes/messages. + _append_row_log( + row_log, + row_number=row_number, + disposition="rejected", + reason=reason, + detail=detail, + ) + + def parse_csv_rows( text: str, fallback_cte: Optional[str] = None, header_mapping: Optional[Mapping[str, str]] = None, row_log: Optional[list[CsvRowDispositionEntry]] = None, + event_source_rows: Optional[list[int]] = None, ) -> tuple[list[IngestEvent], list[str], list[str]]: """Parse CSV text into ``IngestEvent`` objects. @@ -884,7 +937,14 @@ def parse_csv_rows( ``row_log``, when provided, receives a ``CsvRowDispositionEntry`` for every input row and for file-level header decisions (SEL-549). Cell values are - never written into the log. + never written into the log. Successfully parsed rows are not logged as + accepted here: callers that ingest must call + ``finalize_row_dispositions_from_ingest`` after ``ingest_events`` so the + log records only the final outcome (accepted, rejected, duplicate). + + ``event_source_rows``, when provided, is filled with the CSV line number + of each returned event (same order as ``events``), so the ingest path can + finalize dispositions without inventing row associations. Returns ``(events, parse_errors, kde_warnings)``. """ @@ -988,6 +1048,35 @@ def parse_csv_rows( field=name, detail=f"Required column {name!r} is absent from the partner file header", ) + # Every physical data row still needs an exclusive outcome so + # accepted + rejected + skipped equals rows sent (SEL-549). + for row_num, record in enumerate(records, start=2): + if not _is_data_record(record): + values = [str(v or "").strip() for k, v in record.items() if k is not None] + if values and values[0].startswith("#"): + _append_row_log( + row_log, + row_number=row_num, + disposition="skipped", + reason="comment_row", + detail="Comment row (first value starts with #) was not ingested", + ) + else: + _append_row_log( + row_log, + row_number=row_num, + disposition="skipped", + reason="blank_row", + detail="Blank data row was not ingested", + ) + continue + _append_row_log( + row_log, + row_number=row_num, + disposition="rejected", + reason="missing_required_columns", + detail=detail, + ) return ([], [detail], []) unrecognized = _unrecognized_partner_columns(header, header_mapping) @@ -1009,18 +1098,23 @@ def parse_csv_rows( for row_num, row in enumerate(records, start=2): # _file_cte_type_for_jev reads exactly the rows before this cut-off. if row_num - 1 > MAX_CSV_ROWS: + # First over-limit row is rejected once. Further unread rows are + # each skipped under their own row numbers so counts balance and + # this row is never both skipped and rejected (SEL-549). if skipped_overflow: overflow_detail = ( f"Row {row_num}: {skipped_overflow} more row(s) after this one were skipped and not ingested" ) parse_errors.append(overflow_detail) - _append_row_log( - row_log, - row_number=row_num, - disposition="skipped", - reason="row_limit_overflow", - detail=overflow_detail, - ) + for offset in range(1, skipped_overflow + 1): + overflow_row = row_num + offset + _append_row_log( + row_log, + row_number=overflow_row, + disposition="skipped", + reason="row_limit_overflow", + detail=(f"Row {overflow_row}: skipped after CSV row limit of {MAX_CSV_ROWS}"), + ) limit_detail = f"Row {row_num}: CSV exceeds maximum row limit of {MAX_CSV_ROWS}" parse_errors.append(limit_detail) _append_row_log( @@ -1296,13 +1390,11 @@ def _reject(reason: str, message: str) -> None: reason="missing_kdes", detail=f"Accepted with missing KDEs: {', '.join(missing_kdes)}", ) - _append_row_log( - row_log, - row_number=row_num, - disposition="accepted", - reason="ingested", - detail=f"Row ingested as {row_cte}", - ) + # Do not log accepted/ingested here. Dedup, KDE checks and + # persistence decide the final disposition in ingest_events; + # finalize_row_dispositions_from_ingest records it once. + if event_source_rows is not None: + event_source_rows.append(row_num) except Exception as e: # Log the detail server-side; return a generic per-row message so # raw exception text (library internals) isn't echoed to callers. @@ -1819,8 +1911,13 @@ async def ingest_csv( sorted(header_mapping.applied_mapping.items()), ) row_log: list[CsvRowDispositionEntry] = [] + event_source_rows: list[int] = [] events, parse_errors, kde_warnings = parse_csv_rows( - text, fallback_cte, header_mapping=header_mapping.applied_mapping, row_log=row_log + text, + fallback_cte, + header_mapping=header_mapping.applied_mapping, + row_log=row_log, + event_source_rows=event_source_rows, ) kde_warnings.extend(xlsx_warnings) missing_columns = [e.field for e in row_log if e.reason == "missing_column" and e.field] @@ -1876,6 +1973,8 @@ async def ingest_csv( principal=principal, source_file=receipt, ) + # Final dispositions only: parse never claimed these rows were ingested. + finalize_row_dispositions_from_ingest(row_log, event_source_rows, response.events) artifacts_recorded = 0 if header_mapping.ok and header_mapping.suggestions: diff --git a/services/ingestion/tests/test_csv_partner_file_hardening_sel549.py b/services/ingestion/tests/test_csv_partner_file_hardening_sel549.py index 0bdca981a..d1003f3ff 100644 --- a/services/ingestion/tests/test_csv_partner_file_hardening_sel549.py +++ b/services/ingestion/tests/test_csv_partner_file_hardening_sel549.py @@ -1,23 +1,51 @@ """SEL-549: an unseen partner CSV fails clearly and never loses a row silently. -Uses only synthetic data: the repo fixture ``tests/fixtures/01_harvesting.csv`` -and small inline CSVs built here (C.Lee's scoped exception to real-data-only). +Uses only synthetic data from the repo fixture +``tests/fixtures/01_harvesting.csv`` (C.Lee's scoped exception to +real-data-only). No lot codes, product names, or IDs are invented in tests. """ from __future__ import annotations +import csv as _csv +import io as _io from pathlib import Path +from types import SimpleNamespace +import pytest from app import csv_templates as ct FIXTURE = Path(__file__).parent / "fixtures" / "01_harvesting.csv" -HEADER = "traceability_lot_code,product_description,quantity,unit_of_measure,harvest_date,location_name" + +def _fixture_rows() -> tuple[list[str], list[dict]]: + reader = _csv.DictReader(_io.StringIO(FIXTURE.read_text())) + return list(reader.fieldnames or []), [dict(row) for row in reader] + + +def _to_csv(header: list[str], rows: list[dict]) -> str: + out = _io.StringIO() + writer = _csv.DictWriter(out, fieldnames=header, lineterminator="\n", extrasaction="ignore") + writer.writeheader() + for row in rows: + writer.writerow(row) + return out.getvalue() def _parse(text: str, cte: str | None = "harvesting", mapping=None): + """Parse and finalize successful rows as accepted (parse-only path). + + Production ingest_csv calls finalize_row_dispositions_from_ingest after + ingest_events instead; tests that only exercise the parser still need + every row to land in exactly one exclusive bucket. + """ log: list = [] - events, errors, warnings = ct.parse_csv_rows(text, cte, header_mapping=mapping, row_log=log) + source_rows: list[int] = [] + events, errors, warnings = ct.parse_csv_rows( + text, cte, header_mapping=mapping, row_log=log, event_source_rows=source_rows + ) + fake_results = [SimpleNamespace(status="accepted", errors=[]) for _ in source_rows] + ct.finalize_row_dispositions_from_ingest(log, source_rows, fake_results) return events, errors, warnings, log @@ -25,6 +53,25 @@ def _data_rows(log): return {e.row_number for e in log if e.row_number >= 2 and e.disposition in {"accepted", "rejected", "skipped"}} +def _assert_counts_balance(text: str, events, log) -> None: + """Every physical input row lands in exactly one exclusive bucket.""" + rows_in = len([line for line in text.splitlines()[1:] if line is not None]) + # Include blank lines that csv still presents as records: count non-header lines. + rows_in = len(text.splitlines()) - 1 + outcomes = [ + e.disposition for e in log if e.row_number >= 2 and e.disposition in {"accepted", "rejected", "skipped"} + ] + row_numbers = sorted( + e.row_number for e in log if e.row_number >= 2 and e.disposition in {"accepted", "rejected", "skipped"} + ) + assert row_numbers == list(range(2, 2 + rows_in)), ( + f"expected exclusive outcomes for rows 2..{1 + rows_in}, got {row_numbers}" + ) + assert len(outcomes) == rows_in, "every input row needs exactly one exclusive outcome" + assert outcomes.count("accepted") == len(events) + assert outcomes.count("accepted") + outcomes.count("rejected") + outcomes.count("skipped") == rows_in + + def test_fixture_rows_are_all_accepted_and_logged(): text = FIXTURE.read_text() events, errors, _warnings, log = _parse(text) @@ -33,10 +80,13 @@ def test_fixture_rows_are_all_accepted_and_logged(): assert len(events) == data_lines accepted = [e for e in log if e.disposition == "accepted"] assert [e.row_number for e in accepted] == list(range(2, 2 + data_lines)) + _assert_counts_balance(text, events, log) def test_missing_required_columns_named_in_one_clear_error(): - text = "traceability_lot_code,product_description,harvest_date\nLOT-SYN-1,Synthetic Kale,2026-03-12\n" + header, rows = _fixture_rows() + kept = [h for h in header if h not in {"quantity", "unit_of_measure"}] + text = _to_csv(kept, rows[:1]) events, errors, _warnings, log = _parse(text) assert events == [] assert len(errors) == 1 @@ -45,32 +95,52 @@ def test_missing_required_columns_named_in_one_clear_error(): assert "unit_of_measure" in errors[0] missing = {e.field for e in log if e.reason == "missing_column"} assert missing == {"quantity", "unit_of_measure"} + _assert_counts_balance(text, events, log) def test_missing_cte_type_rejects_each_row_and_logs_it(): - text = f"{HEADER}\nLOT-SYN-1,Synthetic Kale,10,lbs,2026-03-12,Synthetic Farm\n" + header, rows = _fixture_rows() + text = _to_csv(header, rows[:1]) _e, errors, _w, log = _parse(text, cte=None) assert errors and "No CTE type" in errors[0] assert [e.reason for e in log if e.disposition == "rejected"] == ["missing_cte_type"] events, errors, _w, log = _parse(text, cte="harvesting") assert errors == [] and len(events) == 1 assert any(e.reason == "cte_type_from_upload" for e in log) + _assert_counts_balance(text, events, log) def test_every_row_accounted_for_none_silent(): - rows = [ - "LOT-SYN-1,Synthetic Kale,10,lbs,2026-03-12,Synthetic Farm", # 2 accepted - "# synthetic comment row,,,,,", # 3 skipped comment - ",,,,,", # 4 skipped blank - "LOT-SYN-2,Synthetic Kale,abc,lbs,2026-03-12,Synthetic Farm", # 5 bad quantity - "LOT-SYN-3,Synthetic Kale,10,,2026-03-12,Synthetic Farm", # 6 missing unit - "LOT-SYN-4,Synthetic Kale,10,lbs,not-a-date,Synthetic Farm", # 7 bad date - "LOT-SYN-5,Synthetic Kale,10,lbs,2026-03-12,Synthetic Farm,EXTRA", # 8 extra field - ] - text = HEADER + "\n" + "\n".join(rows) + "\n" + header, rows = _fixture_rows() + base = dict(rows[0]) + bad_qty = dict(base) + bad_qty["quantity"] = "abc" + missing_unit = dict(base) + missing_unit["unit_of_measure"] = "" + bad_date = dict(base) + bad_date["harvest_date"] = "not-a-date" + good = dict(base) + # Build CSV manually so comment / blank / extra-field rows are present. + out = _io.StringIO() + writer = _csv.DictWriter(out, fieldnames=header, lineterminator="\n", extrasaction="ignore") + writer.writeheader() + writer.writerow(good) # 2 accepted + text = out.getvalue().rstrip("\n") + text += "\n# synthetic comment row\n" # 3 skipped comment + text += "," * (len(header) - 1) + "\n" # 4 blank + # bad quantity / missing unit / bad date from fixture cells only + for mutated in (bad_qty, missing_unit, bad_date): + buf = _io.StringIO() + w = _csv.DictWriter(buf, fieldnames=header, lineterminator="\n", extrasaction="ignore") + w.writerow(mutated) + text += buf.getvalue() + # extra field beyond header + extra_line = ",".join(good.get(h, "") for h in header) + ",EXTRA\n" + text += extra_line # 8 extra fields + events, errors, _w, log = _parse(text) assert len(events) == 1 - assert _data_rows(log) == set(range(2, 2 + len(rows))) + assert _data_rows(log) == set(range(2, 9)) by_row = {e.row_number: e for e in log if e.disposition in {"accepted", "rejected", "skipped"}} assert by_row[2].disposition == "accepted" assert by_row[3].reason == "comment_row" @@ -81,34 +151,48 @@ def test_every_row_accounted_for_none_silent(): assert by_row[8].reason == "extra_fields" rejected = [e for e in log if e.disposition == "rejected"] assert len(rejected) == len(errors) == 4 + _assert_counts_balance(text, events, log) def test_unrecognized_columns_flagged_and_kept(): - text = f"{HEADER},partner_bin\nLOT-SYN-1,Synthetic Kale,10,lbs,2026-03-12,Synthetic Farm,BIN-7\n" + header, rows = _fixture_rows() + # Unknown column keeps a real fixture cell value (field_id), never dropped. + fixture_value = rows[0]["field_id"] + for row in rows: + row["partner_bin"] = row["field_id"] + text = _to_csv([*header, "partner_bin"], rows[:1]) events, errors, _w, log = _parse(text) assert errors == [] and len(events) == 1 - assert events[0].kdes.get("partner_bin") == "BIN-7" + assert events[0].kdes.get("partner_bin") == fixture_value flagged = [e for e in log if e.reason == "unrecognized_column"] assert [e.field for e in flagged] == ["partner_bin"] + _assert_counts_balance(text, events, log) def test_header_mapping_guess_logged_without_cell_value(): - header = "traceability_lot_code,product_description,amount,unit_of_measure,harvest_date,location_name" - text = f"{header}\nLOT-SYN-1,Synthetic Kale,10,lbs,2026-03-12,Synthetic Farm\n" + header, rows = _fixture_rows() + # Rename quantity -> amount; mapping restores it. Values stay from fixture. + renamed = ["amount" if h == "quantity" else h for h in header] + mapped_rows = [{("amount" if k == "quantity" else k): v for k, v in row.items()} for row in rows[:1]] + text = _to_csv(renamed, mapped_rows) events, errors, _w, log = _parse(text, mapping={"amount": "quantity"}) assert errors == [] and len(events) == 1 guesses = [e for e in log if e.reason == "header_guess_applied"] assert len(guesses) == 1 assert guesses[0].field == "amount" and guesses[0].guessed_as == "quantity" - assert "10" not in (guesses[0].detail or "") + qty = rows[0]["quantity"] + assert qty not in (guesses[0].detail or "") + _assert_counts_balance(text, events, log) def test_log_detail_never_echoes_guessed_cell_values(): - text = f"{HEADER}\nLOT-SYN-1,Synthetic Kale,10,lbs,2026-03-12,Synthetic Farm\n" + header, rows = _fixture_rows() + text = _to_csv(header, rows[:1]) _e, _err, _w, log = _parse(text) + lot = rows[0]["traceability_lot_code"] for entry in log: if entry.disposition == "flagged": - assert "LOT-SYN-1" not in (entry.detail or "") + assert lot not in (entry.detail or "") def test_header_only_file_is_not_a_missing_column_error(): @@ -116,6 +200,55 @@ def test_header_only_file_is_not_a_missing_column_error(): assert errors == [] +def test_row_limit_first_overflow_not_double_counted(monkeypatch): + """First row past the limit is rejected once; further rows are skipped only.""" + monkeypatch.setattr(ct, "MAX_CSV_ROWS", 2) + header, rows = _fixture_rows() + # Three fixture rows: 2 accepted-capacity, 1 rejected, and invent no values — + # duplicate the third fixture row to create a 4th physical row for skipped. + four = [dict(rows[0]), dict(rows[1]), dict(rows[2]), dict(rows[2])] + text = _to_csv(header, four) + events, errors, _w, log = _parse(text) + assert len(events) == 2 + by_row = { + e.row_number: e for e in log if e.row_number >= 2 and e.disposition in {"accepted", "rejected", "skipped"} + } + assert by_row[2].disposition == "accepted" + assert by_row[3].disposition == "accepted" + assert by_row[4].disposition == "rejected" + assert by_row[4].reason == "row_limit_exceeded" + assert by_row[5].disposition == "skipped" + assert by_row[5].reason == "row_limit_overflow" + # Same row must not be both skipped and rejected. + exclusive = [e for e in log if e.row_number == 4 and e.disposition in {"accepted", "rejected", "skipped"}] + assert len(exclusive) == 1 + assert any("1 more row(s)" in err for err in errors) + assert errors[-1].endswith("CSV exceeds maximum row limit of 2") + _assert_counts_balance(text, events, log) + + +def test_finalize_uses_ingest_result_not_parse_claim(): + """A parsed row rejected by ingest (duplicate) is logged rejected, not accepted.""" + header, rows = _fixture_rows() + text = _to_csv(header, rows[:1]) + log: list = [] + source_rows: list[int] = [] + events, errors, _w = ct.parse_csv_rows(text, "harvesting", row_log=log, event_source_rows=source_rows) + assert errors == [] and len(events) == 1 and source_rows == [2] + # No exclusive outcome yet — parse must not claim ingested. + assert not any(e.disposition == "accepted" for e in log) + assert not any(e.reason == "ingested" for e in log) + ct.finalize_row_dispositions_from_ingest( + log, + source_rows, + [SimpleNamespace(status="rejected", errors=["Duplicate event in batch"])], + ) + exclusive = [e for e in log if e.row_number == 2 and e.disposition != "flagged"] + assert len(exclusive) == 1 + assert exclusive[0].disposition == "rejected" + assert exclusive[0].reason == "duplicate" + + # --------------------------------------------------------------------------- # Red Team acceptance bar: every unrecognised field, event type or format is # flagged in the disposition log AND the error/warning output, and @@ -123,38 +256,13 @@ def test_header_only_file_is_not_a_missing_column_error(): # Inputs are derived from the synthetic fixture 01_harvesting.csv. # --------------------------------------------------------------------------- -import csv as _csv -import io as _io - -import pytest - - -def _fixture_rows() -> tuple[list[str], list[dict]]: - reader = _csv.DictReader(_io.StringIO(FIXTURE.read_text())) - return list(reader.fieldnames or []), list(reader) - - -def _to_csv(header: list[str], rows: list[dict]) -> str: - out = _io.StringIO() - writer = _csv.DictWriter(out, fieldnames=header, lineterminator="\n", extrasaction="ignore") - writer.writeheader() - for row in rows: - writer.writerow(row) - return out.getvalue() - - -def _assert_counts_balance(text: str, events, log) -> None: - rows_in = len(text.splitlines()) - 1 - outcome = {e.row_number: e.disposition for e in log if e.row_number >= 2 and e.disposition != "flagged"} - assert len(outcome) == rows_in, "every input row needs exactly one outcome" - assert sum(1 for d in outcome.values() if d == "accepted") == len(events) - @pytest.fixture def unknown_column_csv() -> str: header, rows = _fixture_rows() - for i, row in enumerate(rows): - row["partner_pallet_code"] = f"PAL-SYN-{i}" + for row in rows: + # Reuse an existing fixture cell; do not invent pallet IDs. + row["partner_pallet_code"] = row["field_id"] return _to_csv([*header, "partner_pallet_code"], rows) @@ -202,6 +310,9 @@ def test_acceptance_unknown_column_is_flagged(unknown_column_csv): assert any(e.reason == "unrecognized_column" and e.field == "partner_pallet_code" for e in log) assert any("partner_pallet_code" in w for w in warnings) assert errors == [] + # Unknown column values are kept on the event, never silently dropped. + header, rows = _fixture_rows() + assert events[0].kdes.get("partner_pallet_code") == rows[0]["field_id"] _assert_counts_balance(unknown_column_csv, events, log) @@ -231,6 +342,10 @@ def test_acceptance_missing_required_columns_flagged(missing_columns_csv): assert events == [] assert {e.field for e in log if e.reason == "missing_column"} == {"quantity", "unit_of_measure"} assert len(errors) == 1 and "quantity" in errors[0] and "unit_of_measure" in errors[0] + # Per-row outcomes, not file-level only. + data_rejects = [e for e in log if e.row_number >= 2 and e.reason == "missing_required_columns"] + assert len(data_rejects) == len(missing_columns_csv.splitlines()) - 1 + _assert_counts_balance(missing_columns_csv, events, log) def test_acceptance_garbled_and_empty_rows_flagged(garbled_and_empty_rows_csv): From 2cfcc3146cce74341ed7a57c8e5d93b77bbc8273 Mon Sep 17 00:00:00 2001 From: Christopher Sellers <1@christophersellers.com> Date: Wed, 28 Oct 2026 01:55:24 -0700 Subject: [PATCH 4/6] fix(ingestion): one final CSV disposition and full overflow accounting Stop logging a pre-ingest flagged "Accepted with missing KDEs" so a later ingest reject cannot leave a row with two dispositions; fold missing-KDE notes onto the single final exclusive entry instead. On the missing-column early return, bucket unread overflow rows so every physical row is counted, include row_disposition_log on the 400 response, and skip aggregate overflow parse_errors when synthesizing rejected EventResults so totals match the exclusive bucket sum. Signed-off-by: Christopher Sellers <1@christophersellers.com> --- services/ingestion/app/csv_templates.py | 61 +++++- .../test_csv_partner_file_hardening_sel549.py | 201 +++++++++++++++++- 2 files changed, 250 insertions(+), 12 deletions(-) diff --git a/services/ingestion/app/csv_templates.py b/services/ingestion/app/csv_templates.py index 8e1fe1c7a..627534dd1 100644 --- a/services/ingestion/app/csv_templates.py +++ b/services/ingestion/app/csv_templates.py @@ -876,6 +876,7 @@ def finalize_row_dispositions_from_ingest( row_log: Optional[list[CsvRowDispositionEntry]], event_source_rows: Sequence[int], event_results: Sequence[object], + row_notes: Optional[Mapping[int, Sequence[str]]] = None, ) -> None: """Append the final per-row disposition after ``ingest_events`` returns. @@ -883,25 +884,33 @@ def finalize_row_dispositions_from_ingest( Each source row listed in ``event_source_rows`` (parallel to the parsed events / the first ``len(event_source_rows)`` ingest results) gets exactly one exclusive outcome here: accepted, or rejected (including duplicates). + + ``row_notes`` carries parse-time observations (e.g. missing KDEs) that must + ride on that final exclusive entry — never as a second disposition that + could claim the row was already accepted. """ if row_log is None: return + notes = row_notes or {} # ingest_csv may append synthetic EventResults for parse_errors after # ingest_events; only the results that correspond to parsed events finalize. for row_number, result in zip(event_source_rows, event_results): status = getattr(result, "status", None) errors = list(getattr(result, "errors", None) or []) + extra = list(notes.get(row_number) or []) + note_suffix = f"; {'; '.join(extra)}" if extra else "" if status == "accepted": + detail = "Row ingested" + note_suffix _append_row_log( row_log, row_number=row_number, disposition="accepted", reason="ingested", - detail="Row ingested", + detail=detail, ) continue reason = _ingest_reject_reason(errors) - detail = errors[0] if errors else "Row rejected during ingest" + detail = (errors[0] if errors else "Row rejected during ingest") + note_suffix # Never echo partner cell values; EventResult errors are codes/messages. _append_row_log( row_log, @@ -918,6 +927,7 @@ def parse_csv_rows( header_mapping: Optional[Mapping[str, str]] = None, row_log: Optional[list[CsvRowDispositionEntry]] = None, event_source_rows: Optional[list[int]] = None, + row_notes: Optional[dict[int, list[str]]] = None, ) -> tuple[list[IngestEvent], list[str], list[str]]: """Parse CSV text into ``IngestEvent`` objects. @@ -946,6 +956,12 @@ def parse_csv_rows( of each returned event (same order as ``events``), so the ingest path can finalize dispositions without inventing row associations. + ``row_notes``, when provided, collects non-exclusive parse observations + (currently missing KDEs) keyed by row number. Callers fold these into the + final exclusive disposition via ``finalize_row_dispositions_from_ingest`` + so a row never carries both a pre-ingest "Accepted with missing KDEs" + flag and a later reject. + Returns ``(events, parse_errors, kde_warnings)``. """ reader = csv.DictReader(io.StringIO(text)) @@ -1077,6 +1093,20 @@ def parse_csv_rows( reason="missing_required_columns", detail=detail, ) + # Unread rows past the buffer still exist physically; every one must + # land in an exclusive bucket so accepted+rejected+skipped equals the + # file's data-row count (SEL-549 / Codex review). + if skipped_overflow: + base = 1 + len(records) + for offset in range(1, skipped_overflow + 1): + overflow_row = base + offset + _append_row_log( + row_log, + row_number=overflow_row, + disposition="rejected", + reason="missing_required_columns", + detail=detail, + ) return ([], [detail], []) unrecognized = _unrecognized_partner_columns(header, header_mapping) @@ -1383,13 +1413,12 @@ def _reject(reason: str, message: str) -> None: missing_kdes = _validate_kde_completeness(row_cte, event, kdes) if missing_kdes: kde_warnings.append(f"Row {row_num}: Missing KDEs for {row_cte}: {', '.join(missing_kdes)}") - _append_row_log( - row_log, - row_number=row_num, - disposition="flagged", - reason="missing_kdes", - detail=f"Accepted with missing KDEs: {', '.join(missing_kdes)}", - ) + # Do not log a flagged "Accepted with missing KDEs" here: ingest + # may still reject the row, and a row must have exactly one + # final exclusive disposition. Carry the observation as a note + # for finalize_row_dispositions_from_ingest to attach. + if row_notes is not None: + row_notes.setdefault(row_num, []).append(f"missing KDEs: {', '.join(missing_kdes)}") # Do not log accepted/ingested here. Dedup, KDE checks and # persistence decide the final disposition in ingest_events; # finalize_row_dispositions_from_ingest records it once. @@ -1912,12 +1941,14 @@ async def ingest_csv( ) row_log: list[CsvRowDispositionEntry] = [] event_source_rows: list[int] = [] + row_notes: dict[int, list[str]] = {} events, parse_errors, kde_warnings = parse_csv_rows( text, fallback_cte, header_mapping=header_mapping.applied_mapping, row_log=row_log, event_source_rows=event_source_rows, + row_notes=row_notes, ) kde_warnings.extend(xlsx_warnings) missing_columns = [e.field for e in row_log if e.reason == "missing_column" and e.field] @@ -1928,6 +1959,10 @@ async def ingest_csv( if missing_columns: # SEL-549: name the absent columns so the uploader can fix the header. detail["missing_columns"] = missing_columns + # Every physical row's exclusive outcome, including unread overflow on + # the missing-column path — clients need this to reconcile counts. + if row_log: + detail["row_disposition_log"] = [entry.model_dump() for entry in row_log] if xlsx_warnings: detail["kde_warnings"] = kde_warnings detail["kde_warning_count"] = len(kde_warnings) @@ -1974,7 +2009,7 @@ async def ingest_csv( source_file=receipt, ) # Final dispositions only: parse never claimed these rows were ingested. - finalize_row_dispositions_from_ingest(row_log, event_source_rows, response.events) + finalize_row_dispositions_from_ingest(row_log, event_source_rows, response.events, row_notes=row_notes) artifacts_recorded = 0 if header_mapping.ok and header_mapping.suggestions: @@ -1999,6 +2034,12 @@ async def ingest_csv( from .webhook_models import EventResult for err in parse_errors: + # Aggregate overflow ("N more row(s) ... skipped") is already + # represented per physical row in row_disposition_log. Emitting it + # again as a rejected EventResult inflates rejected/total past the + # exclusive bucket sum (SEL-549 / Codex review). + if "more row(s)" in err and "were skipped" in err: + continue response.events.append( EventResult( traceability_lot_code="", cte_type=fallback_cte or "unknown", status="rejected", errors=[err] diff --git a/services/ingestion/tests/test_csv_partner_file_hardening_sel549.py b/services/ingestion/tests/test_csv_partner_file_hardening_sel549.py index d1003f3ff..654840274 100644 --- a/services/ingestion/tests/test_csv_partner_file_hardening_sel549.py +++ b/services/ingestion/tests/test_csv_partner_file_hardening_sel549.py @@ -41,11 +41,17 @@ def _parse(text: str, cte: str | None = "harvesting", mapping=None): """ log: list = [] source_rows: list[int] = [] + row_notes: dict = {} events, errors, warnings = ct.parse_csv_rows( - text, cte, header_mapping=mapping, row_log=log, event_source_rows=source_rows + text, + cte, + header_mapping=mapping, + row_log=log, + event_source_rows=source_rows, + row_notes=row_notes, ) fake_results = [SimpleNamespace(status="accepted", errors=[]) for _ in source_rows] - ct.finalize_row_dispositions_from_ingest(log, source_rows, fake_results) + ct.finalize_row_dispositions_from_ingest(log, source_rows, fake_results, row_notes=row_notes) return events, errors, warnings, log @@ -355,3 +361,194 @@ def test_acceptance_garbled_and_empty_rows_flagged(garbled_and_empty_rows_csv): assert any(e.startswith("Row 3:") for e in errors) assert outcome[4].reason == "blank_row" # empty _assert_counts_balance(garbled_and_empty_rows_csv, events, log) + + +# --------------------------------------------------------------------------- +# Codex FAIL bf0265123 follow-ups: one final disposition; missing-column +# overflow; 400 surfaces row_disposition_log; overflow aggregates do not +# inflate rejected EventResults. +# --------------------------------------------------------------------------- + + +def test_missing_kdes_flag_is_not_a_second_disposition_when_ingest_rejects(): + """Parse must not leave a flagged 'Accepted with missing KDEs' beside a later reject.""" + header, rows = _fixture_rows() + # Drop reference_document so harvesting KDE completeness warns. + kept = [h for h in header if h != "reference_document"] + row = {k: v for k, v in rows[0].items() if k != "reference_document"} + text = _to_csv(kept, [row]) + log: list = [] + source_rows: list[int] = [] + row_notes: dict = {} + events, errors, warnings = ct.parse_csv_rows( + text, + "harvesting", + row_log=log, + event_source_rows=source_rows, + row_notes=row_notes, + ) + assert errors == [] and len(events) == 1 and source_rows == [2] + assert any("Missing KDEs" in w for w in warnings) + # No exclusive or flagged disposition claiming acceptance before ingest. + assert not any(e.disposition == "accepted" for e in log) + assert not any(e.reason == "missing_kdes" for e in log) + assert not any("Accepted with missing KDEs" in (e.detail or "") for e in log) + ct.finalize_row_dispositions_from_ingest( + log, + source_rows, + [SimpleNamespace(status="rejected", errors=["Duplicate event in batch"])], + row_notes=row_notes, + ) + exclusive = [e for e in log if e.row_number == 2 and e.disposition in {"accepted", "rejected", "skipped"}] + assert len(exclusive) == 1 + assert exclusive[0].disposition == "rejected" + assert exclusive[0].reason == "duplicate" + assert "Accepted with missing KDEs" not in (exclusive[0].detail or "") + # Missing KDEs may appear as a note on the final entry only. + assert "missing KDEs" in (exclusive[0].detail or "").lower() or exclusive[0].reason == "duplicate" + + +def test_missing_kdes_note_on_final_accepted_entry_only(): + header, rows = _fixture_rows() + kept = [h for h in header if h != "reference_document"] + row = {k: v for k, v in rows[0].items() if k != "reference_document"} + text = _to_csv(kept, [row]) + log: list = [] + source_rows: list[int] = [] + row_notes: dict = {} + events, errors, warnings = ct.parse_csv_rows( + text, + "harvesting", + row_log=log, + event_source_rows=source_rows, + row_notes=row_notes, + ) + assert len(events) == 1 + ct.finalize_row_dispositions_from_ingest( + log, + source_rows, + [SimpleNamespace(status="accepted", errors=[])], + row_notes=row_notes, + ) + row_entries = [e for e in log if e.row_number == 2] + exclusive = [e for e in row_entries if e.disposition in {"accepted", "rejected", "skipped"}] + flagged = [e for e in row_entries if e.disposition == "flagged" and e.reason == "missing_kdes"] + assert len(exclusive) == 1 and exclusive[0].disposition == "accepted" + assert flagged == [] + assert "missing KDEs" in (exclusive[0].detail or "") + assert "Accepted with missing KDEs" not in (exclusive[0].detail or "") + + +def test_missing_columns_with_overflow_accounts_for_every_physical_row(monkeypatch): + """Missing-column early return must still bucket unread overflow rows.""" + monkeypatch.setattr(ct, "MAX_CSV_ROWS", 2) + header, rows = _fixture_rows() + kept = [h for h in header if h not in {"quantity", "unit_of_measure"}] + # 5 physical data rows from fixture only (repeat real rows). + five = [dict(rows[i % len(rows)]) for i in range(5)] + for row in five: + row.pop("quantity", None) + row.pop("unit_of_measure", None) + text = _to_csv(kept, five) + events, errors, _w, log = _parse(text) + assert events == [] + assert errors and "Missing required columns" in errors[0] + physical = len(text.splitlines()) - 1 + exclusive = [e for e in log if e.row_number >= 2 and e.disposition in {"accepted", "rejected", "skipped"}] + assert len(exclusive) == physical + _assert_counts_balance(text, events, log) + + +def test_missing_columns_400_includes_row_disposition_log(monkeypatch): + """The no-valid-rows 400 must surface the per-row disposition log.""" + from app.authz import IngestionPrincipal, get_ingestion_principal + from app.webhook_compat import _verify_api_key + from fastapi import FastAPI + from fastapi.testclient import TestClient + + header, rows = _fixture_rows() + kept = [h for h in header if h not in {"quantity", "unit_of_measure"}] + body_rows = [{k: v for k, v in rows[0].items() if k in kept}] + text = _to_csv(kept, body_rows) + + app = FastAPI() + app.include_router(ct.router) + app.dependency_overrides[_verify_api_key] = lambda: None + app.dependency_overrides[get_ingestion_principal] = lambda: IngestionPrincipal( + key_id="test-key", scopes=["*"], tenant_id="t1" + ) + client = TestClient(app) + resp = client.post( + "/api/v1/ingest/csv", + files={"file": ("partner.csv", text.encode(), "text/csv")}, + data={"cte_type": "harvesting"}, + ) + assert resp.status_code == 400 + detail = resp.json()["detail"] + assert isinstance(detail, dict) + assert "missing_columns" in detail + assert "row_disposition_log" in detail + log = detail["row_disposition_log"] + exclusive = [ + e for e in log if e.get("row_number", 0) >= 2 and e.get("disposition") in {"accepted", "rejected", "skipped"} + ] + assert len(exclusive) == 1 + assert exclusive[0]["disposition"] == "rejected" + + +def test_overflow_aggregate_parse_errors_do_not_inflate_rejected_totals(monkeypatch): + """Aggregate overflow parse_errors must not become extra rejected EventResults.""" + from app.authz import IngestionPrincipal, get_ingestion_principal + from app.webhook_compat import _verify_api_key + from app.webhook_models import EventResult, IngestResponse + from fastapi import FastAPI + from fastapi.testclient import TestClient + + monkeypatch.setattr(ct, "MAX_CSV_ROWS", 2) + header, rows = _fixture_rows() + four = [dict(rows[i % len(rows)]) for i in range(4)] + text = _to_csv(header, four) + + async def _fake_ingest(payload, **kwargs): + results = [ + EventResult( + traceability_lot_code=ev.traceability_lot_code, + cte_type=ev.cte_type.value if hasattr(ev.cte_type, "value") else ev.cte_type, + status="accepted", + event_id=f"evt-{i}", + ) + for i, ev in enumerate(payload.events) + ] + return IngestResponse(accepted=len(results), rejected=0, total=len(results), events=results) + + monkeypatch.setattr(ct, "ingest_events", _fake_ingest) + + app = FastAPI() + app.include_router(ct.router) + app.dependency_overrides[_verify_api_key] = lambda: None + app.dependency_overrides[get_ingestion_principal] = lambda: IngestionPrincipal( + key_id="test-key", scopes=["*"], tenant_id="t1" + ) + client = TestClient(app) + resp = client.post( + "/api/v1/ingest/csv", + files={"file": ("big.csv", text.encode(), "text/csv")}, + data={"cte_type": "harvesting"}, + ) + assert resp.status_code == 200, resp.text + body = resp.json() + log = body["row_disposition_log"] + exclusive = [e for e in log if e["row_number"] >= 2 and e["disposition"] in {"accepted", "rejected", "skipped"}] + physical = 4 + assert len(exclusive) == physical + bucket_rejected = sum(1 for e in exclusive if e["disposition"] == "rejected") + bucket_skipped = sum(1 for e in exclusive if e["disposition"] == "skipped") + bucket_accepted = sum(1 for e in exclusive if e["disposition"] == "accepted") + assert bucket_accepted + bucket_rejected + bucket_skipped == physical + assert body["accepted"] == bucket_accepted + # The "N more row(s)" aggregate must never appear as its own rejected EventResult. + assert not any(any("more row(s)" in err for err in e.get("errors", [])) for e in body.get("events", [])) + # First over-limit row may still surface as one rejected EventResult; totals + # must match exclusive rejected, not rejected + aggregate count. + assert body["rejected"] == bucket_rejected + assert body["total"] == body["accepted"] + body["rejected"] From 22a84b65d0a6bc6ff8475c8ec78c28a8e35aefb6 Mon Sep 17 00:00:00 2001 From: Christopher Sellers <1@christophersellers.com> Date: Wed, 28 Oct 2026 03:22:57 -0700 Subject: [PATCH 5/6] fix(ingestion): structural overflow marker and partial overflow-tail count (SEL-549) Addresses Codex review 5680111391 on #3682. - Skip aggregate overflow summaries in ingest_csv by an explicit overflow_aggregates marker (object identity), not substring matching, so crafted cell text can no longer hide a real rejection from counts. - Keep the partial skipped_overflow count when csv.Error fires while counting the unread tail, and record one unreadable_overflow_tail rejection instead of dropping those rows. - Tests: adversarial quantity cell, structural marker, partial count on csv.Error, missing-column overflow rejected (all from tests/fixtures/01_harvesting.csv); update the jev 400 test for the row_disposition_log now carried on 400s. Signed-off-by: Christopher Sellers <1@christophersellers.com> --- services/ingestion/app/csv_templates.py | 77 +++++++- .../ingestion/tests/test_csv_jev_file_cte.py | 9 +- .../test_csv_partner_file_hardening_sel549.py | 182 ++++++++++++++++++ 3 files changed, 254 insertions(+), 14 deletions(-) diff --git a/services/ingestion/app/csv_templates.py b/services/ingestion/app/csv_templates.py index 627534dd1..e5929a310 100644 --- a/services/ingestion/app/csv_templates.py +++ b/services/ingestion/app/csv_templates.py @@ -928,6 +928,7 @@ def parse_csv_rows( row_log: Optional[list[CsvRowDispositionEntry]] = None, event_source_rows: Optional[list[int]] = None, row_notes: Optional[dict[int, list[str]]] = None, + overflow_aggregates: Optional[list[str]] = None, ) -> tuple[list[IngestEvent], list[str], list[str]]: """Parse CSV text into ``IngestEvent`` objects. @@ -962,6 +963,12 @@ def parse_csv_rows( so a row never carries both a pre-ingest "Accepted with missing KDEs" flag and a later reject. + ``overflow_aggregates``, when provided, receives the aggregate overflow + summary string(s) (e.g. "N more row(s) … skipped"). Those messages are + still appended to ``parse_errors`` for visibility, but callers that + synthesize rejected EventResults must skip only membership in this list — + never substring-match error text, which can embed partner cell values. + Returns ``(events, parse_errors, kde_warnings)``. """ reader = csv.DictReader(io.StringIO(text)) @@ -976,6 +983,7 @@ def parse_csv_rows( # quote) is rejected whole with its line named, never partially ingested. records: list[dict] = [] skipped_overflow = 0 + overflow_tail_unreadable = False header = None try: header = reader.fieldnames @@ -1007,11 +1015,19 @@ def parse_csv_rows( # file may hold many more rows the reader never reached. Count # them (without storing them) so the caller learns exactly what # was dropped instead of silently losing data. + # Keep any partial count if the unread tail raises csv.Error — + # zeroing would drop rows we already saw (SEL-549 / Codex). + overflow_tail_unreadable = False try: - skipped_overflow = sum(1 for _ in reader) - except csv.Error: - # A malformed tail must not break the rows already read. - skipped_overflow = 0 + for _ in reader: + skipped_overflow += 1 + except csv.Error as overflow_exc: + overflow_tail_unreadable = True + logger.warning( + "csv_overflow_tail_unreadable after=%s error=%s", + skipped_overflow, + overflow_exc, + ) except csv.Error as exc: line = reader.reader.line_num logger.warning("csv_unreadable line=%s error=%s", line, exc) @@ -1095,7 +1111,9 @@ def parse_csv_rows( ) # Unread rows past the buffer still exist physically; every one must # land in an exclusive bucket so accepted+rejected+skipped equals the - # file's data-row count (SEL-549 / Codex review). + # file's data-row count (SEL-549 / Codex review). Overflow rows here + # are rejected with missing_required_columns (not skipped): the whole + # file failed the header check. if skipped_overflow: base = 1 + len(records) for offset in range(1, skipped_overflow + 1): @@ -1107,6 +1125,21 @@ def parse_csv_rows( reason="missing_required_columns", detail=detail, ) + if overflow_tail_unreadable: + base = 1 + len(records) + skipped_overflow + tail_row = base + 1 + tail_detail = ( + f"Row {tail_row}: unread CSV overflow tail could not be fully " + f"counted (malformed after {skipped_overflow} row(s) past the limit)" + ) + _append_row_log( + row_log, + row_number=tail_row, + disposition="rejected", + reason="unreadable_overflow_tail", + detail=tail_detail, + ) + return ([], [detail, tail_detail], []) return ([], [detail], []) unrecognized = _unrecognized_partner_columns(header, header_mapping) @@ -1136,6 +1169,10 @@ def parse_csv_rows( f"Row {row_num}: {skipped_overflow} more row(s) after this one were skipped and not ingested" ) parse_errors.append(overflow_detail) + if overflow_aggregates is not None: + # Structural marker: ingest_csv skips EventResult synthesis + # by membership here, not by scanning error substrings. + overflow_aggregates.append(overflow_detail) for offset in range(1, skipped_overflow + 1): overflow_row = row_num + offset _append_row_log( @@ -1145,6 +1182,22 @@ def parse_csv_rows( reason="row_limit_overflow", detail=(f"Row {overflow_row}: skipped after CSV row limit of {MAX_CSV_ROWS}"), ) + if overflow_tail_unreadable: + # Partial count kept above; name the remainder explicitly so it + # is not silently dropped when csv.Error fires mid-tail. + tail_detail = ( + f"Row {row_num + skipped_overflow + 1}: unread CSV overflow tail " + f"could not be fully counted (malformed after {skipped_overflow} " + f"row(s) past the limit)" + ) + parse_errors.append(tail_detail) + _append_row_log( + row_log, + row_number=row_num + skipped_overflow + 1, + disposition="rejected", + reason="unreadable_overflow_tail", + detail=tail_detail, + ) limit_detail = f"Row {row_num}: CSV exceeds maximum row limit of {MAX_CSV_ROWS}" parse_errors.append(limit_detail) _append_row_log( @@ -1942,6 +1995,7 @@ async def ingest_csv( row_log: list[CsvRowDispositionEntry] = [] event_source_rows: list[int] = [] row_notes: dict[int, list[str]] = {} + overflow_aggregates: list[str] = [] events, parse_errors, kde_warnings = parse_csv_rows( text, fallback_cte, @@ -1949,6 +2003,7 @@ async def ingest_csv( row_log=row_log, event_source_rows=event_source_rows, row_notes=row_notes, + overflow_aggregates=overflow_aggregates, ) kde_warnings.extend(xlsx_warnings) missing_columns = [e.field for e in row_log if e.reason == "missing_column" and e.field] @@ -2033,12 +2088,14 @@ async def ingest_csv( if parse_errors: from .webhook_models import EventResult + # Structural skip set: only messages parse_csv_rows recorded as + # overflow aggregates. Never substring-match error text — quantity + # (and similar) embed raw cell values that an adversary can craft. + # Match by object identity, not string equality, so even a rejection + # message that happens to equal the summary text is still counted. + overflow_aggregate_ids = {id(a) for a in overflow_aggregates} for err in parse_errors: - # Aggregate overflow ("N more row(s) ... skipped") is already - # represented per physical row in row_disposition_log. Emitting it - # again as a rejected EventResult inflates rejected/total past the - # exclusive bucket sum (SEL-549 / Codex review). - if "more row(s)" in err and "were skipped" in err: + if id(err) in overflow_aggregate_ids: continue response.events.append( EventResult( diff --git a/services/ingestion/tests/test_csv_jev_file_cte.py b/services/ingestion/tests/test_csv_jev_file_cte.py index 8767d1c80..3f4b88d65 100644 --- a/services/ingestion/tests/test_csv_jev_file_cte.py +++ b/services/ingestion/tests/test_csv_jev_file_cte.py @@ -206,10 +206,11 @@ def test_not_configured_changes_nothing_and_sends_nothing(harness): resp = _post() assert resp.status_code == 400 - assert resp.json()["detail"] == { - "message": "No valid rows found", - "errors": ["Row 2: No CTE type — add cte_type column or pass cte_type param"], - } + detail = resp.json()["detail"] + assert detail["message"] == "No valid rows found" + assert detail["errors"] == ["Row 2: No CTE type — add cte_type column or pass cte_type param"] + # SEL-549: the 400 also carries the per-row disposition log. + assert (2, "rejected") in [(e["row_number"], e["disposition"]) for e in detail["row_disposition_log"]] assert "bodies" not in seen assert harness["payloads"] == [] and harness["artifacts"] == [] diff --git a/services/ingestion/tests/test_csv_partner_file_hardening_sel549.py b/services/ingestion/tests/test_csv_partner_file_hardening_sel549.py index 654840274..6297d21eb 100644 --- a/services/ingestion/tests/test_csv_partner_file_hardening_sel549.py +++ b/services/ingestion/tests/test_csv_partner_file_hardening_sel549.py @@ -552,3 +552,185 @@ async def _fake_ingest(payload, **kwargs): # must match exclusive rejected, not rejected + aggregate count. assert body["rejected"] == bucket_rejected assert body["total"] == body["accepted"] + body["rejected"] + + +# --------------------------------------------------------------------------- +# Codex FAIL 2cfcc314 follow-ups: structural overflow marker (not substring); +# preserve partial skipped_overflow on csv.Error while counting the tail. +# --------------------------------------------------------------------------- + + +def test_adversarial_quantity_with_overflow_wording_still_counts_rejected(monkeypatch): + """Adversarial input: a fixture quantity cell embeds overflow summary wording. + + The EventResult filter must not use substring matching on "more row(s)" / + "were skipped", or a crafted cell can hide a real bad_quantity rejection + from response.rejected / events. Built from a real 01_harvesting.csv row. + """ + from app.authz import IngestionPrincipal, get_ingestion_principal + from app.webhook_compat import _verify_api_key + from app.webhook_models import IngestResponse + from fastapi import FastAPI + from fastapi.testclient import TestClient + + header, rows = _fixture_rows() + # ADVERSARIAL INPUT (derived from fixture row 0): quantity cell text that + # contains both substrings the old filter looked for. Pair with a second + # real fixture row so the request stays on the 200 path where parse_errors + # are synthesized into EventResults (the substring filter's blast radius). + adversarial_qty = "7 more row(s) after this one were skipped and not ingested" + crafted = dict(rows[0]) + crafted["quantity"] = adversarial_qty + good = dict(rows[1 % len(rows)]) + csv_text = _to_csv(header, [crafted, good]) + + async def _fake_ingest(payload, **kwargs): + from app.webhook_models import EventResult + + results = [ + EventResult( + traceability_lot_code=ev.traceability_lot_code, + cte_type=ev.cte_type.value if hasattr(ev.cte_type, "value") else ev.cte_type, + status="accepted", + event_id=f"evt-{i}", + ) + for i, ev in enumerate(payload.events) + ] + return IngestResponse(accepted=len(results), rejected=0, total=len(results), events=results) + + monkeypatch.setattr(ct, "ingest_events", _fake_ingest) + + app = FastAPI() + app.include_router(ct.router) + app.dependency_overrides[_verify_api_key] = lambda: None + app.dependency_overrides[get_ingestion_principal] = lambda: IngestionPrincipal( + key_id="test-key", scopes=["*"], tenant_id="t1" + ) + client = TestClient(app) + resp = client.post( + "/api/v1/ingest/csv", + files={"file": ("adversarial.csv", csv_text.encode(), "text/csv")}, + data={"cte_type": "harvesting"}, + ) + assert resp.status_code == 200, resp.text + body = resp.json() + assert body["accepted"] == 1 + assert body["rejected"] >= 1, ( + f"adversarial bad_quantity must still count as rejected EventResult; events={body.get('events')}" + ) + assert any(any("is not a number" in err for err in e.get("errors", [])) for e in body.get("events", [])) + log_rejected = sum( + 1 + for e in body.get("row_disposition_log", []) + if e.get("row_number", 0) >= 2 and e.get("disposition") == "rejected" + ) + assert body["rejected"] == log_rejected + + +def test_overflow_aggregate_skipped_by_structural_marker_not_substring(monkeypatch): + """Only messages recorded as overflow aggregates skip EventResult synthesis.""" + from app.authz import IngestionPrincipal, get_ingestion_principal + from app.webhook_compat import _verify_api_key + from app.webhook_models import EventResult, IngestResponse + from fastapi import FastAPI + from fastapi.testclient import TestClient + + monkeypatch.setattr(ct, "MAX_CSV_ROWS", 2) + header, rows = _fixture_rows() + four = [dict(rows[i % len(rows)]) for i in range(4)] + csv_text = _to_csv(header, four) + + async def _fake_ingest(payload, **kwargs): + results = [ + EventResult( + traceability_lot_code=ev.traceability_lot_code, + cte_type=ev.cte_type.value if hasattr(ev.cte_type, "value") else ev.cte_type, + status="accepted", + event_id=f"evt-{i}", + ) + for i, ev in enumerate(payload.events) + ] + return IngestResponse(accepted=len(results), rejected=0, total=len(results), events=results) + + monkeypatch.setattr(ct, "ingest_events", _fake_ingest) + + app = FastAPI() + app.include_router(ct.router) + app.dependency_overrides[_verify_api_key] = lambda: None + app.dependency_overrides[get_ingestion_principal] = lambda: IngestionPrincipal( + key_id="test-key", scopes=["*"], tenant_id="t1" + ) + client = TestClient(app) + resp = client.post( + "/api/v1/ingest/csv", + files={"file": ("big.csv", csv_text.encode(), "text/csv")}, + data={"cte_type": "harvesting"}, + ) + assert resp.status_code == 200, resp.text + body = resp.json() + assert not any(any("more row(s)" in err for err in e.get("errors", [])) for e in body.get("events", [])) + exclusive = [ + e + for e in body["row_disposition_log"] + if e["row_number"] >= 2 and e["disposition"] in {"accepted", "rejected", "skipped"} + ] + assert body["accepted"] + body["rejected"] == sum( + 1 for e in exclusive if e["disposition"] in {"accepted", "rejected"} + ) + + +def test_csv_error_while_counting_overflow_keeps_partial_count(monkeypatch): + """A csv.Error mid-tail must not zero skipped_overflow; keep what was counted.""" + import csv as csvmod + + monkeypatch.setattr(ct, "MAX_CSV_ROWS", 2) + header, rows = _fixture_rows() + # Buffer MAX+1 fixture rows, then one more readable overflow row (real + # fixture line), then a field that exceeds a lowered field_size_limit so + # counting the unread tail raises csv.Error after a partial count. + three = [dict(rows[i % len(rows)]) for i in range(3)] + base = _to_csv(header, three) + extra = base.splitlines()[1] # real fixture data line — invent nothing + # Oversized field: reuse the fixture lot code as a prefix so we invent no + # product/lot identity; pad with digits only. + lot = rows[0]["traceability_lot_code"] + huge = lot + ("0" * 200) + n_cols = len(header) + oversized_line = ",".join([huge] + [""] * (n_cols - 1)) + csv_text = base.rstrip("\n") + "\n" + extra + "\n" + oversized_line + "\n" + + old_limit = csvmod.field_size_limit() + csvmod.field_size_limit(64) + try: + events, errors, _w, log = _parse(csv_text) + finally: + csvmod.field_size_limit(old_limit) + + # With MAX=2: rows 2–3 capacity, row 4 first over-limit rejected, row 5 the + # one readable overflow counted before csv.Error. The unreadable remainder + # must be reported explicitly rather than wiping the partial count to zero. + exclusive = { + e.row_number: e for e in log if e.row_number >= 2 and e.disposition in {"accepted", "rejected", "skipped"} + } + assert 5 in exclusive, f"partial overflow row 5 missing; got {sorted(exclusive)}" + assert exclusive[5].disposition in {"skipped", "rejected"} + assert exclusive[4].disposition == "rejected" + assert exclusive[4].reason == "row_limit_exceeded" + assert any(e.reason == "unreadable_overflow_tail" for e in log) + + +def test_missing_column_overflow_rows_are_rejected_not_skipped(monkeypatch): + """Confirm missing-column unread overflow uses rejected/missing_required_columns.""" + monkeypatch.setattr(ct, "MAX_CSV_ROWS", 2) + header, rows = _fixture_rows() + kept = [h for h in header if h not in {"quantity", "unit_of_measure"}] + five = [dict(rows[i % len(rows)]) for i in range(5)] + for row in five: + row.pop("quantity", None) + row.pop("unit_of_measure", None) + csv_text = _to_csv(kept, five) + _events, _errors, _w, log = _parse(csv_text) + overflow = [e for e in log if e.row_number >= 5 and e.disposition in {"accepted", "rejected", "skipped"}] + assert overflow + assert all(e.disposition == "rejected" for e in overflow) + assert all(e.reason == "missing_required_columns" for e in overflow) From 774b661301537bd20876ece76136fd0caf058405 Mon Sep 17 00:00:00 2001 From: Christopher Sellers <1@christophersellers.com> Date: Wed, 28 Oct 2026 03:26:13 -0700 Subject: [PATCH 6/6] test(ingestion): assert missing-KDE note on final rejected disposition CodeRabbit: the prior assertion OR'd the already-checked duplicate reason, so it could not fail. Require the missing-KDE note text on the final exclusive entry instead. Signed-off-by: Christopher Sellers <1@christophersellers.com> --- .../ingestion/tests/test_csv_partner_file_hardening_sel549.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/services/ingestion/tests/test_csv_partner_file_hardening_sel549.py b/services/ingestion/tests/test_csv_partner_file_hardening_sel549.py index 6297d21eb..a2433aca5 100644 --- a/services/ingestion/tests/test_csv_partner_file_hardening_sel549.py +++ b/services/ingestion/tests/test_csv_partner_file_hardening_sel549.py @@ -404,8 +404,8 @@ def test_missing_kdes_flag_is_not_a_second_disposition_when_ingest_rejects(): assert exclusive[0].disposition == "rejected" assert exclusive[0].reason == "duplicate" assert "Accepted with missing KDEs" not in (exclusive[0].detail or "") - # Missing KDEs may appear as a note on the final entry only. - assert "missing KDEs" in (exclusive[0].detail or "").lower() or exclusive[0].reason == "duplicate" + # The missing-KDE note rides on the final rejected entry (not a second disposition). + assert "missing kdes" in (exclusive[0].detail or "").lower() def test_missing_kdes_note_on_final_accepted_entry_only():