k1nq
87 lines
2.4 KiB
Python
87 lines
2.4 KiB
Python
import json
|
|
import uuid
|
|
from datetime import UTC, datetime
|
|
from decimal import Decimal
|
|
|
|
import httpx2
|
|
import pytest
|
|
|
|
from app.config import Settings
|
|
from app.models import Currency, PaymentStatus
|
|
from app.processing import PaymentGateway, WebhookClient
|
|
from app.schemas import PaymentWebhook
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("success_rate", "expected"),
|
|
[
|
|
(1.0, PaymentStatus.SUCCEEDED),
|
|
(0.0, PaymentStatus.FAILED),
|
|
],
|
|
)
|
|
async def test_gateway_returns_configured_result(
|
|
success_rate: float,
|
|
expected: PaymentStatus,
|
|
) -> None:
|
|
gateway = PaymentGateway(
|
|
Settings(
|
|
processing_min_seconds=0,
|
|
processing_max_seconds=0,
|
|
processing_success_rate=success_rate,
|
|
)
|
|
)
|
|
|
|
assert await gateway.charge() is expected
|
|
|
|
|
|
async def test_webhook_client_posts_serialized_payload(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
requests: list[httpx2.Request] = []
|
|
|
|
def handler(request: httpx2.Request) -> httpx2.Response:
|
|
requests.append(request)
|
|
return httpx2.Response(204)
|
|
|
|
transport = httpx2.MockTransport(handler)
|
|
async_client = httpx2.AsyncClient
|
|
monkeypatch.setattr(
|
|
"app.processing.httpx2.AsyncClient",
|
|
lambda **kwargs: async_client(transport=transport, **kwargs),
|
|
)
|
|
payload = PaymentWebhook(
|
|
payment_id=uuid.uuid4(),
|
|
amount=Decimal("10.25"),
|
|
currency=Currency.USD,
|
|
status=PaymentStatus.SUCCEEDED,
|
|
processed_at=datetime.now(UTC),
|
|
metadata={"order_id": 42},
|
|
)
|
|
|
|
await WebhookClient(timeout=1).send("https://example.com/hook", payload)
|
|
|
|
assert len(requests) == 1
|
|
assert json.loads(requests[0].content) == payload.model_dump(mode="json")
|
|
|
|
|
|
async def test_webhook_client_raises_for_server_error(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
transport = httpx2.MockTransport(lambda _: httpx2.Response(500))
|
|
async_client = httpx2.AsyncClient
|
|
monkeypatch.setattr(
|
|
"app.processing.httpx2.AsyncClient",
|
|
lambda **kwargs: async_client(transport=transport, **kwargs),
|
|
)
|
|
payload = PaymentWebhook(
|
|
payment_id=uuid.uuid4(),
|
|
amount=Decimal("1.00"),
|
|
currency=Currency.EUR,
|
|
status=PaymentStatus.FAILED,
|
|
processed_at=datetime.now(UTC),
|
|
metadata={},
|
|
)
|
|
|
|
with pytest.raises(httpx2.HTTPStatusError):
|
|
await WebhookClient(timeout=1).send("https://example.com/hook", payload)
|