f8cd37ac8c
Includes: API (Express/TypeScript), Astro site, Python workers, document generators, FCC compliance tools, Canada CRTC formation, Ansible infrastructure, and deployment scripts. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
475 lines
17 KiB
Python
475 lines
17 KiB
Python
"""
|
||
API-only CRTC pricing matrix test.
|
||
|
||
Verifies pricing math for all province × company-type × option combinations
|
||
by calling POST /api/v1/canada-crtc/orders directly (no browser needed).
|
||
|
||
Tests:
|
||
B1. 6 base pricing combos (2 provinces × 3 company types)
|
||
B2. 2 expedited pricing combos (BC + ON)
|
||
B3. 3 discount code tests (percent, flat, invalid)
|
||
B4. 1 own-Canadian-address test
|
||
B5. 4 validation edge cases (missing required fields → 400)
|
||
|
||
Usage:
|
||
python3 -m scripts.tests.e2e_crtc_pricing
|
||
python3 -m scripts.tests.e2e_crtc_pricing --keep # don't clean up test orders
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import json
|
||
import logging
|
||
import os
|
||
import sys
|
||
import uuid
|
||
from pathlib import Path
|
||
|
||
from dotenv import load_dotenv
|
||
|
||
env_path = Path(__file__).parent / ".env.test"
|
||
if env_path.exists():
|
||
load_dotenv(env_path)
|
||
|
||
import time
|
||
import requests
|
||
|
||
LOG = logging.getLogger("tests.pricing")
|
||
logging.basicConfig(
|
||
level=logging.INFO,
|
||
format="%(asctime)s [%(name)s] %(levelname)s %(message)s",
|
||
stream=sys.stdout,
|
||
)
|
||
|
||
# Use the dev API internal port directly (faster, avoids nginx TLS timeout)
|
||
API_URL = os.environ.get("DEV_API_URL", "http://207.174.124.71:3002")
|
||
DEV_DB_URL = os.environ.get(
|
||
"DEV_DATABASE_URL",
|
||
"postgresql://pw:pw_dev_2026@207.174.124.71:5433/performancewest",
|
||
)
|
||
|
||
# ── Expected constants (must match api/src/routes/canada-crtc.ts) ─────────
|
||
SERVICE_FEE = 389900
|
||
TRADE_NAME_ADDON = 7500
|
||
NAMED_ADDON = 8500
|
||
EXPEDITED_FEE = 50000
|
||
|
||
# Gov fees in CAD cents (from GOV_FEES_CAD in canada-crtc.ts)
|
||
GOV_CAD = {
|
||
"BC": {"numbered": 35000, "numbered_tradename": 39000, "named": 38000, "expedite": 10000},
|
||
"ON": {"numbered": 36000, "numbered_tradename": 40000, "named": 38500, "expedite": 0},
|
||
}
|
||
|
||
TYPE_ADDONS = {
|
||
"numbered": 0,
|
||
"numbered_tradename": TRADE_NAME_ADDON,
|
||
"named": NAMED_ADDON,
|
||
}
|
||
|
||
# FX range: CAD→USD rate is typically 0.68-0.80, plus 10% buffer + ceil to $1.
|
||
# We use a wide range to avoid flakiness.
|
||
FX_LOW = 0.65
|
||
FX_HIGH = 0.85
|
||
|
||
|
||
def cad_to_usd_range(cad_cents: int) -> tuple[int, int]:
|
||
"""Return (min, max) USD cents for a given CAD cents amount, accounting for FX variation."""
|
||
if cad_cents == 0:
|
||
return (0, 0)
|
||
lo = int(cad_cents * FX_LOW)
|
||
hi = int(cad_cents * FX_HIGH * 1.10) + 100 # +10% buffer + $1 rounding headroom
|
||
return (lo, hi)
|
||
|
||
|
||
# ── Test data factory ─────────────────────────────────────────────────────
|
||
|
||
def make_payload(
|
||
province: str = "BC",
|
||
company_type: str = "numbered",
|
||
trade_name: str = "",
|
||
name_choices: tuple[str, str, str] = ("", "", ""),
|
||
legal_ending: str = "Ltd.",
|
||
expedited: bool = False,
|
||
discount_code: str = "",
|
||
has_own_ca_address: bool = False,
|
||
own_ca_fields: dict | None = None,
|
||
amb_location_slug: str | None = None,
|
||
) -> dict:
|
||
uid = uuid.uuid4().hex[:6]
|
||
geo = f"{province} and Worldwide"
|
||
|
||
payload = {
|
||
"customer_name": f"PricingTest {province} {company_type}",
|
||
"customer_email": f"pricing-test+{uid}@performancewest.net",
|
||
"customer_phone": "+10005551234",
|
||
"company_type": company_type,
|
||
"incorporation_province": province,
|
||
"director_first_name": "Pricing",
|
||
"director_last_name": "Tester",
|
||
"director_country": "US",
|
||
"director_street": "100 Test Blvd",
|
||
"director_street2": "",
|
||
"director_city": "Houston",
|
||
"director_province": "TX",
|
||
"director_postal": "77001",
|
||
"director_citizenship": "United States",
|
||
"services_description": f"Pricing matrix test — {province} {company_type}",
|
||
"geographic_coverage": geo,
|
||
"include_bits": True,
|
||
"reg_contact_name": "Pricing Tester",
|
||
"reg_contact_email": f"pricing-test+{uid}@performancewest.net",
|
||
"reg_contact_phone": "+10005551234",
|
||
"expedited": expedited,
|
||
"disclaimer_agreed": True,
|
||
"test_mode": True,
|
||
}
|
||
|
||
if company_type == "numbered_tradename":
|
||
payload["trade_name"] = trade_name or "Test Trade Name"
|
||
payload["add_trade_name"] = True
|
||
elif company_type == "named":
|
||
payload["company_name_choice1"] = name_choices[0] or "Test Telecom Solutions Ltd."
|
||
payload["company_name_choice2"] = name_choices[1] or "Northern Voice Networks Inc."
|
||
payload["company_name_choice3"] = name_choices[2] or "Lakeside Communications Corp."
|
||
payload["legal_ending"] = legal_ending
|
||
|
||
if discount_code:
|
||
payload["discount_code"] = discount_code
|
||
|
||
if has_own_ca_address:
|
||
payload["has_own_ca_address"] = True
|
||
defaults = {
|
||
"own_ca_company": "Test Corp Office",
|
||
"own_ca_street": "100 Test St",
|
||
"own_ca_city": "Vancouver",
|
||
"own_ca_province": province,
|
||
"own_ca_postal": "V6B 1A1",
|
||
}
|
||
payload.update(own_ca_fields or defaults)
|
||
payload["amb_location_slug"] = None
|
||
elif amb_location_slug:
|
||
payload["amb_location_slug"] = amb_location_slug
|
||
|
||
return payload
|
||
|
||
|
||
_last_request_at = 0.0
|
||
|
||
def post_order(payload: dict) -> tuple[int, dict]:
|
||
"""POST to the order creation endpoint. Returns (status_code, response_json)."""
|
||
global _last_request_at
|
||
# Rate limit: wait at least 1.5s between requests to avoid 429
|
||
elapsed = time.time() - _last_request_at
|
||
if elapsed < 1.5:
|
||
time.sleep(1.5 - elapsed)
|
||
_last_request_at = time.time()
|
||
|
||
url = f"{API_URL}/api/v1/canada-crtc/orders"
|
||
r = requests.post(url, json=payload, timeout=15)
|
||
try:
|
||
data = r.json()
|
||
except Exception:
|
||
data = {"raw": r.text[:500]}
|
||
return r.status_code, data
|
||
|
||
|
||
# ── Result tracking ───────────────────────────────────────────────────────
|
||
|
||
PASS = 0
|
||
FAIL = 0
|
||
RESULTS: list[dict] = []
|
||
|
||
|
||
def check(label: str, ok: bool, detail: str = ""):
|
||
global PASS, FAIL
|
||
if ok:
|
||
PASS += 1
|
||
LOG.info(" PASS: %s", label)
|
||
else:
|
||
FAIL += 1
|
||
LOG.info(" FAIL: %s — %s", label, detail)
|
||
RESULTS.append({"label": label, "ok": ok, "detail": detail})
|
||
|
||
|
||
def in_range(value: int, lo: int, hi: int) -> bool:
|
||
return lo <= value <= hi
|
||
|
||
|
||
# ── B1: Base pricing per province × company type ─────────────────────────
|
||
|
||
def test_base_pricing():
|
||
LOG.info("\n=== B1: Base Pricing Matrix (6 combos) ===")
|
||
|
||
for prov in ("BC", "ON"):
|
||
for ctype in ("numbered", "numbered_tradename", "named"):
|
||
label = f"{prov}/{ctype}"
|
||
LOG.info("\n --- %s ---", label)
|
||
|
||
payload = make_payload(province=prov, company_type=ctype)
|
||
status, data = post_order(payload)
|
||
|
||
check(f"{label}: HTTP 201", status == 201, f"got {status}: {data.get('error', '')}")
|
||
if status != 201:
|
||
continue
|
||
|
||
p = data.get("pricing", {})
|
||
expected_addon = TYPE_ADDONS[ctype]
|
||
expected_service = SERVICE_FEE + expected_addon
|
||
|
||
check(f"{label}: service_fee_cents = {expected_service}",
|
||
p.get("service_fee_cents") == expected_service,
|
||
f"got {p.get('service_fee_cents')}")
|
||
|
||
check(f"{label}: type_addon_cents = {expected_addon}",
|
||
p.get("type_addon_cents") == expected_addon,
|
||
f"got {p.get('type_addon_cents')}")
|
||
|
||
# Gov fees — range check (CAD→USD varies by FX rate)
|
||
gov_cad = GOV_CAD[prov][ctype]
|
||
gov_lo, gov_hi = cad_to_usd_range(gov_cad)
|
||
gov_actual = p.get("government_fee_cents", 0)
|
||
check(f"{label}: gov_fee in range [{gov_lo}, {gov_hi}]",
|
||
in_range(gov_actual, gov_lo, gov_hi),
|
||
f"got {gov_actual} (C${gov_cad / 100:.0f})")
|
||
|
||
# Total = service + gov + mailbox - discount
|
||
total = p.get("subtotal_cents", 0)
|
||
mailbox = p.get("mailbox_annual_cents", 0)
|
||
discount = p.get("discount_cents", 0)
|
||
expected_total = expected_service + gov_actual + mailbox - discount
|
||
check(f"{label}: total_cents math correct",
|
||
total == expected_total,
|
||
f"total={total}, expected={expected_total} (svc={expected_service}+gov={gov_actual}+mb={mailbox}-disc={discount})")
|
||
|
||
|
||
# ── B2: Expedited pricing ─────────────────────────────────────────────────
|
||
|
||
def test_expedited_pricing():
|
||
LOG.info("\n=== B2: Expedited Pricing (BC + ON) ===")
|
||
|
||
for prov in ("BC", "ON"):
|
||
label = f"{prov}/expedited"
|
||
LOG.info("\n --- %s ---", label)
|
||
|
||
payload = make_payload(province=prov, expedited=True)
|
||
status, data = post_order(payload)
|
||
|
||
check(f"{label}: HTTP 201", status == 201, f"got {status}: {data.get('error', '')}")
|
||
if status != 201:
|
||
continue
|
||
|
||
p = data.get("pricing", {})
|
||
expedite_actual = p.get("expedite_fee_cents", 0)
|
||
|
||
# BC: $500 + cadToUsd(C$100) ≈ $500 + ~$72-$94
|
||
# ON: $500 + cadToUsd(C$0) = $500 exactly
|
||
prov_expedite_cad = GOV_CAD[prov]["expedite"]
|
||
if prov_expedite_cad > 0:
|
||
exp_lo = EXPEDITED_FEE + cad_to_usd_range(prov_expedite_cad)[0]
|
||
exp_hi = EXPEDITED_FEE + cad_to_usd_range(prov_expedite_cad)[1]
|
||
else:
|
||
exp_lo = EXPEDITED_FEE
|
||
exp_hi = EXPEDITED_FEE
|
||
|
||
check(f"{label}: expedite_fee in [{exp_lo}, {exp_hi}]",
|
||
in_range(expedite_actual, exp_lo, exp_hi),
|
||
f"got {expedite_actual}")
|
||
|
||
# Verify total includes expedite fee
|
||
total = p.get("subtotal_cents", 0)
|
||
check(f"{label}: total includes expedite fee",
|
||
total >= SERVICE_FEE + expedite_actual,
|
||
f"total={total}")
|
||
|
||
|
||
# ── B3: Discount codes ────────────────────────────────────────────────────
|
||
|
||
def test_discount_codes():
|
||
LOG.info("\n=== B3: Discount Codes ===")
|
||
|
||
# Test 1: LAUNCH25 — 25% off service fee
|
||
LOG.info("\n --- LAUNCH25 (25%% percent) ---")
|
||
payload = make_payload(discount_code="LAUNCH25")
|
||
status, data = post_order(payload)
|
||
check("LAUNCH25: HTTP 201", status == 201, f"got {status}: {data.get('error', '')}")
|
||
if status == 201:
|
||
p = data.get("pricing", {})
|
||
expected_discount = round(SERVICE_FEE * 25 / 100)
|
||
check("LAUNCH25: discount_cents = 25% of service fee",
|
||
p.get("discount_cents") == expected_discount,
|
||
f"got {p.get('discount_cents')}, expected {expected_discount}")
|
||
check("LAUNCH25: total reduced by discount",
|
||
p.get("subtotal_cents", 0) < SERVICE_FEE + p.get("government_fee_cents", 0) + p.get("mailbox_annual_cents", 0),
|
||
f"total={p.get('subtotal_cents')}")
|
||
|
||
# Test 2: FIRST50 — $50 flat off service fee
|
||
LOG.info("\n --- FIRST50 ($50 flat) ---")
|
||
payload = make_payload(discount_code="FIRST50")
|
||
status, data = post_order(payload)
|
||
check("FIRST50: HTTP 201", status == 201, f"got {status}: {data.get('error', '')}")
|
||
if status == 201:
|
||
p = data.get("pricing", {})
|
||
check("FIRST50: discount_cents = 5000",
|
||
p.get("discount_cents") == 5000,
|
||
f"got {p.get('discount_cents')}")
|
||
|
||
# Test 3: INVALID999 — nonexistent code
|
||
LOG.info("\n --- INVALID999 (nonexistent) ---")
|
||
payload = make_payload(discount_code="INVALID999")
|
||
status, data = post_order(payload)
|
||
check("INVALID999: order still creates (201)", status == 201,
|
||
f"got {status}: {data.get('error', '')}")
|
||
if status == 201:
|
||
p = data.get("pricing", {})
|
||
check("INVALID999: discount_cents = 0",
|
||
p.get("discount_cents", 0) == 0,
|
||
f"got {p.get('discount_cents')}")
|
||
|
||
# Test 4: REF-EXAMPLE — referral partner code (15% off)
|
||
LOG.info("\n --- REF-EXAMPLE (15%% referral) ---")
|
||
payload = make_payload(discount_code="REF-EXAMPLE")
|
||
status, data = post_order(payload)
|
||
check("REF-EXAMPLE: HTTP 201", status == 201, f"got {status}: {data.get('error', '')}")
|
||
if status == 201:
|
||
p = data.get("pricing", {})
|
||
expected_discount = round(SERVICE_FEE * 15 / 100)
|
||
check("REF-EXAMPLE: discount = 15% of service fee",
|
||
p.get("discount_cents") == expected_discount,
|
||
f"got {p.get('discount_cents')}, expected {expected_discount}")
|
||
|
||
|
||
# ── B4: Own Canadian address ──────────────────────────────────────────────
|
||
|
||
def test_own_address():
|
||
LOG.info("\n=== B4: Own Canadian Address ===")
|
||
|
||
payload = make_payload(has_own_ca_address=True)
|
||
status, data = post_order(payload)
|
||
|
||
check("own-addr: HTTP 201", status == 201, f"got {status}: {data.get('error', '')}")
|
||
if status != 201:
|
||
return
|
||
|
||
p = data.get("pricing", {})
|
||
check("own-addr: mailbox_annual_cents = 0",
|
||
p.get("mailbox_annual_cents", -1) == 0,
|
||
f"got {p.get('mailbox_annual_cents')}")
|
||
|
||
# Verify in PG
|
||
order_number = data.get("order_number")
|
||
if order_number:
|
||
try:
|
||
import psycopg2
|
||
conn = psycopg2.connect(DEV_DB_URL)
|
||
cur = conn.cursor()
|
||
cur.execute(
|
||
"SELECT has_own_ca_address, amb_location_slug, amb_annual_price_cents, own_ca_company "
|
||
"FROM canada_crtc_orders WHERE order_number = %s",
|
||
(order_number,),
|
||
)
|
||
row = cur.fetchone()
|
||
conn.close()
|
||
if row:
|
||
check("own-addr PG: has_own_ca_address = true", row[0] is True, str(row[0]))
|
||
check("own-addr PG: amb_location_slug = null", row[1] is None, str(row[1]))
|
||
check("own-addr PG: amb_annual_price_cents = 0", row[2] == 0, str(row[2]))
|
||
check("own-addr PG: own_ca_company set", bool(row[3]), str(row[3]))
|
||
except Exception as e:
|
||
LOG.warning(" PG check skipped: %s", e)
|
||
|
||
|
||
# ── B5: Validation edge cases ─────────────────────────────────────────────
|
||
|
||
def test_validation():
|
||
LOG.info("\n=== B5: Validation Edge Cases ===")
|
||
|
||
# Missing customer_name
|
||
LOG.info("\n --- Missing customer_name ---")
|
||
payload = make_payload()
|
||
payload.pop("customer_name")
|
||
status, data = post_order(payload)
|
||
check("no customer_name: 400", status == 400, f"got {status}")
|
||
|
||
# Named company without name_choice_1
|
||
LOG.info("\n --- Named without name_choice_1 ---")
|
||
payload = make_payload(company_type="named")
|
||
payload.pop("company_name_choice1", None)
|
||
status, data = post_order(payload)
|
||
check("named no choice1: 400", status == 400, f"got {status}: {data.get('error', '')}")
|
||
|
||
# Tradename without trade_name
|
||
LOG.info("\n --- Tradename without trade_name ---")
|
||
payload = make_payload(company_type="numbered_tradename")
|
||
payload.pop("trade_name", None)
|
||
payload.pop("add_trade_name", None)
|
||
status, data = post_order(payload)
|
||
check("tradename no name: 400", status == 400, f"got {status}: {data.get('error', '')}")
|
||
|
||
# Invalid province
|
||
LOG.info("\n --- Invalid province ---")
|
||
payload = make_payload()
|
||
payload["incorporation_province"] = "QC"
|
||
status, data = post_order(payload)
|
||
check("invalid province: 400", status == 400, f"got {status}: {data.get('error', '')}")
|
||
|
||
|
||
# ── Cleanup ───────────────────────────────────────────────────────────────
|
||
|
||
def cleanup():
|
||
LOG.info("\n=== Cleanup ===")
|
||
try:
|
||
import psycopg2
|
||
conn = psycopg2.connect(DEV_DB_URL)
|
||
cur = conn.cursor()
|
||
cur.execute("DELETE FROM canada_crtc_orders WHERE customer_email LIKE 'pricing-test+%%@performancewest.net'")
|
||
deleted = cur.rowcount
|
||
conn.commit()
|
||
conn.close()
|
||
LOG.info(" Deleted %d test order(s)", deleted)
|
||
except Exception as e:
|
||
LOG.warning(" Cleanup failed: %s", e)
|
||
|
||
|
||
# ── Main ──────────────────────────────────────────────────────────────────
|
||
|
||
def main():
|
||
parser = argparse.ArgumentParser(description="CRTC API pricing matrix test")
|
||
parser.add_argument("--keep", action="store_true", help="Don't delete test orders after run")
|
||
args = parser.parse_args()
|
||
|
||
LOG.info("=" * 60)
|
||
LOG.info(" CRTC API PRICING MATRIX TEST")
|
||
LOG.info(" Target: %s", API_URL)
|
||
LOG.info("=" * 60)
|
||
|
||
test_base_pricing()
|
||
test_expedited_pricing()
|
||
test_discount_codes()
|
||
test_own_address()
|
||
test_validation()
|
||
|
||
if not args.keep:
|
||
cleanup()
|
||
|
||
LOG.info("\n" + "=" * 60)
|
||
LOG.info(" RESULTS: %d passed, %d failed, %d total", PASS, FAIL, PASS + FAIL)
|
||
LOG.info("=" * 60)
|
||
|
||
if FAIL > 0:
|
||
LOG.info("\n Failed checks:")
|
||
for r in RESULTS:
|
||
if not r["ok"]:
|
||
LOG.info(" ✗ %s — %s", r["label"], r["detail"])
|
||
|
||
LOG.info("")
|
||
if FAIL == 0:
|
||
LOG.info(" ALL CHECKS PASSED")
|
||
else:
|
||
LOG.info(" %d CHECK(S) FAILED", FAIL)
|
||
LOG.info("=" * 60)
|
||
|
||
sys.exit(1 if FAIL > 0 else 0)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|