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