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
  There is no local password anymore (D15/D16, T10.4) — identity is Okta's job,
  and Okta requires a real browser to complete, which this stdlib script cannot
  do. So instead of signing in over HTTP the way the front end does, this script
  mints a session the same way server/app.py's okta_callback() does after Okta
  hands back an identity: auth.create_token() for an existing account, seeded
  straight into the cookie jar. That means it needs to run somewhere that can
  read the SAME AUTH_SECRET_KEY and reach the SAME database as the server under
  test — inside the api container, or locally against your dev DB. It can no
  longer sign in to an arbitrary remote URL from an unrelated workstation; if
  the target is remote, run it on that host or inside that container instead.

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

  …or pass --user explicitly. The account must already exist — sign it in
  through Okta once first (or create it from the admin console) if it doesn't;
  this script promotes no one and provisions nothing.

  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
  # From inside the api container (has AUTH_SECRET_KEY and DATABASE_URL; hits
  # FastAPI directly):
  docker compose exec -e WP_SMOKE_USER api \
      python /app/server/smoketest.py http://localhost:8000

  # Local dev, against the app you're running yourself:
  export AUTH_SECRET_KEY=... DATABASE_URL=... WP_SMOKE_USER=smoketest
  python3 server/smoketest.py http://localhost:8000

  # Self-signed / internal TLS cert on the HTTP side? skip verification:
  python3 server/smoketest.py https://wp-suite.company.local --insecure

  # Leave the demo project in the database so you can open it in the UI:
  python3 server/smoketest.py http://localhost:8000 --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 credentials, or no account by that
username). 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
from urllib.parse import urlparse

# So `from server import auth, models` / `from server.db import SessionLocal` resolve
# when this file is run directly (`python3 server/smoketest.py`) rather than as
# `python -m server.smoketest` — same reasoning as the sys.path lines in tests/*.py.
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))

# ── 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.
# 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
COOKIE_JAR = None


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


def seed_session_cookie(token: str, base: str) -> None:
    """Put a minted session into the jar directly, the same shape a Set-Cookie
    response from the old /api/auth/login would have produced — so the logout
    check below (which relies on the jar honoring logout()'s Set-Cookie that
    expires it) keeps working unchanged. `base` is explicit rather than read off
    this module's own BASE global, so seed_demo.py (which imports this function
    but has its own BASE) seeds the cookie for the host it's actually targeting."""
    host = urlparse(base).hostname or "localhost"
    COOKIE_JAR.set_cookie(http.cookiejar.Cookie(
        version=0, name="wp_session", value=token,
        port=None, port_specified=False,
        domain=host, domain_specified=True, domain_initial_dot=False,
        path="/", path_specified=True,
        secure=False, expires=None, discard=True,
        comment=None, comment_url=None, rest={"HttpOnly": 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 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="existing account to sign in as (default: $WP_SMOKE_USER). Use an admin account.")
    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 a username rather than running headlong into 401s.
    if not args.user:
        return abort(
            "no account — $WP_SMOKE_USER not set.",
            "  Every /api/ route except /api/health needs a session, so there is nothing\n"
            "  meaningful to test without one. Set it and re-run:\n\n"
            "      export WP_SMOKE_USER=<admin-account>\n\n"
            "  Or pass --user. 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" — mint a session directly (see AUTHENTICATION above) and seed
        # it into OPENER's jar, so it rides every request after this one exactly the
        # way a real Set-Cookie response would have.
        try:
            from server import auth as srv_auth
            from server.db import SessionLocal
        except ImportError as e:
            return abort(f"cannot import the server package to mint a session: {e}",
                         "  This script now needs to run where server/ is importable and\n"
                         "  AUTH_SECRET_KEY / DATABASE_URL match the target server's — see\n"
                         "  AUTHENTICATION above.")
        with SessionLocal() as db:
            user = srv_auth.find_user(db, args.user)
            if not user:
                return abort(f"no account named '{args.user}'.",
                             "  This script signs in as an existing account, it doesn't create one —\n"
                             "  sign in through Okta once first, or create it from the admin console.")
            if not user.is_active:
                return abort(f"'{args.user}' is disabled.", "")
            token = srv_auth.create_token(user)
        seed_session_cookie(token, BASE)
        logged_in = True
        check("session cookie seeded", bool(token))

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