mi
91 lines
2.8 KiB
Python
91 lines
2.8 KiB
Python
import uuid
|
|
|
|
import httpx
|
|
import pytest
|
|
|
|
from app.db import SendStatus
|
|
from app.provider import IdgtlConfig, callback_url_with_credentials, classify_response
|
|
|
|
|
|
def response(status: int, payload=None) -> httpx.Response:
|
|
request = httpx.Request("POST", "https://direct.example/api/v1/message")
|
|
if payload is None:
|
|
return httpx.Response(status, request=request)
|
|
return httpx.Response(status, json=payload, request=request)
|
|
|
|
|
|
@pytest.mark.parametrize("status", [401, 402, 403, 422])
|
|
def test_explicit_business_rejections_are_not_retried(status: int) -> None:
|
|
result = classify_response(response(status), "message-id")
|
|
assert result.send_status == SendStatus.REJECTED
|
|
assert result.retry_safe is False
|
|
|
|
|
|
@pytest.mark.parametrize("status", [500, 502, 503, 504])
|
|
def test_ambiguous_http_results_are_uncertain(status: int) -> None:
|
|
result = classify_response(response(status), "message-id")
|
|
assert result.send_status == SendStatus.UNCERTAIN
|
|
assert result.retry_safe is False
|
|
|
|
|
|
def test_exact_success_contract() -> None:
|
|
message_uuid = str(uuid.uuid4())
|
|
result = classify_response(
|
|
response(
|
|
200,
|
|
{
|
|
"errors": False,
|
|
"response": [
|
|
{
|
|
"code": 201,
|
|
"messageUuid": message_uuid,
|
|
"externalMessageId": "message-id",
|
|
}
|
|
],
|
|
},
|
|
),
|
|
"message-id",
|
|
)
|
|
assert result.send_status == SendStatus.ACCEPTED
|
|
assert result.message_uuid == message_uuid
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"payload",
|
|
[
|
|
{"errors": True, "response": []},
|
|
{"errors": False, "response": []},
|
|
{"errors": False, "response": [{"code": 200}]},
|
|
{
|
|
"errors": False,
|
|
"response": [
|
|
{
|
|
"code": 201,
|
|
"messageUuid": str(uuid.uuid4()),
|
|
"externalMessageId": "wrong",
|
|
}
|
|
],
|
|
},
|
|
],
|
|
)
|
|
def test_malformed_200_is_rejected_contract_violation(payload) -> None:
|
|
result = classify_response(response(200, payload), "message-id")
|
|
assert result.send_status == SendStatus.REJECTED
|
|
assert result.contract_violation is True
|
|
|
|
|
|
def test_callback_credentials_are_url_encoded() -> None:
|
|
config = IdgtlConfig(
|
|
base_url="https://direct.example",
|
|
api_key="api-key",
|
|
callback_url="https://tohin.ru/callbacks/idgtl/sms",
|
|
callback_username="user@example",
|
|
callback_password="p:a/ss", # noqa: S106 - synthetic URL-encoding fixture
|
|
connect_timeout_ms=3000,
|
|
request_timeout_ms=70000,
|
|
callback_enabled=True,
|
|
)
|
|
assert callback_url_with_credentials(config) == (
|
|
"https://user%40example:p%3Aa%2Fss@tohin.ru/callbacks/idgtl/sms"
|
|
)
|