Files

132 lines
6.0 KiB
Python

"""End-to-end smoke tests against a running server (http://127.0.0.1:8000).
Run: .venv/Scripts/python.exe tools/smoke_test.py
"""
import sys
from pprint import pprint
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)
# ---------- catalog ----------
r = c.get("/units", params={"page_size": 5})
check("GET /units", r.status_code == 200 and len(r.json()["data"]) == 5 and "si_factor" in r.json()["data"][0])
r = c.get("/units", params={"dimension": "mass"})
check("GET /units?dimension=mass", r.status_code == 200 and r.json()["meta"]["total"] == 3)
r = c.get("/brands")
check("GET /brands", r.status_code == 200 and r.json()["meta"]["total"] == 10)
r = c.get("/material-categories", params={"q": "سیمان"})
check("GET /material-categories?q", r.status_code == 200 and r.json()["meta"]["total"] >= 1)
r = c.get("/materials", params={"q": "میلگرد A3 سایز ۱۴"})
check("GET /materials?q (unicode)", 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": "میلگرد A3 سایز ۱۴"})
if r.status_code == 200 and r.json()["data"]:
m = r.json()["data"][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 has catalog_code+is_active", 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])
# dynamic EAV filter: attr_rebar_diameter=14
r = c.get("/materials", params={"attr_rebar_diameter": "14", "expand": "attributes"})
if r.status_code == 200:
data = r.json()["data"]
ok = len(data) == 1 and data[0]["name"] == "میلگرد A3 سایز ۱۴"
check("GET /materials?attr_rebar_diameter=14", ok, r.text[:200])
else:
check("GET /materials?attr_rebar_diameter=14", False, r.text[:200])
# EAV boolean filter: attr_rebar_ribbed=true
r = c.get("/materials", params={"attr_rebar_ribbed": "true"})
check("GET /materials?attr_rebar_ribbed=true", r.status_code == 200 and r.json()["meta"]["total"] == 1, r.text[:200])
# brand filter
r = c.get("/brands", params={"q": "فولاد مبارکه"})
brand_id = r.json()["data"][0]["id"]
r = c.get("/materials", params={"brand_id": brand_id})
check("GET /materials?brand_id=", r.status_code == 200 and r.json()["meta"]["total"] == 3, r.text[:200])
# unknown attr filter -> 400
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)
r = c.get("/suppliers")
check("GET /suppliers", r.status_code == 200 and r.json()["meta"]["total"] == 6)
r = c.get("/projects")
check("GET /projects", r.status_code == 200 and r.json()["meta"]["total"] == 5)
r = c.get("/attribute-definitions", params={"expand": "unit"})
check("GET /attribute-definitions", r.status_code == 200 and r.json()["meta"]["total"] == 28, r.text[:200])
r = c.get("/material-categories/{cid}/attributes".format(cid=c.get("/material-categories", params={"q": "میلگرد"}).json()["data"][0]["id"]))
check("GET category attributes (nested attribute)", r.status_code == 200 and len(r.json()["data"]) == 3 and "attribute" in r.json()["data"][0])
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"] == 3)
# ---------- quotes ----------
r = c.get("/quotes", params={"page_size": 5})
check("GET /quotes", r.status_code == 200 and r.json()["meta"]["total"] == 7, r.text[:300])
r = c.get("/quotes", params={"expand": "line_items,termsheets,events,supplier,project", "status": "accepted"})
if r.status_code == 200 and r.json()["data"]:
q = r.json()["data"][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"]]))
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")
check("GET /quotes/stats", r.status_code == 200 and r.json()["data"]["total"] == 7 and "average_lead_time_days" in r.json()["data"], 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)
r = c.get("/quotes", params={"q": "میلگرد"})
check("GET /quotes?q (searches material names)", r.status_code == 200 and r.json()["meta"]["total"] >= 1, r.text[:300])
r = c.get("/projects/{pid}/stats".format(pid=c.get("/projects", params={"q": "ونوس"}).json()["data"][0]["id"]))
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)