qingsongping-AIDevelop/scripts/verify_salary_calculator_blackbox.py

"""N05 salary calculator black-box verification.

Run only against an isolated, disposable MySQL container and an app wired to it.
Required environment variables:
  BASE_URL=http://127.0.0.1:<PORT>/api/v1
  DB_HOST, DB_PORT, DB_USERNAME, DB_PASSWORD, DB_NAME
"""
import os
import sys
import time

import pymysql
import requests

BASE = os.environ.get("BASE_URL", "").rstrip("/")
DB = {
    "host": os.environ["DB_HOST"],
    "port": int(os.environ["DB_PORT"]),
    "user": os.environ["DB_USERNAME"],
    "password": os.environ["DB_PASSWORD"],
    "database": os.environ["DB_NAME"],
}
if not BASE:
    raise SystemExit("BASE_URL is required")


def db():
    return pymysql.connect(**DB, autocommit=True)


def sql_one(query, args=()):
    with db() as conn:
        with conn.cursor() as cur:
            cur.execute(query, args)
            return cur.fetchone()


def sql_all(query, args=()):
    with db() as conn:
        with conn.cursor() as cur:
            cur.execute(query, args)
            return cur.fetchall()


def login_student(phone):
    r = requests.post(f"{BASE}/auth/sms/send", params={"phone": phone, "purpose": "login"}, timeout=10)
    assert r.status_code == 200, r.text
    code = None
    for _ in range(20):
        row = sql_one(
            "SELECT code FROM sms_verification_codes "
            "WHERE phone=%s AND purpose='login' AND used=0 ORDER BY id DESC LIMIT 1",
            (phone,),
        )
        if row:
            code = row[0]
            break
        time.sleep(0.25)
    assert code, "mock SMS code not found"
    r = requests.post(f"{BASE}/auth/login", json={"phone": phone, "code": code, "purpose": "login"}, timeout=10)
    assert r.status_code == 200, r.text
    return r.json()["data"]["accessToken"]


def main():
    versions = {row[0] for row in sql_all("SELECT version FROM flyway_schema_history")}
    assert "40" in versions and "41" in versions, versions

    assert sql_one("SELECT COUNT(*) FROM salary_city_policy")[0] == 359
    levels = dict(sql_all("SELECT data_level, COUNT(*) FROM salary_city_policy GROUP BY data_level"))
    assert levels == {"CITY": 12, "PROVINCE_FALLBACK": 347}, levels

    phone = os.environ.get("BLACKBOX_PHONE", f"139{int(time.time()) % 100000000:08d}")
    token = login_student(phone)
    headers = {"Authorization": f"Bearer {token}", "Content-Type": "application/json"}

    cities = requests.get(f"{BASE}/salary-calculator/cities", headers=headers, timeout=10)
    assert cities.status_code == 200 and len(cities.json()["data"]) == 359, cities.text

    city = requests.get(f"{BASE}/salary-calculator/cities/110000", headers=headers, timeout=10)
    assert city.status_code == 200 and city.json()["data"]["dataLevel"] == "CITY", city.text

    payload = {
        "cityCode": "110000",
        "grossMonthlySalary": 10000.00,
        "monthIndex": 1,
        "specialAdditionalDeduction": 0.00,
        "expectedPolicyVersion": city.json()["data"]["policyVersion"],
        "idempotencyKey": f"blackbox-{phone}",
        "overrides": None,
    }
    first = requests.post(f"{BASE}/salary-calculator/calculations", headers=headers, json=payload, timeout=10)
    assert first.status_code == 200, first.text
    history_id = first.json()["data"]["historyId"]
    assert float(first.json()["data"]["result"]["netSalary"]) == 7667.5, first.text

    replay = requests.post(f"{BASE}/salary-calculator/calculations", headers=headers, json=payload, timeout=10)
    assert replay.status_code == 200 and replay.json()["data"]["historyId"] == history_id, replay.text

    conflict_payload = dict(payload)
    conflict_payload["grossMonthlySalary"] = 12000.00
    conflict = requests.post(f"{BASE}/salary-calculator/calculations", headers=headers, json=conflict_payload, timeout=10)
    assert conflict.status_code == 409 and conflict.json()["code"] == 9804, conflict.text

    stale_payload = dict(payload)
    stale_payload["idempotencyKey"] = "stale-" + phone
    stale_payload["expectedPolicyVersion"] = "stale-version"
    stale = requests.post(f"{BASE}/salary-calculator/calculations", headers=headers, json=stale_payload, timeout=10)
    assert stale.status_code == 409 and stale.json()["code"] == 9805, stale.text
    assert stale.json()["data"]["latestPolicyVersion"] == city.json()["data"]["policyVersion"]

    history = requests.get(f"{BASE}/salary-calculator/calculations", headers=headers, timeout=10)
    assert history.status_code == 200 and history.json()["data"]["total"] == 1, history.text

    detail = requests.get(f"{BASE}/salary-calculator/calculations/{history_id}", headers=headers, timeout=10)
    assert detail.status_code == 200 and detail.json()["data"]["policySnapshot"]["dataLevel"] == "CITY", detail.text

    deleted = requests.delete(f"{BASE}/salary-calculator/calculations/{history_id}", headers=headers, timeout=10)
    assert deleted.status_code == 200, deleted.text
    missing = requests.get(f"{BASE}/salary-calculator/calculations/{history_id}", headers=headers, timeout=10)
    assert missing.status_code == 404 and missing.json()["code"] == 9803, missing.text

    print("PASS: N05 salary calculator disposable-MySQL black box")


if __name__ == "__main__":
    try:
        main()
    except Exception as exc:
        print(f"FAIL: {type(exc).__name__}: {exc}", file=sys.stderr)
        raise