MMD
f7b1f7f751
- main.py: global get_current_user dependency on all routers except /auth/login and /health; prod fail-fast startup guard rejects wildcard CORS, default credentials, DEV_LOGIN_ENABLED - inventory/pm: X-User-Role header identity fallback now only active when PROCUREMENT_DEV_LOGIN_ENABLED=true (was unconditional in production) - item_decisions writes + attachment deletes: require_role( procurement_manager, admin) — require_role was defined but never used - config.py: ENVIRONMENT flag, comma-separated CORS_ORIGINS, no default Casdoor client secret/admin password, DEV_LOGIN_ENABLED=false - frontend: authenticated file downloads (Bearer + blob) — <a href> links cannot send the token; FileList/QuoteDetail/AcceptModal/EditModal/ PurchaseHistory converted; DEV_LOGIN_ENABLED build arg defaults false - smoke tests authenticate via POST /auth/login (SMOKE_USER/SMOKE_PASSWORD) - traefik: security headers, API rate limit, 12MB body cap (prod overlay) - nginx client_max_body_size aligned to 12MB
143 lines
6.3 KiB
Python
143 lines
6.3 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)
|
|
|
|
# ---------- auth: every endpoint now requires a Casdoor JWT ----------
|
|
import os
|
|
|
|
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})
|
|
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)
|