new-site/scripts/tests/test_coupon_ab.py

#!/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_full_price_control_arm():
    """A 0% arm (code "") is a full-price control: returned as ("", "") so the
    email renders the normal-price branch, but carriers are still hash-bucketed
    evenly across all three arms (so the control is measurable)."""
    coupons = {20: "AAAAA", 30: "BBBBB", 0: ""}  # 0 = full-price control
    counts: Counter = Counter()
    arm_for: dict = {}
    for i in range(30000):
        e = f"user{i}@carrier{i % 500}.com"
        code, pct = btc.pick_coupon_for_email(e, coupons)
        # The control arm presents identically to a no-coupon send.
        if code == "":
            assert pct == ""
        else:
            assert code in ("AAAAA", "BBBBB") and pct in ("20", "30")
        # Bucket label: recover arm even for the silent control via re-hash.
        h = __import__("hashlib").sha256(e.encode()).hexdigest()
        arm = sorted(coupons)[int(h, 16) % 3]
        arm_for[e] = arm
        counts[arm] += 1
        # Stable across re-calls.
        assert btc.pick_coupon_for_email(e, coupons) == (code, pct)
    total = sum(counts.values())
    for arm in (0, 20, 30):
        share = counts[arm] / total
        assert 0.31 <= share <= 0.353, (arm, share)
    # The price helper yields no number for the control arm (blank pct).
    assert btc.discounted_price_attribs("mcs150", None, "")["coupon_priceable"] == ""


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": ""
    }


def test_lp_link_well_formed_in_all_states():
    """The CTA URL must be valid whether or not a coupon is active. lp_link owns
    the `?`-query so templates append their own params with a leading `&`."""
    import urllib.parse

    def cta(lp, lead):  # mimic a template appending its own params
        return f"{lp}{lead}utm_source=listmonk&utm_campaign=x"

    for code in (None, "KQMTN"):
        for ct, st in (("mcs150", None), ("irp_ifta", "CT")):
            lp = btc.lp_link_with_coupon(ct, st, code, dot="1228791")
            url = cta(lp, "&")
            parts = urllib.parse.urlsplit(url)
            assert "&" not in parts.path, url          # no stray & in path
            assert parts.query.count("?") == 0, url    # no double ?
            assert "dot=1228791" in parts.query, url
            if code:
                assert f"code={code}" in parts.query, url


def test_discounted_price_matches_checkout_formula():
    """coupon_price_deal must equal full - round(full*pct/100) for discountable
    services, and be blank for non-discountable ones (e.g. boc3-filing)."""
    cat = btc._load_service_catalog()
    assert cat, "service catalog failed to load"
    for ct, st in (("mcs150", None), ("inactive", None), ("irp_ifta", "CT"),
                   ("hazmat", None), ("state_weight_tax", "NY"), ("state_emissions", "CA")):
        slug = btc.price_slug_for(ct, st)
        entry = cat[slug]
        for pct in ("20", "30", "40"):
            a = btc.discounted_price_attribs(ct, st, pct)
            full = entry["price_cents"]
            deal = full - round(full * int(pct) / 100)
            exp_full = f"${full // 100}" if full % 100 == 0 else f"${full / 100:.2f}"
            exp_deal = f"${deal // 100}" if deal % 100 == 0 else f"${deal / 100:.2f}"
            assert a["coupon_price_full"] == exp_full, (ct, pct, a)
            assert a["coupon_price_deal"] == exp_deal, (ct, pct, a)
            assert a["coupon_priceable"] == "1"

    # Non-discountable passthrough -> no price, percent-only fallback.
    boc = btc.discounted_price_attribs("for_hire_boc3", None, "40")
    assert boc == {"coupon_price_full": "", "coupon_price_deal": "", "coupon_priceable": ""}
    # No coupon -> blank.
    assert btc.discounted_price_attribs("mcs150", None, "")["coupon_priceable"] == ""


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.")