Files
Project-SDE-WP-Suite/tests/qa_gate_check.py
n.siegfried 2486f87010 T7.6 - CR-014/D2/D9/D10: the Ready for QA gate, notification only, shipped off
Marlena's ask: QA sees inbound work ahead of time, not after the fact.

The rung: 'Ready for QA' sits between In Progress and QC in BOTH ladders
(wp-creation-app.js STATUS_ORDER, server/app.py STATUS_ORDER) and in Field
View's list - inside the T7.3 transition model, not beside it: entering it
from an unreleased state crosses the release gates, and 'Issue' (hold) stays
a branch. The dashboard filter and the navigator grouping learned the state
from the ladder without their own edits.

Who hears (D2): the QA GROUP, a multi-pick of project members on the SOP
wizard's team step, stored as account ids at data.sop.project.qaGroupIds.
Entering Ready for QA emails that list and nobody else. A rejection emails
the owner AND the same list (amended answer), returns the package to In
Progress, and REQUIRES a fresh comment - server-enforced on both write paths
(the first version accepted any old comment already on the record, which made
every rejection after the first one free; the gate now demands a new entry).
Accept and reject are real buttons on the release banner; the comment modal
enforces its field; qa_ready / qa_rejected / status_changed all land in the
audit history.

The link (X1): wp_link() now opens THE package - wp-creation-index.html
?project&wp=<id>, which the creator boots directly and login.html?next=
round-trips for a signed-out recipient. It previously pointed at the suite
root, which is exactly the failure X1 names; assignment mail inherits the fix.

Email discipline (D10 + standing rules): ships OFF (the stored setting the
admin console already owns; PUT /api/settings is admin-only, 403 for anyone
else, and audited). With it off, transitions write outbox rows marked
'skipped' and the sink receives nothing. With it on, the probe runs a REAL
SMTP conversation against an in-process capture sink and asserts the count
and the exact recipient set. A dead SMTP host leaves a 'failed' outbox row
with the error recorded. The SMTP password exists only in the environment.

DEVIATION, stated: the task's Do-paragraph asks the email to include location
and a scope summary; the done-when list (and CLAUDE.md) says no customer IP in
a message body. The done-when wins: bodies carry the WP number, who moved it,
and the deep link. A location canary planted on the package is asserted absent
from every captured message. If the fuller body is wanted, that is a product
call - needs Nick.

Found while building, logged not fixed (BL-021): project_sop_team() reads
sop.data['project'], a path pushSOP never writes - the critical-reopen email
has never actually reached the PM/CM. One-line fix, owned by T9.9.

Field View (D9): 'Ready for QA' is carried by TEXT on the card at 390px.

Verification (each probe run alone): NEW tests/qa_gate_check.py 40/40.
Regressions: hold_check 50/50, pipeline_check 44/44, aggregates_check 16/16,
frame_check 39/39, validation_check 83/83.

Items: CR-014, D2, D9, D10 (X1, X3 honored)

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
2026-08-19 10:54:07 -07:00

439 lines
20 KiB
Python

#!/usr/bin/env python3
"""Does the QA gate notify the right people, and only when told to? — CR-014, T7.6.
Marlena's ask: QA needs to see inbound work ahead of time, not be told after the
fact. `Ready for QA` is a rung between In Progress and QC; entering it emails the
QA group configured on the SOP (D2); a rejection returns the package to In
Progress with a mandatory comment and emails the owner plus the same group.
Email discipline, per the standing rules and D10:
- ships OFF: with the toggle off, transitions write outbox rows marked
'skipped' and the capture sink receives NOTHING
- only an administrator can turn it on, from the stored setting (not an env
var); the change lands in the audit log
- the capture sink is a real SMTP conversation on localhost - the count and
the recipients of captured messages are asserted, and no real mail exists
- the SMTP password lives in the environment alone; message bodies carry a WP
number and a deep link, never package contents (the location canary proves
it)
Self-contained: throwaway SQLite, its own uvicorn, its own SMTP sink, headless
browser for the client half. Exit 0 all passed, 1 a failure, 2 could not run.
"""
import json
import os
import re
import socket
import sys
import tempfile
import threading
import time
import urllib.error
import urllib.request
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
import cdp # noqa: E402
from browser_check import seed, start_server, chk, _PASS, _FAIL # noqa: E402
from sections_check import set_sop # noqa: E402
from stepper_check import dismiss_dialogs # noqa: E402
ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
CANARY = "FAB-9 SECRET SECTOR" # rides on the package; must never reach a body
def ascii_(v, n=300):
return re.sub(r"\s+", " ", str(v)).encode("ascii", "replace").decode()[:n]
def settle(seconds=0.6):
time.sleep(seconds)
def api(base, path, token, method="GET", body=None):
req = urllib.request.Request(base + path, method=method)
req.add_header("Cookie", "wp_session=" + token)
req.add_header("Accept", "application/json")
data = None
if body is not None:
data = json.dumps(body).encode()
req.add_header("Content-Type", "application/json")
try:
with urllib.request.urlopen(req, data, timeout=15) as r:
return r.status, json.loads(r.read().decode() or "null")
except urllib.error.HTTPError as e:
try:
return e.code, json.loads(e.read().decode() or "null")
except Exception:
return e.code, None
def audit(base, tok, wp_id, action):
_, rows = api(base, "/api/audit?entity_type=wp&entity_id=%s&action=%s" % (wp_id, action), tok)
return rows or []
class SmtpSink(threading.Thread):
"""The smallest SMTP server that satisfies smtplib: enough protocol to accept
a message and record it. TLS is refused by omission (the settings turn it
off), auth is never requested. Everything captured stays in memory."""
daemon = True
def __init__(self):
super().__init__()
self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
self.sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
self.sock.bind(("127.0.0.1", 0))
self.sock.listen(8)
self.port = self.sock.getsockname()[1]
self.messages = [] # {"to": [...], "data": str}
self._stop = False
def run(self):
self.sock.settimeout(0.5)
while not self._stop:
try:
conn, _ = self.sock.accept()
except socket.timeout:
continue
except OSError:
break
try:
self._serve(conn)
except Exception:
pass
finally:
try:
conn.close()
except Exception:
pass
def stop(self):
self._stop = True
try:
self.sock.close()
except Exception:
pass
def _serve(self, conn):
conn.settimeout(10)
f = conn.makefile("rb")
conn.sendall(b"220 sink ready\r\n")
rcpt, in_data, buf = [], False, []
while True:
line = f.readline()
if not line:
return
if in_data:
if line.rstrip(b"\r\n") == b".":
self.messages.append({"to": list(rcpt),
"data": b"".join(buf).decode("utf-8", "replace")})
rcpt, in_data, buf = [], False, []
conn.sendall(b"250 OK\r\n")
else:
buf.append(line)
continue
cmd = line.decode("utf-8", "replace").strip()
up = cmd.upper()
if up.startswith(("EHLO", "HELO")):
conn.sendall(b"250-sink\r\n250 OK\r\n")
elif up.startswith("MAIL FROM"):
conn.sendall(b"250 OK\r\n")
elif up.startswith("RCPT TO"):
m = re.search(r"<([^>]+)>", cmd)
rcpt.append(m.group(1) if m else cmd)
conn.sendall(b"250 OK\r\n")
elif up.startswith("DATA"):
in_data = True
conn.sendall(b"354 go\r\n")
elif up.startswith("QUIT"):
conn.sendall(b"221 bye\r\n")
return
else:
conn.sendall(b"250 OK\r\n")
def wait_for(fn, timeout=12.0):
end = time.time() + timeout
while time.time() < end:
if fn():
return True
time.sleep(0.3)
return False
def set_qa_group(ids):
"""D2's field, as buildSOP() writes it: data['sop']['project']['qaGroupIds']."""
from server.db import SessionLocal
from server import models
with SessionLocal() as db:
sop = db.get(models.Sop, "sopA")
data = json.loads(json.dumps(sop.data or {}))
data.setdefault("sop", {}).setdefault("project", {})["qaGroupIds"] = ids
sop.data = data
db.commit()
def mkwp(base, tok, wp_id, status="In Progress", extra_data=None, assignee=None):
data = {"constraints": [{"name": "Boom lift", "status": "cleared", "comment": ""}],
"location": CANARY, "desc": "600ft of 3/4 EMT through " + CANARY}
data.update(extra_data or {})
return api(base, "/api/wps", tok, "POST", {
"id": wp_id, "project_id": "projA", "number": "QA-" + wp_id[-2:],
"subject": "conduit " + CANARY, "status": status,
"assignee_id": assignee, "data": data})
def main():
exe = cdp.find_browser()
if not exe:
print("no headless-capable browser found; set WP_BROWSER.")
return 2
tmpdir = tempfile.mkdtemp(prefix="wpsuite-qa-")
db_path = os.path.join(tmpdir, "check.db")
server = None
browser = None
sink = SmtpSink()
sink.start()
try:
tok = seed(db_path)
set_sop(db_path, {})
set_qa_group(["user_sue", "user_pat"])
port = cdp.free_port()
base = "http://127.0.0.1:%d" % port
server = start_server(port, db_path)
root, pat = tok["root"], tok["pat"]
# ── 1. the rung, and the ship-off default ────────────────────────────
print("\n1. the new state, with email OFF (as shipped)")
code, _ = mkwp(base, root, "wpQA1", assignee="user_mix")
chk("a package saves at In Progress", code == 200, code)
code, _ = api(base, "/api/wps/wpQA1/status", root, "POST", {"status": "Ready for QA"})
chk("Ready for QA is a legal transition from In Progress", code == 200, code)
chk("...and it is written to history",
bool(audit(base, root, "wpQA1", "qa_ready")), )
_, rows = api(base, "/api/notifications?all=true", root)
qa_rows = [r for r in (rows or []) if r.get("kind") == "qa_ready"]
chk("with email OFF the outbox records the sends as skipped, not silent",
qa_rows and all(r.get("status") == "skipped" for r in qa_rows),
ascii_([(r.get("email"), r.get("status")) for r in qa_rows]))
chk("...and the capture sink received NOTHING", len(sink.messages) == 0,
len(sink.messages))
code, _ = api(base, "/api/wps/wpQA1/status", root, "POST",
{"status": "Draft"})
chk("the rung is inside the transition model: leaving it re-crosses the "
"release gate on the way back up",
api(base, "/api/wps/wpQA1/status", root, "POST",
{"status": "Ready for QA"})[0] == 200)
# ── 2. D10: who may turn it on ────────────────────────────────────────
print("\n2. D10: the switch")
code, _ = api(base, "/api/settings", pat, "PUT", {"email_enabled": True})
chk("a non-administrator cannot turn email on", code == 403, code)
code, _ = api(base, "/api/settings", root, "PUT", {
"email_enabled": True, "smtp_host": "127.0.0.1", "smtp_port": sink.port,
"smtp_use_tls": False, "from_addr": "suite@sink.local",
"app_base_url": base})
chk("an administrator can", code == 200, code)
_, ev = api(base, "/api/audit?entity_type=settings&action=settings_updated", root)
chk("...and the change is in the audit log",
ev and ev[0].get("detail", {}).get("email_enabled") is True, ascii_(ev[:1]))
_, pub = api(base, "/api/settings", root)
chk("the settings API never returns an SMTP password, only whether one is set",
"smtp_password" not in (pub or {}) and "smtp_password_set" in (pub or {}),
ascii_(sorted((pub or {}).keys())))
src = open(os.path.join(ROOT, "server", "notify.py"), encoding="utf-8").read()
chk("the password is read from the environment and stored nowhere",
"SMTP_PASSWORD" in src and "smtp_password" not in json.dumps(
__import__("server.notify", fromlist=["DEFAULTS"]).DEFAULTS))
# ── 3. entering Ready for QA emails the group, and nobody else ───────
print("\n3. the send, against the sink")
code, _ = mkwp(base, root, "wpQA2", assignee="user_mix")
code, _ = api(base, "/api/wps/wpQA2/status", root, "POST", {"status": "Ready for QA"})
chk("the transition succeeds with email on", code == 200, code)
# The wp was created WITH an assignee, so one wp_assigned mail rides along
# - correct and its own feature. This section asserts the QA mails alone.
qa_msgs = lambda: [m for m in sink.messages if "Ready for QA:" in m["data"]]
chk("the sink captures exactly the QA group - two Ready-for-QA messages",
wait_for(lambda: len(qa_msgs()) == 2, 12), len(qa_msgs()))
rcpts = sorted(m["to"][0] for m in qa_msgs())
chk("...addressed to the group members and NOBODY else",
rcpts == ["pat@example.test", "sue@example.test"], ascii_(rcpts))
body = qa_msgs()[0]["data"] if qa_msgs() else ""
chk("the message carries the WP number", "QA-A2" in body, ascii_(body, 200))
chk("...and a link that opens THAT work package, not the app root",
"/wp-creation-index.html?project=projA&wp=wpQA2" in body, ascii_(body, 400))
chk("...and no customer IP: the location canary does not appear",
CANARY not in body and all(CANARY not in m["data"] for m in sink.messages))
chk("...and no SMTP password either", "SMTP_PASSWORD" not in body
and os.getenv("SMTP_PASSWORD", "hunter2-not-set") not in body)
# ── 4. rejection ──────────────────────────────────────────────────────
print("\n4. rejection: comment required, owner + group notified")
sink.messages.clear()
code, _ = api(base, "/api/wps/wpQA2/status", root, "POST", {"status": "In Progress"})
chk("a rejection with no comment is refused", code == 409, code)
code, _ = api(base, "/api/wps/wpQA2/status", root, "POST",
{"status": "In Progress", "comment": "Torque strap missing on run 4"})
chk("with a comment it returns to In Progress", code == 200, code)
rows = audit(base, root, "wpQA2", "qa_rejected")
chk("...the rejection is in history WITH the comment",
rows and "Torque strap" in (rows[0].get("detail", {}).get("comment") or ""),
ascii_(rows[:1]))
chk("...the owner and the same group are emailed - three messages",
wait_for(lambda: len(sink.messages) == 3, 12), len(sink.messages))
rcpts = sorted(m["to"][0] for m in sink.messages)
chk("...exactly them", rcpts == ["mix@example.test", "pat@example.test",
"sue@example.test"], ascii_(rcpts))
chk("...and the comment itself stays on the package, out of the mail",
all("Torque strap" not in m["data"] for m in sink.messages))
# The upsert path enforces the same comment rule (it is how the browser saves).
code, _ = api(base, "/api/wps/wpQA2/status", root, "POST", {"status": "Ready for QA"})
_, wp = api(base, "/api/wps/wpQA2", root)
code, _ = api(base, "/api/wps", root, "POST", {
"id": "wpQA2", "project_id": "projA", "number": wp["number"],
"subject": wp["subject"], "status": "In Progress", "data": wp["data"]})
chk("the upsert refuses the same no-comment rejection - it is not a side door",
code == 409, code)
# ── 5. failure is recorded, not silent ────────────────────────────────
print("\n5. a dead SMTP host is a recorded failure")
sink.messages.clear()
dead = cdp.free_port()
api(base, "/api/settings", root, "PUT", {"smtp_host": "127.0.0.1",
"smtp_port": dead})
code, _ = mkwp(base, root, "wpQA3")
api(base, "/api/wps/wpQA3/status", root, "POST", {"status": "Ready for QA"})
def failed_rows():
_, rows = api(base, "/api/notifications?all=true", root)
return [r for r in (rows or [])
if r.get("kind") == "qa_ready" and r.get("status") == "failed"]
chk("the outbox row is marked failed with the error recorded",
wait_for(lambda: len(failed_rows()) >= 1, 15)
and bool(failed_rows()[0].get("error")), ascii_(failed_rows()[:1]))
api(base, "/api/settings", root, "PUT", {"smtp_host": "127.0.0.1",
"smtp_port": sink.port})
# ── 6. the client: pill, banner, accept/reject, deep link, field view ─
print("\n6. the browser half")
browser = cdp.Browser(exe)
page = browser.page()
page.clear_cookies()
page.set_cookie("wp_session", root)
page.viewport(1440, 900)
page.goto(base + "/wp-creation-index.html?project=projA&wp=wpQA2")
dismiss_dialogs(page)
ok = wait_for(lambda: page.eval("!!window.wpCreatorReady"), 15)
settle(1.5)
chk("the creator boots on an emailed deep link and opens THAT package",
ok and page.eval("editingId") == "wpQA2", page.eval("editingId"))
page.eval("window.prompt=()=>null; window.alert=()=>{}; window.confirm=()=>false;")
chk("the status control has a real Ready for QA pill",
page.eval("!!document.querySelector('.radio-pill[data-val=\"Ready for QA\"] input')"))
chk("the dashboard status filter learned the new state on its own",
page.eval("STATUS_ORDER.includes('Ready for QA')"))
# The creator expands the record to the SOP's constraint set, whose
# defaults are OPEN - so Ready for QA is (correctly) refused until they
# are dealt with. Clear them the way a user would before the QA flow.
n = page.eval("pkgConstraints.length")
for i in range(n):
page.eval("""(() => {
const tr = document.querySelectorAll('#constraint-body tr')[%d];
[...tr.querySelectorAll('.cstatus button')]
.find(b => b.textContent.trim() === 'N/A').click();
})()""" % i)
settle(0.15)
page.eval("document.querySelector('.radio-pill[data-val=\"Ready for QA\"]').click()")
settle(0.5)
chk("selecting it shows the QA banner with accept and reject as real buttons",
page.eval("document.querySelectorAll('#release-banner .rb-act').length") == 2)
page.eval("qaAccept()")
settle(0.3)
chk("accept moves to QC", page.eval("getRadio('status')") == "QC")
page.eval("document.querySelector('.radio-pill[data-val=\"Ready for QA\"]').click()")
settle(0.4)
page.eval("qaRejectOpen()")
chk("reject opens the comment modal",
page.eval("document.getElementById('qa-reject-modal').classList.contains('open')"))
page.eval("qaRejectSubmit()")
settle(0.2)
chk("an empty comment does not pass",
page.eval("getRadio('status')") == "Ready for QA")
page.eval("document.getElementById('qa-reject-comment').value='Redo the strapping'")
page.eval("qaRejectSubmit()")
settle(0.3)
chk("with a comment the package returns to In Progress",
page.eval("getRadio('status')") == "In Progress")
chk("...and the comment is on the package",
"Redo the strapping" in (page.eval(
"JSON.stringify(pkgQaRejections[pkgQaRejections.length-1]||{})") or ""))
# Field View (D9): the state is text on the card, legible at 390px.
page.viewport(390, 900, mobile=True)
page.goto(base + "/field.html?project=projA")
dismiss_dialogs(page)
settle(2.5)
pill = json.loads(page.eval("""JSON.stringify((() => {
const el = [...document.querySelectorAll('.pill')]
.find(x => x.textContent.trim() === 'Ready for QA');
if (!el) return null;
const cs = getComputedStyle(el);
return {size: parseFloat(cs.fontSize), text: el.textContent.trim()};
})())"""))
chk("390px Field View: 'Ready for QA' is carried by TEXT on the card (D9)",
pill is not None, ascii_(pill))
chk("...at a legible size", bool(pill) and pill["size"] >= 11, ascii_(pill))
# The wizard's D2 field: rendered from project members, feeds state.
page.viewport(1440, 900)
page.goto(base + "/work-package-suite.html?tab=sop")
dismiss_dialogs(page)
settle(2.5)
chk("the SOP wizard has the QA group field, populated from project members",
page.eval("(document.getElementById('proj_qagroup')||{options:[]}).options.length") > 0,
page.eval("(document.getElementById('proj_qagroup')||{options:[]}).options.length"))
page.eval("""(() => {
const sel = document.getElementById('proj_qagroup');
[...sel.options].forEach(o => o.selected = ['user_sue','user_pat'].includes(o.value));
sel.dispatchEvent(new Event('change'));
})()""")
chk("picking people lands in the SOP state that completeSOP() persists",
sorted(json.loads(page.eval("JSON.stringify(state.qaGroupIds||[])")))
== ["user_pat", "user_sue"],
page.eval("JSON.stringify(state.qaGroupIds||[])"))
js_errors = [e for e in page.js_errors()]
chk("no JavaScript errors anywhere in the browser half", not js_errors,
ascii_(js_errors[:2]))
finally:
sink.stop()
if browser is not None:
try:
browser.close()
except Exception:
pass
if server is not None:
try:
server.terminate()
except Exception:
pass
print("\n" + "-" * 54)
print("%d/%d checks passed." % (len(_PASS), len(_PASS) + len(_FAIL)))
for f in _FAIL:
print(" - " + f)
return 1 if _FAIL else 0
if __name__ == "__main__":
sys.exit(main())