channel-exit/tests/test_providers.py

"""What actually goes on the wire to SendGrid and Twilio.

Asserted at the seam rather than mocked at the module boundary, because the
shape of these two requests is the part a provider silently rejects: a From
that is not the verified sender, a field Twilio spells `To` and not `to`.
"""

from __future__ import annotations

import base64
import json
import urllib.parse

from app import providers


def recorder(status: int, body: str = "{}", reply_headers: dict | None = None):
    calls: list[dict] = []

    def fake_post(url: str, *, data: bytes, headers: dict, timeout: float):
        calls.append({"url": url, "data": data, "headers": headers, "timeout": timeout})
        return providers.Response(status, body, reply_headers or {})

    fake_post.calls = calls  # type: ignore[attr-defined]
    return fake_post


def test_email_is_addressed_from_the_verified_sender() -> None:
    post = recorder(202, "", {"X-Message-Id": "abc"})
    outcome = providers.send_email(
        to="[email protected]",
        subject="Due",
        body="Bins",
        api_key="sg-key",
        sender="[email protected]",
        timeout=5,
        post=post,
    )
    sent = json.loads(post.calls[0]["data"])
    assert post.calls[0]["url"] == providers.SENDGRID_URL
    assert sent["from"]["email"] == "[email protected]"
    assert sent["personalizations"][0]["to"] == [{"email": "[email protected]"}]
    assert sent["content"][0]["type"] == "text/plain"
    assert post.calls[0]["headers"]["Authorization"] == "Bearer sg-key"
    assert outcome.ok and outcome.status == 202 and outcome.message_id == "abc"


def test_an_email_with_no_subject_still_sends() -> None:
    post = recorder(202)
    providers.send_email(
        to="[email protected]", subject="", body="Bins", api_key="k", sender="[email protected]", timeout=5, post=post
    )
    assert json.loads(post.calls[0]["data"])["subject"] == "Notification"


def test_sendgrid_refusal_is_reported_not_raised() -> None:
    post = recorder(403, json.dumps({"errors": [{"message": "The from address does not match a verified Sender"}]}))
    outcome = providers.send_email(
        to="[email protected]", subject="s", body="b", api_key="k", sender="[email protected]", timeout=5, post=post
    )
    assert not outcome.ok
    assert outcome.status == 403
    assert "verified Sender" in outcome.detail


def test_sms_is_form_encoded_with_twilio_field_names() -> None:
    post = recorder(201, json.dumps({"sid": "SM9"}))
    outcome = providers.send_sms(
        to="+15551234567",
        subject="Due",
        body="Bins",
        account_sid="AC123",
        auth_token="secret",
        sender="+15550001111",
        timeout=5,
        post=post,
    )
    fields = dict(urllib.parse.parse_qsl(post.calls[0]["data"].decode()))
    assert post.calls[0]["url"].endswith("/Accounts/AC123/Messages.json")
    assert fields == {"To": "+15551234567", "From": "+15550001111", "Body": "Due\n\nBins"}
    expected = "Basic " + base64.b64encode(b"AC123:secret").decode()
    assert post.calls[0]["headers"]["Authorization"] == expected
    assert outcome.ok and outcome.message_id == "SM9"


def test_a_runaway_body_is_truncated_rather_than_billed_by_the_segment() -> None:
    post = recorder(201, json.dumps({"sid": "SM9"}))
    providers.send_sms(
        to="+15551234567",
        subject="",
        body="x" * 50_000,
        account_sid="AC1",
        auth_token="t",
        sender="+1555",
        timeout=5,
        post=post,
    )
    fields = dict(urllib.parse.parse_qsl(post.calls[0]["data"].decode()))
    assert len(fields["Body"]) == providers.SMS_LIMIT


def test_twilio_refusal_carries_its_code() -> None:
    post = recorder(400, json.dumps({"message": "To and From cannot be the same", "code": 21266}))
    outcome = providers.send_sms(
        to="+1555", subject="s", body="b", account_sid="AC1", auth_token="t", sender="+1555", timeout=5, post=post
    )
    assert not outcome.ok
    assert outcome.status == 400
    assert "21266" in outcome.detail


def test_an_unparseable_error_body_is_truncated_not_dumped() -> None:
    post = recorder(500, "<html>" + "x" * 5000)
    outcome = providers.send_email(
        to="[email protected]", subject="s", body="b", api_key="k", sender="[email protected]", timeout=5, post=post
    )
    assert not outcome.ok
    assert len(outcome.detail) <= 200