#!/usr/bin/env python3 """End-to-end smoke test for the Work Package Suite API + PostgreSQL. Exercises the real HTTP endpoints the way the front end does, proving that NGINX → FastAPI → PostgreSQL all work and that the Python logic (the AWP release gate, metrics, cascade delete) behaves. Stdlib only — no pip, no jq. USAGE # Against the deployed site (through the NGINX proxy): python3 server/smoketest.py https://wp-suite.company.local # Self-signed / internal TLS cert? skip verification: python3 server/smoketest.py https://wp-suite.company.local --insecure # From inside the api container (hits FastAPI directly): docker compose exec api python /app/server/smoketest.py http://localhost:8000 # Leave the demo project in the database so you can open it in the UI: python3 server/smoketest.py https://wp-suite.company.local --keep The base URL is the SITE root (no /api). Default: http://localhost:8000 Exit code 0 = all checks passed, 1 = one or more failed. """ import argparse import json import ssl import sys import urllib.error import urllib.request # ── tiny colored reporter ───────────────────────────────────────────────────── _PASS, _FAIL = [], [] def _c(s, code): # color if a TTY return f"\033[{code}m{s}\033[0m" if sys.stdout.isatty() else s def ok(msg): _PASS.append(msg); print(" " + _c("PASS", "32") + " " + msg) def bad(msg): _FAIL.append(msg); print(" " + _c("FAIL", "31") + " " + msg) def check(name, cond, detail=""): (ok if cond else bad)(name + (f" ({detail})" if detail and not cond else "")) return cond BASE = "" CTX = None def call(method, path, body=None): """Returns (status_code, parsed_body). Never raises on HTTP status.""" url = BASE + path data = json.dumps(body).encode() if body is not None else None req = urllib.request.Request( url, data=data, method=method, headers={"Content-Type": "application/json", "Accept": "application/json"}, ) try: with urllib.request.urlopen(req, context=CTX, timeout=20) as r: raw = r.read().decode(); status = r.status except urllib.error.HTTPError as e: raw = e.read().decode(); status = e.code try: parsed = json.loads(raw) if raw else None except ValueError: parsed = raw return status, parsed def main(): global BASE, CTX ap = argparse.ArgumentParser(description="Work Package Suite API smoke test") ap.add_argument("base_url", nargs="?", default="http://localhost:8000", help="Site root, no /api (default: http://localhost:8000)") ap.add_argument("--insecure", action="store_true", help="skip TLS verification") ap.add_argument("--keep", action="store_true", help="keep the demo project (don't delete)") args = ap.parse_args() BASE = args.base_url.rstrip("/") if args.insecure: CTX = ssl.create_default_context(); CTX.check_hostname = False; CTX.verify_mode = ssl.CERT_NONE print(f"\nWork Package Suite — API smoke test\nTarget: {BASE}\n") project_id = None try: # 1) Health — API is up and reachable through the proxy. try: st, body = call("GET", "/api/health") except urllib.error.URLError as e: print(_c("\nABORT", "31") + f" cannot reach {BASE}/api/health — {e}\n" " Is the stack up (docker compose ps) and the URL correct?\n") return 1 check("health endpoint returns ok", st == 200 and isinstance(body, dict) and body.get("ok") is True, f"status={st} body={body}") # 2) Create a project (writes to the projects table). st, proj = call("POST", "/api/projects", { "name": "ZZ Smoke Test Project", "number": "SMOKE-001", "client": "Internal QA", "division": "Controls", "site": "Test Host", "created_by": "smoketest", }) project_id = proj.get("id") if isinstance(proj, dict) else None check("create project", st == 200 and bool(project_id), f"status={st}") # 3) Read it back + confirm it's in the list (SQL round-trip). st, got = call("GET", f"/api/projects/{project_id}") check("fetch project by id", st == 200 and got.get("number") == "SMOKE-001", f"status={st}") st, lst = call("GET", "/api/projects") check("project appears in list", st == 200 and any(p.get("id") == project_id for p in lst), f"status={st} count={len(lst) if isinstance(lst, list) else '?'}") # 4) Create a SOP linked to the project. st, sop = call("POST", "/api/sops", { "project_id": project_id, "name": "ZZ Smoke SOP", "number": "SMOKE-001", "complete": True, "created_by": "smoketest", "data": {"governance": {"woFormat": "WP##-[Sector]-[TYPE]", "disciplines": ["Mechanical", "Electrical", "Tech"]}}, }) sop_id = sop.get("id") if isinstance(sop, dict) else None check("create SOP linked to project", st == 200 and bool(sop_id) and sop.get("project_id") == project_id, f"status={st}") st, latest = call("GET", f"/api/sops/latest?project_id={project_id}") check("latest SOP for project resolves", st == 200 and latest.get("id") == sop_id, f"status={st}") # 5) Create a Work Package with one OPEN constraint (not release-ready). st, wp = call("POST", "/api/wps", { "project_id": project_id, "sop_id": sop_id, "number": "WP01-SMOKE", "subject": "Smoke test package", "type": "Conduit Install", "status": "Scheduled", "created_by": "smoketest", "data": {"disciplines": ["Electrical"], "hours": "40", "actualHrs": "", "constraints": [{"name": "Materials", "status": "open", "comment": "awaiting delivery"}, {"name": "Safety & Permitting", "status": "cleared", "comment": ""}]}, }) wp_id = wp.get("id") if isinstance(wp, dict) else None check("create work package", st == 200 and bool(wp_id), f"status={st}") # 6) The AWP release gate: issuing with an open constraint must be REFUSED (409). st, refused = call("POST", f"/api/wps/{wp_id}/issue") check("issue is blocked while a constraint is open (409)", st == 409, f"status={st} body={refused}") # 7) Clear the constraint (upsert), then issue must SUCCEED (200, status Issued). call("POST", "/api/wps", { "id": wp_id, "project_id": project_id, "sop_id": sop_id, "number": "WP01-SMOKE", "subject": "Smoke test package", "type": "Conduit Install", "status": "Scheduled", "data": {"disciplines": ["Electrical"], "hours": "40", "actualHrs": "", "constraints": [{"name": "Materials", "status": "cleared", "comment": ""}, {"name": "Safety & Permitting", "status": "cleared", "comment": ""}]}, }) st, issued = call("POST", f"/api/wps/{wp_id}/issue") check("issue succeeds once constraints clear", st == 200 and issued.get("status") == "Issued", f"status={st}") check("issued_at timestamp is set", isinstance(issued, dict) and bool(issued.get("issued_at"))) # 8) Status transition endpoint. st, prog = call("POST", f"/api/wps/{wp_id}/status", {"status": "In Progress"}) check("status transition endpoint", st == 200 and prog.get("status") == "In Progress", f"status={st}") # 9) Metrics aggregate for the project (Python aggregation over SQL rows). st, m = call("GET", f"/api/wps/metrics?project_id={project_id}") check("metrics endpoint aggregates", st == 200 and isinstance(m, dict) and m.get("total", 0) >= 1, f"status={st} metrics={m}") # 10) Comment / feedback write + read. st, c = call("POST", "/api/feedback", { "type": "wp_review_comment", "name": "smoketest", "wp_id": wp_id, "text": "SMOKE TEST comment — safe to delete", "page": "/smoketest"}) check("post comment/feedback", st == 200 and isinstance(c, dict) and bool(c.get("id")), f"status={st}") st, comments = call("GET", f"/api/comments?wp_id={wp_id}") check("comment is queryable", st == 200 and any("SMOKE TEST" in (x.get("text") or "") for x in comments), f"status={st}") # 11) WPs filter by project. st, wps = call("GET", f"/api/wps?project_id={project_id}") check("list WPs by project", st == 200 and any(w.get("id") == wp_id for w in wps), f"status={st}") # 12) Archiving a project: it leaves the default list, stays reachable with # archived=all, and freezes read-only — then unarchiving restores all three. # The freeze is the whole point of the feature, so it is asserted, not assumed. st, arch = call("POST", f"/api/projects/{project_id}/archive", {"archived": True}) check("archive project", st == 200 and arch.get("archived") is True, f"status={st}") st, lst = call("GET", "/api/projects") check("archived project drops out of the default list", st == 200 and not any(p.get("id") == project_id for p in lst), f"status={st}") st, lst = call("GET", "/api/projects?archived=all") check("archived project is still there with archived=all", st == 200 and any(p.get("id") == project_id for p in lst), f"status={st}") st, refused = call("POST", "/api/wps", { "id": wp_id, "project_id": project_id, "sop_id": sop_id, "number": "WP01-SMOKE", "subject": "edited while archived", "type": "Conduit Install", "status": "Scheduled", "data": {"disciplines": ["Electrical"], "hours": "40"}}) check("writing to an archived project is refused (409)", st == 409, f"status={st} body={refused}") st, unarch = call("POST", f"/api/projects/{project_id}/archive", {"archived": False}) check("unarchive project", st == 200 and unarch.get("archived") is False, f"status={st}") st, _ = call("POST", "/api/wps", { "id": wp_id, "project_id": project_id, "sop_id": sop_id, "number": "WP01-SMOKE", "subject": "Smoke test package", "type": "Conduit Install", "status": "In Progress", "data": {"disciplines": ["Electrical"], "hours": "40"}}) check("writing succeeds again once unarchived", st == 200, f"status={st}") finally: # 13) Cleanup — deleting the project cascades to its SOPs and WPs (FK ON DELETE CASCADE). if project_id and not args.keep: st, _ = call("DELETE", f"/api/projects/{project_id}") check("delete project (cascades SOP + WPs)", st == 200, f"status={st}") st, after = call("GET", f"/api/wps?project_id={project_id}") check("WPs removed by cascade", st == 200 and isinstance(after, list) and len(after) == 0, f"status={st} remaining={after}") elif project_id and args.keep: print(f"\n --keep: left demo project {project_id} ('ZZ Smoke Test Project') in the database.") # ── summary ──────────────────────────────────────────────────────────────── total = len(_PASS) + len(_FAIL) print(f"\n{'-'*52}\n{len(_PASS)}/{total} checks passed.") if _FAIL: print(_c(f"FAILED ({len(_FAIL)}):", "31")) for f in _FAIL: print(" - " + f) print("\nResult: " + _c("FAIL", "31") + "\n") return 1 print("\nResult: " + _c("ALL PASS — API, Python logic, and SQL are working.", "32") + "\n") return 0 if __name__ == "__main__": sys.exit(main())