← back to Dw Five Field Step0
Anchor SKU guard to live variants
1726ebd2eafc73b528f5c73738118f8e46bcd850 · 2026-08-29 06:37:19 -0700 · Steve Abrams
Files touched
M bulk-fivefield-exec.pyM sku_guard.pyM test/test_sku_guard.pyM verification/e2e-proof.json
Diff
commit 1726ebd2eafc73b528f5c73738118f8e46bcd850
Author: Steve Abrams <steve@designerwallcoverings.com>
Date: Sat Aug 29 06:37:19 2026 -0700
Anchor SKU guard to live variants
---
bulk-fivefield-exec.py | 2 +-
sku_guard.py | 80 ++++++++++++++++++++++++++++++---------------
test/test_sku_guard.py | 46 +++++++++++++++++---------
verification/e2e-proof.json | 6 ++--
4 files changed, 88 insertions(+), 46 deletions(-)
diff --git a/bulk-fivefield-exec.py b/bulk-fivefield-exec.py
index 6aa8d45..9276c1c 100644
--- a/bulk-fivefield-exec.py
+++ b/bulk-fivefield-exec.py
@@ -265,7 +265,7 @@ def process(item, results, dry_run=False):
"requires_shipping": True, "taxable": True}}
# Identity guard immediately before the only variant-create boundary.
- sku_check = validate_dw_sku(target, item.get("mfr_sku"), dw)
+ sku_check = validate_dw_sku(target, fix, variants)
if not sku_check["ok"]:
rec["status"] = "errored"; rec["reason"] = f"sku_preflight:{sku_check['reason']}"
results.append(rec); print(f"ERR {dw}: SKU preflight {sku_check['reason']}"); return 0
diff --git a/sku_guard.py b/sku_guard.py
index 7a7202b..9c47cc1 100644
--- a/sku_guard.py
+++ b/sku_guard.py
@@ -1,34 +1,62 @@
-"""Fail-closed DW SKU identity guard; never allocates or infers a SKU."""
+"""Fail-closed DW SKU identity guard anchored to fetched live variants."""
import re
-UNIT_SUFFIX = re.compile(r"-(?:sample|roll|yard|panel)$", re.IGNORECASE)
+ASCII_SKU = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._/-]*$", re.ASCII)
+NON_SAMPLE_UNIT = re.compile(r"-(?:roll|yard|panel)$", re.IGNORECASE | re.ASCII)
+SAMPLE_SUFFIX = re.compile(r"-sample$", re.IGNORECASE | re.ASCII)
-def _base(value):
- if value is None:
- return ""
- text = str(value).strip()
- if not text or text.lower() in {"none", "null", "null-sample"}:
- return ""
- return UNIT_SUFFIX.sub("", text).strip().upper()
+def _strict_sku(value):
+ if not isinstance(value, str) or value != value.strip() or not ASCII_SKU.fullmatch(value):
+ return None
+ if value.lower() in {"none", "null", "null-sample"}:
+ return None
+ return value
-def validate_dw_sku(candidate, mfr_sku=None, existing_dw_sku=None):
- """Allow only an identity already present on the product or manufacturer record."""
- base = _base(candidate)
- if not base:
+def validate_dw_sku(candidate, lane, live_variants):
+ """Require a lane-shaped target derived from an actual fetched live variant SKU."""
+ target = _strict_sku(candidate)
+ if not target:
return {"ok": False, "reason": "blank-candidate", "basis": None}
- sources = []
- existing = _base(existing_dw_sku)
- manufacturer = _base(mfr_sku)
- if existing:
- sources.append(("existing-dw-sku", existing))
- if manufacturer:
- sources.append(("manufacturer-sku", manufacturer))
- if not sources:
- return {"ok": False, "reason": "no-existing-identity", "basis": None}
- for basis, source in sources:
- if base == source:
- return {"ok": True, "reason": "preserved-existing-identity", "basis": basis}
- return {"ok": False, "reason": "candidate-not-derived-from-existing-identity", "basis": None}
+ if not isinstance(live_variants, list):
+ return {"ok": False, "reason": "invalid-live-variants", "basis": None}
+
+ if lane == "add-sample":
+ if not target.endswith("-Sample"):
+ return {"ok": False, "reason": "invalid-sample-target", "basis": None}
+ target_base = target[:-7]
+ if target_base != target_base.upper():
+ return {"ok": False, "reason": "noncanonical-sample-base", "basis": None}
+ trusted = set()
+ for variant in live_variants:
+ if not isinstance(variant, dict) or str(variant.get("option1") or "").strip().lower() == "sample":
+ continue
+ live_sku = _strict_sku(variant.get("sku"))
+ if live_sku:
+ live_base = NON_SAMPLE_UNIT.sub("", live_sku)
+ if live_base == live_base.upper():
+ trusted.add(live_base)
+ if target_base in trusted:
+ return {"ok": True, "reason": "derived-from-live-non-sample", "basis": "live-variant"}
+ return {"ok": False, "reason": "sample-base-not-live", "basis": None}
+
+ if lane == "build-roll":
+ if target != target.upper():
+ return {"ok": False, "reason": "noncanonical-roll-base", "basis": None}
+ trusted = set()
+ for variant in live_variants:
+ if not isinstance(variant, dict):
+ continue
+ live_sku = _strict_sku(variant.get("sku"))
+ is_sample = str(variant.get("option1") or "").strip().lower() == "sample"
+ if live_sku and is_sample and SAMPLE_SUFFIX.search(live_sku):
+ live_base = SAMPLE_SUFFIX.sub("", live_sku)
+ if live_base == live_base.upper():
+ trusted.add(live_base)
+ if target in trusted:
+ return {"ok": True, "reason": "recovered-from-live-sample", "basis": "live-sample-variant"}
+ return {"ok": False, "reason": "roll-base-not-live-sample", "basis": None}
+
+ return {"ok": False, "reason": "unknown-lane", "basis": None}
diff --git a/test/test_sku_guard.py b/test/test_sku_guard.py
index b747f73..002b81c 100644
--- a/test/test_sku_guard.py
+++ b/test/test_sku_guard.py
@@ -7,25 +7,39 @@ from sku_guard import validate_dw_sku
class SkuGuardTests(unittest.TestCase):
- def test_allows_existing_identity_and_unit_suffixes(self):
- for candidate in ["DWKK-102147", "DWKK-102147-Sample", "dwkk-102147-roll"]:
- result = validate_dw_sku(candidate, "128012", "DWKK-102147")
- self.assertTrue(result["ok"])
- self.assertEqual(result["basis"], "existing-dw-sku")
-
- def test_allows_recovered_real_manufacturer_identity(self):
- result = validate_dw_sku("128012-Sample", "128012", None)
+ def test_add_sample_requires_a_fetched_live_non_sample_base(self):
+ live = [{"sku": "DWKK-102147", "option1": "Roll"}]
+ result = validate_dw_sku("DWKK-102147-Sample", "add-sample", live)
self.assertTrue(result["ok"])
- self.assertEqual(result["basis"], "manufacturer-sku")
+ self.assertEqual(result["basis"], "live-variant")
+ self.assertFalse(validate_dw_sku("DWKK-999999-Sample", "add-sample", live)["ok"])
- def test_rejects_new_sequential_or_unrelated_identity(self):
- result = validate_dw_sku("DWKK-999999", "128012", "DWKK-102147")
- self.assertEqual(result, {"ok": False, "reason": "candidate-not-derived-from-existing-identity", "basis": None})
+ def test_build_roll_requires_a_fetched_live_sample_base(self):
+ result = validate_dw_sku("DWKK-102147", "build-roll", [
+ {"sku": "DWKK-102147-Sample", "option1": "Sample"},
+ ])
+ self.assertTrue(result["ok"])
+ self.assertEqual(result["basis"], "live-sample-variant")
+ self.assertFalse(validate_dw_sku("DWKK-999999", "build-roll", [
+ {"sku": "DWKK-102147-Sample", "option1": "Sample"},
+ ])["ok"])
+
+ def test_legitimate_base_ending_sample_is_anchored_without_lossy_double_strip(self):
+ result = validate_dw_sku("ART-SAMPLE-Sample", "add-sample", [
+ {"sku": "ART-SAMPLE", "option1": "Roll"},
+ ])
+ self.assertTrue(result["ok"])
+
+ def test_rejects_coercion_unicode_controls_whitespace_and_unknown_lane(self):
+ live = [{"sku": "ABC", "option1": "Roll"}]
+ for candidate in [None, "", "None", "null", 12345, " ABC-Sample", "ABC\nDEF-Sample", "straße-Sample", "abc-Sample"]:
+ self.assertFalse(validate_dw_sku(candidate, "add-sample", live)["ok"])
+ self.assertEqual(validate_dw_sku("ABC-Sample", "mystery", live)["reason"], "unknown-lane")
- def test_rejects_blank_literal_null_and_source_less_candidates(self):
- for candidate in [None, "", "None", "null", "null-Sample"]:
- self.assertFalse(validate_dw_sku(candidate, "128012", "DWKK-102147")["ok"])
- self.assertEqual(validate_dw_sku("DWKK-102147", None, None)["reason"], "no-existing-identity")
+ def test_rejects_missing_or_malformed_live_provenance(self):
+ self.assertEqual(validate_dw_sku("ABC-Sample", "add-sample", None)["reason"], "invalid-live-variants")
+ self.assertFalse(validate_dw_sku("ABC-Sample", "add-sample", [{"sku": "ABC-Sample", "option1": "Sample"}])["ok"])
+ self.assertFalse(validate_dw_sku("ABC", "build-roll", [{"sku": "ABC-Sample", "option1": "Roll"}])["ok"])
if __name__ == "__main__":
diff --git a/verification/e2e-proof.json b/verification/e2e-proof.json
index 894035d..6a4ca3e 100644
--- a/verification/e2e-proof.json
+++ b/verification/e2e-proof.json
@@ -1,16 +1,16 @@
{
- "intent": "Block any five-field variant write whose SKU is not derived from an existing product or manufacturer identity.",
+ "intent": "Block any five-field variant write whose SKU is not derived from a fetched live variant on that product.",
"risk_tier": "R1 isolated validation and pre-write integration; no Shopify or DB call",
"environment": "local Python unittest and static source boundary assertion",
"timestamp": "2026-08-29T06:42:00Z",
"ticket": "TK-10956-block-invented-skus-before-five-field-va",
"build_identity": "git parent b51daf4 plus owned Cycle 15 diff",
"checks": [
- {"verdict":"PASS","boundary":"identity validation","command":"python3 -m unittest discover -s test -p 'test_*.py'","assertions":"existing and manufacturer identities pass; unrelated sequential, blank, literal-null, and source-less candidates fail"},
+ {"verdict":"PASS","boundary":"identity validation","command":"python3 -m unittest discover -s test -p 'test_*.py'","assertions":"sample targets derive from fetched live non-sample variants; roll targets derive from fetched live sample variants; unrelated worklist values and malformed identities fail"},
{"verdict":"PASS","boundary":"pre-write ordering","command":"static source assertion","assertions":"validate_dw_sku call and fail-closed return precede Shopify variant POST"},
{"verdict":"PASS","boundary":"syntax/diff","command":"python3 -m py_compile sku_guard.py bulk-fivefield-exec.py && git diff --check","assertions":"modules compile and diff is clean"}
],
- "negative_checks": ["unrelated sequential candidate", "blank candidate", "literal null candidate", "no existing identity"],
+ "negative_checks": ["self-authorizing worklist candidate", "unrelated sequential candidate", "blank/literal-null/coercible candidate", "Unicode or control characters", "missing or malformed live variant provenance", "wrong lane suffix"],
"side_effects": "none; no DB, Shopify, provider, customer-facing, schedule, restart, deploy, or send action",
"cleanup": "pycache files are ignored and removed after the test",
"verdict": "PASS for the local pre-write boundary"
← 9a4296c Block invented SKUs before variant writes
·
back to Dw Five Field Step0
·
Require unambiguous live SKU provenance e63b94f →