mail-address-semantics/tests/test_modes_slave.py

"""Email as slave (CONTRACT.md 3.2) under a stub outside master (MAS-WP-0010 T06)."""
import pytest

from mas_modes.harness import Harness, ROOT

STATUTE = ROOT / "mas-modes-v0.1-spec/conformance/statute.json"
D = "modes.example"


def harness(**cfg):
    return Harness(STATUTE, {"case": {"role": "slave", "master_url": "{master}", **cfg}})


@pytest.fixture
def h():
    harness_ = harness()
    yield harness_
    harness_.close()


def intake(h, local="case", text="My order is late", **kw):
    d = h.mail(local, text, **kw)
    assert d.status == 202 and d.response["status"] == "intake-reported"
    [(code, body)] = h.master.pump()
    assert code == 200, body
    return body


def test_master_issues_the_case_id(h):
    d = h.mail("case", "My order is late")
    [msg] = h.master.of_type("case.intake")
    assert msg["intake_ref"] == d.response["intake_ref"] and msg["event"]["mas"]["intake_for"] == "case"
    assert h.outbox() == []                       # nothing sent before the master decides
    [(code, body)] = h.master.pump()
    assert body["case"] == {"address": f"case=EXT-100@{D}", "key": "case=ext-100"}
    d = h.mail("case=EXT-100", "more details")
    assert d.response["status"] == "reported"
    [m] = h.master.of_type("case.message")
    assert m["case"]["key"] == "case=ext-100" and m["event"]["message"]["text"].startswith("more details")
    assert h.outbox() == []                       # still silent: no autonomous sending


def test_email_never_invents_ids(h):
    h.mail("case", "hi")
    assert h.mail("case=EXT-999", "guess").response["status"] == "unknown-resource"
    assert h.master.of_type("case.message") == []


def test_assign_refusals(h):
    h.master.auto_assign = False
    ref = h.mail("case", "hi").response["intake_ref"]
    assert h.master.assign("nope", "X1")[0] == 404
    assert h.master.assign(ref, "bad+id")[0] == 400
    assert h.master.assign(ref, "X1")[0] == 200
    assert h.master.assign(ref, "X2")[1]["error"] == "already-assigned"
    ref2 = h.mail("case", "again").response["intake_ref"]
    assert h.master.assign(ref2, "x1")[1]["error"] == "case-exists"   # ids are case-insensitive here


def test_send_only_on_instruction(h):
    body = intake(h)
    case, mref = body["case"], body["message_ref"]
    code, resp = h.master.send(case, mref, "We are looking into it.")
    assert (code, resp) == (200, {"ok": True, "sent": True})
    [mail] = h.outbox()
    assert mail["mail_from"] == f"case=EXT-100@{D}" and mail["mail_to"] == "customer@example.net"
    assert mail["subject"] == "Re: Help needed"


@pytest.mark.parametrize("kw, reason", [
    ({"dmarc": "fail"}, "unauthenticated-sender"),
    ({"dmarc": None}, "unauthenticated-sender"),
    ({"headers": "Auto-Submitted: auto-replied\r\n"}, "automated-sender"),
])
def test_principle_10_refuses_master_sends(h, kw, reason):
    body = intake(h, **kw)
    code, resp = h.master.send(body["case"], body["message_ref"], "hello")
    assert code == 409 and resp["error"] == "refused" and resp["reason"] == reason
    assert h.outbox() == []


def test_send_cap_and_unknown_message():
    h = Harness(STATUTE, {"case": {"role": "slave", "master_url": "{master}"}}, max_sent=0)
    try:
        body = intake(h)
        assert h.master.send(body["case"], body["message_ref"], "x")[1]["reason"] == "send-cap-reached"
        assert h.master.send(body["case"], "m-unknown", "x")[1]["reason"] == "unknown-message"
    finally:
        h.close()


def test_autonomous_policy_acknowledges():
    h = harness(autonomous_send=True)
    try:
        body = intake(h)
        assert body["acknowledged"] is True
        [ack] = h.outbox()
        assert "case EXT-100" in ack["body"] and ack["mail_from"] == f"case=EXT-100@{D}"
    finally:
        h.close()


def test_transfer_blocked_by_default(h):
    body = intake(h, local="case!callback")
    [req] = h.master.of_type("handoff.requested")
    assert req["forwarded"] is False and req["handoff_id"] == body["handoff_id"]
    [st] = h.master.of_type("handoff.status")
    assert (st["status"], st["reason"]) == ("refused", "blocked")
    assert h.callback.received == [] and h.outbox() == []


def test_transfer_to_primed_webhook():
    h = harness(transfers={"webhook": "{callback}"})
    try:
        body = intake(h, local="case!callback", text="Phone: 123\nTime: now")
        assert h.master.of_type("handoff.requested")[0]["forwarded"] is True
        [req] = h.callback.received
        assert req["case"]["key"] == "case=ext-100" and req["details"] == {"phone": "123", "time": "now"}
        h.callback.run()
        relayed = h.master.of_type("handoff.status")
        assert [m["status"] for m in relayed] == ["scheduled", "running", "completed"]
        assert [m["seq"] for m in relayed] == [1, 2, 3]
        assert h.outbox() == []                   # resuming is the master's job
        assert h.advance(10 ** 6) == []           # and so are timeouts
    finally:
        h.close()


def test_policy_from_assign_and_later_update(h):
    h.master.auto_assign = False
    ref = h.mail("case!callback", "Phone: 1\nTime: 2").response["intake_ref"]
    code, body = h.master.command("case.assign", intake_ref=ref, resource_id="A1",
                                  policy={"transfers": "block", "autonomous_send": False})
    assert code == 200 and h.master.of_type("handoff.requested")[0]["forwarded"] is False
    code, body = h.master.policy(body["case"], autonomous_send=True)
    assert body["policy"] == {"autonomous_send": True, "transfers": "block"}


def test_negotiation_only_when_autonomous():
    for autonomous, mails in ((False, 0), (True, 2)):     # acknowledgement + question
        h = harness(transfers={"webhook": "{callback}"}, autonomous_send=autonomous)
        try:
            h.callback.default = "negotiate"
            intake(h, local="case!callback", text="call me")
            h.callback.run()
            assert h.master.of_type("handoff.status")[-1]["status"] == "negotiating"
            assert len(h.outbox()) == mails
        finally:
            h.close()


def test_master_cancels_on_its_own_timeout():
    h = harness(transfers={"webhook": "{callback}"}, fields=("phone", "time", "scenario"))
    try:
        body = intake(h, local="case!callback", text="Phone: 1\nTime: 2\nScenario: silent")
        assert h.callback.run() == []
        code, resp = h.master.cancel(body["case"], body["handoff_id"])
        assert code == 200 and h.callback.received[-1]["type"] == "handoff.cancel"
        h.callback.run()
        assert h.master.of_type("handoff.status")[-1]["status"] == "cancelled"
    finally:
        h.close()