mail-address-semantics/mas_modes/runner.py

"""Run the mode-contract conformance bundle against this email side.

Message fixtures and the transition table are checked against the schema
and `contract.allowed`; flows run through `harness.Harness` with the real
gateway, the callback stub and the stub outside master. Used by `mas test`
and pytest.
"""

from __future__ import annotations

import json
from collections import Counter
from dataclasses import dataclass, field
from importlib import resources

from . import flow_fixtures, message_errors, message_fixtures, transitions
from .contract import allowed
from .harness import Harness

MASTER = {"role": "master", "modes": {"callback": "{callback}"},
          "fields": ("phone", "time", "scenario"), "timeout": 3600}
SLAVE = {"role": "slave", "master_url": "{master}"}


@dataclass
class Report:
    passed: int = 0
    failures: list = field(default_factory=list)

    @property
    def ok(self) -> bool:
        return not self.failures


def check_messages(report: Report):
    fx = message_fixtures()
    for c in fx["valid"]:
        errs = message_errors(c["message"])
        (report.failures.append((c["case_id"], f"should be valid: {errs}")) if errs
         else _ok(report))
    for c in fx["invalid"]:
        (report.failures.append((c["case_id"], f"should be invalid ({c['why']})"))
         if not message_errors(c["message"]) else _ok(report))
    t = transitions()
    statuses = set(t["terminal"]) | set(t["transitions"]) - {"(new)"}
    for old in [None, *sorted(statuses)]:
        for new in sorted(statuses):
            want = new in t["transitions"].get(old or "(new)", [])
            if allowed(old, new) != want:
                report.failures.append((f"transition {old}->{new}", f"expected allowed={want}"))
            else:
                _ok(report)


def _ok(report):
    report.passed += 1


def _fill(value, ctx):
    if isinstance(value, str):
        for k, v in ctx.items():
            value = value.replace("{" + k + "}", str(v))
        return value
    if isinstance(value, dict):
        return {k: _fill(v, ctx) for k, v in value.items()}
    return value


def _statute_path():
    return resources.files(__package__).joinpath("conformance/statute.json")


def run_flow(flow: dict) -> list[str]:
    """Returns the problems found; empty when the flow passes."""
    with resources.as_file(_statute_path()) as statute:
        if flow["role"] == "master":
            roles = {"ticket": MASTER}
        else:
            roles = {"case": {**SLAVE, **flow.get("config", {})}}
        h = Harness(statute, roles, scenario=flow.get("scenario", "completed"), auto_assign=True)
    ctx, problems = {}, []
    last_command = None
    timed_out: list = []
    try:
        for n, step in enumerate(flow["steps"], 1):
            resp = None
            if "mail" in step:
                m = _fill(step["mail"], ctx)
                d = h.mail(m["to"], m.get("text", ""), dmarc=m.get("dmarc", "pass"))
                resp = d.response if isinstance(d.response, dict) else {}
            elif "mode" in step:
                h.callback.run()
            elif "advance" in step:
                timed_out = h.advance(step["advance"])
            elif "pump" in step:
                results = h.master.pump()
                resp = results[-1][1] if results else {}
            elif "master" in step:
                cmd = _fill(step["master"], {**ctx, "callback": ""})
                case = {"address": ctx["case_address"], "key": ctx["case_key"]}
                if cmd["command"] == "send":
                    last_command = h.master.send(case, cmd["in_reply_to"], cmd["body"])
                elif cmd["command"] == "cancel":
                    last_command = h.master.cancel(case, cmd["handoff_id"])
            if isinstance(resp, dict):
                if resp.get("ticket_id"):
                    ctx["case_id"] = resp["ticket_id"]
                if isinstance(resp.get("case"), dict):
                    ctx["case_id"] = resp["case"]["address"].split("=", 1)[1].split("@")[0]
                    ctx["case_address"], ctx["case_key"] = resp["case"]["address"], resp["case"]["key"]
                for k in ("message_ref", "handoff_id"):
                    if resp.get(k):
                        ctx[k] = resp[k]
            for key, want in step.get("expect", {}).items():
                got = _observe(h, key, resp, last_command, timed_out)
                if key == "master_received":
                    got = {t: got.get(t, 0) for t in want}
                elif key == "last_mail_contains":
                    got, want = (want in (got or "")), True
                if got != want:
                    problems.append(f"step {n} {key}: got {got!r}, expected {want!r}")
    finally:
        h.close()
    return problems


def _observe(h, key, resp, last_command, timed_out):
    if key == "response_status":
        return (resp or {}).get("status")
    if key == "statuses":
        return [m["status"] for hd in h.callback.handoffs.values() for m in hd.sent]
    if key == "relayed":
        return [m["status"] for m in h.master.of_type("handoff.status")]
    if key == "outbox_count":
        return len(h.outbox())
    if key == "last_mail_contains":
        mails = h.outbox()
        return mails[-1]["body"] if mails else None
    if key == "master_received":
        return dict(Counter(m["type"] for m in h.master.received))
    if key == "mode_received":
        return len(h.callback.received)
    if key == "mode_received_last":
        return h.callback.received[-1]["type"] if h.callback.received else None
    if key == "timed_out":
        return len(timed_out)
    if key == "case_address":
        return (resp or {}).get("case", {}).get("address")
    if key == "command_ok":
        return (last_command[1] or {}).get("ok")
    if key == "command_reason":
        return (last_command[1] or {}).get("reason")
    raise KeyError(f"unknown expectation {key!r}")


def run_mode_corpus() -> Report:
    report = Report()
    check_messages(report)
    for flow in flow_fixtures():
        problems = run_flow(flow)
        if problems:
            report.failures.append((flow["id"], "; ".join(problems)))
        else:
            report.passed += 1
    return report


if __name__ == "__main__":
    r = run_mode_corpus()
    print(json.dumps({"passed": r.passed, "failures": r.failures}, indent=2))