← back to Designerwallcoverings
TK-11471: weight gate on the DWPW->GRS migration (the actual live leak)
4d31a1ad6ae1ba8f61556a0d72e255ab51223755 · 2026-09-11 11:42:38 -0700 · Steve Abrams
This job — not fentucci-naturals — created the 80 zero-weight ACTIVE Fentucci
variants measured on the live store at 2026-09-11T18:18:58Z. Evidence: all 40
offender GRS SKUs appear in dwpw-grs-migrate.lastrun.json (mode APPLY, 37
ensure=create + 3 ensure=update, every one result=PUBLISHED_NO_TWIN), and
`grep -n "weight\|grams" dwpw-grs-migrate.py` returned nothing — it created and
published ACTIVE with no weight handling at all. It was invisible to the earlier
sweeps because it is Python and sits outside scripts/*-onboard/.
CREATE: every productVariantsBulk{Update,Create} inventoryItem now carries
measurement.weight — Per Yard 3.0 lb, Sample 0.25 lb, both SOURCED FROM THE
GUARD via resolve_weight_lb() rather than hand-typed, so the create payload and
the pre-publish gate cannot disagree. The update path stamps the existing
Per Yard variant too (3 of the 40 offenders came in through it).
PUBLISH: new STEP 2b weight_gate() runs before publish_active(). It re-reads
weights through a query that ACTUALLY returns them, checks the response really
carried the field, heals any zero/unset variant to the approved default,
RE-VERIFIES, and on failure returns LEFT_DRAFT_NEEDS_WEIGHT — product stays
DRAFT, is not published, and the DWPW twin is NOT archived (same hard interlock
as the existing image gate: a half-migrated pair is recoverable, a zero-weight
live product mis-costs freight on every order).
Tests (lib/test_weight_guard.py, 14 offline, no network/DB/cost) drive the REAL
process() apply path with a mocked gql. Mutation-verified red: disabling the
gate -> 3 failures; stripping measurement.weight from the create payloads -> 1
failure. No --apply run, no Shopify writes, no launchctl change.
Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01FJHxAzaEMMxado57mFjiCk
Files touched
M scripts/dwpw-grs-migrate.pyA scripts/lib/test_weight_guard.py
Diff
commit 4d31a1ad6ae1ba8f61556a0d72e255ab51223755
Author: Steve Abrams <steve@designerwallcoverings.com>
Date: Fri Sep 11 11:42:38 2026 -0700
TK-11471: weight gate on the DWPW->GRS migration (the actual live leak)
This job — not fentucci-naturals — created the 80 zero-weight ACTIVE Fentucci
variants measured on the live store at 2026-09-11T18:18:58Z. Evidence: all 40
offender GRS SKUs appear in dwpw-grs-migrate.lastrun.json (mode APPLY, 37
ensure=create + 3 ensure=update, every one result=PUBLISHED_NO_TWIN), and
`grep -n "weight\|grams" dwpw-grs-migrate.py` returned nothing — it created and
published ACTIVE with no weight handling at all. It was invisible to the earlier
sweeps because it is Python and sits outside scripts/*-onboard/.
CREATE: every productVariantsBulk{Update,Create} inventoryItem now carries
measurement.weight — Per Yard 3.0 lb, Sample 0.25 lb, both SOURCED FROM THE
GUARD via resolve_weight_lb() rather than hand-typed, so the create payload and
the pre-publish gate cannot disagree. The update path stamps the existing
Per Yard variant too (3 of the 40 offenders came in through it).
PUBLISH: new STEP 2b weight_gate() runs before publish_active(). It re-reads
weights through a query that ACTUALLY returns them, checks the response really
carried the field, heals any zero/unset variant to the approved default,
RE-VERIFIES, and on failure returns LEFT_DRAFT_NEEDS_WEIGHT — product stays
DRAFT, is not published, and the DWPW twin is NOT archived (same hard interlock
as the existing image gate: a half-migrated pair is recoverable, a zero-weight
live product mis-costs freight on every order).
Tests (lib/test_weight_guard.py, 14 offline, no network/DB/cost) drive the REAL
process() apply path with a mocked gql. Mutation-verified red: disabling the
gate -> 3 failures; stripping measurement.weight from the create payloads -> 1
failure. No --apply run, no Shopify writes, no launchctl change.
Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01FJHxAzaEMMxado57mFjiCk
---
scripts/dwpw-grs-migrate.py | 74 ++++++++++++-
scripts/lib/test_weight_guard.py | 231 +++++++++++++++++++++++++++++++++++++++
2 files changed, 301 insertions(+), 4 deletions(-)
diff --git a/scripts/dwpw-grs-migrate.py b/scripts/dwpw-grs-migrate.py
index 69fb2d1..89aa725 100644
--- a/scripts/dwpw-grs-migrate.py
+++ b/scripts/dwpw-grs-migrate.py
@@ -14,6 +14,12 @@ PER PRODUCT, in this EXACT order (the ordering IS the safety property):
sellable ALWAYS pos1 (productVariantsBulkReorder in one op).
2. VERIFY draft: image HTTP 200, price == 3*cost_yd (+/-0.02), variants present.
Fail -> SKIP product, leave DWPW untouched.
+ 2b. WEIGHT GATE (TK-11414 hard rule, wired TK-11471): every variant must carry a
+ real weight. Create stamps it; before publish the weights are re-read through a
+ query that ACTUALLY returns them, any zero/unset variant is healed to the
+ approved default (Sample 0.25 lb / Wallcovering 3.0 lb) and RE-VERIFIED. Still
+ zero -> LEFT_DRAFT_NEEDS_WEIGHT, no publish, no DWPW archive. Zero weight
+ collapses the order into the lowest weight tier / free band and mis-costs freight.
3. PUBLISH the GRS (status ACTIVE) + add to Online Store + Google & YouTube channel.
4. VERIFY GRS live on the public storefront (/products/<handle> -> 200).
5. HARD INTERLOCK: ONLY if step 4 passed, find the ACTIVE DWPW twin by mfr
@@ -43,6 +49,12 @@ import time
import urllib.error
import urllib.request
+import sys
+sys.path.insert(0, os.path.join(os.path.dirname(os.path.abspath(__file__)), "lib"))
+from weight_guard import ( # noqa: E402 (TK-11471 weight gate)
+ WEIGHT_REQUERY, heal_and_verify_weights, resolve_weight_lb,
+ response_carries_weight_field)
+
# ---------------------------------------------------------------- config
DOMAIN = "designer-laboratory-sandbox.myshopify.com"
API = "2024-10"
@@ -57,6 +69,12 @@ PUB_GOOGLE_YT = "gid://shopify/Publication/29646651457"
VENDOR = "Fentucci"
PRODUCT_TYPE = "Wallcovering"
+# TK-11414 weight defaults, SOURCED FROM THE GUARD (never hand-typed here, so the
+# create payload and the pre-publish gate can never disagree about a compliant weight).
+_WEIGHT_SELLABLE_LB = resolve_weight_lb({"sku": "sellable", "title": "Per Yard"},
+ {"product_type": PRODUCT_TYPE})
+_WEIGHT_SAMPLE_LB = resolve_weight_lb({"sku": "x-Sample", "title": "Sample"},
+ {"product_type": PRODUCT_TYPE})
COLLECTION = "TWIL Naturals"
WIDTH_METAFIELD = '36" Wide (trim to 34")'
DESC_TPL = ("{name} is a natural woven grasscloth wallcovering by Fentucci, with a "
@@ -326,14 +344,18 @@ def create_grs(row, image_ok, res):
q = '''mutation($pid:ID!,$vars:[ProductVariantsBulkInput!]!){
productVariantsBulkUpdate(productId:$pid, variants:$vars){ userErrors{ field message } } }'''
d = gql(q, {"pid": pid, "vars": [{"id": default_variant, "price": row["dw_price"],
- "inventoryItem": {"sku": row["grs"], "tracked": True}}]})
+ "inventoryItem": {"sku": row["grs"], "tracked": True,
+ "measurement": {"weight": {"value": _WEIGHT_SELLABLE_LB,
+ "unit": "POUNDS"}}}}]})
ue = d["data"]["productVariantsBulkUpdate"]["userErrors"]
if ue: raise RuntimeError(f"variant update {row['grs']}: {ue}")
# Sample variant
q = '''mutation($pid:ID!,$vars:[ProductVariantsBulkInput!]!){
productVariantsBulkCreate(productId:$pid, variants:$vars){ userErrors{ field message } } }'''
d = gql(q, {"pid": pid, "vars": [{"optionValues": [{"optionName": "Size", "name": "Sample"}],
- "price": "4.25", "inventoryItem": {"sku": row["grs"] + "-Sample", "tracked": False}}]})
+ "price": "4.25", "inventoryItem": {"sku": row["grs"] + "-Sample", "tracked": False,
+ "measurement": {"weight": {"value": _WEIGHT_SAMPLE_LB,
+ "unit": "POUNDS"}}}}]})
ue = d["data"]["productVariantsBulkCreate"]["userErrors"]
if ue: raise RuntimeError(f"sample create {row['grs']}: {ue}")
_set_metafields(pid, row)
@@ -367,7 +389,9 @@ def update_grs(row, existing, image_ok, res):
q = '''mutation($pid:ID!,$vars:[ProductVariantsBulkInput!]!){
productVariantsBulkUpdate(productId:$pid, variants:$vars){ userErrors{ field message } } }'''
d = gql(q, {"pid": pid, "vars": [{"id": sell["id"], "price": row["dw_price"],
- "inventoryItem": {"sku": row["grs"], "tracked": True}}]})
+ "inventoryItem": {"sku": row["grs"], "tracked": True,
+ "measurement": {"weight": {"value": _WEIGHT_SELLABLE_LB,
+ "unit": "POUNDS"}}}}]})
ue = d["data"]["productVariantsBulkUpdate"]["userErrors"]
if ue: raise RuntimeError(f"variant update {row['grs']}: {ue}")
# ensure Sample variant exists
@@ -375,7 +399,9 @@ def update_grs(row, existing, image_ok, res):
q = '''mutation($pid:ID!,$vars:[ProductVariantsBulkInput!]!){
productVariantsBulkCreate(productId:$pid, variants:$vars){ userErrors{ field message } } }'''
d = gql(q, {"pid": pid, "vars": [{"optionValues": [{"optionName": "Size", "name": "Sample"}],
- "price": "4.25", "inventoryItem": {"sku": row["grs"] + "-Sample", "tracked": False}}]})
+ "price": "4.25", "inventoryItem": {"sku": row["grs"] + "-Sample", "tracked": False,
+ "measurement": {"weight": {"value": _WEIGHT_SAMPLE_LB,
+ "unit": "POUNDS"}}}}]})
ue = d["data"]["productVariantsBulkCreate"]["userErrors"]
if ue: raise RuntimeError(f"sample create {row['grs']}: {ue}")
_set_metafields(pid, row)
@@ -425,6 +451,38 @@ def _reorder_sellable_first(pid):
ue = d["data"]["productVariantsBulkReorder"]["userErrors"]
if ue: raise RuntimeError(f"reorder: {ue}")
+def weight_gate(pid, grs):
+ """TK-11414 hard rule: NO product goes ACTIVE with a missing/zero variant weight.
+
+ SELF-HEAL then VERIFY (the pattern Steve approved in sanderson-onboard
+ create_sdg.mjs 8d09eed): stranding product is worse than assigning the
+ already-approved default, but a heal that silently fails must never publish.
+
+ Returns [] to proceed, or hold reasons. The re-read runs WEIGHT_REQUERY, which
+ explicitly asks for inventoryItem{measurement{weight{value unit}}}, and the
+ RESPONSE IS CHECKED FOR THAT FIELD before any verdict — a response that never
+ carried weight can never be mistaken for a clean reading. Fails CLOSED.
+ Idempotent: one read and nothing else when every weight is already positive.
+ """
+ try:
+ resp = gql(WEIGHT_REQUERY, {"id": pid})
+ except Exception as e: # noqa: BLE001 - fail closed
+ return ["weight>0: re-read failed: %s" % str(e)[:80]]
+ if (resp or {}).get("errors"):
+ return ["weight>0: re-read errored: %s" % str(resp["errors"])[:100]]
+ fresh = ((resp or {}).get("data") or {}).get("product")
+ if not fresh:
+ return ["weight>0: product not readable for weights"]
+ if not response_carries_weight_field(fresh):
+ return ["weight>0: NOT MEASURED — re-read response carried no weight field"]
+ res = heal_and_verify_weights(gql, pid, fresh)
+ if res["ok"]:
+ return []
+ reasons = (["weight>0: still zero after heal: %s" % ", ".join(res["stillZero"])]
+ if res["stillZero"] else ["weight>0: heal could not be confirmed"])
+ return reasons + ["weight>0: " + e for e in res["errs"][:3]]
+
+
def publish_active(pid, grs, prev_status):
# status ACTIVE
q = '''mutation($input:ProductInput!){ productUpdate(input:$input){ product{ id status } userErrors{ field message } } }'''
@@ -569,6 +627,14 @@ def process(row, apply):
# no image attached (or ingest FAILED) -> cannot go ACTIVE; leave DRAFT +
# Needs-Image, do NOT archive DWPW.
res["result"] = "LEFT_DRAFT_NEEDS_IMAGE"; return res
+ # STEP 2b WEIGHT GATE — heal then verify. A product still zero-weight stays DRAFT,
+ # is not published, and the DWPW twin is NOT archived (same interlock as the image
+ # gate above): a half-migrated pair is recoverable, a zero-weight live product is
+ # mis-costed freight on every order.
+ w_reasons = weight_gate(pid, grs)
+ if w_reasons:
+ res["result"] = "LEFT_DRAFT_NEEDS_WEIGHT"; res["weight_block"] = w_reasons
+ return res
# STEP 3 publish
publish_active(pid, grs, prev_status)
# STEP 4 verify storefront
diff --git a/scripts/lib/test_weight_guard.py b/scripts/lib/test_weight_guard.py
new file mode 100644
index 0000000..243b5a8
--- /dev/null
+++ b/scripts/lib/test_weight_guard.py
@@ -0,0 +1,231 @@
+#!/usr/bin/env python3
+"""test_weight_guard.py — TK-11471. OFFLINE: zero network, zero DB, zero cost.
+
+ python3 scripts/lib/test_weight_guard.py
+
+CLAUDE.md TK-11431 amendment 3: a check ships with a NEGATIVE test proving it goes
+red on an injected fault, or it does not ship.
+
+Covers the Python guard (lib/weight_guard.py) AND the real publish path in
+dwpw-grs-migrate.py — the job that actually created 80 zero-weight ACTIVE variants
+on 2026-09-11 (all 40 products appear in dwpw-grs-migrate.lastrun.json with
+result=PUBLISHED_NO_TWIN, 37 ensure=create + 3 ensure=update).
+"""
+import importlib.util
+import os
+import sys
+import unittest
+
+HERE = os.path.dirname(os.path.abspath(__file__))
+SCRIPTS = os.path.dirname(HERE)
+sys.path.insert(0, HERE)
+
+import weight_guard as wg # noqa: E402
+
+MJS = os.path.join(HERE, "weight-guard.mjs")
+
+
+def sellable(**kw):
+ v = {"id": "gid://shopify/ProductVariant/1", "sku": "GRS-27580", "title": "Per Yard",
+ "price": "63.00"}
+ v.update(kw)
+ return v
+
+
+def sample(**kw):
+ v = {"id": "gid://shopify/ProductVariant/2", "sku": "GRS-27580-Sample", "title": "Sample",
+ "price": "4.25"}
+ v.update(kw)
+ return v
+
+
+def node(vid, sku, title, lb, iid=None, drop_field=False):
+ inv = {"id": iid or ("gid://shopify/InventoryItem/%s" % vid)}
+ if not drop_field:
+ inv["measurement"] = None if lb is None else {"weight": {"value": lb, "unit": "POUNDS"}}
+ return {"id": "gid://shopify/ProductVariant/%s" % vid, "sku": sku, "title": title,
+ "price": "4.25" if "Sample" in title else "63.00", "inventoryItem": inv}
+
+
+WALLCOVERING = {"product_type": "Wallcovering"}
+
+
+class TestDetector(unittest.TestCase):
+ def test_injected_zero_weight_sellable_is_flagged(self):
+ p = {"productType": "Wallcovering",
+ "variants": [node(1, "GRS-27580", "Per Yard", 0), node(2, "GRS-27580-Sample", "Sample", 0.25)]}
+ self.assertEqual([v["sku"] for v in wg.all_zero_weight_variants(p)], ["GRS-27580"])
+
+ def test_injected_zero_weight_SAMPLE_is_flagged(self):
+ """dw-active-weight-canary FAILs on a zero-weight SAMPLE too (its live run split
+ the offenders 43 sample / 40 sellable). The .mjs zeroWeightBlockers() filters
+ samples out; that bug must not exist here."""
+ p = {"productType": "Wallcovering",
+ "variants": [node(1, "GRS-27580", "Per Yard", 3.0), node(2, "GRS-27580-Sample", "Sample", 0)]}
+ self.assertEqual([v["sku"] for v in wg.all_zero_weight_variants(p)], ["GRS-27580-Sample"])
+
+ def test_missing_weight_field_is_flagged_fails_safe(self):
+ p = {"productType": "Wallcovering", "variants": [sellable(), sample()]}
+ self.assertEqual(len(wg.all_zero_weight_variants(p)), 2)
+ self.assertEqual(len(wg.unmeasured_variants(p)), 2)
+
+ def test_none_and_junk_weight_are_flagged(self):
+ self.assertTrue(wg.has_zero_weight(sellable(weight=None)))
+ self.assertTrue(wg.has_zero_weight(sellable(weight="junk", weight_unit="lb")))
+ self.assertTrue(wg.has_zero_weight(node(1, "GRS-1", "Per Yard", None)))
+
+ def test_units_convert_and_positive_is_not_flagged(self):
+ self.assertAlmostEqual(wg.current_weight_lb(node(1, "x", "Per Yard", 1361, )), 1361) # POUNDS as given
+ self.assertAlmostEqual(wg.current_weight_lb(sellable(weight=1361, weight_unit="g")), 3.0, places=2)
+ self.assertAlmostEqual(wg.current_weight_lb(sellable(weight=1.36, weight_unit="kg")), 2.998, places=2)
+ self.assertAlmostEqual(wg.current_weight_lb(sellable(weight=48, weight_unit="oz")), 3.0, places=3)
+ self.assertFalse(wg.has_zero_weight(sellable(weight=1361, weight_unit="g")))
+ v = {"id": "v", "sku": "x", "title": "Per Yard",
+ "inventoryItem": {"id": "i", "measurement": {"weight": {"value": 1361, "unit": "GRAMS"}}}}
+ self.assertAlmostEqual(wg.current_weight_lb(v), 3.0, places=2)
+ self.assertFalse(wg.has_zero_weight(v))
+
+ def test_all_weighted_returns_empty(self):
+ p = {"productType": "Wallcovering",
+ "variants": [node(1, "GRS-27580", "Per Yard", 3.0), node(2, "GRS-27580-Sample", "Sample", 0.25)]}
+ self.assertEqual(wg.all_zero_weight_variants(p), [])
+
+ def test_resolve_preserves_positive_and_fills_default(self):
+ self.assertAlmostEqual(wg.resolve_weight_lb(sellable(weight=7.5, weight_unit="lb"), WALLCOVERING), 7.5)
+ self.assertEqual(wg.resolve_weight_lb(sellable(), WALLCOVERING), 3.0)
+ self.assertEqual(wg.resolve_weight_lb(sample(), WALLCOVERING), 0.25)
+ self.assertEqual(wg.resolve_weight_lb(sellable(), {"product_type": "Mural"}), 4.0)
+ self.assertEqual(wg.resolve_weight_lb(sellable(), {"product_type": "Nonesuch"}), 2.0)
+
+ @unittest.skipUnless(os.path.exists(MJS), "weight-guard.mjs missing")
+ def test_constants_match_the_mjs_twin(self):
+ with open(MJS) as fh:
+ src = fh.read()
+ self.assertIn("SAMPLE_WEIGHT_LB = 0.25", src)
+ self.assertIn("FALLBACK_LB = 2.0", src)
+ table = src.split("TYPE_DEFAULT_LB = {", 1)[1].split("};", 1)[0]
+ pairs = {}
+ for chunk in table.replace("\n", " ").split(","):
+ if ":" not in chunk:
+ continue
+ k, v = chunk.split(":", 1)
+ try:
+ pairs[k.strip().strip("'\"")] = float(v.strip().rstrip(","))
+ except ValueError:
+ pass
+ self.assertEqual(pairs, wg.TYPE_DEFAULT_LB, "TYPE_DEFAULT_LB drifted from weight-guard.mjs")
+
+
+def load_migrate():
+ os.environ.setdefault("SHOPIFY_FULL_ACCESS_TOKEN", "test-token-not-real")
+ spec = importlib.util.spec_from_file_location("dwpw_grs_migrate",
+ os.path.join(SCRIPTS, "dwpw-grs-migrate.py"))
+ m = importlib.util.module_from_spec(spec)
+ spec.loader.exec_module(m)
+ return m
+
+
+class TestMigrateWeightsAreStamped(unittest.TestCase):
+ def test_create_payload_weights_come_from_the_guard(self):
+ m = load_migrate()
+ self.assertEqual(m._WEIGHT_SELLABLE_LB, 3.0) # Wallcovering default
+ self.assertEqual(m._WEIGHT_SAMPLE_LB, 0.25) # sample default
+ self.assertGreater(m._WEIGHT_SELLABLE_LB, 0)
+ self.assertGreater(m._WEIGHT_SAMPLE_LB, 0)
+
+ def test_every_variant_mutation_carries_a_weight(self):
+ """INJECTED-FAULT SHAPE CHECK on the real shipped source: no
+ productVariantsBulk{Create,Update} may send an inventoryItem without a
+ measurement.weight — that is exactly how 80 variants shipped at zero."""
+ with open(os.path.join(SCRIPTS, "dwpw-grs-migrate.py")) as fh:
+ src = fh.read()
+ chunks = [c for c in src.split('"inventoryItem": {')[1:]]
+ self.assertGreaterEqual(len(chunks), 4, "expected >=4 variant inventoryItem payloads")
+ for c in chunks:
+ head = c[:260]
+ self.assertIn("measurement", head,
+ "an inventoryItem variant payload ships with no weight:\n" + head)
+
+
+class TestMigratePublishGate(unittest.TestCase):
+ """Mocked harness over the REAL process() apply path. No network, no Shopify."""
+
+ ROW = {"grs": "GRS-27580", "mfr": "T-1", "title": "Oliva Khaki Grasscloth Wallcovering",
+ "image": "https://example.invalid/x.jpg", "dw_price": "63.00", "cost_yd": "21.00"}
+
+ def _harness(self, heal_ok):
+ m = load_migrate()
+ state = {"weights": {"gid://shopify/InventoryItem/1": 0.0,
+ "gid://shopify/InventoryItem/2": 0.0},
+ "published": False, "archived": False, "mutations": []}
+
+ def fake_gql(query, variables=None):
+ variables = variables or {}
+ if "inventoryItemUpdate" in query:
+ state["mutations"].append((variables["id"], variables["w"]))
+ if not heal_ok:
+ return {"data": {"inventoryItemUpdate":
+ {"userErrors": [{"message": "injected heal failure"}]}}}
+ state["weights"][variables["id"]] = variables["w"]
+ return {"data": {"inventoryItemUpdate": {"userErrors": []}}}
+ # WEIGHT_REQUERY — the response DOES carry the weight field
+ edges = []
+ for iid, lb in state["weights"].items():
+ is_sample = iid.endswith("/2")
+ edges.append({"node": node(iid.rsplit("/", 1)[-1],
+ "GRS-27580-Sample" if is_sample else "GRS-27580",
+ "Sample" if is_sample else "Per Yard",
+ None if lb <= 0 else lb, iid=iid)})
+ return {"data": {"product": {"productType": "Wallcovering",
+ "variants": {"edges": edges}}}}
+
+ def fake_publish(pid, grs, prev_status):
+ state["published"] = True
+
+ m.gql = fake_gql
+ m.find_grs = lambda grs: None
+ m.http_status = lambda url, method="GET": 200
+ m.find_active_dwpw_twin = lambda mfr: []
+ m.create_grs = lambda row, image_ok, res: ("gid://shopify/Product/9", "oliva-khaki")
+ m.verify_read = lambda pid, grs, **kw: {"variants": [sellable(), sample()]}
+ m.verify_image = lambda pid, url, **kw: (True, "READY", ["READY"])
+ m.publish_active = fake_publish
+ m.archive_dwpw = lambda twin, grs: state.__setitem__("archived", True)
+ return m, state
+
+ def test_zero_weight_product_is_HEALED_then_PUBLISHED(self):
+ m, state = self._harness(heal_ok=True)
+ res = m.process(dict(self.ROW), apply=True)
+ self.assertEqual(res["result"], "PUBLISHED_NO_TWIN", res)
+ self.assertTrue(state["published"])
+ self.assertEqual(sorted(w for _, w in state["mutations"]), [0.25, 3.0])
+
+ def test_product_whose_heal_FAILS_is_NOT_PUBLISHED(self):
+ m, state = self._harness(heal_ok=False)
+ res = m.process(dict(self.ROW), apply=True)
+ self.assertEqual(res["result"], "LEFT_DRAFT_NEEDS_WEIGHT", res)
+ self.assertFalse(state["published"], "a zero-weight product must NEVER be published ACTIVE")
+ self.assertFalse(state["archived"], "the DWPW twin must not be archived on a held GRS")
+ self.assertTrue(any("weight>0" in r for r in res["weight_block"]), res.get("weight_block"))
+
+ def test_gate_fails_closed_when_the_requery_carries_no_weight_field(self):
+ """MEASURE WHAT YOU CLAIM — a response that never contained weight must not
+ be read as a clean pass."""
+ m, state = self._harness(heal_ok=True)
+ m.gql = lambda q, v=None: {"data": {"product": {"productType": "Wallcovering", "variants": {
+ "edges": [{"node": node(1, "GRS-27580", "Per Yard", 0, drop_field=True)}]}}}}
+ res = m.process(dict(self.ROW), apply=True)
+ self.assertEqual(res["result"], "LEFT_DRAFT_NEEDS_WEIGHT", res)
+ self.assertFalse(state["published"])
+
+ def test_gate_is_a_noop_when_weights_are_already_positive(self):
+ m, state = self._harness(heal_ok=True)
+ state["weights"] = {"gid://shopify/InventoryItem/1": 3.0,
+ "gid://shopify/InventoryItem/2": 0.25}
+ res = m.process(dict(self.ROW), apply=True)
+ self.assertEqual(res["result"], "PUBLISHED_NO_TWIN", res)
+ self.assertEqual(state["mutations"], [], "idempotent: no heal writes when already weighted")
+
+
+if __name__ == "__main__":
+ unittest.main(verbosity=2)
← e5b2795 TK-11471: Python twin of the weight guard (lib/weight_guard.
·
back to Designerwallcoverings
·
TK-11471: weight gate on the JD rolling publish (the 4th ung ea8f871 →