246 lines
15 KiB
Python
246 lines
15 KiB
Python
"""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)
|
|
|
|
# ---------- 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"]["id"] > 19, 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)
|
|
r = c.get("/brands", params={"q": "فولاد مبارکه"})
|
|
brand = r.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": "دسته تست"})
|
|
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)
|
|
r = c.get("/material-categories", params={"q": "میلگرد"})
|
|
cat = r.json()["data"][0]
|
|
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
|
|
cat = c.get("/material-categories", params={"q": "میلگرد"}).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})
|
|
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"})
|
|
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])
|
|
r = c.post(f"/materials/{mat['id']}/units", json={"unit_id": 6, "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": 6, "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 ----------
|
|
r = c.get("/materials", params={"q": "سیمان پرتلند تیپ ۲"})
|
|
m2 = r.json()["data"][0]
|
|
r = c.get("/brands", params={"q": "Bosch"})
|
|
b2 = r.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)
|
|
|
|
# ---------- supplier/project patches ----------
|
|
r = c.get("/suppliers", params={"q": "آریا"})
|
|
s1 = r.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"})
|
|
r = c.get("/projects", params={"q": "ونوس"})
|
|
p1 = r.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": "ساختمان ۱۸ طبقه مسکونی در زعفرانیه"})
|
|
|
|
# ---------- 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 ----------
|
|
r = c.get(f"/quotes/{qid}/termsheets")
|
|
ts = r.json()["data"]
|
|
if len(ts) == 1:
|
|
r = c.post(f"/quotes/{qid}/termsheets", json={"label": "ثانویه", "schedule_steps": [{"due_date": "2026-12-01", "amount": 10}]})
|
|
check("POST second termsheet", r.status_code == 201, r.text[:200])
|
|
ts = r.json()["data"]
|
|
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])
|
|
r = c.delete(f"/quotes/{qid}/termsheets/{ts[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 (on a pending quote) ----------
|
|
r = c.get(f"/quotes/{qid}/line-items")
|
|
li = r.json()["data"][0]
|
|
r = c.put(f"/quotes/{qid}/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/{qid}/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/{qid}/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/{qid}/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])
|
|
|
|
# ---------- item termsheets ----------
|
|
r = c.get(f"/quotes/{qid}/line-items/{li['id']}/termsheets")
|
|
its = r.json()["data"]
|
|
r = c.patch(f"/quotes/{qid}/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/{qid}/line-items/{li['id']}/termsheets/{its[0]['id']}", json={"unit_price": its[0]["unit_price"]})
|
|
|
|
# ---------- 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]
|
|
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)
|