hi
"]), "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}" ``` - [ ] **Step 2: Run to see what breaks** Run: `.venv/bin/python -m pytest backend/tests/test_products_serialize.py::test_inv12_roundtrip_is_noop_over_generated_catalogs -q` Expected: This is the integration moment. It may already PASS (Tasks 1–2 built the grammar correctly), or FAIL on a specific seed exposing a serialize/parse asymmetry. **If it fails, that is a real INV-12 bug** — debug with `superpowers:systematic-debugging`: print the offending product's `result.records` entry to see which field/variant/image the diff flagged, fix `serialize.py` (or, if the spec is genuinely wrong, amend the spec then replan), and re-run. Likely culprits to check in order: 1. **Option values for no-option products** — the all-NULL variant must emit no `Option*Value` (Task 1 `_write_variant` guards this). 2. **`description_html`** — `nh3.clean("hi
")` must equal the stored value. The generator uses already-sanitized HTML so this holds; if it flags, the catalog's stored HTML wasn't canonical (not an export bug). 3. **Image position** — exported positions must match stored; the generator uses sequential 1..n matching `_write_image`. 4. **Tags ordering/whitespace** — `", ".join` then split-and-strip must reproduce the list; the generator avoids commas-in-tags. - [ ] **Step 3: Fix any asymmetry found (no fix needed if it already passes)** Apply the minimal `serialize.py` change the failure points to. Re-run until green. - [ ] **Step 4: Run the whole serializer file** Run: `.venv/bin/python -m pytest backend/tests/test_products_serialize.py -q` Expected: PASS (all serializer tests, incl. INV-12 over 200 catalogs) - [ ] **Step 5: Commit** ```bash git add backend/tests/test_products_serialize.py backend/app/domains/products/serialize.py git commit -m "test(products): INV-12 property test — export round-trips to a no-op (SD-0002 §6.8)" ``` --- ## Task 4: Repo — status-filtered catalog export query **Files:** - Modify: `backend/app/domains/products/repo.py` - Modify: `backend/tests/test_products_serialize.py` (add a DB-backed test) — or a new `backend/tests/test_products_export.py` (preferred; keep serializer unit tests DB-free). - Create: `backend/tests/test_products_export.py` `load_catalog` already builds the full snapshot but takes no status filter. Rather than thread a filter through it (variants/images join off products), add a thin `export_catalog` that reuses `load_catalog` and filters products by status in Python — the catalog fits in memory (≤5k rows, INV-18) and this keeps one snapshot-builder. Status `all` returns everything. - [ ] **Step 1: Write the failing test** ```python # backend/tests/test_products_export.py """Export: status-filtered catalog snapshot + the streamed service (PUC-9, TEL-3).""" import csv import io import json import logging import psycopg import pytest from app.domains import products from app.domains.products import repo, serialize from app.platform import db CSV = ( b"Handle,Title,Vendor,Status,Variant Price\n" b"active-mug,Active Mug,Acme,active,18.00\n" b"draft-tee,Draft Tee,Acme,draft,24.00\n" ) @pytest.fixture() def migrated_conn(fresh_db_url): with psycopg.connect(fresh_db_url) as conn: db.migrate(conn) yield conn @pytest.fixture() def merchant(migrated_conn): acct = migrated_conn.execute( "INSERT INTO account (email) VALUES ('m@example.com') RETURNING id").fetchone()[0] sf = migrated_conn.execute( "INSERT INTO storefront (name) VALUES ('Shop') RETURNING id").fetchone()[0] migrated_conn.execute( "INSERT INTO storefront_membership (account_id, storefront_id) VALUES (%s,%s)", (acct, sf)) migrated_conn.commit() return {"account_id": acct, "storefront_id": sf} def _seed(conn, merchant): draft = products.import_validate(conn, merchant["storefront_id"], merchant["account_id"], "c.csv", CSV) products.confirm_draft(conn, merchant["storefront_id"], merchant["account_id"], draft["id"]) def test_export_all_returns_both(migrated_conn, merchant): _seed(migrated_conn, merchant) snap = repo.export_catalog(migrated_conn, merchant["storefront_id"], "all") assert {p.handle for p in snap} == {"active-mug", "draft-tee"} def test_export_status_filter(migrated_conn, merchant): _seed(migrated_conn, merchant) active = repo.export_catalog(migrated_conn, merchant["storefront_id"], "active") assert [p.handle for p in active] == ["active-mug"] draft = repo.export_catalog(migrated_conn, merchant["storefront_id"], "draft") assert [p.handle for p in draft] == ["draft-tee"] ``` - [ ] **Step 2: Run to verify it fails** Run: `.venv/bin/python -m pytest backend/tests/test_products_export.py -q` Expected: FAIL — `AttributeError: module ... has no attribute 'export_catalog'` - [ ] **Step 3: Add `export_catalog` to repo.py** ```python # in backend/app/domains/products/repo.py, after load_catalog() _EXPORT_STATUSES = ("all", "active", "draft", "archived") def export_catalog( conn: psycopg.Connection, storefront_id: int, status_filter: str ) -> list[CatalogProduct]: """The storefront's catalog as an ordered snapshot list, optionally filtered by product status (PUC-9). Reuses load_catalog's snapshot builder; the catalog fits in memory (≤5k rows, INV-18). Ordered by handle for a stable, deterministic export.""" catalog = load_catalog(conn, storefront_id) products = sorted(catalog.values(), key=lambda p: p.handle) if status_filter and status_filter != "all": products = [p for p in products if p.fields.get("status") == status_filter] return products ``` - [ ] **Step 4: Run to verify it passes** Run: `.venv/bin/python -m pytest backend/tests/test_products_export.py -q` Expected: PASS (2 passed) - [ ] **Step 5: Commit** ```bash git add backend/app/domains/products/repo.py backend/tests/test_products_export.py git commit -m "feat(products): status-filtered export_catalog snapshot query (PUC-9)" ``` --- ## Task 5: Service — streamed export + TEL-3, and the `EmptyCatalog` error **Files:** - Modify: `backend/app/domains/products/errors.py` - Modify: `backend/app/domains/products/service.py` - Modify: `backend/app/domains/products/__init__.py` - Modify: `backend/tests/test_products_export.py` The service streams the serialized CSV and emits TEL-3 once, after the stream completes, with `product_count` and `duration_ms`. An empty (post-filter) catalog raises `EmptyCatalog` → the BFF maps it to `409 empty_catalog` (PUC-9: Export disabled with a note when empty; the endpoint defends the same rule). Streaming + telemetry-after-completion means the service returns a generator that emits TEL-3 as its final act. The count is known up front (the snapshot list), so we capture it before yielding. - [ ] **Step 1: Write the failing test** ```python # append to backend/tests/test_products_export.py @pytest.fixture() def telemetry_propagation(): """ecomm.telemetry has propagate=False in app logging; re-enable for caplog.""" lg = logging.getLogger("ecomm.telemetry") old = lg.propagate lg.propagate = True try: yield finally: lg.propagate = old def test_export_streams_canonical_csv(migrated_conn, merchant): _seed(migrated_conn, merchant) text = "".join(products.export_catalog(migrated_conn, merchant["storefront_id"], "all")) rows = list(csv.DictReader(io.StringIO(text))) assert {r["Handle"] for r in rows} == {"active-mug", "draft-tee"} assert rows[0]["Handle"] == "active-mug" # sorted by handle def test_export_empty_raises(migrated_conn, merchant): # No catalog at all → empty. with pytest.raises(products.EmptyCatalog): list(products.export_catalog(migrated_conn, merchant["storefront_id"], "all")) def test_export_empty_after_filter_raises(migrated_conn, merchant): _seed(migrated_conn, merchant) # only active + draft exist with pytest.raises(products.EmptyCatalog): list(products.export_catalog(migrated_conn, merchant["storefront_id"], "archived")) def test_tel3_emitted(migrated_conn, merchant, caplog, telemetry_propagation): _seed(migrated_conn, merchant) with caplog.at_level(logging.INFO, logger="ecomm.telemetry"): list(products.export_catalog(migrated_conn, merchant["storefront_id"], "all")) events = [json.loads(r.message) for r in caplog.records if r.name == "ecomm.telemetry"] exported = [e for e in events if e["event"] == "catalog_exported"] assert len(exported) == 1 assert exported[0]["product_count"] == 2 assert exported[0]["status_filter"] == "all" assert "duration_ms" in exported[0] ``` - [ ] **Step 2: Run to verify it fails** Run: `.venv/bin/python -m pytest backend/tests/test_products_export.py -q` Expected: FAIL — `AttributeError: module 'app.domains.products' has no attribute 'export_catalog'` / `EmptyCatalog`. - [ ] **Step 3a: Add `EmptyCatalog` to errors.py** ```python # in backend/app/domains/products/errors.py, after NothingToApply class EmptyCatalog(ProductsError): """PUC-9: nothing to export (no products, or none matching the status filter).""" ``` - [ ] **Step 3b: Add the streaming service to service.py** ```python # in backend/app/domains/products/service.py # add to the existing imports: from collections.abc import Iterator from . import codec, diff, repo, serialize, validate from .errors import ( DraftExpired, DraftNotFound, EmptyCatalog, NothingToApply, PreviewStale, RunNotFound, ) # add this function (e.g. after summary()): def export_catalog( conn: psycopg.Connection, storefront_id: int, status_filter: str ) -> Iterator[str]: """Stream the storefront's catalog as canonical CSV (PUC-9; INV-12 codec). Read-only: builds the snapshot, then streams the serializer over it. TEL-3 is emitted once the stream is exhausted, with the product count and elapsed time. Raises EmptyCatalog before yielding anything if the (filtered) catalog is empty, so the BFF can answer 409 cleanly with no partial body. """ started = time.monotonic() snapshot = repo.export_catalog(conn, storefront_id, status_filter) if not snapshot: raise EmptyCatalog() def _stream() -> Iterator[str]: yield from serialize.catalog_to_csv(snapshot) telemetry.emit( "catalog_exported", storefront_id=storefront_id, status_filter=status_filter, product_count=len(snapshot), duration_ms=int((time.monotonic() - started) * 1000), ) return _stream() ``` Note: `repo.export_catalog` (snapshot, Task 4) and the service `export_catalog` (stream, here) share a name across layers — same intent, different layer, like `list_runs`/`get_run` already do. The empty check runs **eagerly** (before the generator is consumed) so `409` is decided before any bytes stream. - [ ] **Step 3c: Export from the package surface** ```python # in backend/app/domains/products/__init__.py # add EmptyCatalog to the errors import and __all__, and export_catalog to the # service import and __all__: from .errors import ( DraftExpired, DraftNotFound, EmptyCatalog, FileRejected, NothingToApply, PreviewStale, ProductsError, RunNotFound, ) from .service import ( confirm_draft, discard_draft, export_catalog, get_draft, get_draft_records, get_run, import_validate, list_runs, summary, ) # __all__ gains: "EmptyCatalog", "export_catalog" ``` - [ ] **Step 4: Run to verify it passes** Run: `.venv/bin/python -m pytest backend/tests/test_products_export.py -q` Expected: PASS (6 passed) - [ ] **Step 5: Commit** ```bash git add backend/app/domains/products/errors.py backend/app/domains/products/service.py backend/app/domains/products/__init__.py backend/tests/test_products_export.py git commit -m "feat(products): streamed export service + TEL-3 + EmptyCatalog (PUC-9, §9.1)" ``` --- ## Task 6: Decimal / numeric round-trip unit tests (the property test's blind spot) **Files:** - Modify: `backend/tests/test_products_serialize.py` The INV-12 property test deliberately uses text fields. Numeric fields (`price`, `cost`, `weight`, `volume`, `inventory_qty`, variant `position`) need explicit round-trip coverage because their serialize→parse is value-identity, not string-identity. We assert the **diff** is empty (the real invariant), not the bytes. This catches e.g. a serializer dropping a Decimal's trailing zero in a way the parser then reads back differently. - [ ] **Step 1: Write the failing/covering test** ```python # append to backend/tests/test_products_serialize.py 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 ``` - [ ] **Step 2: Run it** Run: `.venv/bin/python -m pytest backend/tests/test_products_serialize.py::test_decimal_and_int_variant_fields_roundtrip -q` Expected: PASS if Decimals round-trip cleanly (psycopg loads `NUMERIC` as `Decimal`, the diff compares `Decimal == Decimal`). **If it FAILS**, the diff flagged a numeric field — inspect `result.records[0]["detail"]`; the cause is almost always a `str(Decimal)` form the parser re-parses to a different `Decimal` (it shouldn't: `Decimal(str(d)) == d` always). Fix `serialize._cell` only if a real asymmetry exists; otherwise the test documents the guarantee. - [ ] **Step 3: (only if it failed) Fix `_cell`** No change expected; `Decimal(str(d)) == d` holds. If weight stored as `Decimal("0.250")` re-parses to `Decimal("0.250")` (it does — equal value), the diff's `!=` is False. Leave `_cell` as is. - [ ] **Step 4: Run the full serializer suite** Run: `.venv/bin/python -m pytest backend/tests/test_products_serialize.py -q` Expected: PASS (all) - [ ] **Step 5: Commit** ```bash git add backend/tests/test_products_serialize.py git commit -m "test(products): decimal/int variant fields round-trip clean (INV-12 coverage)" ``` --- ## Task 7: BFF — `GET /api/products/export` (streamed, gated, 409 empty) **Files:** - Modify: `backend/app/main.py` - Modify: `backend/tests/test_products_endpoints.py` The endpoint: `_merchant_gate` → `products.export_catalog(...)` → stream as `text/csv` with a `content-disposition` attachment filename. `409 empty_catalog` on `EmptyCatalog`. The `status` query param is validated to the four allowed values (default `all`). Uses FastAPI's `StreamingResponse`. - [ ] **Step 1: Write the failing test** ```python # append to backend/tests/test_products_endpoints.py def test_export_returns_canonical_csv(fresh_db_url): with _merchant_client(fresh_db_url) as client: draft = _upload(client).json() client.post(f"/api/products/imports/drafts/{draft['id']}/confirm") resp = client.get("/api/products/export") assert resp.status_code == 200 assert resp.headers["content-type"].startswith("text/csv") assert "attachment" in resp.headers["content-disposition"] body = resp.text assert body.splitlines()[0].startswith("Handle,") assert "moon-mug" in body def test_export_status_filter_respected(fresh_db_url): with _merchant_client(fresh_db_url) as client: # GOOD_CSV's moon-mug has no Status column → defaults to active. draft = _upload(client).json() client.post(f"/api/products/imports/drafts/{draft['id']}/confirm") assert "moon-mug" in client.get("/api/products/export?status=active").text # No archived products → 409. assert client.get("/api/products/export?status=archived").status_code == 409 def test_export_empty_catalog_409(fresh_db_url): with _merchant_client(fresh_db_url) as client: resp = client.get("/api/products/export") assert resp.status_code == 409 assert resp.json()["error"]["code"] == "empty_catalog" def test_export_requires_merchant(fresh_db_url): with TestClient(create_app(database_url=fresh_db_url)) as client: assert client.get("/api/products/export").status_code == 401 def test_export_bad_status_422(fresh_db_url): with _merchant_client(fresh_db_url) as client: assert client.get("/api/products/export?status=bogus").status_code == 422 ``` - [ ] **Step 2: Run to verify it fails** Run: `.venv/bin/python -m pytest backend/tests/test_products_endpoints.py -k export -q` Expected: FAIL — 404 (route not defined). - [ ] **Step 3: Add the endpoint to main.py** ```python # in backend/app/main.py # add StreamingResponse to the fastapi.responses import: from fastapi.responses import JSONResponse, PlainTextResponse, StreamingResponse # add this route, next to the other /api/products/* routes (e.g. after products_sample_csv # or before it — order is irrelevant for distinct paths): @app.get("/api/products/export") def export_products( status: str = Query(default="all", pattern="^(all|active|draft|archived)$"), conn: psycopg.Connection = Depends(get_conn), sess: dict | None = Depends(get_session), ): """Stream the catalog as canonical CSV, optionally status-filtered (§6.4; PUC-9).""" gate = _merchant_gate(conn, sess) if isinstance(gate, JSONResponse): return gate _account, sf = gate try: stream = products.export_catalog(conn, sf.id, status) except products.EmptyCatalog: return _error(409, "empty_catalog", "There are no products to export.") return StreamingResponse( stream, media_type="text/csv", headers={"content-disposition": 'attachment; filename="ecomm-products-export.csv"'}, ) ``` Note: `EmptyCatalog` is raised **eagerly** by `export_catalog` (Task 5) before the generator yields, so the `try/except` around the call catches it and we can answer `409` with no partial stream. Good. - [ ] **Step 4: Run to verify it passes** Run: `.venv/bin/python -m pytest backend/tests/test_products_endpoints.py -k export -q` Expected: PASS (5 passed) - [ ] **Step 5: Run the whole backend suite + import linter** Run: `( cd backend && ../.venv/bin/lint-imports ) && .venv/bin/python -m pytest backend -q` Expected: PASS — import boundaries clean (serialize.py is in `domains`, imports only `models`/`diff`), all backend tests green. - [ ] **Step 6: Commit** ```bash git add backend/app/main.py backend/tests/test_products_endpoints.py git commit -m "feat(api): GET /api/products/export — streamed canonical CSV, 409 empty_catalog (§6.4, PUC-9)" ``` --- ## Task 8: Frontend API — export URL helper **Files:** - Modify: `frontend/src/productsApi.ts` - Create: `frontend/src/productsApi.export.test.ts` The export is a **browser download**, not a fetch (we want the browser's native save dialog and streaming). So the API module contributes a URL builder + the status list, not a `request()` wrapper. - [ ] **Step 1: Write the failing test** ```typescript // frontend/src/productsApi.export.test.ts import { describe, expect, it } from "vitest"; import { EXPORT_STATUSES, exportUrl } from "./productsApi"; describe("export url", () => { it("lists the four status filters with 'all' first", () => { expect(EXPORT_STATUSES.map((s) => s.value)).toEqual(["all", "active", "draft", "archived"]); }); it("builds the endpoint url with the status query", () => { expect(exportUrl("all")).toBe("/api/products/export?status=all"); expect(exportUrl("archived")).toBe("/api/products/export?status=archived"); }); }); ``` - [ ] **Step 2: Run to verify it fails** Run: `( cd frontend && npx vitest run src/productsApi.export.test.ts )` Expected: FAIL — `exportUrl` / `EXPORT_STATUSES` not exported. - [ ] **Step 3: Add the helper to productsApi.ts** ```typescript // in frontend/src/productsApi.ts, after dialectLabel() export type ExportStatus = "all" | "active" | "draft" | "archived"; // PUC-9 status filter, 'all' first (the default). Labels drive the menu. export const EXPORT_STATUSES: { value: ExportStatus; label: string }[] = [ { value: "all", label: "All products" }, { value: "active", label: "Active" }, { value: "draft", label: "Draft" }, { value: "archived", label: "Archived" }, ]; export function exportUrl(status: ExportStatus): string { return `/api/products/export?status=${status}`; } ``` - [ ] **Step 4: Run to verify it passes** Run: `( cd frontend && npx vitest run src/productsApi.export.test.ts )` Expected: PASS (2 passed) - [ ] **Step 5: Commit** ```bash git add frontend/src/productsApi.ts frontend/src/productsApi.export.test.ts git commit -m "feat(frontend): export URL helper + status-filter list (PUC-9)" ``` --- ## Task 9: Frontend — the Export status-filter menu on the Products page **Files:** - Modify: `frontend/src/screens/products/ProductsPage.tsx` - Modify: `frontend/src/styles/products.css` - Create: `frontend/src/screens/products/ProductsPage.export.test.tsx` Replace the disabled Export button (lines ~66–71) with a small menu: a primary "Export" toggle that opens the four status options; choosing one navigates the browser to `exportUrl(status)` (triggering the download). When the catalog is empty, keep the **disabled** button + existing note (PUC-9). The menu is a native `