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.

AUTHENTICATION
  Every /api/ route except /api/health requires a session (auth_gate in
  server/app.py), so the script signs in first and keeps the session cookie for
  the rest of the run. Credentials come from the environment by preference, so a
  password never has to appear in a command line or shell history:

      export WP_SMOKE_USER=smoketest
      export WP_SMOKE_PASSWORD='…'
      python3 server/smoketest.py https://wp-suite.company.local

  …or pass --user / --password explicitly.

  Use an ADMIN account. The script creates a project and deletes it again at the
  end, and deleting one takes Project Admin on that project (require_project_admin);
  a plain project_user can create a project but not clean it up. The script checks
  the signed-in role up front and warns if it is too low, rather than letting you
  discover it in the cleanup step.

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 -e WP_SMOKE_USER -e WP_SMOKE_PASSWORD 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 codes: 0 = all checks passed · 1 = one or more checks failed · 2 = the run
could not start (unreachable host, missing or rejected credentials). 2 is kept
distinct on purpose: "I could not test this" is not the same answer as "this is
broken", and conflating them is what made an unauthenticated version of this
script report a wall of failures against a perfectly healthy stack.
"""
import argparse
import http.cookiejar
import json
import os
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
# One opener for the whole run, carrying the cookie jar that holds the session
# issued by /api/auth/login. urlopen() has no cookie support, which is why the
# session used to be dropped on the floor and every data route answered 401.
OPENER = None


def build_opener(ctx=None):
    handlers = [urllib.request.HTTPCookieProcessor(http.cookiejar.CookieJar())]
    if ctx is not None:
        handlers.append(urllib.request.HTTPSHandler(context=ctx))
    return urllib.request.build_opener(*handlers)


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 OPENER.open(req, 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 abort(msg, hint=""):
    """Could not run — distinct from 'ran and found problems'. See exit codes above."""
    print(_c("\nABORT", "31") + "  " + msg)
    if hint:
        print(hint)
    print()
    return 2


def main():
    global BASE, CTX, OPENER
    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)")
    ap.add_argument("--user", default=os.getenv("WP_SMOKE_USER", ""),
                    help="account to sign in as (default: $WP_SMOKE_USER). Use an admin account.")
    ap.add_argument("--password", default=os.getenv("WP_SMOKE_PASSWORD", ""),
                    help="its password (default: $WP_SMOKE_PASSWORD — preferred, "
                         "so it stays out of shell history)")
    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
    OPENER = build_opener(CTX)

    print(f"\nWork Package Suite — API smoke test\nTarget: {BASE}\n")

    # Refuse to start without credentials rather than running headlong into 401s.
    if not args.user or not args.password:
        missing = " and ".join(
            n for n, v in (("WP_SMOKE_USER", args.user), ("WP_SMOKE_PASSWORD", args.password)) if not v)
        return abort(
            f"no credentials — {missing} not set.",
            "  Every /api/ route except /api/health needs a session, so there is nothing\n"
            "  meaningful to test without one. Set them and re-run:\n\n"
            "      export WP_SMOKE_USER=<admin-account>\n"
            "      export WP_SMOKE_PASSWORD='…'\n\n"
            "  Or pass --user/--password. Use an admin account: the run creates a project\n"
            "  and deletes it again, and the delete needs Project Admin on it.")

    project_id = None
    # Guards the sign-out in `finally`. Without it an ABORT on a rejected login still
    # ran the logout checks, which "passed" — a session that never existed is trivially
    # refused after logout — and printed PASS lines underneath an abort message.
    logged_in = False
    try:
        # 1) Health — API is up and reachable through the proxy. Exempt from auth,
        # so this also isolates "host unreachable" from "credentials rejected".
        try:
            st, body = call("GET", "/api/health")
        except urllib.error.URLError as e:
            return abort(f"cannot reach {BASE}/api/health — {e}",
                         "  Is the stack up (docker compose ps) and the URL correct?")
        check("health endpoint returns ok", st == 200 and isinstance(body, dict) and body.get("ok") is True,
              f"status={st} body={body}")

        # 2) Sign in. The cookie the response sets is held by OPENER's jar and rides
        # every request after this one.
        st, body = call("POST", "/api/auth/login",
                        {"username": args.user, "password": args.password})
        if st != 200:
            detail = body.get("detail") if isinstance(body, dict) else body
            hint = ("  The account may be locked: the API locks an account for a while after\n"
                    "  a few consecutive failures (AUTH_MAX_ATTEMPTS / AUTH_LOCKOUT_MINUTES),\n"
                    "  so re-running with the wrong password makes this worse, not better.\n"
                    "  Check the password, then wait out the lockout window."
                    if st in (401, 403, 423, 429) else
                    "  Unexpected status from the login endpoint — check the API logs.")
            return abort(f"could not sign in as '{args.user}' (HTTP {st}): {detail}", hint)
        logged_in = True
        check("login issues a session", st == 200)

        # 3) Prove the session actually travels — this is the check whose absence let
        # an unauthenticated version of this script look like a broken stack.
        st, me = call("GET", "/api/auth/me")
        who = (me or {}).get("user", {}) if isinstance(me, dict) else {}
        check("session is accepted on an authenticated route",
              st == 200 and who.get("username", "").lower() == args.user.lower(),
              f"status={st} body={me}")
        role = who.get("role", "?")
        print(f"  ..... signed in as {who.get('username', args.user)} (role: {role})")
        if role not in ("admin", "project_super_user", "project_admin"):
            print(_c("  NOTE", "33") + f"  '{role}' cannot archive or delete a project, so the "
                  "archive checks and the\n         cleanup step will fail and a stray test project "
                  "will be left behind.\n         Re-run with an admin account for a clean pass.")

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

        # 5) 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 '?'}")

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

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

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

        # 9) 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")))

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

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

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

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

        # 14) 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:
        # 15) 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.")

        # 16) Sign out. Exercises the logout endpoint, and means a run does not end
        # holding a live session — which matters when this is run from a shared
        # jump host or a CI worker. Only if we got one: see `logged_in`.
        if logged_in:
            st, _ = call("POST", "/api/auth/logout")
            check("logout clears the session", st == 200, f"status={st}")
            st, _ = call("GET", "/api/auth/me")
            check("session is refused after logout (401)", st == 401, f"status={st}")

    # ── 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())