justin/new-site
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Initial commit — Performance West telecom compliance platform

Browse source 
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>
This commit is contained in:
justin 2026-04-27 06:54:22 -05:00
commit f8cd37ac8c
1823 changed files with 145167 additions and 0 deletions
Download patch file Download diff file Expand all files Collapse all files

475
scripts/tests/e2e_crtc_pricing.py Normal file
View file

@ -0,0 +1,475 @@
"""
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()