Добавлен адаптер к tbank и формирование ссылки на оплату
This commit is contained in:
@@ -0,0 +1,170 @@
|
||||
import asyncio
|
||||
import hashlib
|
||||
from typing import Any
|
||||
|
||||
import httpx
|
||||
import pytest
|
||||
|
||||
from app.adapters.tbank.client import TBankAdapter
|
||||
from app.adapters.tbank.base import (
|
||||
TBankPaymentAdapterError,
|
||||
TBankPaymentRequestError,
|
||||
)
|
||||
|
||||
|
||||
class SequenceHTTPClient:
|
||||
def __init__(self, results: list[Any]) -> None:
|
||||
self._results = results
|
||||
self.calls: list[dict[str, Any]] = []
|
||||
|
||||
def _next_result(self) -> Any:
|
||||
return self._results[len(self.calls) - 1]
|
||||
|
||||
async def post(
|
||||
self,
|
||||
url: str,
|
||||
*,
|
||||
json: dict[str, Any] | None = None,
|
||||
data: dict[str, Any] | None = None,
|
||||
headers: dict[str, str] | None = None,
|
||||
timeout: float | None = None,
|
||||
) -> httpx.Response:
|
||||
self.calls.append(
|
||||
{
|
||||
"method": "POST",
|
||||
"url": url,
|
||||
"json": json,
|
||||
"data": data,
|
||||
"headers": headers,
|
||||
"timeout": timeout,
|
||||
}
|
||||
)
|
||||
result = self._next_result()
|
||||
if isinstance(result, Exception):
|
||||
raise result
|
||||
return result
|
||||
|
||||
|
||||
def _make_adapter(
|
||||
http_client: SequenceHTTPClient,
|
||||
*,
|
||||
retry_attempts: int = 0,
|
||||
retry_backoff_seconds: float = 0.2,
|
||||
sleep_calls: list[float] | None = None,
|
||||
) -> TBankAdapter:
|
||||
async def fake_sleep(seconds: float) -> None:
|
||||
if sleep_calls is not None:
|
||||
sleep_calls.append(seconds)
|
||||
|
||||
return TBankAdapter(
|
||||
http_client=http_client, # type: ignore[arg-type]
|
||||
init_url="https://securepay.tinkoff.ru/v2/Init",
|
||||
terminal_key="TBankTest",
|
||||
password="test-password",
|
||||
timeout_seconds=7.5,
|
||||
retry_attempts=retry_attempts,
|
||||
retry_backoff_seconds=retry_backoff_seconds,
|
||||
sleep=fake_sleep,
|
||||
)
|
||||
|
||||
|
||||
def test_create_payment_link_posts_signed_payload_and_maps_payment_url() -> None:
|
||||
response = httpx.Response(
|
||||
200,
|
||||
json={"Success": True, "PaymentURL": "https://securepay.test/pay/1"},
|
||||
request=httpx.Request("POST", "https://securepay.tinkoff.ru/v2/Init"),
|
||||
)
|
||||
http_client = SequenceHTTPClient([response])
|
||||
adapter = _make_adapter(http_client)
|
||||
|
||||
result = asyncio.run(
|
||||
adapter.create_payment_link(
|
||||
order_uuid="order-uuid-1",
|
||||
amount_kopecks=125000,
|
||||
)
|
||||
)
|
||||
|
||||
expected_token = hashlib.sha256(
|
||||
"125000order-uuid-1test-passwordTBankTest".encode("utf-8")
|
||||
).hexdigest()
|
||||
assert result == "https://securepay.test/pay/1"
|
||||
assert http_client.calls == [
|
||||
{
|
||||
"method": "POST",
|
||||
"url": "https://securepay.tinkoff.ru/v2/Init",
|
||||
"json": {
|
||||
"TerminalKey": "TBankTest",
|
||||
"Amount": 125000,
|
||||
"OrderId": "order-uuid-1",
|
||||
"Token": expected_token,
|
||||
},
|
||||
"data": None,
|
||||
"headers": {
|
||||
"Accept": "application/json",
|
||||
"Content-Type": "application/json",
|
||||
},
|
||||
"timeout": 7.5,
|
||||
}
|
||||
]
|
||||
|
||||
|
||||
def test_create_payment_link_maps_4xx_to_request_error() -> None:
|
||||
response = httpx.Response(
|
||||
400,
|
||||
json={"Success": False, "Message": "bad request"},
|
||||
request=httpx.Request("POST", "https://securepay.tinkoff.ru/v2/Init"),
|
||||
)
|
||||
http_client = SequenceHTTPClient([response])
|
||||
adapter = _make_adapter(http_client)
|
||||
|
||||
with pytest.raises(TBankPaymentRequestError, match="status 400"):
|
||||
asyncio.run(adapter.create_payment_link("order-uuid-1", 125000))
|
||||
|
||||
|
||||
def test_create_payment_link_maps_unsuccessful_payload_to_request_error() -> None:
|
||||
response = httpx.Response(
|
||||
200,
|
||||
json={"Success": False, "Message": "bad request"},
|
||||
request=httpx.Request("POST", "https://securepay.tinkoff.ru/v2/Init"),
|
||||
)
|
||||
http_client = SequenceHTTPClient([response])
|
||||
adapter = _make_adapter(http_client)
|
||||
|
||||
with pytest.raises(TBankPaymentRequestError):
|
||||
asyncio.run(adapter.create_payment_link("order-uuid-1", 125000))
|
||||
|
||||
|
||||
def test_create_payment_link_maps_transport_errors_to_client_error() -> None:
|
||||
request = httpx.Request("POST", "https://securepay.tinkoff.ru/v2/Init")
|
||||
http_client = SequenceHTTPClient([httpx.ConnectError("down", request=request)])
|
||||
adapter = _make_adapter(http_client)
|
||||
|
||||
with pytest.raises(TBankPaymentAdapterError, match="retry attempts"):
|
||||
asyncio.run(adapter.create_payment_link("order-uuid-1", 125000))
|
||||
|
||||
|
||||
def test_create_payment_link_retries_5xx_and_raises_client_error() -> None:
|
||||
first_response = httpx.Response(
|
||||
503,
|
||||
json={"Message": "temporary failure"},
|
||||
request=httpx.Request("POST", "https://securepay.tinkoff.ru/v2/Init"),
|
||||
)
|
||||
second_response = httpx.Response(
|
||||
503,
|
||||
json={"Message": "temporary failure"},
|
||||
request=httpx.Request("POST", "https://securepay.tinkoff.ru/v2/Init"),
|
||||
)
|
||||
http_client = SequenceHTTPClient([first_response, second_response])
|
||||
sleep_calls: list[float] = []
|
||||
adapter = _make_adapter(
|
||||
http_client,
|
||||
retry_attempts=1,
|
||||
retry_backoff_seconds=0.25,
|
||||
sleep_calls=sleep_calls,
|
||||
)
|
||||
|
||||
with pytest.raises(TBankPaymentAdapterError, match="retriable status 503"):
|
||||
asyncio.run(adapter.create_payment_link("order-uuid-1", 125000))
|
||||
|
||||
assert len(http_client.calls) == 2
|
||||
assert sleep_calls == [0.25]
|
||||
Reference in New Issue
Block a user