Project-SDE-WP-Suite/tests/critical_reopen_check.py

#!/usr/bin/env python3
"""Does the critical-reopen mail reach the PM and CM? — BL-021, fixed 2026-08-20.

`project_sop_team()` read `sop.data['project']`, but `pushSOP` stores every row
as `data={sop, state}` — the project block is one level deeper. The lookup
returned `[]` for every real row, so the on-hold email's recipient list was
silently reduced to assignee + distribution: the PM and CM named in
`notify_critical_reopen`'s own docstring never got it, from the day it shipped.

The fixture writes the PRODUCTION shape (nested under 'sop'), because a
hand-built flat row would have passed against the bug — which is exactly how it
went unverified this long. Sink pattern from qa_gate_check, one implementation.

Boots its own throwaway SQLite + uvicorn; run it alone, not back to back.
Exit 0 all passed, 1 a failure, 2 could not run.
"""
import io
import json
import os
import re
import sys
import tempfile

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 qa_gate_check import SmtpSink, api, wait_for                    # noqa: E402

ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))


def ascii_(v, n=280):
    return re.sub(r"\s+", " ", str(v)).encode("ascii", "replace").decode()[:n]


def set_team(pm_id, cm_id):
    """The PRODUCTION shape: data['sop']['project'], as pushSOP writes it."""
    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 {}))
        proj = data.setdefault("sop", {}).setdefault("project", {})
        proj["pmId"], proj["cmId"] = pm_id, cm_id
        sop.data = data
        db.commit()


def main():
    print("\n1. the read matches the written shape (static)")
    src = io.open(os.path.join(ROOT, "server", "app.py"), encoding="utf-8").read()
    fn = src[src.index("def project_sop_team"):src.index("def project_qa_group")]
    chk("project_sop_team reads the nested data['sop']['project'] first",
        '.get("sop")' in fn and '.get("project")' in fn)

    tmpdir = tempfile.mkdtemp(prefix="wpsuite-reopen-")
    db_path = os.path.join(tmpdir, "check.db")
    server = None
    sink = SmtpSink()
    sink.start()
    try:
        tok = seed(db_path)
        set_sop(db_path, {})
        set_team("user_sue", "user_pat")
        port = cdp.free_port()
        base = "http://127.0.0.1:%d" % port
        server = start_server(port, db_path)
        root = tok["root"]

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

        print("\n2. a released package, its critical constraint reopened")
        body = {
            "id": "wpCR1", "project_id": "projA", "number": "CR-01",
            "subject": "energize MCC-4", "status": "In Progress",
            "assignee_id": "user_mix",
            "data": {"location": "B-100 / Level 2",
                     "constraints": [{"name": "Power shutdown", "status": "cleared",
                                      "critical": True, "comment": ""}]}}
        code, _ = api(base, "/api/wps", root, "POST", body)
        chk("the package saves", code == 200, code)
        # The creation enqueues a wp_assigned mail delivered by a background
        # task; wait for it BEFORE clearing or it leaks into the reopen count.
        wait_for(lambda: len(sink.messages) >= 1, 10)
        code, _ = api(base, "/api/wps/wpCR1/status", root, "POST", {"status": "Issued"})
        chk("...and releases (the critical constraint is cleared)", code == 200, code)
        sink.messages.clear()

        body["status"] = "Issued"
        body["data"]["constraints"][0]["status"] = "open"
        code, wp = api(base, "/api/wps", root, "POST", body)
        chk("reopening the critical constraint saves through the normal upsert",
            code == 200, code)

        print("\n3. the mail reaches everyone the docstring promises")
        chk("three messages: assignee + PM + CM (the actor is excluded)",
            wait_for(lambda: len(sink.messages) == 3, 15), len(sink.messages))
        rcpts = sorted(m["to"][0] for m in sink.messages)
        chk("...the PM and CM are among them - THE BL-021 fix, sink-verified",
            "sue@example.test" in rcpts and "pat@example.test" in rcpts, ascii_(rcpts))
        chk("...and the assignee, exactly them, nobody twice",
            rcpts == ["mix@example.test", "pat@example.test", "sue@example.test"],
            ascii_(rcpts))
        text = next((m["text"] for m in sink.messages if "On hold:" in m["data"]), "")
        chk("the body names the constraint and the package, title included (CR-014)",
            "Power shutdown" in text and "CR-01" in text and "energize MCC-4" in text,
            ascii_(text, 300))
        chk("...and where the work happens, and a deep link to THAT package",
            "B-100 / Level 2" in text
            and "/wp-creation-index.html?project=projA&wp=wpCR1" in text, ascii_(text, 300))
        chk("...in the house convention (the footer this body alone used to lack)",
            "automated message from the Work Package Suite" in text)
        _, ev = api(base, "/api/audit?entity_type=wp&entity_id=wpCR1"
                          "&action=constraint_reopened", root)
        chk("the reopen is in the audit history", bool(ev), ascii_(ev[:1] if ev else ev))
    finally:
        sink.stop()
        if server:
            server.terminate()

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