← back to Designerwallcoverings
scripts/c34-osborne-gaston/test_enqueue_specs.py
236 lines
#!/usr/bin/env python3
"""
RIGOROUS smoke test for the c34 Osborne & Little + Gaston Y Daniela
specs-metafield enqueue generator (enqueue-specs.py).
Hermetic + $0 + READ-ONLY: live_metafields() and token() are monkeypatched,
so the test NEVER calls the live Shopify API, never reads a real secret, and
never touches any DB. The generator's file writes (restore-map + enqueue SQL)
go to a tempdir.
Asserts BEHAVIOR, not "it ran" (Steve's THEMES/validation rule):
(a) every emitted row is blank-fill-only — never targets a product whose
metafield is already non-null in the (mocked) live store
(b) idempotent — re-running over the same inputs emits the identical set
(c) width formatted exactly "{n} inches", trailing zeros trimmed
(d) material canonical "Non-Woven" (no Non woven/Nonwoven), and
Wallpaper/Holographic/N/A are EXCLUDED from material while their
width is STILL emitted
(e) counts match 58 width + 39 material across 59 products (all-blank live)
Run: python3 test_enqueue_specs.py
Exit 0 = PASS, non-zero = BLOCK.
"""
import os, re, sys, csv, json, runpy, tempfile, importlib.util, contextlib, io
HERE = os.path.dirname(os.path.abspath(__file__))
GEN = os.path.join(HERE, "enqueue-specs.py")
WORKLIST = os.path.expanduser(
"~/.claude/yolo-queue/pending-approval/done/c34-osborne-gaston-specs-worklist.tsv"
)
EXPECT_PRODUCTS = 59
EXPECT_WIDTH = 58
EXPECT_MATERIAL = 39
CANON_MATERIALS = {"Non-Woven", "Paper", "Vinyl", "Vinyl-Coated Non-Woven"}
BAD_MATERIAL_VARIANTS = {"Non woven", "Nonwoven", "non-woven", "NON-WOVEN", "Non Woven"}
WIDTH_RE = re.compile(r"^\d+(\.\d+)? inches$")
def load_generator():
"""Import enqueue-specs.py as a module without running main()."""
spec = importlib.util.spec_from_file_location("c34_enqueue_specs", GEN)
mod = importlib.util.module_from_spec(spec)
spec.loader.exec_module(mod)
return mod
def run_generator(mod, live_map, limit=0, workdir=None):
"""
Run the generator's main() against the real worklist TSV but with
live_metafields + token mocked. live_map: {product_id: {key: value}} for
the simulated live store. Returns (stdout, emitted_sql_path, restore_path).
"""
mod.token = lambda: "MOCK-TOKEN-no-network"
mod.live_metafields = lambda pid, tok: dict(live_map.get(str(pid), {}))
argv = ["enqueue-specs.py", "--worklist", WORKLIST]
if limit:
argv += ["--limit", str(limit)]
cwd0 = os.getcwd()
os.chdir(workdir)
buf = io.StringIO()
try:
sys.argv = argv
with contextlib.redirect_stdout(buf):
mod.main()
finally:
os.chdir(cwd0)
out = buf.getvalue()
sqls = sorted(f for f in os.listdir(workdir) if f.startswith("enqueue-") and f.endswith(".sql"))
rms = sorted(f for f in os.listdir(workdir) if f.startswith("restore-map-") and f.endswith(".csv"))
return out, os.path.join(workdir, sqls[-1]), os.path.join(workdir, rms[-1])
def parse_inserts(sql_path):
"""
Parse the emitted INSERT...SELECT statements into structured rows:
[{product_id, key, value, endpoint, token_alias, source_agent, has_not_exists}]
"""
text = open(sql_path).read()
blocks = re.split(r"INSERT INTO shopify_api_queue", text)[1:]
out = []
for b in blocks:
ep = re.search(r"/products/(\d+)/metafields\.json", b)
# the payload JSON literal
pay = re.search(r"'(\{\"metafield\".*?\})'::jsonb", b)
ta = re.search(r"'product', \d+, 'pending', '([^']+)'", b)
if not (ep and pay):
continue
payload = json.loads(pay.group(1))
mf = payload["metafield"]
out.append({
"product_id": ep.group(1),
"key": mf["key"],
"value": mf["value"],
"type": mf["type"],
"namespace": mf["namespace"],
"source_agent": ta.group(1) if ta else None,
"has_not_exists": "NOT EXISTS" in b,
})
return out
PASS, FAIL = [], []
def check(cond, label):
(PASS if cond else FAIL).append(label)
print(f" [{'PASS' if cond else 'FAIL'}] {label}")
def main():
print(f"Generator : {GEN}")
print(f"Worklist : {WORKLIST}")
if not os.path.exists(GEN):
sys.exit("BLOCK: generator enqueue-specs.py not present yet (vp-dw-commerce authoring in parallel)")
if not os.path.exists(WORKLIST):
sys.exit("BLOCK: worklist TSV not found")
wl = list(csv.DictReader(open(WORKLIST), delimiter="\t"))
product_ids = [r["product_id"] for r in wl]
mod = load_generator()
# ---- Scenario 1: ALL-BLANK live store -> full expected set (counts gate e) ----
print("\n[1] All-blank live store (full expected emission)")
all_blank = {pid: {} for pid in product_ids}
with tempfile.TemporaryDirectory() as td:
out1, sql1, rm1 = run_generator(mod, all_blank, workdir=td)
ins1 = parse_inserts(sql1)
w = [i for i in ins1 if i["key"] == "width"]
m = [i for i in ins1 if i["key"] == "material"]
prods_checked = len({r["product_id"] for r in wl})
check(prods_checked == EXPECT_PRODUCTS, f"(e) 59 products in worklist (got {prods_checked})")
check(len(w) == EXPECT_WIDTH, f"(e) 58 width writes (got {len(w)})")
check(len(m) == EXPECT_MATERIAL, f"(e) 39 material writes (got {len(m)})")
# (c) width format exactly "{n} inches", trailing zeros trimmed
bad_fmt = [i["value"] for i in w if not WIDTH_RE.match(i["value"])]
check(not bad_fmt, f"(c) all widths match '{{n}} inches' (violations: {bad_fmt})")
trailing_zero = [i["value"] for i in w if re.match(r"^\d+\.0 inches$", i["value"])]
check(not trailing_zero, f"(c) no trailing-.0 widths (found: {trailing_zero})")
# (d) material canonical + no bad variants
bad_mat = [i["value"] for i in m if i["value"] not in CANON_MATERIALS]
check(not bad_mat, f"(d) all materials canonical {sorted(CANON_MATERIALS)} (violations: {bad_mat})")
leaked_variant = [i["value"] for i in m if i["value"] in BAD_MATERIAL_VARIANTS]
check(not leaked_variant, f"(d) no Non woven/Nonwoven variants leaked (found: {leaked_variant})")
nonwoven_present = any(i["value"] == "Non-Woven" for i in m)
check(nonwoven_present, "(d) canonical 'Non-Woven' present in emitted materials")
# (d) Wallpaper/Holographic/N/A excluded from material BUT width still emitted
excl_rows = [r for r in wl if r["proposed_material"].startswith("EXCLUDE-")
and r["proposed_material"] != "EXCLUDE-none"]
excl_pids = {r["product_id"] for r in excl_rows}
emitted_mat_pids = {i["product_id"] for i in m}
check(not (excl_pids & emitted_mat_pids),
f"(d) Wallpaper/Holographic/N/A products got NO material write "
f"(leaked: {excl_pids & emitted_mat_pids})")
# their width must still be emitted (these are real wallpaper rolls)
emitted_width_pids = {i["product_id"] for i in w}
# only the single SKIP-has-value row (W7980-01) legitimately has no width
skip_width_pids = {r["product_id"] for r in wl if r["proposed_width"].startswith("SKIP")}
missing_width = (excl_pids - skip_width_pids) - emitted_width_pids
check(not missing_width,
f"(d) excluded-material products STILL get a width write (missing: {missing_width})")
# (a) blank-only: every emitted target was blank in the (all-blank) live store -> trivially true,
# asserted strongly in Scenario 2.
check(all(i["has_not_exists"] for i in ins1), "(structure) every INSERT carries the NOT-EXISTS dedup guard")
check(all(i["source_agent"] == "c34-osborne-gaston-specs" for i in ins1),
"(structure) every row tagged source_agent=c34-osborne-gaston-specs")
check(all(i["namespace"] == "custom" and i["type"] == "single_line_text_field" for i in ins1),
"(structure) every metafield is custom / single_line_text_field")
sig1 = sorted((i["product_id"], i["key"], i["value"]) for i in ins1)
# ---- Scenario 1b: idempotent re-run (gate b) ----
print("\n[2] Idempotent re-run over identical inputs")
with tempfile.TemporaryDirectory() as td:
_, sql1b, _ = run_generator(mod, all_blank, workdir=td)
sig1b = sorted((i["product_id"], i["key"], i["value"]) for i in parse_inserts(sql1b))
check(sig1 == sig1b, f"(b) re-run emits the IDENTICAL set ({len(sig1)} rows both runs)")
# ---- Scenario 2: live store ALREADY FILLED -> blank-only must emit ZERO (gate a) ----
print("\n[3] Live store already filled (blank-only guard must skip everything)")
filled = {}
for r in wl:
mp = {}
if r["proposed_width"] not in ("(blank)", "") and not r["proposed_width"].startswith("SKIP"):
mp["width"] = "PRE-EXISTING 99 inches"
if r["proposed_material"] and not r["proposed_material"].startswith(("EXCLUDE", "SKIP")):
mp["material"] = "PRE-EXISTING Silk"
filled[r["product_id"]] = mp
with tempfile.TemporaryDirectory() as td:
out3, sql3, _ = run_generator(mod, filled, workdir=td)
ins3 = parse_inserts(sql3)
check(len(ins3) == 0,
f"(a) blank-only: 0 writes when live already has values (got {len(ins3)})")
# ---- Scenario 3: ONE product blank, rest filled -> only that one emits (gate a, fine-grained) ----
print("\n[4] Single genuinely-blank product among filled peers")
target = product_ids[0]
mixed = dict(filled)
mixed[target] = {} # this one is blank for both keys
with tempfile.TemporaryDirectory() as td:
_, sql4, _ = run_generator(mod, mixed, workdir=td)
ins4 = parse_inserts(sql4)
emitted_pids = {i["product_id"] for i in ins4}
check(emitted_pids == {target},
f"(a) only the genuinely-blank product emits (expected {{{target}}}, got {emitted_pids})")
# ---- Scenario 4: canary --limit honored ----
print("\n[5] --limit canary mode")
with tempfile.TemporaryDirectory() as td:
out5, sql5, _ = run_generator(mod, all_blank, limit=1, workdir=td)
ins5 = parse_inserts(sql5)
canary_pids = {i["product_id"] for i in ins5}
check(len(canary_pids) <= 1, f"(canary) --limit 1 touches at most 1 product (got {canary_pids})")
print("\n" + "=" * 60)
print(f"RESULT: {len(PASS)} passed, {len(FAIL)} failed")
if FAIL:
print("BLOCK — failing checks:")
for f in FAIL:
print(" -", f)
sys.exit(1)
print("PASS — testing gate satisfied.")
if __name__ == "__main__":
main()