trucking: A/B/C coupon price test (20/30/40% off) + SpamAssassin harness

- CAMPAIGN_COUPON_AB_PCTS="20,30,40" mints one daily code per arm; each
  carrier is bucketed by a stable sha256(email) hash so the split is even
  (~33/33/33 verified over 30k) and stable across re-sends (no arm-hopping).
- Each arm's code stores its own percent in discount_codes, so the advertised
  discount always matches what checkout applies; redemptions are countable per
  code (marker campaign-daily:<date>:<pct>).
- Empty/unset keeps legacy single-arm behavior (COUPON_PCT, legacy marker).
- coupon_attribs() now takes per-recipient pct.
- Tests: scripts/tests/test_coupon_ab.py (5 pass). SpamAssassin: both main
  campaigns (186/188) score 0.0 HAM across all 3 arms, coupon block renders
  clean; harness saved for re-runs.

scripts/tests/sa_coupon_test.py

@@ -0,0 +1,123 @@
+#!/usr/bin/env python3
+"""Render a trucking campaign body with coupon merge-tags filled in and score it
+through SpamAssassin. Run on the prod host (has spamassassin + listmonk DB).
+
+Usage: python3 /tmp/sa_coupon_test.py <campaign_id> <pct> <code>
+"""
+import html
+import re
+import subprocess
+import sys
+from email.message import EmailMessage
+from email.utils import formatdate, make_msgid
+
+CID = sys.argv[1] if len(sys.argv) > 1 else "186"
+PCT = sys.argv[2] if len(sys.argv) > 2 else "40"
+CODE = sys.argv[3] if len(sys.argv) > 3 else "KQ7MTN".replace("6", "K")[:5] or "KQMTN"
+
+
+def psql(q):
+    out = subprocess.check_output(
+        ["docker", "exec", "performancewest-api-postgres-1",
+         "psql", "-U", "pw", "-d", "listmonk", "-tA", "-c", q],
+        text=True,
+    )
+    return out.rstrip("\n")
+
+
+subject = psql(f"SELECT subject FROM campaigns WHERE id={CID};")
+from_email = psql(f"SELECT from_email FROM campaigns WHERE id={CID};")
+body = psql(f"SELECT body FROM campaigns WHERE id={CID};")
+
+# --- Sample subscriber attribs (mirrors build_trucking_campaigns.coupon_attribs) ---
+attribs = {
+    "dot_number": "1228791",
+    "company": "BLUE RIDGE FREIGHT LLC",
+    "state": "CT",
+    "lp_link": f"https://performancewest.net/order/mcs150-update?code={CODE}",
+    "coupon_code": CODE,
+    "coupon_pct": PCT,
+    "coupon_expires": "11:59 PM ET tonight",
+}
+
+# --- Minimal listmonk template renderer: {{ if .Subscriber.Attribs.X }}..{{ else }}..{{ end }} + {{ .Subscriber.Attribs.X }} ---
+def render(tpl: str, at: dict) -> str:
+    # Resolve {{ if .Subscriber.Attribs.key }}TRUE{{ else }}FALSE{{ end }} (no nesting).
+    pat = re.compile(
+        r"\{\{\s*if\s+\.Subscriber\.Attribs\.(\w+)\s*\}\}(.*?)"
+        r"(?:\{\{\s*else\s*\}\}(.*?))?\{\{\s*end\s*\}\}",
+        re.DOTALL,
+    )
+    def repl(m):
+        key, t_branch, f_branch = m.group(1), m.group(2), m.group(3) or ""
+        return t_branch if str(at.get(key, "")).strip() else f_branch
+    prev = None
+    while prev != tpl:
+        prev = tpl
+        tpl = pat.sub(repl, tpl)
+    # Resolve simple value tags.
+    def val(m):
+        return html.escape(str(at.get(m.group(1), "")))
+    tpl = re.sub(r"\{\{\s*\.Subscriber\.Attribs\.(\w+)\s*\}\}", val, tpl)
+    # Listmonk unsubscribe/tracking placeholders -> realistic stand-ins.
+    tpl = tpl.replace("{{ UnsubscribeURL }}", "https://performancewest.net/unsubscribe/abc123")
+    tpl = re.sub(r"\{\{[^}]*\}\}", "", tpl)  # strip any remaining tags
+    return tpl
+
+rendered_subject = render(subject, attribs)
+rendered_body = render(body, attribs)
+
+# --- Assemble a realistic MIME email (what the recipient's filter sees) ---
+m_from = from_email
+# from_email is like 'Performance West <noreply@performancewest.net>'
+msg = EmailMessage()
+msg["From"] = m_from
+msg["To"] = "dispatch@blueridgefreight.com"
+msg["Subject"] = rendered_subject
+msg["Date"] = formatdate(localtime=True)
+msg["Message-ID"] = make_msgid(domain="performancewest.net")
+msg["List-Unsubscribe"] = "<https://performancewest.net/unsubscribe/abc123>, <mailto:unsub@performancewest.net>"
+msg["List-Unsubscribe-Post"] = "List-Unsubscribe=One-Click"
+msg["Precedence"] = "bulk"
+
+# plain-text altbody (strip tags crudely) + html
+text_alt = re.sub(r"<[^>]+>", "", rendered_body)
+text_alt = html.unescape(re.sub(r"\n{3,}", "\n\n", text_alt)).strip()
+msg.set_content(text_alt)
+msg.add_alternative(rendered_body, subtype="html")
+
+raw = msg.as_bytes()
+
+# --- Score through SpamAssassin (-t test mode; network tests skipped for determinism) ---
+proc = subprocess.run(
+    ["spamassassin", "-t", "-L"],  # -L = local tests only (no DNS/network), -t = test mode
+    input=raw, capture_output=True,
+)
+scored = proc.stdout.decode("utf-8", "replace")
+
+# Pull the score + the rule hits from the rewritten headers.
+score_line = ""
+rules = []
+for line in scored.splitlines():
+    if line.startswith("X-Spam-Status:"):
+        score_line = line
+    if line.startswith("X-Spam-Report:") or line.startswith("\tpts") or re.match(r"^\s+[-0-9.]+\s+[A-Z0-9_]+\s", line):
+        rules.append(line.rstrip())
+
+print("=" * 70)
+print(f"Campaign {CID}  arm={PCT}%  code={CODE}")
+print(f"Subject: {rendered_subject}")
+print("=" * 70)
+print(score_line or "(no X-Spam-Status header found)")
+print("-" * 70)
+m = re.search(r"score=([-0-9.]+)", score_line)
+if m:
+    print(f"SCORE: {m.group(1)}   (SpamAssassin default spam threshold = 5.0)")
+# Show the detailed report block.
+rep = re.search(r"X-Spam-Report:(.*?)(?:\nX-Spam-|\n\n)", scored, re.DOTALL)
+if rep:
+    print(rep.group(0))
+else:
+    # Fallback: print any rule lines we collected.
+    for r in rules:
+        print(r)

scripts/tests/sa_coupon_test_local.py

@@ -0,0 +1,100 @@
+#!/usr/bin/env python3
+"""Render a trucking campaign body with coupon merge-tags filled in and score it
+through local SpamAssassin (4.0.1). Reads the body HTML from a file.
+
+Usage: sa_coupon_local.py <body.html> <subject> <pct> <code>
+"""
+import html
+import re
+import subprocess
+import sys
+from email.message import EmailMessage
+from email.utils import formatdate, make_msgid
+
+BODY_FILE = sys.argv[1]
+SUBJECT_TPL = sys.argv[2]
+PCT = sys.argv[3] if len(sys.argv) > 3 else "40"
+CODE = sys.argv[4] if len(sys.argv) > 4 else "KQMTN"
+
+body = open(BODY_FILE, encoding="utf-8").read()
+
+attribs = {
+    "dot_number": "1228791",
+    "company": "BLUE RIDGE FREIGHT LLC",
+    "state": "CT",
+    "lp_link": f"https://performancewest.net/order/mcs150-update?code={CODE}",
+    "coupon_code": CODE,
+    "coupon_pct": PCT,
+    "coupon_expires": "11:59 PM ET tonight",
+}
+
+
+def render(tpl: str, at: dict) -> str:
+    pat = re.compile(
+        r"\{\{\s*if\s+\.Subscriber\.Attribs\.(\w+)\s*\}\}(.*?)"
+        r"(?:\{\{\s*else\s*\}\}(.*?))?\{\{\s*end\s*\}\}",
+        re.DOTALL,
+    )
+
+    def repl(m):
+        key, t_branch, f_branch = m.group(1), m.group(2), m.group(3) or ""
+        return t_branch if str(at.get(key, "")).strip() else f_branch
+
+    prev = None
+    while prev != tpl:
+        prev = tpl
+        tpl = pat.sub(repl, tpl)
+
+    def val(m):
+        return html.escape(str(at.get(m.group(1), "")))
+
+    tpl = re.sub(r"\{\{\s*\.Subscriber\.Attribs\.(\w+)\s*\}\}", val, tpl)
+    tpl = tpl.replace("{{ UnsubscribeURL }}", "https://performancewest.net/unsubscribe/abc123")
+    tpl = re.sub(r"\{\{[^}]*\}\}", "", tpl)
+    return tpl
+
+
+rendered_subject = render(SUBJECT_TPL, attribs)
+rendered_body = render(body, attribs)
+
+msg = EmailMessage()
+msg["From"] = "Performance West <noreply@performancewest.net>"
+msg["To"] = "dispatch@blueridgefreight.com"
+msg["Subject"] = rendered_subject
+msg["Date"] = formatdate(localtime=True)
+msg["Message-ID"] = make_msgid(domain="performancewest.net")
+msg["List-Unsubscribe"] = "<https://performancewest.net/unsubscribe/abc123>, <mailto:unsub@performancewest.net>"
+msg["List-Unsubscribe-Post"] = "List-Unsubscribe=One-Click"
+msg["Precedence"] = "bulk"
+
+text_alt = re.sub(r"<[^>]+>", "", rendered_body)
+text_alt = html.unescape(re.sub(r"\n{3,}", "\n\n", text_alt)).strip()
+msg.set_content(text_alt)
+msg.add_alternative(rendered_body, subtype="html")
+
+raw = msg.as_bytes()
+
+proc = subprocess.run(["spamassassin", "-t", "-L"], input=raw, capture_output=True)
+scored = proc.stdout.decode("utf-8", "replace")
+
+score_line = ""
+for line in scored.splitlines():
+    if line.startswith("X-Spam-Status:"):
+        score_line = line
+        break
+
+m = re.search(r"score=([-0-9.]+)", score_line)
+score = m.group(1) if m else "?"
+verdict = "SPAM" if score_line.startswith("X-Spam-Status: Yes") else "HAM"
+
+print(f"arm={PCT}% code={CODE}  ->  SCORE {score}  [{verdict}]   subj: {rendered_subject}")
+
+# Detailed rule report.
+rep = re.search(r"X-Spam-Report:(.*?)\n(?:[A-Za-z-]+:|\n)", scored, re.DOTALL)
+if "--verbose" in sys.argv:
+    if rep:
+        print(rep.group(1))
+    else:
+        for line in scored.splitlines():
+            if re.match(r"^\s*[-0-9.]+\s+[A-Z0-9_]+", line):
+                print("   ", line.strip())

scripts/tests/test_coupon_ab.py

@@ -0,0 +1,76 @@
+#!/usr/bin/env python3
+"""Unit tests for the trucking same-day coupon A/B/C price test.
+
+Verifies that CAMPAIGN_COUPON_AB_PCTS produces an even, stable per-email split,
+that the advertised percent always matches the arm's actual code (so the email
+never promises a discount checkout won't honor), and that the off / single-arm
+states behave. Pure-function tests, no DB required (psycopg2 is stubbed).
+
+Run: CAMPAIGN_COUPON_AB_PCTS="20,30,40" python3 scripts/tests/test_coupon_ab.py
+"""
+from __future__ import annotations
+
+import importlib.util
+import os
+import sys
+import types
+from collections import Counter
+from pathlib import Path
+
+os.environ.setdefault("CAMPAIGN_COUPON_AB_PCTS", "20,30,40")
+
+# Stub the DB driver so the builder imports without a live Postgres.
+sys.modules.setdefault("psycopg2", types.ModuleType("psycopg2"))
+
+_SCRIPT = Path(__file__).resolve().parents[1] / "build_trucking_campaigns.py"
+_spec = importlib.util.spec_from_file_location("btc", _SCRIPT)
+btc = importlib.util.module_from_spec(_spec)
+_spec.loader.exec_module(btc)
+
+
+def test_arms_parsed():
+    assert btc.COUPON_AB_PCTS == (20, 30, 40)
+
+
+def test_even_stable_and_code_matches_pct():
+    coupons = {20: "AAAAA", 30: "BBBBB", 40: "CCCCC"}
+    emails = [f"user{i}@carrier{i % 500}.com" for i in range(30000)]
+    counts: Counter = Counter()
+    for e in emails:
+        code, pct = btc.pick_coupon_for_email(e, coupons)
+        counts[pct] += 1
+        # Advertised percent must match the arm's real code.
+        assert coupons[int(pct)] == code
+        # Stable: same input -> same arm.
+        assert btc.pick_coupon_for_email(e, coupons) == (code, pct)
+    total = sum(counts.values())
+    # Each arm should be within ~2 points of an even third.
+    for pct in ("20", "30", "40"):
+        share = counts[pct] / total
+        assert 0.31 <= share <= 0.353, (pct, share)
+
+
+def test_off_state():
+    assert btc.pick_coupon_for_email("x@y.com", None) == ("", "")
+    assert btc.pick_coupon_for_email("x@y.com", {}) == ("", "")
+
+
+def test_single_arm():
+    assert btc.pick_coupon_for_email("a@b.com", {40: "ZZZZZ"}) == ("ZZZZZ", "40")
+
+
+def test_coupon_attribs_reflects_pct():
+    a = btc.coupon_attribs("BBBBB", "30")
+    assert a["coupon_code"] == "BBBBB"
+    assert a["coupon_pct"] == "30"
+    assert btc.coupon_attribs("", "30") == {
+        "coupon_code": "", "coupon_pct": "", "coupon_expires": ""
+    }
+
+
+if __name__ == "__main__":
+    fns = [v for k, v in sorted(globals().items()) if k.startswith("test_")]
+    for fn in fns:
+        fn()
+        print(f"ok  {fn.__name__}")
+    print(f"\nAll {len(fns)} coupon A/B tests passed.")