197 lines
8.6 KiB
Python
197 lines
8.6 KiB
Python
"""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}"
|
|
|
|
|
|
from decimal import Decimal
|
|
|
|
|
|
def test_decimal_and_int_variant_fields_roundtrip():
|
|
catalog = {
|
|
"priced": CatalogProduct(
|
|
id=1, handle="priced", title="Priced",
|
|
option_names=(None, None, None),
|
|
fields={"title": "Priced", "description_html": None, "vendor": None,
|
|
"product_type": "standalone", "google_product_category": None,
|
|
"tags": [], "status": "active", "published": True},
|
|
variants=[CatalogVariant(
|
|
id=1, options=(None, None, None), position=1,
|
|
fields={"sku": "P1", "price": Decimal("18.00"),
|
|
"cost": Decimal("9.5"), "weight": Decimal("0.250"),
|
|
"inventory_qty": 40, "variant_image": None})],
|
|
images=[],
|
|
)
|
|
}
|
|
result = _roundtrip_diff(catalog)
|
|
assert result.summary["unchanged"] == 1, result.records
|
|
assert result.summary["updates"] == 0
|