← back to Fentucci Naturals
scripts/lib/test_weight_guard.py
366 lines
#!/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)