ERP/backend/tools/smoke_test_misc.py

"""Coverage pass 3: remaining CRUD corners not exercised by the other two suites."""
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 ----------
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']}"

# ---------- units ----------
r = c.post("/units", json={"code": "test-unit", "name_fa": "واحد تست", "dimension": "count", "si_factor": 1.0})
check("POST /units", r.status_code == 201 and r.json()["data"]["si_factor"] == 1.0, r.text[:200])
uid = r.json()["data"]["id"]
r = c.post("/units", json={"code": "test-unit", "si_factor": 1.0})
check("POST /units duplicate code -> 409", r.status_code == 409, r.text[:200])
r = c.patch(f"/units/{uid}", json={"name_fa": "واحد تست ویرایش", "si_factor": 2.0})
check("PATCH /units", r.status_code == 200 and r.json()["data"]["si_factor"] == 2.0)
r = c.delete(f"/units/{uid}")
check("DELETE /units (unused)", r.status_code == 200)
r = c.delete("/units/1")  # 'm' is referenced everywhere
check("DELETE /units in use -> 409", r.status_code == 409 and "استفاده" in r.json()["error"]["message"])

# ---------- brands ----------
r = c.post("/brands", json={"name": "برند تست", "country": "ایران"})
check("POST /brands", r.status_code == 201, r.text[:200])
bid = r.json()["data"]["id"]
r = c.post("/brands", json={"name": "برند تست", "country": "ایران"})
check("POST /brands duplicate -> 409", r.status_code == 409)
r = c.patch(f"/brands/{bid}", json={"name": "برند تست ۲"})
check("PATCH /brands", r.status_code == 200 and r.json()["data"]["name"] == "برند تست ۲")
r = c.delete(f"/brands/{bid}")
check("DELETE /brands (unused)", r.status_code == 200)
_bml = c.get("/brand-materials", params={"page_size": 1}).json()["data"]
if _bml:
    brand = {"id": _bml[0]["brand_id"]}
else:
    brand = c.get("/brands", params={"page_size": 1}).json()["data"][0]
r = c.delete(f"/brands/{brand['id']}")
check("DELETE /brands with associations -> 409", r.status_code == 409, r.text[:200])

# ---------- categories ----------
r = c.post("/material-categories", json={"slug": "test-cat", "name": "دسته تست"})
if r.status_code == 409:  # rerun safety — fetch the existing test category
    cid = c.get("/material-categories", params={"q": "test-cat"}).json()["data"][0]["id"]
    check("POST /material-categories (exists from previous run)", True)
else:
    check("POST /material-categories", r.status_code == 201, r.text[:200])
    cid = r.json()["data"]["id"]
r = c.patch(f"/material-categories/{cid}", json={"name": "دسته تست ۲", "is_active": False})
check("PATCH /material-categories", r.status_code == 200 and r.json()["data"]["is_active"] is False)
r = c.delete(f"/material-categories/{cid}")
check("DELETE /material-categories (empty)", r.status_code == 200)
_mat0 = c.get("/materials", params={"page_size": 1, "expand": "category"}).json()["data"][0]
cat = _mat0["category"]
r = c.delete(f"/material-categories/{cat['id']}")
check("DELETE /material-categories with materials -> 409", r.status_code == 409, r.text[:200])

# ---------- attribute definitions ----------
r = c.post("/attribute-definitions", json={"attribute_key": "test_attr", "label": "ویژگی تست", "data_type": "enum",
                                           "enum_values": ["الف", "ب"]})
check("POST /attribute-definitions (enum)", r.status_code == 201, r.text[:200])
adid = r.json()["data"]["id"]
r = c.post("/attribute-definitions", json={"attribute_key": "test_attr", "label": "ت", "data_type": "string"})
check("POST dup attribute_key -> 409", r.status_code == 409)
r = c.post("/attribute-definitions", json={"attribute_key": "test_attr2", "label": "ت", "data_type": "enum"})
check("POST enum without enum_values -> 400", r.status_code == 400 and r.json()["error"]["field"] == "enum_values")
r = c.patch(f"/attribute-definitions/{adid}", json={"label": "ویژگی تست جدید"})
check("PATCH /attribute-definitions", r.status_code == 200 and r.json()["data"]["label"] == "ویژگی تست جدید")
r = c.patch(f"/attribute-definitions/{adid}", json={"data_type": "number"})
check("PATCH data_type with no values -> OK", r.status_code == 200 and r.json()["data"]["data_type"] == "number", r.text[:200])

# use a DEDICATED test material so the seed material's EAV values stay intact
# (dataset-independent: any category; rerun reuses the existing test material)
cat = c.get("/material-categories", params={"page_size": 1}).json()["data"][0]
unit_kg = c.get("/units", params={"q": "kg"}).json()["data"][0]["id"]
r = c.post("/materials", json={"category_id": cat["id"], "name": "متریال تست EAV", "canonical_unit_id": unit_kg})
if r.status_code == 409:  # rerun
    tmat = c.get("/materials", params={"q": "متریال تست EAV"}).json()["data"][0]
    check("create test material for EAV tests (exists from previous run)", True)
else:
    check("create test material for EAV tests", r.status_code == 201, r.text[:200])
    tmat = r.json()["data"]
mat = tmat
r = c.put(f"/materials/{mat['id']}/attribute-values", json={"attributes": [
    {"attribute_definition_id": adid, "value": 5},
]})
check("PUT attribute-values on material", r.status_code == 200, r.text[:200])
r = c.patch(f"/attribute-definitions/{adid}", json={"data_type": "string"})
check("PATCH data_type with existing values -> 409", r.status_code == 409, r.text[:200])
r = c.delete(f"/attribute-definitions/{adid}")
check("DELETE attribute-definition with values -> 409", r.status_code == 409)
r = c.get(f"/materials/{mat['id']}/attribute-values")
check("GET attribute-values flattened", r.status_code == 200 and r.json()["data"][0]["value"] == 5.0, r.text[:200])
r = c.put(f"/materials/{mat['id']}/attribute-values", json={"attributes": []})
check("PUT attribute-values empty -> deletes all", r.status_code == 200)
r = c.delete(f"/attribute-definitions/{adid}")
check("DELETE attribute-definition (no values) -> 200", r.status_code == 200, r.text[:200])

# ---------- category attributes ----------
# use a FRESH definition so the pair is not already in the seed junction
r = c.post("/attribute-definitions", json={"attribute_key": "test_ca_attr", "label": "ویژگی دسته", "data_type": "string"})
if r.status_code == 409:  # rerun safety — fetch the existing definition
    newdef = next(d for d in c.get("/attribute-definitions", params={"page_size": 100}).json()["data"]
                  if d["attribute_key"] == "test_ca_attr")
else:
    newdef = r.json()["data"]
r = c.post(f"/material-categories/{cat['id']}/attributes", json={"attribute_definition_id": newdef["id"], "is_required": True, "sort_order": 9})
check("POST category attribute", r.status_code == 201 and r.json()["data"]["attribute"]["attribute_key"] == "test_ca_attr", r.text[:200])
r = c.post(f"/material-categories/{cat['id']}/attributes", json={"attribute_definition_id": newdef["id"]})
check("POST category attribute duplicate -> 409", r.status_code == 409)
r = c.patch(f"/material-categories/{cat['id']}/attributes/{newdef['id']}", json={"is_required": False, "sort_order": 5})
check("PATCH category attribute", r.status_code == 200 and r.json()["data"]["sort_order"] == 5)
r = c.delete(f"/material-categories/{cat['id']}/attributes/{newdef['id']}")
check("DELETE category attribute", r.status_code == 200)
r = c.delete(f"/attribute-definitions/{newdef['id']}")
check("cleanup: DELETE unused attribute-definition", r.status_code == 200)

# ---------- material units ----------
r = c.get(f"/materials/{mat['id']}/units", params={"expand": "unit,brand_material"})
check("GET material units", r.status_code == 200, r.text[:200])
_unit_other = next(u for u in c.get("/units", params={"page_size": 100}).json()["data"] if u["code"] != "kg")
r = c.post(f"/materials/{mat['id']}/units", json={"unit_id": _unit_other["id"], "conversion_factor": 0.001, "is_default": False})
check("POST material unit", r.status_code == 201, r.text[:200])
muid = r.json()["data"]["id"]
r = c.post(f"/materials/{mat['id']}/units", json={"unit_id": _unit_other["id"], "conversion_factor": 2})
check("POST material unit duplicate -> 409", r.status_code == 409, r.text[:200])
r = c.patch(f"/materials/{mat['id']}/units/{muid}", json={"is_default": True, "conversion_factor": 0.002})
check("PATCH material unit (is_default)", r.status_code == 200 and r.json()["data"]["is_default"] is True)
r = c.delete(f"/materials/{mat['id']}/units/{muid}")
check("DELETE material unit", r.status_code == 200)

# ---------- brand materials ----------
# dataset-independent: any material + a freshly created brand (no existing link)
m2 = c.get("/materials", params={"page_size": 1}).json()["data"][0]
r = c.post("/brands", json={"name": "برند لینک تست", "country": "ایران"})
b2 = r.json()["data"] if r.status_code == 201 else c.get("/brands", params={"q": "برند لینک تست"}).json()["data"][0]
r = c.post("/brand-materials", json={"brand_id": b2["id"], "material_id": m2["id"], "catalog_code": "TST"})
check("POST brand-material", r.status_code == 201, r.text[:200])
bmid = r.json()["data"]["id"]
r = c.post("/brand-materials", json={"brand_id": b2["id"], "material_id": m2["id"]})
check("POST brand-material duplicate -> 409", r.status_code == 409)
r = c.patch(f"/brand-materials/{bmid}", json={"catalog_code": "TST2", "is_active": False})
check("PATCH brand-material", r.status_code == 200 and r.json()["data"]["catalog_code"] == "TST2")
r = c.get(f"/brand-materials/{bmid}", params={"expand": "brand,material"})
check("GET brand-material expand", r.status_code == 200 and "brand" in r.json()["data"] and "material" in r.json()["data"])
r = c.delete(f"/brand-materials/{bmid}")
check("DELETE brand-material", r.status_code == 200)
r = c.delete(f"/brands/{b2['id']}")
check("cleanup: DELETE link-test brand", r.status_code == 200, r.text[:200])

# ---------- supplier/project patches ----------
s1 = c.get("/suppliers", params={"page_size": 1}).json()["data"][0]
r = c.patch(f"/suppliers/{s1['id']}", json={"phone": "021-00000000"})
check("PATCH /suppliers", r.status_code == 200 and r.json()["data"]["phone"] == "021-00000000")
r = c.patch(f"/suppliers/{s1['id']}", json={"phone": "021-55123456"})
p1 = c.get("/projects", params={"page_size": 1}).json()["data"][0]
r = c.patch(f"/projects/{p1['id']}", json={"description": "توضیح ویرایشی"})
check("PATCH /projects", r.status_code == 200 and r.json()["data"]["description"] == "توضیح ویرایشی")
r = c.patch(f"/projects/{p1['id']}", json={"description": p1.get("description")})

# ---------- quote PATCH metadata ----------
r = c.get("/quotes", params={"status": "offered", "page_size": 1})
qid = r.json()["data"][0]["id"]
r = c.patch(f"/quotes/{qid}", json={"notes": "یادداشت ویرایشی", "is_official": False})
check("PATCH /quotes metadata", r.status_code == 200 and r.json()["data"]["is_official"] is False, r.text[:200])
r = c.patch(f"/quotes/{qid}", json={"notes": None, "is_official": True})
check("PATCH /quotes restore", r.status_code == 200)
r = c.patch(f"/quotes/{qid}", json={"status": "arrived"})
check("PATCH /quotes forward offered->arrived -> 409", r.status_code == 409)
r = c.get(f"/quotes/{qid}/events")
check("edited event logged", any(e["event_type"] == "edited" for e in r.json()["data"]))

# ---------- termsheet delete rules ----------
# termsheet-count tolerance: this suite needs >=2 termsheets on the picked quote
r = c.get(f"/quotes/{qid}/termsheets")
ts = r.json()["data"]
while len(ts) < 2:
    r = c.post(f"/quotes/{qid}/termsheets", json={"label": "ثانویه", "schedule_steps": [{"due_date": "2026-12-01", "amount": 10}]})
    check("POST additional termsheet", r.status_code == 201, r.text[:200])
    ts = c.get(f"/quotes/{qid}/termsheets").json()["data"]
r = c.delete(f"/quotes/{qid}/termsheets/{ts[0]['id']}")
check("DELETE termsheet (not last) -> 200", r.status_code == 200, r.text[:200])
_remaining = c.get(f"/quotes/{qid}/termsheets").json()["data"]
r = c.delete(f"/quotes/{qid}/termsheets/{_remaining[-1]['id']}")
check("DELETE last termsheet -> 409", r.status_code == 409, r.text[:200])
# restore a second termsheet for later tests
r = c.post(f"/quotes/{qid}/termsheets", json={"label": "ثانویه ۲", "schedule_steps": [{"due_date": "2026-12-01", "amount": 10}]})
check("POST third termsheet", r.status_code == 201)

# ---------- schedule steps delete-last rule ----------
r = c.get(f"/quotes/{qid}/termsheets")
ts0 = r.json()["data"][0]
r = c.get(f"/quotes/{qid}/termsheets/{ts0['id']}/schedule-steps")
steps = r.json()["data"]
if len(steps) > 1:
    r = c.delete(f"/quotes/{qid}/termsheets/{ts0['id']}/schedule-steps/{steps[0]['id']}")
    check("DELETE schedule-step (not last)", r.status_code == 200, r.text[:200])
r = c.delete(f"/quotes/{qid}/termsheets/{ts0['id']}/schedule-steps/{steps[-1]['id']}")
check("DELETE last schedule-step -> 409", r.status_code == 409, r.text[:200])

# ---------- decision rules (need a PENDING decision — rerun-safe pick) ----------
_pending = None
for _qo in c.get("/quotes", params={"status": "offered", "page_size": 20, "expand": "line_items"}).json()["data"]:
    for _lix in _qo["line_items"]:
        if _lix.get("decision") and _lix["decision"]["status"] == "pending":
            _pending = (_qo["id"], _lix)
            break
    if _pending:
        break
if _pending:
    _qid2, li = _pending
    r = c.put(f"/quotes/{_qid2}/line-items/{li['id']}/decision", json={"status": "chosen", "chosen_item_termsheet_id": "00000000-0000-0000-0000-000000000000"})
    check("PUT decision chosen with foreign link -> 400", r.status_code == 400, r.text[:200])
    r = c.put(f"/quotes/{_qid2}/line-items/{li['id']}/decision", json={"status": "rejected"})
    check("PUT decision rejected", r.status_code == 200 and r.json()["data"]["status"] == "rejected", r.text[:200])
    r = c.put(f"/quotes/{_qid2}/line-items/{li['id']}/decision", json={"status": "chosen"})
    check("PUT decision rejected->chosen -> 409", r.status_code == 409, r.text[:200])
    r = c.patch(f"/quotes/{_qid2}/line-items/{li['id']}/decision/status", json={"status": "pending"})
    # pending is NOT an accepted payload value — the endpoint only transitions to chosen/rejected
    check("PATCH decision status to pending -> 400", r.status_code == 400, r.text[:200])
else:
    print("  SKIP  decision rules — no offered quote with a pending decision")

# ---------- item termsheets ----------
# termsheet-count tolerance marker
if _pending:
    _its_quote, li = _pending
    r = c.get(f"/quotes/{_its_quote}/line-items/{li['id']}/termsheets")
    its = r.json().get("data", [])
    if its:
        r = c.patch(f"/quotes/{_its_quote}/line-items/{li['id']}/termsheets/{its[0]['id']}", json={"unit_price": 9999})
        check("PATCH item-termsheet", r.status_code == 200 and r.json()["data"]["unit_price"] == 9999.0, r.text[:200])
        r = c.patch(f"/quotes/{_its_quote}/line-items/{li['id']}/termsheets/{its[0]['id']}", json={"unit_price": its[0]["unit_price"]})
    else:
        print("  SKIP  PATCH item-termsheet — picked line item has no item pricing rows")
else:
    print("  SKIP  PATCH item-termsheet — no pending decision available")

# ---------- invoice upload on accepted/arrived (multi-file) ----------
r = c.get("/quotes", params={"status": "arrived", "page_size": 1})
arrived = r.json()["data"]
if arrived:
    qa = arrived[0]
    # rerun safety — remove invoices left by previous runs
    for _inv in c.get(f"/quotes/{qa['id']}/invoice").json()["data"]:
        c.delete(f"/quotes/{qa['id']}/invoice/{_inv['id']}")
    files = {"file": ("inv.pdf", b"%PDF-1.4 invoice", "application/pdf")}
    r = c.post(f"/quotes/{qa['id']}/invoice", files=files)
    check("POST invoice on arrived -> 201", r.status_code == 201 and r.json()["data"]["mime_type"] == "application/pdf", r.text[:200])
    r = c.post(f"/quotes/{qa['id']}/invoice", files={"file": ("inv2.png", b"\x89PNG\r\n", "image/png")})
    check("POST second invoice -> 201 (multi-file)", r.status_code == 201 and r.json()["data"]["file_name"] == "inv2.png", r.text[:200])
    r = c.get(f"/quotes/{qa['id']}/invoice")
    lst = r.json()["data"]
    check("GET invoice list has 2", r.status_code == 200 and len(lst) == 2, r.text[:200])
    first = lst[0]
    r = c.get(f"/quotes/{qa['id']}/invoice/{first['id']}")
    check("GET invoice download by id", r.status_code == 200 and r.content == b"%PDF-1.4 invoice", r.text[:200])
    r = c.delete(f"/quotes/{qa['id']}/invoice/{first['id']}")
    check("DELETE invoice by id", r.status_code == 200 and r.json()["data"]["deleted"] is True, r.text[:200])
    r = c.get(f"/quotes/{qa['id']}/invoice")
    check("GET invoice list has 1 after delete", r.status_code == 200 and len(r.json()["data"]) == 1, r.text[:200])
    r = c.get(f"/quotes/{qa['id']}/invoice/{first['id']}")
    check("GET deleted invoice -> 404", r.status_code == 404)
else:
    check("invoice tests (no arrived quote)", False, "no arrived quote in seed")

# cleanup: remove the dedicated test material
r = c.delete(f"/materials/{tmat['id']}")
check("cleanup: DELETE test material", r.status_code == 200, r.text[:200])

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