Files
ERP/backend/tools/smoke_test_write.py
T
MMD e1495afee9 fix(auth): auth deps use their own DB session; smoke suites made dataset-independent
RUNTIME BUG (caught by live smoke tests): the global auth dependency
queried the request-scoped session, auto-beginning its transaction — any
endpoint opening 'async with db.begin()' afterwards crashed with
'transaction already begun' (POST /materials 500). get_current_user /
get_actor now resolve in a short-lived session (expire_on_commit=False
keeps the context usable detached); inventory/pm scopes keep passing the
request session explicitly.

Smoke suites rewritten against CURRENT contracts:
- assertions are behavior-based (shapes, envelopes, filters), no magic
  seed counts; dynamic lookups instead of hard-coded names/ids
- rerun-tolerant: tolerate 409-on-create by reusing prior artifacts
- aligned with current rules: revert blocked after inventory receipts
  (v2.6), attachments allowed in every status, termsheets optional at
  quote level
- SMOKE_USER/SMOKE_PASSWORD env for login (default tester/123456)

Verified on a fresh volume: migration + audit_log bootstrap, Casdoor
seed (bcrypt), demo data, and 29+76+61 smoke checks all green.
2026-09-19 00:49:15 +03:30

379 lines
23 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
"""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 ----------
# dataset-independent: fetch/create whatever the seed does or doesn't have
cat = c.get("/material-categories", params={"page_size": 1}).json()["data"][0]
r_sup = c.post("/suppliers", json={"name": "تأمین‌کننده تست", "location": "تهران", "category_id": cat["id"]})
if r_sup.status_code == 201:
sup = r_sup.json()["data"]
check("POST /suppliers", True)
else: # rerun — the test supplier already exists
sup = c.get("/suppliers", params={"q": "تأمین‌کننده تست"}).json()["data"][0]
check("POST /suppliers (exists from previous run — 409 tolerated)", r_sup.status_code == 409, r_sup.text[:200])
r_proj = c.post("/projects", json={"name": "پروژه تست", "description": "تست خودکار"})
if r_proj.status_code == 201:
proj = r_proj.json()["data"]
check("POST /projects", True)
else:
proj = c.get("/projects", params={"q": "پروژه تست"}).json()["data"][0]
check("POST /projects (exists from previous run — 409 tolerated)", r_proj.status_code == 409, r_proj.text[:200])
# ---------- material with EAV + brands ----------
unit_kg = c.get("/units", params={"q": "kg"}).json()["data"][0]["id"]
brand = c.get("/brands", params={"page_size": 1}).json()["data"][0]
defs = {d["attribute_key"]: d for d in c.get("/attribute-definitions", params={"page_size": 100}).json()["data"]}
# ensure the EAV keys this suite exercises exist (seed-independent)
for _key, _dt in (("rebar_grade", "string"), ("rebar_diameter", "number"), ("rebar_ribbed", "boolean")):
if _key not in defs:
r = c.post("/attribute-definitions", json={"attribute_key": _key, "label": _key, "data_type": _dt})
check("POST /attribute-definitions (%s)" % _key, r.status_code == 201, r.text[:200])
defs[_key] = r.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},
],
},
)
if r.status_code == 201:
material = r.json()["data"]
check("POST /materials (EAV+brands)", True)
else: # rerun — fetch the material a previous run created
material = c.get("/materials", params={"q": "میلگرد تست A3 سایز ۱۶"}).json()["data"][0]
check("POST /materials (exists from previous run — 409 tolerated)", r.status_code == 409, r.text[:200])
r = c.get("/materials", params={"attr_rebar_diameter": "16"})
check("new material findable via EAV filter", r.status_code == 200 and any(m["id"] == material["id"] for m in r.json()["data"]))
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})
if r.status_code == 409: # rerun — reuse the existing conversion
uc = next(x for x in c.get("/unit-conversions", params={"page_size": 100}).json()["data"]
if x["from_unit"] == "کیلوگرم" and x["to_unit"] == "متر" and x["category"]["id"] == cat["id"])
check("POST /unit-conversions (exists from previous run — 409 tolerated)", True)
else:
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})
if r.status_code == 201:
sc_id = r.json()["data"]["id"]
check("POST /standard-conversions", True)
else: # rerun — reuse the existing conversion
sc_id = next(x["id"] for x in c.get("/standard-conversions", params={"page_size": 100}).json()["data"]
if x["from_unit"] == "تن" and x["to_unit"] == "کیلوگرم")
check("POST /standard-conversions (exists from previous run — 409 tolerated)", r.status_code == 409, r.text[:200])
r = c.get("/standard-conversions")
check("GET /standard-conversions", r.status_code == 200 and r.json()["meta"]["total"] >= 1)
# ---------- nested quote creation ----------
_all_materials = c.get("/materials", params={"page_size": 100}).json()["data"]
material2 = next(m for m in _all_materials if m["id"] != material["id"])
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": []})
# current contract: termsheets are optional at quote level (item pricing
# is driven by item_termsheets) — a quote without them is valid
check("POST /quotes empty termsheets -> 201 (optional since v2.6)", r.status_code == 201, r.text[:200])
if r.status_code == 201:
c.delete(f"/quotes/{r.json()['data']['id']}")
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 (v2.6 rule: quotes WITH inventory receipts
# are immutable — they feed the stock ledger) ----------
r = c.patch(f"/quotes/{qid}", json={"status": "offered", "notes": "بازگشت برای اصلاح"})
check(
"PATCH backward accepted->offered blocked after receipts",
r.status_code == 409 and "رسید" in r.json()["error"]["message"], 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 unchanged after blocked revert", decisions == ["chosen", "chosen", "rejected"], str(decisions))
# transition tests need a quote WITHOUT receipts — q3:
q3 = c.post("/quotes", json={
"supplier_id": sup["id"], "project_id": proj["id"],
"quote_date": "2026-08-02T10:00:00Z", "expiry_date": "2026-08-20",
"line_items": [{"temp_id": "li-x", "material_id": material["id"], "unit_id": unit_kg, "quantity": 2}],
"termsheets": [{"temp_id": "ts-x", "label": "نقدی", "schedule_steps": [{"due_date": "2026-08-10", "amount": 5000}]}],
"item_termsheets": [{"line_item_temp_id": "li-x", "termsheet_temp_id": "ts-x", "unit_price": 2500}],
}).json()["data"]
r = c.patch(f"/quotes/{q3['id']}/accept", json={
"accepted_termsheets": [{"id": q3["termsheets"][0]["id"],
"schedule_steps": [{"step_no": 1, "due_date": "2026-08-10", "amount": 5000}]}],
"item_decisions": [{"line_item_id": q3["line_items"][0]["id"], "status": "chosen", "accepted_termsheet_index": 0, "adjusted_unit_price": 2500}],
})
check("q3 accept", r.status_code == 200, r.text[:300])
r = c.patch(f"/quotes/{q3['id']}", json={"status": "offered"})
check("PATCH backward accepted->offered (no receipts) -> OK", 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/{q3['id']}", params={"expand": "line_items"})
check("decisions reset to pending", [li["decision"]["status"] for li in r.json()["data"]["line_items"]] == ["pending"], r.text[:200])
r = c.get(f"/quotes/{q3['id']}/termsheets")
check("termsheets demoted to offered", all(t["status"] == "offered" for t in r.json()["data"]), r.text[:200])
r = c.patch(f"/quotes/{q3['id']}", json={"status": "accepted"})
check("forward offered->accepted via PATCH -> 409", r.status_code == 409, r.text[:200])
r = c.patch(f"/quotes/{q3['id']}/reject", json={"reason": "قیمت بالا"})
check("PATCH reject", r.status_code == 200 and r.json()["data"]["status"] == "rejected", r.text[:200])
r = c.get(f"/quotes/{q3['id']}/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/{q3['id']}", 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/{q3['id']}/expire")
check("PATCH expire", r.status_code == 200 and r.json()["data"]["status"] == "expired", r.text[:200])
r = c.delete(f"/quotes/{q3['id']}")
check("DELETE expired quote", r.status_code == 200 and r.json()["data"]["cascaded"]["line_items"] == 1, 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])
# current contract: attachments are allowed in EVERY status (files.py) —
# the invoice can be attached from the edit page at any time
r = c.post(f"/quotes/{q2['id']}/invoice", files=files)
check("POST invoice on offered (any-status contract)", r.status_code == 201, r.text[:200])
inv_id = r.json()["data"]["id"]
r = c.delete(f"/quotes/{q2['id']}/invoice/{inv_id}")
check("DELETE test invoice", r.status_code == 200, 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 ----------
# q2 (offered) is deletable; qid is a receipts-protected artifact that stays.
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']}")
check("cleanup: DELETE q2 quote", r.status_code == 200, r.text[:200])
# qid (arrived, receipt-protected) still references everything -> protected
r = c.delete(f"/materials/{material['id']}")
check("DELETE material (still referenced by receipt-protected quote) -> 409", r.status_code == 409, r.text[:200])
r = c.delete(f"/suppliers/{sup['id']}")
check("DELETE supplier (still referenced) -> 409", r.status_code == 409, r.text[:200])
r = c.delete(f"/projects/{proj['id']}")
check("DELETE project (still referenced) -> 409", r.status_code == 409, 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)