← back to Fentucci Naturals
TK-11471: vendor a Python weight guard + its negative tests
81a477ad9f8b263ab8d0b6370db2f5a945b1c95c · 2026-09-11 11:34:51 -0700 · Steve Abrams
Python port of designerwallcoverings/scripts/lib/weight-guard.mjs, constants
byte-identical (SAMPLE_WEIGHT_LB 0.25, FALLBACK_LB 2.0, the 17-entry
TYPE_DEFAULT_LB table). This pipeline is Python + REST and lives outside
scripts/*-onboard/, so every .js/.mjs-only or GraphQL-only sweep of this
lineage structurally misses it — third recorded occurrence of that omission.
Deliberate divergence from the .mjs: all_zero_weight_variants() counts SAMPLE
variants. The .mjs zeroWeightBlockers() filters them out, which made the gate
narrower than the invariant dw-active-weight-canary enforces (its live run
split the offenders 43 sample / 40 sellable). That bug is not ported.
heal_and_verify_weights() fails CLOSED: an unreadable re-query, a re-query
response that never carried the weight field, a missing inventoryItem id, or a
userError all return ok=False. response_carries_weight_field() is the guard
against the false-green class — a gate reading a response that never contained
weight proves nothing.
test_weight_guard.py: 30 offline tests (no network, no DB, no cost), every one
an injected fault or an asserted clean state. Verified by mutation: removing
the go-live gate -> 3 failures; porting the .mjs sample-filter bug -> 7
failures; neutering resolve_weight_lb -> 4 failures.
Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01FJHxAzaEMMxado57mFjiCk
Files touched
A scripts/lib/test_weight_guard.pyA scripts/lib/weight_guard.py
Diff
commit 81a477ad9f8b263ab8d0b6370db2f5a945b1c95c
Author: Steve Abrams <steve@designerwallcoverings.com>
Date: Fri Sep 11 11:34:51 2026 -0700
TK-11471: vendor a Python weight guard + its negative tests
Python port of designerwallcoverings/scripts/lib/weight-guard.mjs, constants
byte-identical (SAMPLE_WEIGHT_LB 0.25, FALLBACK_LB 2.0, the 17-entry
TYPE_DEFAULT_LB table). This pipeline is Python + REST and lives outside
scripts/*-onboard/, so every .js/.mjs-only or GraphQL-only sweep of this
lineage structurally misses it — third recorded occurrence of that omission.
Deliberate divergence from the .mjs: all_zero_weight_variants() counts SAMPLE
variants. The .mjs zeroWeightBlockers() filters them out, which made the gate
narrower than the invariant dw-active-weight-canary enforces (its live run
split the offenders 43 sample / 40 sellable). That bug is not ported.
heal_and_verify_weights() fails CLOSED: an unreadable re-query, a re-query
response that never carried the weight field, a missing inventoryItem id, or a
userError all return ok=False. response_carries_weight_field() is the guard
against the false-green class — a gate reading a response that never contained
weight proves nothing.
test_weight_guard.py: 30 offline tests (no network, no DB, no cost), every one
an injected fault or an asserted clean state. Verified by mutation: removing
the go-live gate -> 3 failures; porting the .mjs sample-filter bug -> 7
failures; neutering resolve_weight_lb -> 4 failures.
Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01FJHxAzaEMMxado57mFjiCk
---
scripts/lib/test_weight_guard.py | 365 +++++++++++++++++++++++++++++++++++++++
scripts/lib/weight_guard.py | 297 +++++++++++++++++++++++++++++++
2 files changed, 662 insertions(+)
diff --git a/scripts/lib/test_weight_guard.py b/scripts/lib/test_weight_guard.py
new file mode 100644
index 0000000..fb411d6
--- /dev/null
+++ b/scripts/lib/test_weight_guard.py
@@ -0,0 +1,365 @@
+#!/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. A positive-only test on a detector
+proves nothing. Every case below injects a fault and asserts the gate FAILS, or
+injects a clean state and asserts it does not.
+
+Also proves the constants have not drifted from the .mjs original, and that the
+go-live gate itself (a) heals then activates, and (b) does NOT activate when the
+heal fails.
+"""
+import importlib.util
+import json
+import os
+import sys
+import unittest
+
+HERE = os.path.dirname(os.path.abspath(__file__))
+SCRIPTS = os.path.dirname(HERE)
+ROOT = os.path.dirname(SCRIPTS)
+sys.path.insert(0, HERE)
+
+import weight_guard as wg # noqa: E402
+
+MJS = os.path.join(os.path.expanduser("~"),
+ "Projects/designerwallcoverings/scripts/lib/weight-guard.mjs")
+
+
+def sellable(**kw):
+ v = {"id": 1, "sku": "DWFN-100001", "title": "Per Yard", "price": "0.00"}
+ v.update(kw)
+ return v
+
+
+def sample(**kw):
+ v = {"id": 2, "sku": "DWFN-100001-Sample", "title": "Sample", "price": "4.25"}
+ v.update(kw)
+ return v
+
+
+def gql_variant(weight_value, unit="POUNDS", **kw):
+ """A GraphQL-shaped node: weight lives on inventoryItem.measurement."""
+ v = {"id": "gid://shopify/ProductVariant/1", "sku": "DWFN-100001",
+ "title": "Per Yard", "price": "0.00",
+ "inventoryItem": {"id": "gid://shopify/InventoryItem/1",
+ "measurement": None if weight_value is None
+ else {"weight": {"value": weight_value, "unit": unit}}}}
+ v.update(kw)
+ return v
+
+
+WALLCOVERING = {"product_type": "Wallcovering"}
+
+
+class TestDetector(unittest.TestCase):
+ # ---- INJECTED FAULT: zero weight must be FLAGGED -----------------------
+ def test_injected_zero_weight_sellable_is_flagged(self):
+ p = {"product_type": "Wallcovering",
+ "variants": [sellable(weight=0, weight_unit="lb"), sample(weight=0.25, weight_unit="lb")]}
+ bad = wg.all_zero_weight_variants(p)
+ self.assertEqual([v["sku"] for v in bad], ["DWFN-100001"])
+
+ def test_injected_zero_weight_SAMPLE_is_flagged(self):
+ """The .mjs zeroWeightBlockers() filters samples out; the canary does not.
+ A zero-weight sample MUST be flagged here or the gate is narrower than the
+ invariant it enforces."""
+ p = {"product_type": "Wallcovering",
+ "variants": [sellable(weight=3, weight_unit="lb"), sample(weight=0, weight_unit="lb")]}
+ bad = wg.all_zero_weight_variants(p)
+ self.assertEqual([v["sku"] for v in bad], ["DWFN-100001-Sample"])
+
+ def test_missing_weight_field_is_flagged_fails_safe(self):
+ """No weight key at all (the live build-drafts.py payload) => flagged."""
+ p = {"product_type": "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_weight_is_flagged(self):
+ self.assertTrue(wg.has_zero_weight(sellable(weight=None)))
+ self.assertTrue(wg.has_zero_weight(sellable(weight="", weight_unit="lb")))
+ self.assertTrue(wg.has_zero_weight(sellable(weight="junk", weight_unit="lb")))
+
+ def test_negative_weight_is_flagged(self):
+ self.assertTrue(wg.has_zero_weight(sellable(weight=-1, weight_unit="lb")))
+
+ # ---- UNIT CONVERSION ---------------------------------------------------
+ def test_grams_convert_and_positive_is_not_flagged(self):
+ v = sellable(weight=1361, weight_unit="g") # 1361 g = 3.0 lb
+ self.assertAlmostEqual(wg.current_weight_lb(v), 3.0, places=2)
+ self.assertFalse(wg.has_zero_weight(v))
+
+ def test_kilograms_convert_and_positive_is_not_flagged(self):
+ v = sellable(weight=1.36, weight_unit="kg") # 1.36 kg ~ 3.0 lb
+ self.assertAlmostEqual(wg.current_weight_lb(v), 2.998, places=2)
+ self.assertFalse(wg.has_zero_weight(v))
+
+ def test_ounces_and_grams_field_convert(self):
+ self.assertAlmostEqual(wg.current_weight_lb(sellable(weight=48, weight_unit="oz")), 3.0, places=3)
+ self.assertAlmostEqual(wg.current_weight_lb({"grams": 453.59237}), 1.0, places=6)
+
+ def test_tiny_gram_weight_still_positive_not_flagged(self):
+ """1 g is a silly weight but it is MEASURED and positive — not our defect."""
+ self.assertFalse(wg.has_zero_weight(sellable(weight=1, weight_unit="g")))
+
+ def test_graphql_shape_units(self):
+ self.assertAlmostEqual(wg.current_weight_lb(gql_variant(3.0, "POUNDS")), 3.0, places=6)
+ self.assertAlmostEqual(wg.current_weight_lb(gql_variant(1361, "GRAMS")), 3.0, places=2)
+ self.assertAlmostEqual(wg.current_weight_lb(gql_variant(1.36, "KILOGRAMS")), 2.998, places=2)
+ self.assertTrue(wg.has_zero_weight(gql_variant(0, "POUNDS")))
+ self.assertTrue(wg.has_zero_weight(gql_variant(None)))
+
+ def test_graphql_connection_shape_is_understood(self):
+ p = {"productType": "Wallcovering",
+ "variants": {"edges": [{"node": gql_variant(0)}, {"node": gql_variant(3.0)}]}}
+ self.assertEqual(len(wg.all_zero_weight_variants(p)), 1)
+
+ # ---- CLEAN STATE: must NOT be flagged ---------------------------------
+ def test_all_weighted_returns_empty(self):
+ p = {"product_type": "Wallcovering",
+ "variants": [sellable(weight=3, weight_unit="lb"), sample(weight=0.25, weight_unit="lb")]}
+ self.assertEqual(wg.all_zero_weight_variants(p), [])
+
+ # ---- SAMPLE CLASSIFICATION --------------------------------------------
+ def test_is_sample_variant(self):
+ self.assertTrue(wg.is_sample_variant(sample()))
+ self.assertTrue(wg.is_sample_variant({"sku": "X-1", "price": "4.25"}))
+ self.assertTrue(wg.is_sample_variant({"title": "Memo", "sku": "X-1", "price": "0"}))
+ self.assertFalse(wg.is_sample_variant(sellable()))
+
+ # ---- RESOLVE ----------------------------------------------------------
+ def test_resolve_preserves_existing_positive_weight(self):
+ self.assertAlmostEqual(
+ wg.resolve_weight_lb(sellable(weight=7.5, weight_unit="lb"), WALLCOVERING), 7.5)
+
+ def test_resolve_fills_per_type_default(self):
+ self.assertEqual(wg.resolve_weight_lb(sellable(), WALLCOVERING), 3.0) # Wallcovering
+ self.assertEqual(wg.resolve_weight_lb(sellable(), {"product_type": "Mural"}), 4.0)
+ self.assertEqual(wg.resolve_weight_lb(sellable(), {"productType": "Fabric"}), 1.0)
+
+ def test_resolve_fills_sample_weight_regardless_of_type(self):
+ self.assertEqual(wg.resolve_weight_lb(sample(), WALLCOVERING), 0.25)
+ self.assertEqual(wg.resolve_weight_lb(sample(), {"product_type": "Furniture"}), 0.25)
+
+ def test_resolve_falls_back_for_unknown_type(self):
+ self.assertEqual(wg.resolve_weight_lb(sellable(), {"product_type": "Nonesuch"}), 2.0)
+ self.assertEqual(wg.resolve_weight_lb(sellable(), {}), 2.0)
+
+ def test_resolve_never_returns_zero(self):
+ for v in (sellable(), sample(), sellable(weight=0, weight_unit="lb"),
+ sellable(weight="junk"), gql_variant(0)):
+ self.assertGreater(wg.resolve_weight_lb(v, WALLCOVERING), 0)
+
+ # ---- CONSTANTS MUST NOT DRIFT FROM THE .mjs ---------------------------
+ @unittest.skipUnless(os.path.exists(MJS), "weight-guard.mjs not present on this machine")
+ def test_constants_match_the_mjs_original(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)
+ k = k.strip().strip("'\"")
+ try:
+ pairs[k] = float(v.strip().rstrip(","))
+ except ValueError:
+ pass
+ self.assertEqual(pairs, wg.TYPE_DEFAULT_LB,
+ "TYPE_DEFAULT_LB drifted from weight-guard.mjs")
+
+
+class FakeGql(object):
+ """Minimal in-memory Shopify: an inventoryItem id -> weight-in-lb store."""
+
+ def __init__(self, weights, heal_ok=True, drop_weight_field=False, requery_fails=False):
+ self.weights = dict(weights) # {inventoryItemId: lb}
+ self.heal_ok = heal_ok
+ self.drop_weight_field = drop_weight_field
+ self.requery_fails = requery_fails
+ self.mutations = []
+
+ def __call__(self, query, variables):
+ if "inventoryItemUpdate" in query:
+ self.mutations.append((variables["id"], variables["w"]))
+ if not self.heal_ok:
+ return {"data": {"inventoryItemUpdate": {"userErrors": [{"message": "denied"}]}}}
+ self.weights[variables["id"]] = variables["w"]
+ return {"data": {"inventoryItemUpdate": {"userErrors": []}}}
+ if self.requery_fails:
+ return {"errors": [{"message": "Throttled"}]}
+ nodes = []
+ for iid, lb in self.weights.items():
+ inv = {"id": iid}
+ if not self.drop_weight_field:
+ inv["measurement"] = {"weight": {"value": lb, "unit": "POUNDS"}}
+ nodes.append({"node": {"id": iid, "sku": "DWFN-100001-Sample", "title": "Sample",
+ "price": "4.25", "inventoryItem": inv}})
+ return {"data": {"product": {"productType": "Wallcovering",
+ "variants": {"edges": nodes}}}}
+
+
+class TestHealAndVerify(unittest.TestCase):
+ GID = "gid://shopify/Product/1"
+
+ def test_noop_when_all_weights_positive(self):
+ g = FakeGql({"ii1": 3.0})
+ p = {"productType": "Wallcovering", "variants": [gql_variant(3.0)]}
+ r = wg.heal_and_verify_weights(g, self.GID, p)
+ self.assertTrue(r["ok"])
+ self.assertEqual(g.mutations, [], "no-op must make ZERO network calls")
+
+ def test_heals_zero_then_passes(self):
+ g = FakeGql({"gid://shopify/InventoryItem/1": 0})
+ p = {"productType": "Wallcovering", "variants": [gql_variant(0)]}
+ r = wg.heal_and_verify_weights(g, self.GID, p)
+ self.assertTrue(r["ok"], r["errs"])
+ self.assertEqual(g.mutations, [("gid://shopify/InventoryItem/1", 3.0)])
+
+ def test_heals_sample_at_quarter_pound(self):
+ g = FakeGql({"gid://shopify/InventoryItem/2": 0})
+ p = {"productType": "Wallcovering",
+ "variants": [gql_variant(0, sku="DWFN-100001-Sample", title="Sample", price="4.25",
+ inventoryItem={"id": "gid://shopify/InventoryItem/2",
+ "measurement": None})]}
+ r = wg.heal_and_verify_weights(g, self.GID, p)
+ self.assertTrue(r["ok"], r["errs"])
+ self.assertEqual(g.mutations, [("gid://shopify/InventoryItem/2", 0.25)])
+
+ # ---- INJECTED FAULTS: the gate must go RED ----------------------------
+ def test_FAILS_when_heal_errors(self):
+ g = FakeGql({"gid://shopify/InventoryItem/1": 0}, heal_ok=False)
+ p = {"productType": "Wallcovering", "variants": [gql_variant(0)]}
+ r = wg.heal_and_verify_weights(g, self.GID, p)
+ self.assertFalse(r["ok"])
+ self.assertTrue(r["stillZero"])
+
+ def test_FAILS_when_variant_has_no_inventory_item_id(self):
+ g = FakeGql({})
+ p = {"productType": "Wallcovering",
+ "variants": [{"sku": "DWFN-1", "title": "Per Yard", "weight": 0, "weight_unit": "lb"}]}
+ r = wg.heal_and_verify_weights(g, self.GID, p)
+ self.assertFalse(r["ok"])
+ self.assertIn("weight:no-inventory-item:DWFN-1", r["errs"])
+
+ def test_FAILS_when_requery_errors(self):
+ g = FakeGql({"gid://shopify/InventoryItem/1": 0}, requery_fails=True)
+ p = {"productType": "Wallcovering", "variants": [gql_variant(0)]}
+ r = wg.heal_and_verify_weights(g, self.GID, p)
+ self.assertFalse(r["ok"])
+
+ def test_FAILS_when_reverify_response_never_carried_the_weight_field(self):
+ """MEASURE WHAT YOU CLAIM. If the re-read response has no weight field, the
+ gate must not pass on a clean-looking read."""
+ g = FakeGql({"gid://shopify/InventoryItem/1": 0}, drop_weight_field=True)
+ p = {"productType": "Wallcovering", "variants": [gql_variant(0)]}
+ r = wg.heal_and_verify_weights(g, self.GID, p)
+ self.assertFalse(r["ok"])
+ self.assertIn("weight:reverify-response-carried-no-weight-field", r["errs"])
+
+ def test_rest_fallback_heals_when_graphql_denied(self):
+ g = FakeGql({"gid://shopify/InventoryItem/1": 0}, heal_ok=False)
+
+ def rest_heal(variant, lb):
+ g.weights["gid://shopify/InventoryItem/1"] = lb
+ return True
+ product = {"productType": "Wallcovering", "variants": [gql_variant(0)]}
+ r = wg.heal_and_verify_weights(g, self.GID, product, rest_heal=rest_heal)
+ self.assertTrue(r["ok"], r["errs"])
+
+
+def _load_golive():
+ os.environ.setdefault("SHOPIFY_ADMIN_TOKEN", "test-token-not-real")
+ spec = importlib.util.spec_from_file_location("golive", os.path.join(SCRIPTS, "go-live.py"))
+ mod = importlib.util.module_from_spec(spec)
+ spec.loader.exec_module(mod)
+ return mod
+
+
+class TestGoLiveGate(unittest.TestCase):
+ """Mocked harness over the REAL go_live_one(). No network, no DB, no Shopify."""
+
+ IID = "gid://shopify/InventoryItem/2"
+
+ def _harness(self, heal_ok):
+ gl = _load_golive()
+ state = {"weight_lb": 0.0, "activated": False, "mutations": [], "deleted": []}
+
+ product = {
+ "id": 99, "status": "draft", "title": "Tellaro Flax | Fentucci Naturals",
+ "body_html": "<p>desc</p>", "tags": "Fentucci Naturals, Grasscloth",
+ "images": [{"id": 1}],
+ "variants": [
+ {"id": 11, "sku": "DWFN-100001-Sample", "title": "Sample", "price": "4.25",
+ "inventory_item_id": 2},
+ {"id": 12, "sku": "DWFN-100001", "title": "Per Yard", "price": "0.00",
+ "inventory_item_id": 3},
+ ],
+ }
+
+ def fake_rest(path, method="GET", body=None):
+ if path == "products/99.json" and method == "GET":
+ return {"product": product}
+ if path == "products/99.json" and method == "PUT":
+ pr = (body or {}).get("product", {})
+ if pr.get("status") == "active":
+ state["activated"] = True
+ return {"product": {"status": "active"}}
+ return {"product": {"status": "draft"}}
+ if method == "DELETE":
+ state["deleted"].append(path)
+ return {}
+ return {}
+
+ def fake_graphql(query, variables):
+ if "inventoryItemUpdate" in query:
+ state["mutations"].append((variables["id"], variables["w"]))
+ if not heal_ok:
+ return {"data": {"inventoryItemUpdate":
+ {"userErrors": [{"message": "no write_inventory"}]}}}
+ state["weight_lb"] = variables["w"]
+ return {"data": {"inventoryItemUpdate": {"userErrors": []}}}
+ if "publishablePublish" in query:
+ return {"data": {"publishablePublish": {"userErrors": []}}}
+ # WEIGHT_REQUERY — note the response DOES carry the weight field
+ return {"data": {"product": {"productType": "Wallcovering", "variants": {"edges": [
+ {"node": {"id": "gid://shopify/ProductVariant/11", "sku": "DWFN-100001-Sample",
+ "title": "Sample", "price": "4.25",
+ "inventoryItem": {"id": self.IID, "measurement": (
+ None if state["weight_lb"] <= 0 else
+ {"weight": {"value": state["weight_lb"], "unit": "POUNDS"}})}}}]}}}}
+
+ gl.rest = fake_rest
+ gl.graphql = fake_graphql
+ gl.pg_activate = lambda sku: None
+ return gl, state
+
+ def test_zero_weight_product_is_HEALED_then_ACTIVATED(self):
+ gl, state = self._harness(heal_ok=True)
+ status, errs = gl.go_live_one(99, "DWFN-100001")
+ self.assertEqual(status, "active", errs)
+ self.assertTrue(state["activated"])
+ self.assertEqual(state["mutations"], [(self.IID, 0.25)])
+
+ def test_product_whose_heal_FAILS_is_NOT_activated(self):
+ gl, state = self._harness(heal_ok=False)
+ status, errs = gl.go_live_one(99, "DWFN-100001")
+ self.assertEqual(status, "held", (status, errs))
+ self.assertFalse(state["activated"], "a zero-weight product must NEVER go ACTIVE")
+ self.assertTrue(any("weight>0" in str(e) for e in errs), errs)
+
+ def test_the_held_reason_is_json_serialisable_for_the_held_file(self):
+ gl, _ = self._harness(heal_ok=False)
+ status, errs = gl.go_live_one(99, "DWFN-100001")
+ json.dumps({"sku": "DWFN-100001", "reason": errs}) # must not raise
+ self.assertEqual(status, "held")
+
+
+if __name__ == "__main__":
+ unittest.main(verbosity=2)
diff --git a/scripts/lib/weight_guard.py b/scripts/lib/weight_guard.py
new file mode 100644
index 0000000..721157c
--- /dev/null
+++ b/scripts/lib/weight_guard.py
@@ -0,0 +1,297 @@
+#!/usr/bin/env python3
+"""weight_guard.py — TK-11471 PYTHON port of ~/Projects/designerwallcoverings/scripts/lib/weight-guard.mjs
+
+Steve's rule (TK-11414): NO product may go ACTIVE with a missing/zero product WEIGHT
+(zero weight collapses orders into the lowest weight tier / free-shipping band and
+mis-costs DW freight).
+
+Why a Python twin exists at all: this pipeline is PYTHON + REST and lives OUTSIDE
+scripts/*-onboard/, so every .js/.mjs-only or GraphQL-only sweep of this lineage
+structurally misses it. Constants are kept BYTE-IDENTICAL to the .mjs so the two
+implementations cannot drift into disagreeing about what a compliant weight is.
+
+Two call sites:
+ 1. create payload: weight = resolve_weight_lb(variant, product) (unit POUNDS)
+ 2. before activate: r = heal_and_verify_weights(gql, gid, product)
+ if not r["ok"]: HOLD as draft (never flip ACTIVE at zero weight)
+
+SAMPLES COUNT. dw-active-weight-canary FAILs on ANY zero-weight ACTIVE variant —
+its last live run split the offenders 43 sample / 40 sellable. The .mjs's
+zeroWeightBlockers() filters samples OUT and is DEPRECATED there; that bug is
+deliberately NOT ported. all_zero_weight_variants() is the only population view here.
+"""
+import math
+
+SAMPLE_WEIGHT_LB = 0.25
+FALLBACK_LB = 2.0
+
+# product_type -> sellable default weight (POUNDS). Identical to the .mjs table.
+TYPE_DEFAULT_LB = {
+ 'Wallcovering': 3.0, 'Wallcoverings': 3.0, 'Wallpaper': 3.0,
+ 'Metallic Wallcovering': 3.0, 'Commercial Wallcovering': 3.0,
+ 'Mural': 4.0,
+ 'Fabric': 1.0, 'Commercial Fabric': 1.0, 'Commercial Drapery': 1.0,
+ 'Trim': 0.5, 'Acoustic Panel': 6.0, 'Pillow': 1.5,
+ 'Upholstered Walls/Panels': 6.0, 'Tin Ceiling Tile': 2.0,
+ 'Hardware': 1.0, 'Furniture': 15.0, 'Memo Sample': 0.25,
+}
+
+NAN = float('nan')
+
+
+def _norm(t):
+ return ('' if t is None else str(t)).strip().lower()
+
+
+def _num(v):
+ """float(v), or NaN when unparseable — mirrors JS Number() -> NaN. NaN is
+ treated as zero-weight downstream, so a junk value FAILS SAFE."""
+ try:
+ f = float(v)
+ except (TypeError, ValueError):
+ return NAN
+ return f
+
+
+def _d(x):
+ return x if isinstance(x, dict) else {}
+
+
+def _gql_weight(variant):
+ """inventoryItem.measurement.weight from a GraphQL node (camel or snake key)."""
+ iv = _d(variant).get('inventoryItem')
+ if iv is None:
+ iv = _d(variant).get('inventory_item')
+ return _d(_d(iv).get('measurement')).get('weight')
+
+
+def is_sample_variant(variant=None):
+ """Sample variant? (importers create `Sample` @ $4.25 + the real unit)."""
+ v = _d(variant)
+ label = _norm(v.get('title') if v.get('title') is not None else v.get('option1'))
+ sku = _norm(v.get('sku'))
+ if 'sample' in label or 'memo' in label:
+ return True
+ if sku.endswith('-sample') or 'sample' in sku:
+ return True
+ p = _num(v.get('price'))
+ return math.isfinite(p) and abs(p - 4.25) < 0.01
+
+
+def weight_field_present(variant=None):
+ """Did the payload we were handed CARRY a weight field at all?
+
+ TK-11431 amendment 1 — an UNMEASURED input is never a PASS. A response that
+ simply never contained weight must not be read as 'weight is zero' by luck,
+ nor as 'fine'. has_zero_weight() flags both, but a caller that needs to
+ distinguish 'measured 0' from 'never measured' asks here.
+ """
+ v = _d(variant)
+ if _d(_gql_weight(v)).get('value') is not None:
+ return True
+ return v.get('grams') is not None or v.get('weight') is not None
+
+
+def current_weight_lb(variant=None):
+ """Current weight in POUNDS, or 0 if missing. Accepts variant.weight (+weight_unit,
+ REST), variant.grams, or inventoryItem.measurement.weight{value,unit} (GraphQL)."""
+ v = _d(variant)
+ g = _d(_gql_weight(v))
+ if g.get('value') is not None:
+ val = _num(g.get('value'))
+ u = _norm(g.get('unit')) # Shopify WeightUnit enum
+ if u == 'kilograms' or u.startswith('kg'):
+ return val * 2.20462
+ if u == 'grams' or u == 'g':
+ return val / 453.59237
+ if u == 'ounces' or u == 'oz':
+ return val / 16
+ return val # POUNDS (unit-less => assume lb)
+ if v.get('grams') is not None:
+ return _num(v.get('grams')) / 453.59237
+ if v.get('weight') is not None:
+ val = _num(v.get('weight'))
+ u = _norm(v.get('weight_unit') or 'lb')
+ if u.startswith('kg'):
+ return val * 2.20462
+ if u == 'g' or u.startswith('gram'):
+ return val / 453.59237
+ if u == 'oz':
+ return val / 16
+ return val # lb
+ return 0.0
+
+
+def has_zero_weight(variant=None):
+ w = current_weight_lb(variant)
+ return (not math.isfinite(w)) or w <= 0
+
+
+def default_weight_lb(variant=None, product=None):
+ """The default weight (lb) to assign a variant that has none."""
+ if is_sample_variant(variant):
+ return SAMPLE_WEIGHT_LB
+ p = _d(product)
+ ptype = p.get('productType') if p.get('productType') is not None else p.get('product_type')
+ return TYPE_DEFAULT_LB.get(ptype, FALLBACK_LB)
+
+
+def resolve_weight_lb(variant=None, product=None):
+ """THE CREATE-SIDE GUARD: keep a real positive weight, else fill the default.
+ Drop-in for the REST variant `weight` field (returns POUNDS; pair with
+ weight_unit='lb')."""
+ w = current_weight_lb(variant)
+ return w if (math.isfinite(w) and w > 0) else default_weight_lb(variant, product)
+
+
+def variants_of(product=None):
+ """Normalize variants out of either a GraphQL connection or a plain REST array."""
+ p = _d(product)
+ edges = _d(p.get('variants')).get('edges')
+ if isinstance(edges, list):
+ return [e.get('node') for e in edges if isinstance(e, dict)]
+ v = p.get('variants')
+ return v if isinstance(v, list) else []
+
+
+def all_zero_weight_variants(product=None):
+ """THE ACTIVATE-SIDE GUARD: EVERY zero-weight variant, SAMPLE INCLUDED.
+ Matches dw-active-weight-canary exactly. Non-empty => DO NOT flip ACTIVE."""
+ return [v for v in variants_of(product) if has_zero_weight(v)]
+
+
+def unmeasured_variants(product=None):
+ """Variants whose payload carried NO weight field at all — the response never
+ measured them. Returned separately so a caller can report NOT-MEASURED rather
+ than claiming a clean zero reading."""
+ return [v for v in variants_of(product) if not weight_field_present(v)]
+
+
+def variant_label(variant=None):
+ v = _d(variant)
+ return v.get('sku') or v.get('title') or v.get('id') or '?'
+
+
+# ---------------------------------------------------------------- activate gate
+# The re-query an activate site MUST run so the guard MEASURES something. A product
+# query that omits inventoryItem{measurement{weight}} makes every variant look
+# zero-weight to current_weight_lb, and one that omits productType silently defaults
+# every heal to FALLBACK_LB. Both fields are required.
+WEIGHT_REQUERY = (
+ "query($id:ID!){ product(id:$id){ productType variants(first:100){edges{node{ "
+ "id sku title price inventoryItem{ id measurement{ weight{ value unit } } } }}} } }"
+)
+
+M_WEIGHT_SET = (
+ "mutation($id:ID!,$w:Float!){ inventoryItemUpdate(id:$id, "
+ "input:{measurement:{weight:{value:$w, unit:POUNDS}}}){ userErrors{message} } }"
+)
+
+
+def response_carries_weight_field(product=None):
+ """Did the re-query response ACTUALLY contain the weight field we asked for?
+
+ A gate that reads a response which never carried weight silently passes
+ everything (or, here, fails everything) for the wrong reason — the false-green
+ class CLAUDE.md warns about. Every variant node must carry an `inventoryItem`
+ key holding a `measurement` key. `measurement.weight = null` is a legitimate
+ 'no weight set' reading and is NOT a shape failure.
+ """
+ vs = variants_of(product)
+ if not vs:
+ return False
+ for v in vs:
+ d = _d(v)
+ iv = d.get('inventoryItem', d.get('inventory_item'))
+ if not isinstance(iv, dict) or 'measurement' not in iv:
+ return False
+ return True
+
+
+def heal_and_verify_weights(gql, product_gid, product=None, requery=None, mutation=None,
+ rest_heal=None):
+ """SELF-HEAL then VERIFY — the pattern Steve approved in
+ sanderson-onboard/scripts/create_sdg.mjs (8d09eed). Stranding product is worse
+ than assigning the already-approved default, but a heal that silently fails must
+ NEVER activate.
+
+ 1. every zero-weight variant (SAMPLE INCLUDED) is written default_weight_lb() in POUNDS
+ 2. the product is RE-READ and re-checked — the mutation's own 200 is not evidence
+ 3. ok is False => caller must HOLD the product as draft and name `weight>0`
+
+ Idempotent and a NO-OP (zero network calls) when every weight is already positive.
+ Fails CLOSED: an unreadable re-query, a response that never carried the weight
+ field, a missing inventoryItem id, or a userError all yield ok=False.
+
+ gql -- the call site's own gql(query, variables) -> parsed JSON
+ rest_heal -- optional fallback heal fn (variant_dict, lb) -> True/False, used when
+ the GraphQL inventoryItemUpdate path is unavailable (e.g. a token
+ without write_inventory). Verification stays GraphQL-side either way.
+ """
+ requery = requery or WEIGHT_REQUERY
+ mutation = mutation or M_WEIGHT_SET
+ errs, healed, still_zero = [], [], []
+ heal_failures = 0
+ p = _d(product)
+ product_type = p.get('productType') if p.get('productType') is not None else p.get('product_type')
+
+ zero = all_zero_weight_variants(p)
+ if not zero:
+ return {"ok": True, "healed": healed, "stillZero": [], "errs": errs} # no-op
+
+ for v in zero:
+ label = variant_label(v)
+ lb = default_weight_lb(v, {"productType": product_type})
+ iid = _d(_d(v).get('inventoryItem') or _d(v).get('inventory_item')).get('id')
+ wrote = False
+ if iid:
+ try:
+ r = gql(mutation, {"id": iid, "w": lb})
+ ue = (((r or {}).get('data') or {}).get('inventoryItemUpdate') or {}).get('userErrors') or []
+ top = (r or {}).get('errors') or []
+ for e in ue:
+ errs.append("weight:%s:%s" % (label, e.get('message')))
+ for e in top:
+ errs.append("weight:%s:%s" % (label, str(e.get('message'))[:80]))
+ wrote = not ue and not top
+ except Exception as e: # noqa: BLE001 - fail closed
+ errs.append("weight:%s:%s" % (label, str(e)[:80]))
+ else:
+ errs.append("weight:no-inventory-item:%s" % label)
+
+ if not wrote and rest_heal is not None:
+ try:
+ wrote = bool(rest_heal(v, lb))
+ except Exception as e: # noqa: BLE001 - fail closed
+ errs.append("weight:rest:%s:%s" % (label, str(e)[:80]))
+
+ if wrote:
+ healed.append({"sku": label, "lb": lb})
+ else:
+ heal_failures += 1
+
+ # RE-VERIFY against the live record. Never trust the write.
+ fresh = None
+ try:
+ resp = gql(requery, {"id": product_gid})
+ if (resp or {}).get('errors'):
+ errs.append("weight:reverify:%s" % str(resp['errors'])[:80])
+ fresh = ((resp or {}).get('data') or {}).get('product')
+ except Exception as e: # noqa: BLE001 - fail closed
+ errs.append("weight:reverify:%s" % str(e)[:80])
+ if not fresh:
+ errs.append("weight:reverify-failed")
+ return {"ok": False, "healed": healed, "stillZero": [], "errs": errs}
+ if not response_carries_weight_field(fresh):
+ # NOT MEASURED. The re-query came back without the field it asked for, so a
+ # clean-looking read proves nothing. Never a PASS.
+ errs.append("weight:reverify-response-carried-no-weight-field")
+ return {"ok": False, "healed": healed, "stillZero": [], "errs": errs}
+
+ still_zero = [variant_label(v) for v in all_zero_weight_variants(fresh)]
+ # FAIL CLOSED on an UNHEALED variant even when the re-verify comes back clean: a
+ # variant we could not write is UNMEASURED with respect to our own action, and a
+ # clean re-verify that happens to disagree is not licence to activate. Holding is
+ # reversible and the next run is a no-op, so the conservative branch costs nothing.
+ return {"ok": (not still_zero) and heal_failures == 0,
+ "healed": healed, "stillZero": still_zero, "errs": errs}
← 7bb6ba0 TK-10801: recover 3 DWNAT (Winfield Thybony) images locally;
·
back to Fentucci Naturals
·
TK-11471: wire the weight guard into the Fentucci create + a 9874a31 →