Project-SDE-WP-Suite/server/smoketest.py

#!/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}")

    finally:
        # 12) 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())