96 lines
3.7 KiB
Python
96 lines
3.7 KiB
Python
"""Wire contract with MAX, using an in-process HTTP fake (no real bot/token)."""
|
|
|
|
import json
|
|
import logging
|
|
|
|
import httpx
|
|
import pytest
|
|
|
|
from app.channels.max_client import API_URL, UPDATE_TYPES, MaxApiError, MaxClient
|
|
from app.config import Settings
|
|
|
|
|
|
def settings():
|
|
return Settings(
|
|
_env_file=None,
|
|
max_enabled=True,
|
|
max_bot_token="synthetic-secret-token",
|
|
max_bot_username="synthetic_bot",
|
|
max_webhook_secret="synthetic-webhook-secret-" * 2,
|
|
max_webhook_url="https://api.example.org/api/v1/webhooks/max",
|
|
max_site_url="https://example.org",
|
|
)
|
|
|
|
|
|
async def test_message_wire_contract_and_no_identifiers_in_logs(caplog):
|
|
def handler(request):
|
|
assert str(request.url).startswith(API_URL + "/messages?")
|
|
assert request.url.params["user_id"] == "1234567890123"
|
|
assert "access_token" not in request.url.params
|
|
assert request.headers["Authorization"] == "synthetic-secret-token"
|
|
body = json.loads(request.content)
|
|
assert body["text"] == "neutral reminder" and "format" not in body
|
|
button = body["attachments"][0]["payload"]["buttons"][0][0]
|
|
assert button == {"type": "link", "text": "Open", "url": "https://example.org"}
|
|
return httpx.Response(200, json={"message": {"body": {"mid": "synthetic-id"}}})
|
|
|
|
client = MaxClient(settings(), transport=httpx.MockTransport(handler))
|
|
with caplog.at_level(logging.INFO):
|
|
await client.send(1234567890123, "neutral reminder", site_url="https://example.org", button="Open")
|
|
await client.close()
|
|
assert "1234567890123" not in caplog.text and "synthetic-secret-token" not in caplog.text
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"status,retry", [(429, True), (500, True), (503, True), (408, True), (401, False), (403, False)]
|
|
)
|
|
async def test_error_codes_are_redacted_and_classified(status, retry):
|
|
client = MaxClient(
|
|
settings(),
|
|
transport=httpx.MockTransport(
|
|
lambda r: httpx.Response(
|
|
status, json={"message": "sensitive provider text"}, headers={"Retry-After": "120"}
|
|
)
|
|
),
|
|
)
|
|
with pytest.raises(MaxApiError) as error:
|
|
await client.send(100, "neutral", site_url="https://example.org", button="Open")
|
|
assert error.value.retryable is retry and error.value.retry_after == 120
|
|
assert "sensitive" not in str(error.value)
|
|
await client.close()
|
|
|
|
|
|
@pytest.mark.parametrize("body", [{"success": False, "message": "private"}, {}, {"success": True}, []])
|
|
async def test_http_200_is_not_enough_to_mark_delivered(body):
|
|
client = MaxClient(settings(), transport=httpx.MockTransport(lambda r: httpx.Response(200, json=body)))
|
|
with pytest.raises(MaxApiError):
|
|
await client.send(100, "neutral", site_url="https://example.org", button="Open")
|
|
await client.close()
|
|
|
|
|
|
async def test_timeout_is_retryable_without_request_details():
|
|
def handler(request):
|
|
raise httpx.ReadTimeout("private URL", request=request)
|
|
|
|
client = MaxClient(settings(), transport=httpx.MockTransport(handler))
|
|
with pytest.raises(MaxApiError, match="^transport$") as error:
|
|
await client.request("GET", "/me")
|
|
assert error.value.retryable
|
|
await client.close()
|
|
|
|
|
|
async def test_subscription_includes_secret_and_exact_event_types():
|
|
def handler(request):
|
|
assert request.url.path == "/subscriptions"
|
|
body = json.loads(request.content)
|
|
assert body == {
|
|
"url": settings().max_webhook_url,
|
|
"update_types": UPDATE_TYPES,
|
|
"secret": settings().max_webhook_secret.get_secret_value(),
|
|
}
|
|
return httpx.Response(200, json={"success": True})
|
|
|
|
client = MaxClient(settings(), transport=httpx.MockTransport(handler))
|
|
await client.subscribe(settings())
|
|
await client.close()
|