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

#!/usr/bin/env python3
"""Does a kitting change tell the right people, once? — CR-011 / D10, T8.3.

The field learns when material moves without chasing it in Teams. A kitting
status change emails the package's distribution list plus its warehouse owner
(CR-010's default recipient), through the SAME gate T7.6 built - off by
default, admin-only, every send an outbox row, no real mail anywhere.

Coalescing: rapid consecutive changes rewrite the held (unsent) notification
to the newest transition instead of stacking near-identical siblings.

Reuses qa_gate_check's SMTP sink - one sink implementation, not two.
Exit 0 all passed, 1 a failure, 2 could not run.
"""
import json
import os
import re
import sys
import tempfile
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 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=300):
    return re.sub(r"\s+", " ", str(v)).encode("ascii", "replace").decode()[:n]


def save_wp(base, tok, kit_status, extra=None):
    data = {"constraints": [{"name": "Materials", "status": "cleared", "comment": ""}],
            "kitStatus": kit_status, "kitOwner": "Sue", "kitOwnerId": "user_sue",
            "distributionIds": ["user_pat"], "mimoLoc": "Staging 04, dock B"}
    data.update(extra or {})
    return api(base, "/api/wps", tok, "POST", {
        "id": "wpKN1", "project_id": "projA", "number": "KN-1",
        "subject": "kit notify host", "status": "In Progress", "data": data})


def main():
    exe = cdp.find_browser()   # not used, but keeps environment parity checks
    tmpdir = tempfile.mkdtemp(prefix="wpsuite-kitnotify-")
    db_path = os.path.join(tmpdir, "check.db")
    server = None
    sink = SmtpSink()
    sink.start()
    try:
        tok = seed(db_path)
        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. shipped OFF, and the burst coalesces while held ───────────────
        print("\n1. off by default, and no burst")
        save_wp(base, root, "Not Started")
        for st in ("Picking", "Staged", "In Transit"):     # three rapid changes
            code, _ = save_wp(base, root, st)
            chk("kitting change to %s saves" % st, code == 200, code)
        _, rows = api(base, "/api/notifications?all=true", root)
        kit_rows = [r for r in (rows or []) if r.get("kind") == "kitting_status"]
        chk("with email OFF the rows are skipped, not silent, and COALESCED - "
            "one per recipient, not one per change",
            len(kit_rows) == 2 and all(r.get("status") == "skipped" for r in kit_rows),
            ascii_([(r.get("email"), r.get("status")) for r in kit_rows]))
        chk("...rewritten to the NEWEST transition",
            all("In Transit" in (r.get("subject") or "") for r in kit_rows),
            ascii_([r.get("subject") for r in kit_rows]))
        chk("...and the sink received nothing", len(sink.messages) == 0)
        _, ev = api(base, "/api/audit?entity_type=wp&entity_id=wpKN1"
                          "&action=kitting_status_changed", root)
        chk("every change is in the audit history, uncoalesced",
            len(ev or []) == 3, len(ev or []))

        # ── 2. the T7.6 gate, reused ─────────────────────────────────────────
        print("\n2. the same gate")
        src = open(os.path.join(ROOT, "server", "notify.py"), encoding="utf-8").read()
        chk("no second email flag was invented - the T7.6 gate is THE gate",
            src.count("email_enabled") >= 1
            and "kitting_email" not in src
            and "kitting_email" not in open(os.path.join(ROOT, "server", "app.py"),
                                            encoding="utf-8").read())
        code, _ = api(base, "/api/settings", pat, "PUT", {"email_enabled": True})
        chk("a non-administrator still cannot turn it on (D10)", 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)

        # ── 3. the send ──────────────────────────────────────────────────────
        print("\n3. the send, against the sink")
        code, _ = save_wp(base, root, "Delivered")
        chk("the change lands", code == 200, code)
        chk("the sink captures the distribution list + the warehouse owner: two",
            wait_for(lambda: len(sink.messages) == 2, 12), len(sink.messages))
        rcpts = sorted(m["to"][0] for m in sink.messages)
        chk("...exactly them, actor excluded",
            rcpts == ["pat@example.test", "sue@example.test"], ascii_(rcpts))
        body = sink.messages[0]["data"] if sink.messages else ""
        chk("the mail says old status, new status and who",
            "In Transit" in body and "Delivered" in body and "Root" in body,
            ascii_(body, 260))
        chk("...and the delivery location", "Staging 04, dock B" in body)
        chk("...and a deep link to THAT package, not the app root",
            "/wp-creation-index.html?project=projA&wp=wpKN1" in body)
        chk("...in the house convention (greeting + automated-message footer)",
            body.count("Hi ") >= 1 and "automated message from the Work Package Suite" in body)

    finally:
        sink.stop()
        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())