ERP/backend/tools/smoke_test.py

"""End-to-end smoke tests against a running server (http://127.0.0.1:8000).

Dataset-independent: assertions verify BEHAVIOR (status codes, envelopes,
shapes, filtering) rather than exact seed counts, so the suite works with
any seed version. Run from the api container or a host with access:

Run:  python tools/smoke_test.py
Env:  SMOKE_USER / SMOKE_PASSWORD (default tester / 123456 — seeded users)
"""
import os
import sys

import httpx

BASE = "http://127.0.0.1:8000/api/v1"
PASS, FAIL = 0, 0


def check(name, ok, extra=""):
    global PASS, FAIL
    if ok:
        PASS += 1
        print(f"  PASS  {name}")
    else:
        FAIL += 1
        print(f"  FAIL  {name}  {extra}")


c = httpx.Client(base_url=BASE, timeout=30)

# ---------- auth: every endpoint now requires a Casdoor JWT ----------
r = c.post("/auth/login", json={
    "username": os.environ.get("SMOKE_USER", "tester"),
    "password": os.environ.get("SMOKE_PASSWORD", "123456"),
})
check("POST /auth/login", r.status_code == 200, r.text[:200])
if r.status_code == 200:
    c.headers["Authorization"] = f"Bearer {r.json()['data']['access_token']}"

# ---------- catalog ----------
r = c.get("/units", params={"page_size": 5})
units = r.json()["data"] if r.status_code == 200 else []
check("GET /units", r.status_code == 200 and len(units) == 5 and "si_factor" in units[0])

if units:
    dim = units[0]["dimension"]
    r = c.get("/units", params={"dimension": dim, "page_size": 100})
    body = r.json()
    check(
        "GET /units?dimension filter",
        r.status_code == 200 and body["meta"]["total"] >= 1
        and all(u["dimension"] == dim for u in body["data"]),
        r.text[:200],
    )

r = c.get("/brands")
brands = r.json()["data"] if r.status_code == 200 else []
check("GET /brands", r.status_code == 200 and brands and r.json()["meta"]["total"] >= 1)

# category q-search (only when categories exist)
r = c.get("/material-categories", params={"page_size": 100})
cats = r.json()["data"] if r.status_code == 200 else []
check("GET /material-categories", r.status_code == 200 and isinstance(cats, list))
if cats:
    needle = cats[0]["name"][:3]
    r = c.get("/material-categories", params={"q": needle})
    check(
        "GET /material-categories?q (unicode search)",
        r.status_code == 200 and r.json()["meta"]["total"] >= 1, r.text[:200],
    )

# materials: q-search against a REAL seeded name (fetched, not hard-coded)
r = c.get("/materials", params={"page_size": 100})
materials = r.json()["data"] if r.status_code == 200 else []
check("GET /materials", r.status_code == 200 and materials and r.json()["meta"]["total"] >= 1)

if materials:
    target = materials[0]
    needle = target["name"].split()[0]
    r = c.get("/materials", params={"q": needle})
    check(
        "GET /materials?q (unicode search)",
        r.status_code == 200 and r.json()["meta"]["total"] >= 1, r.text[:200],
    )

    r = c.get("/materials", params={"expand": "category,brands,canonical_unit,attributes", "q": needle})
    expanded = [m for m in r.json().get("data", []) if m["id"] == target["id"]]
    if r.status_code == 200 and expanded:
        m = expanded[0]
        check(
            "GET /materials expand",
            "category" in m and "brands" in m and "canonical_unit" in m and "attributes" in m,
        )
        check("expand category shape", m["category"].keys() == {"id", "slug", "name"})
        check("expand brands shape", all(set(b) >= {"id", "name", "catalog_code", "is_active"} for b in m["brands"]))
        check("expand canonical_unit shape", m["canonical_unit"].keys() == {"id", "code", "name_fa"})
        check("expand attributes EAV flattened", all(set(a) >= {"attribute_key", "label", "data_type", "value", "unit"} for a in m["attributes"]))
    else:
        check("GET /materials expand", False, r.text[:300])

# ---------- EAV filters (positive filter only when the seed has values) ----------
r = c.get("/attribute-definitions", params={"page_size": 100})
attr_defs = r.json()["data"] if r.status_code == 200 else []

if attr_defs and materials:
    r = c.get("/materials", params={"expand": "attributes", "page_size": 100})
    with_attrs = [m for m in r.json().get("data", []) if m.get("attributes")]
    if with_attrs:
        m0 = with_attrs[0]
        key = m0["attributes"][0]["attribute_key"]
        value = str(m0["attributes"][0]["value"])
        r = c.get("/materials", params={f"attr_{key}": value, "expand": "attributes"})
        check(
            f"GET /materials?attr_{key} filter",
            r.status_code == 200 and any(x["id"] == m0["id"] for x in r.json().get("data", [])),
            r.text[:200],
        )
    else:
        print("  SKIP  EAV positive filter — seed defines no material attribute values")
else:
    print("  SKIP  EAV positive filter — seed defines no attribute definitions")

# unknown attr filter -> 400 (dataset-independent)
r = c.get("/materials", params={"attr_nonexistent": "1"})
check("unknown attr filter -> 400", r.status_code == 400 and r.json()["error"]["code"] == "VALIDATION_ERROR")

r = c.get("/materials?page_size=999")
check("page_size>100 -> 400", r.status_code == 400)

# brand filter via a brand that actually has material links
r = c.get("/brand-materials", params={"page_size": 100})
brand_links = r.json()["data"] if r.status_code == 200 else []
if brand_links:
    brand_id = brand_links[0]["brand_id"]
    r = c.get("/materials", params={"brand_id": brand_id})
    check("GET /materials?brand_id=", r.status_code == 200 and r.json()["meta"]["total"] >= 1, r.text[:200])
    r = c.get("/brand-materials", params={"brand_id": brand_id})
    check("GET /brand-materials?brand_id", r.status_code == 200 and r.json()["meta"]["total"] >= 1, r.text[:200])
else:
    check("GET /materials?brand_id=", False, "seed has no brand-material links")

# ---------- suppliers / projects ----------
r = c.get("/suppliers")
suppliers = r.json()["data"] if r.status_code == 200 else []
check("GET /suppliers", r.status_code == 200 and suppliers)

r = c.get("/projects")
projects = r.json()["data"] if r.status_code == 200 else []
check("GET /projects", r.status_code == 200 and projects)

r = c.get("/attribute-definitions")
attr_shape_ok = r.status_code == 200 and all(
    set(a) >= {"attribute_key", "label", "data_type"} for a in r.json().get("data", [])
)
check("GET /attribute-definitions", attr_shape_ok, r.text[:200])

if cats:
    r = c.get(f"/material-categories/{cats[0]['id']}/attributes")
    check(
        "GET category attributes (nested)",
        r.status_code == 200 and isinstance(r.json()["data"], list)
        and all("attribute" in a for a in r.json()["data"]),
        r.text[:200],
    )

# ---------- quotes ----------
r = c.get("/quotes", params={"page_size": 5})
quotes = r.json()["data"] if r.status_code == 200 else []
check("GET /quotes", r.status_code == 200 and quotes and r.json()["meta"]["total"] >= 1)

r = c.get("/quotes", params={"expand": "line_items,termsheets,events,supplier,project"})
accepted = [q for q in r.json().get("data", []) if q.get("line_items")] if r.status_code == 200 else []
if accepted:
    q = accepted[0]
    li = q["line_items"][0]
    check(
        "GET /quotes expand full shape",
        "supplier" in q and "project" in q and "termsheets" in q and "events" in q
        and li["material"]["category"].keys() == {"id", "slug", "name"}
        and "brands" in li["material"] and "specs" in li and "decision" in li
        and "termsheet_ids" in li and "total_price" in li and "unit_price" in li,
        r.text[:400],
    )
    check("events sorted ASC", [e["event_type"] for e in q["events"]] == sorted([e["event_type"] for e in q["events"]]))
    if q["termsheets"]:
        check("termsheet has schedule_steps", "schedule_steps" in q["termsheets"][0])
else:
    check("GET /quotes expand full shape", False, r.text[:400])

r = c.get("/quotes/stats")
stats_ok = r.status_code == 200 and "average_lead_time_days" in r.json()["data"]
if quotes:
    stats_ok = stats_ok and r.json()["data"]["total"] == r.json()["data"]["offered"] + \
        r.json()["data"]["accepted"] + r.json()["data"]["arrived"] + \
        r.json()["data"]["rejected"] + r.json()["data"]["expired"]
check("GET /quotes/stats", stats_ok, r.text[:300])

r = c.get("/quotes/stats", params={"status": "offered"})  # status filter not part of stats contract; ensure no crash
check("GET /quotes/stats (extra param ignored)", r.status_code == 200)

if materials:
    needle = materials[0]["name"].split()[0]
    r = c.get("/quotes", params={"q": needle})
    check("GET /quotes?q (searches material names)", r.status_code == 200, r.text[:300])

if projects:
    r = c.get(f"/projects/{projects[0]['id']}/stats")
    check(
        "GET /projects/{id}/stats",
        r.status_code == 200 and set(r.json()["data"]) >= {"total_quotes", "offered", "accepted", "arrived", "rejected", "expired"},
        r.text[:300],
    )

print(f"\n===== {PASS} passed, {FAIL} failed =====")
sys.exit(1 if FAIL else 0)