n.siegfried/Project-SDE-WP-Suite
3
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

Make the smoke test sign in; enforce SQLite foreign keys

Browse Source 
Closes known issue 3. server/smoketest.py predated the login portal and had no
login step at all, so auth_gate refused every route after /api/health and the
documented way to verify a deploy reported a wall of failures against a healthy
stack.

  - Signs in first, holding the session in an http.cookiejar on a shared opener.
    urlopen() has no cookie support, which is why the session was dropped.
  - Credentials from WP_SMOKE_USER / WP_SMOKE_PASSWORD, or --user/--password, so
    a password need not land in shell history. Refuses to start without them
    rather than running headlong into 401s.
  - Checks the signed-in role up front and warns when it cannot archive or delete
    a project, instead of failing six checks later for an unexplained reason.
  - New exit code 2 for "could not run" (unreachable, or credentials missing or
    rejected), kept distinct from 1 "ran and found problems".
  - Also asserts the session is accepted on an authenticated route and refused
    after sign-out; signs out at the end so a run on a shared host leaves none.

The working smoke test immediately caught a real bug: SQLite ships with foreign
keys disabled and the pragma is per-connection, so every ondelete="CASCADE" was
silently a no-op on dev while working on Postgres. Deleting a project orphaned its
SOPs, work packages and membership rows; deleting a user orphaned theirs. db.py
now sets PRAGMA foreign_keys=ON for SQLite, so dev matches production.

Enforcing them exposed two things that had been getting away with it:

  - create_user adds an account and its ProjectMember rows in one flush, and the
    ORM takes flush order from relationship() declarations. models.py has none by
    design, so it emitted the child INSERT first and the database rejected it.
    Fixed with a db.flush() after the account, and documented at the top of
    models.py so the next same-flush pair does not rediscover it. The other three
    call sites already commit the parent first.
  - A write aimed at a since-deleted project used to leave an orphan row; with FKs
    enforced it would have been an IntegrityError surfacing as a 500, which the
    browser outbox retries forever (it only retires 4xx). require_project_writable
    now refuses a vanished project with 409, like the archived case beside it.

Verified: smoke test 27/27 exit 0 against a live server (the cascade assertion now
passes on SQLite, which is what used to fail); credentials missing and credentials
rejected both abort cleanly with exit 2 and no stray PASS lines; a project_user run
warns up front and fails as described. Scope tests 93/93, live HTTP checks 29/29,
static JS checks 33/33. No orphan rows left in the database afterwards.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
Nick Siegfried
2026-08-14 15:19:36 -05:00
parent c99ef08cf1
commit 64a5fd5612
6 changed files with 209 additions and 95 deletions
Download Patch File Download Diff File Expand all files Collapse all files

149
server/smoketest.py
View File

@@ -5,6 +5,24 @@ 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
@@ -13,16 +31,24 @@ USAGE
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
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 code 0 = all checks passed, 1 = one or more failed.
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
@@ -40,6 +66,18 @@ def check(name, cond, detail=""):
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."""
@@ -50,7 +88,7 @@ def call(method, path, body=None):
headers={"Content-Type": "application/json", "Accept": "application/json"},
)
try:
with urllib.request.urlopen(req, context=CTX, timeout=20) as r:
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
@@ -61,33 +99,95 @@ def call(method, path, body=None):
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
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.
# 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:
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
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) Create a project (writes to the projects table).
# 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",
@@ -96,14 +196,14 @@ def main():
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).
# 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 '?'}")
# 4) Create a SOP linked to the project.
# 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",
@@ -116,7 +216,7 @@ def main():
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).
# 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",
@@ -128,11 +228,11 @@ def main():
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).
# 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}")
# 7) Clear the constraint (upsert), then issue must SUCCEED (200, status Issued).
# 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",
@@ -146,16 +246,16 @@ def main():
f"status={st}")
check("issued_at timestamp is set", isinstance(issued, dict) and bool(issued.get("issued_at")))
# 8) Status transition endpoint.
# 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}")
# 9) Metrics aggregate for the project (Python aggregation over SQL rows).
# 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}")
# 10) Comment / feedback write + read.
# 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"})
@@ -164,11 +264,11 @@ def main():
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.
# 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}")
# 12) Archiving a project: it leaves the default list, stays reachable with
# 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})
@@ -193,7 +293,7 @@ def main():
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).
# 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}")
@@ -203,6 +303,15 @@ def main():
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.")