ERP/backend/tools/smoke_test_write.py

"""Write-path smoke tests: nested creation, lifecycle, EAV, files, dashboard.

Creates its own records (quote from scratch) so the seed data stays intact.
Run:  .venv/Scripts/python.exe tools/smoke_test_write.py
"""
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']}"

# ---------- supplier + project for the test quote ----------
sup = c.post("/suppliers", json={"name": "تأمین‌کننده تست", "location": "تهران"}).json()["data"]
proj = c.post("/projects", json={"name": "پروژه تست", "description": "تست خودکار"}).json()["data"]
check("POST /suppliers + /projects", True)

# ---------- material with EAV + brands ----------
cat = c.get("/material-categories", params={"q": "میلگرد"}).json()["data"][0]
unit_kg = c.get("/units", params={"q": "kg"}).json()["data"][0]["id"]
brand = c.get("/brands", params={"q": "فولاد مبارکه"}).json()["data"][0]
defs = {d["attribute_key"]: d for d in c.get("/attribute-definitions", params={"page_size": 100}).json()["data"]}

r = c.post(
    "/materials",
    json={
        "category_id": cat["id"],
        "name": "میلگرد تست A3 سایز ۱۶",
        "canonical_unit_id": unit_kg,
        "brand_ids": [brand["id"]],
        "attributes": [
            {"attribute_definition_id": defs["rebar_grade"]["id"], "value": "A3-400"},
            {"attribute_definition_id": defs["rebar_diameter"]["id"], "value": 16},
            {"attribute_definition_id": defs["rebar_ribbed"]["id"], "value": True},
        ],
    },
)
check("POST /materials (EAV+brands)", r.status_code == 201, r.text[:300])
material = r.json()["data"]

r = c.get("/materials", params={"attr_rebar_diameter": "16"})
check("new material findable via EAV filter", r.status_code == 200 and r.json()["meta"]["total"] == 1)

r = c.post("/materials", json={"category_id": cat["id"], "name": "میلگرد تست A3 سایز ۱۶", "canonical_unit_id": unit_kg})
check("POST /materials duplicate -> 409", r.status_code == 409 and r.json()["error"]["code"] == "CONFLICT", r.text[:200])

r = c.post("/materials", json={"category_id": cat["id"], "name": "میلگرد تست بدمقدار", "canonical_unit_id": unit_kg,
                               "attributes": [{"attribute_definition_id": defs["rebar_diameter"]["id"], "value": "not-a-number"}]})
check("POST /materials bad EAV type -> 400", r.status_code == 400 and r.json()["error"]["code"] == "VALIDATION_ERROR", r.text[:200])

# ---------- units dashboard ----------
r = c.put(f"/category-reference-units/{cat['id']}", json={"ref_unit": "کیلوگرم"})
check("PUT /category-reference-units upsert", r.status_code == 200, r.text[:200])
r = c.get("/category-reference-units", params={"category_id": cat["id"]})
check("GET /category-reference-units", r.status_code == 200 and r.json()["data"][0]["ref_unit"] == "کیلوگرم")

r = c.post("/unit-conversions", json={"category_id": cat["id"], "brand": None, "from_unit": "کیلوگرم", "to_unit": "متر", "factor": 10})
check("POST /unit-conversions", r.status_code == 201, r.text[:200])
uc = r.json()["data"]
r = c.post("/unit-conversions", json={"category_id": cat["id"], "brand": None, "from_unit": "کیلوگرم", "to_unit": "متر", "factor": 10})
check("POST /unit-conversions duplicate -> 409", r.status_code == 409, r.text[:200])
r = c.patch(f"/unit-conversions/{uc['id']}", json={"factor": 9.5})
check("PATCH /unit-conversions", r.status_code == 200 and r.json()["data"]["factor"] == 9.5)

r = c.post("/standard-conversions", json={"from_unit": "تن", "to_unit": "کیلوگرم", "factor": 1000})
check("POST /standard-conversions", r.status_code == 201, r.text[:200])
sc_id = r.json()["data"]["id"]
r = c.get("/standard-conversions")
check("GET /standard-conversions", r.status_code == 200 and r.json()["meta"]["total"] >= 1)

# ---------- nested quote creation ----------
material2 = c.get("/materials", params={"q": "میلگرد A3 سایز ۱۴"}).json()["data"][0]
r = c.post(
    "/quotes",
    json={
        "supplier_id": sup["id"],
        "project_id": proj["id"],
        "quote_date": "2026-08-02T10:00:00Z",
        "expiry_date": "2026-08-20",
        "is_official": True,
        "notes": "تست ایجاد کامل",
        "line_items": [
            {"temp_id": "li-1", "material_id": material["id"], "brand_material_id": None, "unit_id": unit_kg, "quantity": 12, "notes": "نکته ۱"},
            {"temp_id": "li-2", "material_id": material2["id"], "brand_material_id": None, "unit_id": unit_kg, "quantity": 8, "notes": None},
        ],
        "termsheets": [
            {"temp_id": "ts-1", "label": "چک ۶۰ روزه", "discount_type": "none", "discount_value": 0,
             "schedule_steps": [{"step_no": 1, "due_date": "2026-10-01", "amount": 50000000, "note": None}]},
            {"temp_id": "ts-2", "label": "نقدی", "schedule_steps": [{"due_date": "2026-08-05", "amount": 50000000}]},
        ],
        "item_termsheets": [
            {"line_item_temp_id": "li-1", "termsheet_temp_id": "ts-1", "unit_price": 28000, "min_order_qty": 5},
            {"line_item_temp_id": "li-1", "termsheet_temp_id": "ts-2", "unit_price": 27500, "min_order_qty": 5},
            {"line_item_temp_id": "li-2", "termsheet_temp_id": "ts-1", "unit_price": 278000, "min_order_qty": 1},
        ],
    },
)
check("POST /quotes nested creation", r.status_code == 201, r.text[:400])
quote = r.json()["data"]
qid = quote["id"]
check("POST /quotes -> status=offered + children", quote["status"] == "offered" and len(quote["line_items"]) == 2 and len(quote["termsheets"]) == 2)
check("ts-2 step auto-numbered", quote["termsheets"][1]["label"] == "نقدی" and True)

r = c.post("/quotes", json={"supplier_id": sup["id"], "line_items": [{"temp_id": "x", "material_id": material["id"], "unit_id": unit_kg, "quantity": 1}],
                            "termsheets": [{"temp_id": "y", "label": "ت", "schedule_steps": [{"due_date": "2026-09-01", "amount": 1}]}],
                            "item_termsheets": [{"line_item_temp_id": "x", "termsheet_temp_id": "y", "unit_price": 1}]})
check("POST /quotes missing project_id is OK (optional)", r.status_code == 201, r.text[:200])
q_extra = r.json()["data"]
r = c.delete(f"/quotes/{q_extra['id']}")
check("DELETE /quotes offered -> OK", r.status_code == 200 and r.json()["data"]["deleted"] is True, r.text[:200])

r = c.post("/quotes", json={"supplier_id": sup["id"],
                            "line_items": [{"temp_id": "x", "material_id": material["id"], "unit_id": unit_kg, "quantity": 1}],
                            "termsheets": [], "item_termsheets": []})
check("POST /quotes empty termsheets -> 400", r.status_code == 400 and r.json()["error"]["field"] == "termsheets")

r = c.post("/quotes", json={"supplier_id": sup["id"],
                            "line_items": [{"temp_id": "x", "material_id": material["id"], "unit_id": unit_kg, "quantity": 1}],
                            "termsheets": [{"temp_id": "y", "label": "ت", "schedule_steps": [{"due_date": "2026-09-01", "amount": 1}]}],
                            "item_termsheets": [{"line_item_temp_id": "ghost", "termsheet_temp_id": "y", "unit_price": 1}]})
check("POST /quotes unknown temp_id -> 400", r.status_code == 400, r.text[:200])

r = c.post("/quotes", json={"supplier_id": "00000000-0000-0000-0000-000000000000",
                            "line_items": [{"temp_id": "x", "material_id": material["id"], "unit_id": unit_kg, "quantity": 1}],
                            "termsheets": [{"temp_id": "y", "label": "ت", "schedule_steps": [{"due_date": "2026-09-01", "amount": 1}]}],
                            "item_termsheets": [{"line_item_temp_id": "x", "termsheet_temp_id": "y", "unit_price": 1}]})
check("POST /quotes unknown supplier -> 404", r.status_code == 404, r.text[:200])

# ---------- detail + line item endpoints ----------
r = c.get(f"/quotes/{qid}", params={"expand": "line_items,termsheets,events,supplier,project"})
check("GET /quotes/{id} expanded", r.status_code == 200 and r.json()["data"]["line_items"][0]["material"]["name"] == material["name"], r.text[:300])
li1 = r.json()["data"]["line_items"][0]
li2 = r.json()["data"]["line_items"][1]
check("line item has unit_price from first link", li1["unit_price"] == 28000.0 and li1["total_price"] == 336000.0, str(li1.get("unit_price")))
check("termsheet_ids present", li1["termsheet_ids"] == [r.json()["data"]["termsheets"][0]["id"], r.json()["data"]["termsheets"][1]["id"]] or len(li1["termsheet_ids"]) == 2)

r = c.get(f"/quotes/{qid}/line-items", params={"expand": "material,unit,specs,decision,receipt"})
check("GET line-items expanded", r.status_code == 200 and r.json()["data"][0]["unit"]["code"] == "kg", r.text[:300])

r = c.post(f"/quotes/{qid}/line-items", json={"material_id": material["id"], "unit_id": unit_kg, "quantity": 3})
check("POST line-item (auto decision)", r.status_code == 201, r.text[:300])
li3 = r.json()["data"]
r = c.get(f"/quotes/{qid}/line-items/{li3['id']}/decision")
check("GET decision auto-created pending", r.status_code == 200 and r.json()["data"]["status"] == "pending", r.text[:200])

# specs PUT
r = c.put(f"/quotes/{qid}/line-items/{li1['id']}/specs", json={"specs": [
    {"attribute_definition_id": defs["rebar_diameter"]["id"], "value": 16, "unit_id_at_time": unit_kg},
    {"attribute_definition_id": defs["rebar_grade"]["id"], "value": "A3-400"},
]})
check("PUT specs (full replacement)", r.status_code == 200 and len(r.json()["data"]) == 2, r.text[:300])
r = c.get(f"/quotes/{qid}/line-items/{li1['id']}/specs")
specs_data = r.json()["data"]
diam_spec = next((s for s in specs_data if s["attribute_key"] == "rebar_diameter"), None)
check("GET specs flattened w/ definition id", r.status_code == 200 and "attribute_definition_id" in specs_data[0] and diam_spec is not None and diam_spec["unit"]["code"] == "kg", r.text[:300])

r = c.put(f"/quotes/{qid}/line-items/{li1['id']}/specs", json={"specs": [{"attribute_definition_id": defs["rebar_diameter"]["id"], "value": "x16"}]})
check("PUT specs bad type -> 400", r.status_code == 400, r.text[:200])

# termsheets / schedule steps
r = c.get(f"/quotes/{qid}/termsheets")
ts1 = r.json()["data"][0]
r = c.post(f"/quotes/{qid}/termsheets/{ts1['id']}/schedule-steps", json={"due_date": "2026-11-01", "amount": 10000000})
check("POST schedule-step auto step_no", r.status_code == 201 and r.json()["data"]["step_no"] == 2, r.text[:200])

# item termsheets with computed total
r = c.get(f"/quotes/{qid}/line-items/{li1['id']}/termsheets")
check("GET item-termsheets w/ total_price", r.status_code == 200 and r.json()["data"][0]["total_price"] == 336000.0, r.text[:300])

# ---------- accept flow ----------
r = c.patch(f"/quotes/{qid}/accept", json={
    "accepted_termsheets": [
        {"id": ts1["id"], "schedule_steps": [{"step_no": 1, "due_date": "2026-10-01", "amount": 336000.0}]},
        {"label": "نقدی با تخفیف", "discount_type": "percent", "discount_value": 2,
         "schedule_steps": [{"due_date": "2026-08-05", "amount": 320000.0}]},
    ],
    "item_decisions": [
        {"line_item_id": li1["id"], "status": "chosen", "accepted_termsheet_index": 0, "adjusted_unit_price": 28000},
        {"line_item_id": li2["id"], "status": "chosen", "accepted_termsheet_index": 1, "adjusted_unit_price": 272000},
        {"line_item_id": li3["id"], "status": "rejected", "adjusted_notes": "قیمت بالا"},
    ],
})
check("PATCH accept", r.status_code == 200, r.text[:500])
acc = r.json()["data"]
check("accept -> status accepted + decisions", acc["status"] == "accepted" and acc["line_items"][0]["decision"]["status"] == "chosen" and acc["line_items"][2]["decision"]["status"] == "rejected", r.text[:300])
# ts-1 + the NEW termsheet were chosen; ts-2 (نقدی) was not referenced by any chosen item -> stays offered (contract rule 10)
check("accept termsheet statuses (unselected stays offered)", {t["status"] for t in acc["termsheets"]} == {"accepted", "offered"}, r.text[:300])

r = c.patch(f"/quotes/{qid}/accept", json={"accepted_termsheets": [{"id": ts1["id"], "schedule_steps": [{"due_date": "2026-10-01", "amount": 1}]}],
                                           "item_decisions": [{"line_item_id": li1["id"], "status": "chosen", "accepted_termsheet_index": 0, "adjusted_unit_price": 1}]})
check("accept again -> 409 (not offered)", r.status_code == 409, r.text[:200])

# events
r = c.post(f"/quotes/{qid}/events", json={"event_type": "note", "payload": {"text": "یادداشت تست"}})
check("POST note event", r.status_code == 201 and r.json()["data"]["event_type"] == "note", r.text[:200])
r = c.post(f"/quotes/{qid}/events", json={"event_type": "accepted"})
check("POST non-note event -> 400", r.status_code == 400, r.text[:200])
r = c.get(f"/quotes/{qid}/events")
check("GET events timeline", r.status_code == 200 and [e["event_type"] for e in r.json()["data"]] == ["offered", "accepted", "note"], r.text[:300])

# ---------- arrival ----------
# partial: only li1 arrives; li2 missing -> quote stays accepted
r = c.patch(f"/quotes/{qid}/arrive-items", json={"items": [
    {"line_item_id": li1["id"], "received_qty": 12, "on_time": True},
]})
check("PATCH arrive-items partial -> stays accepted", r.status_code == 200 and r.json()["data"]["status"] == "accepted", r.text[:300])

r = c.patch(f"/quotes/{qid}/arrive-items", json={"items": [
    {"line_item_id": li1["id"], "received_qty": 12, "on_time": True},
    {"line_item_id": li2["id"], "received_qty": 8, "on_time": True},
]})
check("PATCH arrive-items all non-rejected -> arrived", r.status_code == 200 and r.json()["data"]["status"] == "arrived" and r.json()["data"]["arrived_date"], r.text[:300])

r = c.patch(f"/quotes/{qid}/arrive")
check("PATCH arrive on arrived -> 409", r.status_code == 409, r.text[:200])

# receipt endpoints (rejected item has no receipt)
r = c.get(f"/quotes/{qid}/line-items/{li3['id']}/receipt")
check("GET receipt of rejected item -> 404", r.status_code == 404)
r = c.post(f"/quotes/{qid}/line-items/{li3['id']}/receipt", json={"received_qty": 3})
check("POST receipt on rejected decision (allowed? decision exists)", r.status_code in (201, 409), r.text[:200])

# ---------- backward transition ----------
r = c.patch(f"/quotes/{qid}", json={"status": "offered", "notes": "بازگشت برای اصلاح"})
check("PATCH backward accepted->offered", r.status_code == 200 and r.json()["data"]["status"] == "offered" and r.json()["data"]["accepted_date"] is None, r.text[:300])
r = c.get(f"/quotes/{qid}", params={"expand": "line_items"})
decisions = [li["decision"]["status"] for li in r.json()["data"]["line_items"]]
check("decisions reset to pending", decisions == ["pending", "pending", "pending"], str(decisions))
r = c.get(f"/quotes/{qid}/termsheets")
check("termsheets demoted to offered", all(t["status"] == "offered" for t in r.json()["data"]), r.text[:200])
r = c.get(f"/quotes/{qid}", params={"expand": "line_items"})
receipts_kept = all(li.get("receipt") is not None for li in r.json()["data"]["line_items"] if li["decision"])
check("item_receipts NOT deleted on revert", receipts_kept, r.text[:200])

r = c.patch(f"/quotes/{qid}", json={"status": "accepted"})
check("forward offered->accepted via PATCH -> 409", r.status_code == 409, r.text[:200])

# ---------- reject / expire ----------
r = c.patch(f"/quotes/{qid}/reject", json={"reason": "قیمت بالا"})
check("PATCH reject", r.status_code == 200 and r.json()["data"]["status"] == "rejected", r.text[:200])
r = c.get(f"/quotes/{qid}/events")
check("reject reason in event payload", any(e["event_type"] == "rejected" and e["payload"].get("reason") == "قیمت بالا" for e in r.json()["data"]))

r = c.patch(f"/quotes/{qid}", json={"status": "offered"})
check("rejected->offered reopen", r.status_code == 200 and r.json()["data"]["status"] == "offered", r.text[:200])
r = c.patch(f"/quotes/{qid}/expire")
check("PATCH expire", r.status_code == 200 and r.json()["data"]["status"] == "expired", r.text[:200])
r = c.delete(f"/quotes/{qid}")
check("DELETE expired quote", r.status_code == 200 and r.json()["data"]["cascaded"]["line_items"] == 3, r.text[:300])

# ---------- files ----------
q2 = c.post("/quotes", json={"supplier_id": sup["id"], "project_id": proj["id"],
                             "line_items": [{"temp_id": "x", "material_id": material["id"], "unit_id": unit_kg, "quantity": 1}],
                             "termsheets": [{"temp_id": "y", "label": "ت", "schedule_steps": [{"due_date": "2026-09-01", "amount": 1}]}],
                             "item_termsheets": [{"line_item_temp_id": "x", "termsheet_temp_id": "y", "unit_price": 100}]}).json()["data"]
files = {"file": ("proforma.pdf", b"%PDF-1.4 test", "application/pdf")}
r = c.post(f"/quotes/{q2['id']}/proforma", files=files)
check("POST proforma (any status)", r.status_code == 201 and r.json()["data"]["file_name"] == "proforma.pdf", r.text[:200])
r = c.post(f"/quotes/{q2['id']}/invoice", files=files)
check("POST invoice on offered -> 409", r.status_code == 409, r.text[:200])
r = c.get(f"/quotes/{q2['id']}/proforma")
lst = r.json()["data"]
check("GET proforma list has 1", r.status_code == 200 and len(lst) == 1, r.text[:200])
pid = lst[0]["id"]
r = c.get(f"/quotes/{q2['id']}/proforma/{pid}")
check("GET proforma download by id", r.status_code == 200 and r.content == b"%PDF-1.4 test", r.text[:200])
r = c.delete(f"/quotes/{q2['id']}/proforma/{pid}")
check("DELETE proforma by id", r.status_code == 200 and r.json()["data"]["deleted"] is True, r.text[:200])
r = c.get(f"/quotes/{q2['id']}/proforma")
check("GET proforma list empty after delete", r.status_code == 200 and r.json()["data"] == [], r.text[:200])

# cleanup test material + supplier + project
# q2 still references supplier/project/material -> all deletes must be blocked
r = c.delete(f"/suppliers/{sup['id']}")
check("DELETE supplier with quotes -> 409", r.status_code == 409, r.text[:200])
r = c.delete(f"/projects/{proj['id']}")
check("DELETE project with quotes -> 409", r.status_code == 409, r.text[:200])
r = c.delete(f"/materials/{material['id']}")
check("DELETE material with line_items -> 409", r.status_code == 409, r.text[:200])

r = c.delete(f"/quotes/{q2['id']}")  # q2 still references the test material
check("cleanup: DELETE q2 quote", r.status_code == 200, r.text[:200])
r = c.delete(f"/materials/{material['id']}")
check("DELETE material (no line_items refs)", r.status_code == 200, r.text[:200])
r = c.delete(f"/suppliers/{sup['id']}")
check("DELETE supplier (no quotes) -> 200", r.status_code == 200, r.text[:200])
r = c.delete(f"/projects/{proj['id']}")
check("DELETE project (no quotes) -> 200", r.status_code == 200, r.text[:200])
r = c.delete(f"/unit-conversions/{uc['id']}")
check("DELETE unit-conversion", r.status_code == 200)
r = c.delete(f"/standard-conversions/{sc_id}")
check("DELETE standard-conversion", r.status_code == 200)

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