Раис Юсупалиев
121 lines
3.7 KiB
Python
121 lines
3.7 KiB
Python
import asyncio
|
|
|
|
import httpx
|
|
|
|
from app.controllers.v1.delivery import get_aggregator_service
|
|
from app.main import create_app
|
|
from app.schemas.payment import TBankPaymentNotification
|
|
from app.services.aggregator import (
|
|
InvalidTBankPaymentNotificationError,
|
|
TBankPaymentNotificationProcessingError,
|
|
)
|
|
|
|
|
|
class StubAggregatorService:
|
|
def __init__(self, *, response: str = "OK", error: Exception | None = None) -> None:
|
|
self._response = response
|
|
self._error = error
|
|
self.calls: list[TBankPaymentNotification] = []
|
|
|
|
async def handle_tbank_payment_notification(
|
|
self,
|
|
notification: TBankPaymentNotification,
|
|
) -> str:
|
|
self.calls.append(notification)
|
|
if self._error is not None:
|
|
raise self._error
|
|
return self._response
|
|
|
|
|
|
def _install_service_override(app, service: StubAggregatorService) -> None:
|
|
async def override_service() -> StubAggregatorService:
|
|
return service
|
|
|
|
app.dependency_overrides[get_aggregator_service] = override_service
|
|
|
|
|
|
def _valid_payload() -> dict[str, object]:
|
|
return {
|
|
"TerminalKey": "TestTerminal",
|
|
"OrderId": "order-uuid-1",
|
|
"Success": True,
|
|
"Status": "CONFIRMED",
|
|
"PaymentId": 8347568144,
|
|
"ErrorCode": "0",
|
|
"Amount": 125000,
|
|
"Token": "signed-token",
|
|
"Data": {"nested": "accepted by schema"},
|
|
}
|
|
|
|
|
|
def _post_notification(app, payload: dict[str, object]) -> httpx.Response:
|
|
async def run_request() -> httpx.Response:
|
|
transport = httpx.ASGITransport(app=app)
|
|
async with httpx.AsyncClient(
|
|
transport=transport,
|
|
base_url="http://testserver",
|
|
) as client:
|
|
return await client.post(
|
|
"/api/v1/delivery/tbank/notifications",
|
|
json=payload,
|
|
)
|
|
|
|
return asyncio.run(run_request())
|
|
|
|
|
|
def test_post_tbank_notification_returns_plain_text_ok_and_delegates_once() -> None:
|
|
service = StubAggregatorService(response="OK")
|
|
app = create_app()
|
|
_install_service_override(app, service)
|
|
|
|
response = _post_notification(app, _valid_payload())
|
|
|
|
assert response.status_code == 200
|
|
assert response.text == "OK"
|
|
assert response.headers["content-type"].startswith("text/plain")
|
|
assert service.calls == [TBankPaymentNotification(**_valid_payload())]
|
|
|
|
|
|
def test_post_tbank_notification_maps_invalid_token_to_400() -> None:
|
|
service = StubAggregatorService(
|
|
error=InvalidTBankPaymentNotificationError("invalid token")
|
|
)
|
|
app = create_app()
|
|
_install_service_override(app, service)
|
|
|
|
response = _post_notification(app, _valid_payload())
|
|
|
|
assert response.status_code == 400
|
|
assert response.json()["detail"]["code"] == "invalid_tbank_payment_notification"
|
|
assert service.calls == [TBankPaymentNotification(**_valid_payload())]
|
|
|
|
|
|
def test_post_tbank_notification_maps_processing_failure_to_503() -> None:
|
|
service = StubAggregatorService(
|
|
error=TBankPaymentNotificationProcessingError("temporary failure")
|
|
)
|
|
app = create_app()
|
|
_install_service_override(app, service)
|
|
|
|
response = _post_notification(app, _valid_payload())
|
|
|
|
assert response.status_code == 503
|
|
assert (
|
|
response.json()["detail"]["code"]
|
|
== "tbank_payment_notification_processing_unavailable"
|
|
)
|
|
assert service.calls == [TBankPaymentNotification(**_valid_payload())]
|
|
|
|
|
|
def test_post_tbank_notification_rejects_invalid_payload_without_service_call() -> None:
|
|
service = StubAggregatorService()
|
|
app = create_app()
|
|
_install_service_override(app, service)
|
|
payload = _valid_payload()
|
|
del payload["Token"]
|
|
|
|
response = _post_notification(app, payload)
|
|
|
|
assert response.status_code == 422
|
|
assert service.calls == []
|