From 19ee695c204142c6d09c01263db32ee1ef08f0c6 Mon Sep 17 00:00:00 2001 From: Ben Stull Date: Thu, 11 Jun 2026 15:17:01 -0700 Subject: [PATCH] =?UTF-8?q?feat(products):=20row=20validation=20=E2=80=94?= =?UTF-8?q?=20=C2=A76.5.1=20rules=20+=20INV-15=20sanitization?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-Authored-By: Claude Fable 5 --- backend/app/domains/products/validate.py | 297 +++++++++++++++++++++++ backend/tests/test_products_validate.py | 164 +++++++++++++ 2 files changed, 461 insertions(+) create mode 100644 backend/app/domains/products/validate.py create mode 100644 backend/tests/test_products_validate.py diff --git a/backend/app/domains/products/validate.py b/backend/app/domains/products/validate.py new file mode 100644 index 0000000..22e0a64 --- /dev/null +++ b/backend/app/domains/products/validate.py @@ -0,0 +1,297 @@ +"""Row validation — ParsedFile rows → canonical products + row errors (SD-0002 §6.5.1). + +The codec (codec.py) handles file-level gates; this module is the row-semantics +half of the PUC-5 import spine. It groups consecutive rows sharing a Handle into +product blocks (Shopify's grammar), normalizes product/variant/image fields, and +records every rule violation as a merchant-language RowError. Errors never raise: +an error poisons its whole product block (the product previews as kind="error" +and is excluded from apply) while parsing continues so the merchant gets a +complete accounting in one pass (BUC-1a). Description HTML is sanitized with nh3 +on the way in (INV-15). +""" +from __future__ import annotations + +import re +from decimal import Decimal, InvalidOperation + +import nh3 + +from .models import ( + COMPONENT_COLUMNS, + OPTION_VALUE_COLUMNS, + PRODUCT_COLUMNS, + VARIANT_COLUMNS, + CanonicalImage, + CanonicalProduct, + CanonicalVariant, + ParsedFile, + Row, + RowError, +) + +_HANDLE_RE = re.compile(r"^[a-z0-9-]+$") +_STATUSES = {"draft", "active", "archived"} +# Product columns handled as attributes (title, option_names), not via fields{}. +_ATTRIBUTE_COLUMNS = {"Title", "Option1 Name", "Option2 Name", "Option3 Name"} +# Sentinel: the cell failed normalization (the error is already recorded). +_INVALID = object() + + +def build_products(parsed: ParsedFile) -> list[CanonicalProduct]: + """Group rows into product blocks and validate every §6.5.1 rule.""" + products: list[CanonicalProduct] = [] + closed_handles: set[str] = set() + block: list[Row] = [] + + def flush() -> None: + nonlocal block + if block: + closed_handles.add(block[0].cells["Handle"]) + products.append(_build_block(block)) + block = [] + + for row in parsed.rows: + handle = row.cells.get("Handle", "") + if block and handle == block[0].cells["Handle"]: + block.append(row) + continue + flush() + if not handle: + products.append( + _error_block(row, "(missing)", RowError(row.line_number, "Handle", "a row needs a Handle")) + ) + elif not _HANDLE_RE.match(handle): + products.append( + _error_block( + row, + handle, + RowError( + row.line_number, + "Handle", + f"'{handle}' isn't a valid handle — lowercase letters, numbers, and dashes only", + ), + ) + ) + elif handle in closed_handles: + products.append( + _error_block( + row, + handle, + RowError( + row.line_number, + "Handle", + f"rows for '{handle}' must be consecutive — it already appeared earlier in the file", + ), + ) + ) + else: + block = [row] + flush() + return products + + +def all_errors(products: list[CanonicalProduct]) -> list[RowError]: + return [error for product in products for error in product.errors] + + +def _error_block(row: Row, handle: str, error: RowError) -> CanonicalProduct: + return CanonicalProduct(first_line=row.line_number, handle=handle, title="", errors=[error]) + + +def _build_block(rows: list[Row]) -> CanonicalProduct: + first = rows[0] + handle = first.cells["Handle"] + errors: list[RowError] = [] + + title = first.cells.get("Title", "") + if not title: + errors.append(RowError(first.line_number, "Title", f"'{handle}' is missing its Title")) + + option_names = tuple(first.cells.get(f"Option{n} Name") or None for n in (1, 2, 3)) + has_options = any(option_names) + + # Product-level fields come from the first row only; empty cell == clear (None). + fields: dict[str, object] = {} + for column, field_name in PRODUCT_COLUMNS.items(): + if column in _ATTRIBUTE_COLUMNS or column not in first.cells: + continue + cell = first.cells[column] + if not cell: + fields[field_name] = None + continue + value = _product_value(first.line_number, column, field_name, cell, errors) + if value is not _INVALID: + fields[field_name] = value + + variants: list[CanonicalVariant] = [] + images: list[CanonicalImage] = [] + seen_combos: set[tuple[str | None, str | None, str | None]] = set() + + for index, row in enumerate(rows): + cells = row.cells + line = row.line_number + + for column in COMPONENT_COLUMNS: + if cells.get(column): + errors.append( + RowError(line, column, "kits arrive in a coming release — leave the Component columns empty") + ) + + has_image = bool(cells.get("Image Src")) + if has_image: + _collect_image(row, images, errors) + + # A row carries a variant iff any option value / Variant-* cell is filled; + # the first row of a no-option product always carries the single variant. + carries_variant = ( + any(cells.get(c) for c in OPTION_VALUE_COLUMNS) + or any(cells.get(c) for c in VARIANT_COLUMNS) + or (index == 0 and not has_options) + ) + if carries_variant: + options = tuple(cells.get(c) or None for c in OPTION_VALUE_COLUMNS) + for n in (1, 2, 3): + value, name = options[n - 1], option_names[n - 1] + if value and not name: + errors.append( + RowError( + line, + f"Option{n} Value", + f"Option{n} Value given but the product has no Option{n} Name", + ) + ) + elif name and not value: + errors.append( + RowError( + line, + f"Option{n} Value", + f"this variant is missing its Option{n} Value ('{name}')", + ) + ) + if not has_options: + if variants: + errors.append(RowError(line, None, "a product without options can have only one variant")) + elif options in seen_combos: + errors.append( + RowError(line, None, f"duplicate variant — '{handle}' already has a variant with these options") + ) + seen_combos.add(options) + + variant_fields: dict[str, object] = {} + for column, field_name in VARIANT_COLUMNS.items(): + if column not in cells: + continue + cell = cells[column] + if not cell: + variant_fields[field_name] = None + continue + value = _variant_value(line, column, field_name, cell, errors) + if value is _INVALID: + continue + variant_fields[field_name] = value + if field_name == "variant_image" and cell not in {i.source_url for i in images}: + images.append( + CanonicalImage(line_number=line, source_url=cell, position=len(images) + 1, alt_text=None) + ) + variants.append(CanonicalVariant(line_number=line, options=options, fields=variant_fields)) + elif index > 0 and not has_image: + errors.append(RowError(line, None, "this row has no variant or image data")) + + return CanonicalProduct( + first_line=first.line_number, + handle=handle, + title=title, + option_names=option_names, + fields=fields, + variants=variants, + images=images, + errors=errors, + ) + + +def _collect_image(row: Row, images: list[CanonicalImage], errors: list[RowError]) -> None: + cells = row.cells + source_url = cells["Image Src"] + position_cell = cells.get("Image Position", "") + position: int | None = None + if position_cell: + try: + position = int(position_cell) + if position < 1: + raise ValueError + except ValueError: + position = None + errors.append(RowError(row.line_number, "Image Position", f"'{position_cell}' is not a position")) + # Dedupe by source URL within the block — first occurrence wins. + if source_url in {i.source_url for i in images}: + return + images.append( + CanonicalImage( + line_number=row.line_number, + source_url=source_url, + position=position if position is not None else len(images) + 1, + alt_text=cells.get("Image Alt Text") or None, + ) + ) + + +def _product_value(line: int, column: str, field_name: str, cell: str, errors: list[RowError]) -> object: + if field_name == "tags": + return [tag.strip() for tag in cell.split(",") if tag.strip()] + if field_name == "status": + status = cell.lower() + if status not in _STATUSES: + errors.append(RowError(line, column, f"'{cell}' is not a status — use draft, active, or archived")) + return _INVALID + return status + if field_name == "published": + flag = cell.upper() + if flag not in ("TRUE", "FALSE"): + errors.append(RowError(line, column, f"'{cell}' is not TRUE or FALSE")) + return _INVALID + return flag == "TRUE" + if field_name == "description_html": + return nh3.clean(cell) + if field_name == "product_type": + if cell != "standalone": + errors.append(RowError(line, column, "kits arrive in a coming release — Type must be 'standalone'")) + return _INVALID + return cell + return cell + + +def _variant_value(line: int, column: str, field_name: str, cell: str, errors: list[RowError]) -> object: + if field_name in ("price", "cost"): + return _decimal_or_error(line, column, cell, f"'{cell}' is not a price", errors) + if field_name in ("weight", "volume"): + return _decimal_or_error(line, column, cell, f"'{cell}' is not a number", errors) + if field_name == "inventory_qty": + try: + quantity = int(cell) + if quantity < 0: + raise ValueError + except ValueError: + errors.append(RowError(line, column, f"'{cell}' is not a whole number")) + return _INVALID + return quantity + if field_name == "position": + try: + position = int(cell) + if position < 1: + raise ValueError + except ValueError: + errors.append(RowError(line, column, f"'{cell}' is not a position")) + return _INVALID + return position + return cell + + +def _decimal_or_error(line: int, column: str, cell: str, message: str, errors: list[RowError]) -> object: + try: + value = Decimal(cell) + if not value.is_finite() or value < 0: + raise InvalidOperation + except InvalidOperation: + errors.append(RowError(line, column, message)) + return _INVALID + return value diff --git a/backend/tests/test_products_validate.py b/backend/tests/test_products_validate.py new file mode 100644 index 0000000..07cd2f3 --- /dev/null +++ b/backend/tests/test_products_validate.py @@ -0,0 +1,164 @@ +"""§6.5.1 row validation — every row-error rule has a fixture (SD-0002 §6.8).""" +from decimal import Decimal + +import pytest + +from app.domains.products.codec import parse_csv +from app.domains.products.validate import build_products + + +def _products(*lines: str): + return build_products(parse_csv(("\n".join(lines) + "\n").encode())) + + +def _errors(*lines: str): + return [e for p in _products(*lines) for e in p.errors] + + +def test_simple_product_parses_clean(): + [p] = _products( + "Handle,Title,Vendor,Tags,Status,Published,Variant Price,Variant SKU", + "moon-mug,Moon Mug,Acme,\"kitchen, mugs\",active,TRUE,18.00,SKU-1", + ) + assert p.valid and p.handle == "moon-mug" and p.title == "Moon Mug" + assert p.fields["tags"] == ["kitchen", "mugs"] + assert p.fields["status"] == "active" and p.fields["published"] is True + [v] = p.variants + assert v.options == (None, None, None) + assert v.fields["price"] == Decimal("18.00") and v.fields["sku"] == "SKU-1" + + +def test_option_product_groups_consecutive_rows(): + [p] = _products( + "Handle,Title,Option1 Name,Option1 Value,Variant Price", + "tee,Tee,Size,S,24.00", + "tee,,,M,24.00", + "tee,,,L,26.00", + ) + assert p.valid and p.option_names == ("Size", None, None) + assert [v.options[0] for v in p.variants] == ["S", "M", "L"] + + +def test_image_only_rows_and_dedupe(): + [p] = _products( + "Handle,Title,Image Src,Image Position,Image Alt Text", + "mug,Mug,https://x/a.jpg,1,front", + "mug,,https://x/b.jpg,2,back", + "mug,,https://x/a.jpg,3,dupe", + ) + assert p.valid + assert [(i.source_url, i.position) for i in p.images] == [ + ("https://x/a.jpg", 1), ("https://x/b.jpg", 2), + ] + + +def test_variant_image_joins_product_images(): + [p] = _products( + "Handle,Title,Variant Image", + "mug,Mug,https://x/v.jpg", + ) + assert p.variants[0].fields["variant_image"] == "https://x/v.jpg" + assert [i.source_url for i in p.images] == ["https://x/v.jpg"] + + +def test_description_sanitized_inv15(): + [p] = _products( + "Handle,Title,Description", + 'mug,Mug,"

hi

"', + ) + html = p.fields["description_html"] + assert "

" in html and "script" not in html and "onclick" not in html + + +def test_blank_cell_clears_absent_column_missing(): + [p] = _products("Handle,Title,Vendor", "mug,Mug,") + assert p.fields["vendor"] is None # present-but-empty == clear + assert "status" not in p.fields # absent column == untouched + + +@pytest.mark.parametrize( + "header,row,column,fragment", + [ + ("Handle,Title", ",NoHandle", "Handle", "needs a Handle"), + ("Handle,Title", "Bad_Handle!,T", "Handle", "isn't a valid handle"), + ("Handle,Title", "mug,", "Title", "missing its Title"), + ("Handle,Title,Type", "mug,Mug,kit_virtual", "Type", "kits arrive"), + ("Handle,Title,Component 1 SKU", "mug,Mug,ABC", "Component 1 SKU", "kits arrive"), + ("Handle,Title,Status", "mug,Mug,live", "Status", "is not a status"), + ("Handle,Title,Published", "mug,Mug,YES", "Published", "is not TRUE or FALSE"), + ("Handle,Title,Variant Price", 'mug,Mug,"12,50"', "Variant Price", "is not a price"), + ("Handle,Title,Variant Cost", "mug,Mug,-3", "Variant Cost", "is not a price"), + ("Handle,Title,Variant Weight", "mug,Mug,heavy", "Variant Weight", "is not a number"), + ("Handle,Title,Variant Inventory Qty", "mug,Mug,3.5", "Variant Inventory Qty", "is not a whole number"), + ("Handle,Title,Variant Position", "mug,Mug,0", "Variant Position", "is not a position"), + ("Handle,Title,Option1 Value", "mug,Mug,Red", "Option1 Value", "no Option1 Name"), + ], +) +def test_row_error_rules(header, row, column, fragment): + errors = _errors(header, row) + assert any(e.column == column and fragment in e.message for e in errors), errors + + +def test_missing_option_value_for_named_option(): + errors = _errors( + "Handle,Title,Option1 Name,Option1 Value,Variant SKU", + "tee,Tee,Size,S,A", + "tee,,,,B", + ) + assert any("missing its Option1 Value" in e.message and e.line_number == 3 for e in errors) + + +def test_duplicate_option_combo_is_error(): + errors = _errors( + "Handle,Title,Option1 Name,Option1 Value", + "tee,Tee,Size,S", + "tee,,,S", + ) + assert any("duplicate variant" in e.message for e in errors) + + +def test_second_variant_on_no_option_product_is_error(): + errors = _errors( + "Handle,Title,Variant SKU", + "mug,Mug,A", + "mug,,B", + ) + assert any("without options can have only one variant" in e.message for e in errors) + + +def test_non_consecutive_handle_is_error(): + errors = _errors( + "Handle,Title", + "mug,Mug", + "tee,Tee", + "mug,", + ) + assert any("must be consecutive" in e.message and e.line_number == 4 for e in errors) + + +def test_no_data_row_is_error(): + errors = _errors( + "Handle,Title,Variant SKU,Image Src", + "mug,Mug,A,", + "mug,,,", + ) + assert any("no variant or image data" in e.message for e in errors) + + +def test_errors_poison_their_product_only(): + products = _products( + "Handle,Title,Variant Price", + "good-mug,Mug,10.00", + "bad-tee,Tee,not-a-price", + ) + by_handle = {p.handle: p for p in products} + assert by_handle["good-mug"].valid + assert not by_handle["bad-tee"].valid + + +def test_all_errors_collected_not_first_only(): + [p] = _products( + "Handle,Title,Status,Variant Price", + "mug,,bogus,abc", + ) + assert len(p.errors) == 3 # missing Title + bad status + bad price