test(products): INV-12 property test — export round-trips to a no-op (SD-0002 §6.8)

backend/app/domains/products/serialize.py

@@ -0,0 +1,114 @@
+"""Canonical serializer — CatalogProduct snapshot → canonical CSV (SD-0002 §6.5.5).
+
+The export half of "one codec, two directions": this writes exactly the columns
+codec.py/validate.py parse, in a row grammar (§6.5.1) the validator regroups
+identically — so re-importing an unmodified export diffs to nothing (INV-12).
+DB-free, like diff.py: it consumes the same CatalogProduct snapshot the diff
+engine reads (repo.load_catalog), and the service streams the result.
+"""
+from __future__ import annotations
+
+import csv
+import io
+from collections.abc import Iterable, Iterator
+from decimal import Decimal
+
+from .diff import CatalogProduct
+from .models import (
+    IMAGE_COLUMNS,
+    OPTION_VALUE_COLUMNS,
+    PRODUCT_COLUMNS,
+    VARIANT_COLUMNS,
+)
+
+# Canonical column order: Handle, then product, option-value, variant, image
+# columns. A superset of everything the parser knows (models.KNOWN_COLUMNS minus
+# the #15-reserved Component columns, which export never emits).
+HEADER: list[str] = [
+    "Handle",
+    *PRODUCT_COLUMNS,
+    *OPTION_VALUE_COLUMNS,
+    *VARIANT_COLUMNS,
+    *IMAGE_COLUMNS,
+]
+
+# field name -> column name, inverting the model registry (the parser reads
+# column->field; the serializer writes field->column).
+_PRODUCT_FIELD_TO_COL = {field: col for col, field in PRODUCT_COLUMNS.items()}
+_VARIANT_FIELD_TO_COL = {field: col for col, field in VARIANT_COLUMNS.items()}
+
+
+def _cell(value: object) -> str:
+    """Serialize one value to its canonical cell text (the parse inverse)."""
+    if value is None:
+        return ""
+    if isinstance(value, bool):
+        return "TRUE" if value else "FALSE"
+    if isinstance(value, Decimal):
+        return str(value)
+    if isinstance(value, (list, tuple)):
+        return ", ".join(str(v) for v in value)
+    return str(value)
+
+
+def catalog_to_csv(products: Iterable[CatalogProduct]) -> Iterator[str]:
+    """Stream canonical CSV text, header first, one product block at a time."""
+    buf = io.StringIO()
+    writer = csv.writer(buf)
+    writer.writerow(HEADER)
+    yield _drain(buf)
+    for product in products:
+        for row in _product_rows(product):
+            writer.writerow([row.get(col, "") for col in HEADER])
+        yield _drain(buf)
+
+
+def _drain(buf: io.StringIO) -> str:
+    text = buf.getvalue()
+    buf.seek(0)
+    buf.truncate(0)
+    return text
+
+
+def _product_rows(product: CatalogProduct) -> list[dict[str, str]]:
+    """The product's CSV rows (§6.5.1 grammar): product fields + option names on
+    the first row; one variant per row; images interleaved; image-only rows when
+    a product has more images than variants."""
+    base = {"Handle": product.handle, "Title": product.title}
+    for field, col in _PRODUCT_FIELD_TO_COL.items():
+        if field in product.fields:
+            base[col] = _cell(product.fields[field])
+    for slot, name in enumerate(product.option_names, start=1):
+        if name:
+            base[f"Option{slot} Name"] = name
+
+    count = max(len(product.variants), len(product.images))
+    rows: list[dict[str, str]] = []
+    for i in range(count):
+        # Row 0 carries the product-level fields; later rows carry only Handle.
+        row = dict(base) if i == 0 else {"Handle": product.handle}
+        if i < len(product.variants):
+            _write_variant(row, product, product.variants[i])
+        if i < len(product.images):
+            _write_image(row, product.images[i])
+        rows.append(row)
+    return rows
+
+
+def _write_image(row: dict[str, str], image) -> None:
+    row["Image Src"] = image.source_url
+    row["Image Position"] = str(image.position)
+    if image.alt_text is not None:
+        row["Image Alt Text"] = image.alt_text
+
+
+def _write_variant(row: dict[str, str], product: CatalogProduct, variant) -> None:
+    """Fill a row's option-value + variant columns for one variant."""
+    for slot, value in enumerate(variant.options, start=1):
+        # Only emit an option value when the product actually has that option
+        # (a no-option product's single variant carries all-NULL options).
+        if product.option_names[slot - 1] and value is not None:
+            row[f"Option{slot} Value"] = value
+    for field, col in _VARIANT_FIELD_TO_COL.items():
+        if field in variant.fields:
+            row[col] = _cell(variant.fields[field])

backend/tests/test_products_serialize.py

@@ -0,0 +1,172 @@
+"""Canonical serializer + INV-12 round-trip lock (SD-0002 §6.5.5, §6.8)."""
+import csv
+import io
+
+from app.domains.products import serialize
+from app.domains.products.diff import CatalogImage, CatalogProduct, CatalogVariant
+
+
+def _product(**kw) -> CatalogProduct:
+    """A minimal no-option catalog product (one all-NULL variant)."""
+    base = dict(
+        id=1, handle="moon-mug", title="Moon Mug",
+        option_names=(None, None, None),
+        fields={
+            "title": "Moon Mug", "description_html": None, "vendor": "Acme",
+            "product_type": "standalone", "google_product_category": None,
+            "tags": [], "status": "active", "published": True,
+        },
+        variants=[CatalogVariant(
+            id=1, options=(None, None, None), position=1,
+            fields={"sku": "WG-MUG", "barcode": None, "price": None, "cost": None,
+                    "weight": None, "weight_unit": None, "volume": None,
+                    "volume_unit": None, "tax_id_1": None, "tax_id_2": None,
+                    "inventory_tracker": None, "inventory_qty": None,
+                    "variant_image": None})],
+        images=[],
+    )
+    base.update(kw)
+    return CatalogProduct(**base)
+
+
+def _rows(products) -> list[dict]:
+    text = "".join(serialize.catalog_to_csv(products))
+    return list(csv.DictReader(io.StringIO(text)))
+
+
+def test_header_is_full_canonical_set():
+    text = "".join(serialize.catalog_to_csv([_product()]))
+    header = next(csv.reader(io.StringIO(text)))
+    # Handle + Title first; every known column present exactly once.
+    assert header[0] == "Handle"
+    assert "Title" in header and "Variant SKU" in header and "Image Src" in header
+    assert len(header) == len(set(header))
+
+
+def test_single_no_option_product_one_row():
+    rows = _rows([_product()])
+    assert len(rows) == 1
+    assert rows[0]["Handle"] == "moon-mug"
+    assert rows[0]["Title"] == "Moon Mug"
+    assert rows[0]["Variant SKU"] == "WG-MUG"
+    # A no-option product emits no option values.
+    assert rows[0]["Option1 Value"] == ""
+
+
+def _star_tee() -> CatalogProduct:
+    """A 3-variant, 2-image product (the sample.csv shape)."""
+    return CatalogProduct(
+        id=2, handle="star-tee", title="Star Tee",
+        option_names=("Size", "Color", None),
+        fields={"title": "Star Tee", "description_html": None, "vendor": "Acme",
+                "product_type": "standalone", "google_product_category": None,
+                "tags": ["apparel", "tees"], "status": "active", "published": True},
+        variants=[
+            CatalogVariant(id=10, options=("S", "Indigo", None), position=1,
+                           fields={"sku": "WG-TEE-S", "variant_image": None}),
+            CatalogVariant(id=11, options=("M", "Indigo", None), position=2,
+                           fields={"sku": "WG-TEE-M", "variant_image": None}),
+            CatalogVariant(id=12, options=("L", "Indigo", None), position=3,
+                           fields={"sku": "WG-TEE-L", "variant_image": None}),
+        ],
+        images=[
+            CatalogImage(id=1, source_url="https://x/a.jpg", position=1, alt_text="front"),
+            CatalogImage(id=2, source_url="https://x/b.jpg", position=2, alt_text="back"),
+        ],
+    )
+
+
+def test_multivariant_with_images_interleaves():
+    rows = _rows([_star_tee()])
+    assert len(rows) == 3  # max(3 variants, 2 images)
+    # Product-level fields only on the first row.
+    assert rows[0]["Title"] == "Star Tee" and rows[1]["Title"] == ""
+    assert rows[0]["Tags"] == "apparel, tees" and rows[1]["Tags"] == ""
+    # Option names on row 0; option values on every variant row.
+    assert rows[0]["Option1 Name"] == "Size" and rows[1]["Option1 Name"] == ""
+    assert [r["Option1 Value"] for r in rows] == ["S", "M", "L"]
+    assert [r["Variant SKU"] for r in rows] == ["WG-TEE-S", "WG-TEE-M", "WG-TEE-L"]
+    # Two images on the first two rows; third row has no image.
+    assert [r["Image Src"] for r in rows] == ["https://x/a.jpg", "https://x/b.jpg", ""]
+    assert [r["Image Position"] for r in rows] == ["1", "2", ""]
+
+
+def test_more_images_than_variants_emits_image_only_rows():
+    p = _product(images=[
+        CatalogImage(id=1, source_url="https://x/a.jpg", position=1, alt_text=None),
+        CatalogImage(id=2, source_url="https://x/b.jpg", position=2, alt_text=None),
+        CatalogImage(id=3, source_url="https://x/c.jpg", position=3, alt_text=None),
+    ])
+    rows = _rows([p])
+    assert len(rows) == 3  # 1 variant, 3 images
+    assert rows[0]["Variant SKU"] == "WG-MUG"
+    assert rows[1]["Variant SKU"] == "" and rows[1]["Handle"] == "moon-mug"
+    assert [r["Image Src"] for r in rows] == ["https://x/a.jpg", "https://x/b.jpg", "https://x/c.jpg"]
+
+
+import random
+
+from app.domains.products import codec, diff, validate
+
+
+def _gen_catalog(seed: int) -> dict:
+    """A deterministic random text-field catalog, in load_catalog()'s shape."""
+    rng = random.Random(seed)
+    words = ["moon", "star", "river", "cloud", "ember", "fern", "slate", "wave"]
+    catalog: dict[str, CatalogProduct] = {}
+    n_products = rng.randint(1, 6)
+    for pid in range(1, n_products + 1):
+        handle = "-".join(rng.sample(words, rng.randint(1, 3))) + f"-{pid}"
+        if handle in catalog:
+            continue
+        has_opts = rng.random() < 0.6
+        option_names = ("Size", "Color", None) if has_opts else (None, None, None)
+        tags = rng.sample(["apparel", "kitchen", "sale", "new"], rng.randint(0, 3))
+        fields = {
+            "title": f"{handle.title()} Thing",
+            "description_html": rng.choice([None, "<p>hi</p>"]),
+            "vendor": rng.choice([None, "Acme", "Wiggle Goods"]),
+            "product_type": "standalone",
+            "google_product_category": rng.choice([None, "Home & Garden"]),
+            "tags": tags,
+            "status": rng.choice(["draft", "active", "archived"]),
+            "published": rng.choice([True, False]),
+        }
+        variants = []
+        if has_opts:
+            sizes = rng.sample(["S", "M", "L", "XL"], rng.randint(1, 4))
+            for vi, size in enumerate(sizes, start=1):
+                variants.append(CatalogVariant(
+                    id=pid * 100 + vi, options=(size, "Indigo", None), position=vi,
+                    fields={"sku": f"SKU-{pid}-{vi}", "variant_image": None}))
+        else:
+            variants.append(CatalogVariant(
+                id=pid * 100 + 1, options=(None, None, None), position=1,
+                fields={"sku": f"SKU-{pid}", "variant_image": None}))
+        images = []
+        for ii in range(rng.randint(0, 3)):
+            images.append(CatalogImage(
+                id=pid * 10 + ii, source_url=f"https://img/{handle}-{ii}.jpg",
+                position=ii + 1, alt_text=rng.choice([None, f"alt {ii}"])))
+        catalog[handle] = CatalogProduct(
+            id=pid, handle=handle, title=fields["title"], option_names=option_names,
+            fields=fields, variants=variants, images=images)
+    return catalog
+
+
+def _roundtrip_diff(catalog: dict) -> diff.DiffResult:
+    """export → bytes → import pipeline → diff against the same catalog."""
+    text = "".join(serialize.catalog_to_csv(catalog.values()))
+    parsed = codec.parse_csv(text.encode("utf-8"))
+    products = validate.build_products(parsed)
+    return diff.compute_diff(catalog, products)
+
+
+def test_inv12_roundtrip_is_noop_over_generated_catalogs():
+    for seed in range(200):
+        catalog = _gen_catalog(seed)
+        result = _roundtrip_diff(catalog)
+        assert result.summary["adds"] == 0, f"seed {seed}: {result.summary}"
+        assert result.summary["updates"] == 0, f"seed {seed}: {result.summary}"
+        assert result.summary["errors"] == 0, f"seed {seed}: {result.summary}"
+        assert result.summary["unchanged"] == len(catalog), f"seed {seed}"