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, ) from app.config import TBankPaymentAuthConfig, TBankPaymentConfig 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, *, notification_url: str = "https://example.test/tbank/notify", success_url: str = "https://example.test/payment/success", 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", notification_url=notification_url, success_url=success_url, 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( ( "125000" "https://example.test/tbank/notify" "order-uuid-1" "test-password" "https://example.test/payment/success" "TBankTest" ).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", "NotificationURL": "https://example.test/tbank/notify", "SuccessURL": "https://example.test/payment/success", "Token": expected_token, }, "data": None, "headers": { "Accept": "application/json", "Content-Type": "application/json", }, "timeout": 7.5, } ] def test_from_config_posts_configured_urls_and_deterministic_token() -> None: response = httpx.Response( 200, json={"Success": True, "PaymentURL": "https://securepay.test/pay/2"}, request=httpx.Request("POST", "https://securepay.tinkoff.ru/v2/Init"), ) http_client = SequenceHTTPClient([response]) config = TBankPaymentConfig( init_url="https://securepay.tinkoff.ru/v2/Init", notification_url="https://merchant.test/tbank/notification", success_url="https://merchant.test/payment/success", auth=TBankPaymentAuthConfig( terminal_key="ConfigTerminal", password="config-password", ), timeout_seconds=6.25, retry_attempts=0, retry_backoff_seconds=0.1, ) adapter = TBankAdapter.from_config( http_client=http_client, # type: ignore[arg-type] config=config, ) result = asyncio.run( adapter.create_payment_link( order_uuid="order-uuid-2", amount_kopecks=9900, ) ) expected_token = hashlib.sha256( ( "9900" "https://merchant.test/tbank/notification" "order-uuid-2" "config-password" "https://merchant.test/payment/success" "ConfigTerminal" ).encode("utf-8") ).hexdigest() assert result == "https://securepay.test/pay/2" assert http_client.calls[0]["json"] == { "TerminalKey": "ConfigTerminal", "Amount": 9900, "OrderId": "order-uuid-2", "NotificationURL": "https://merchant.test/tbank/notification", "SuccessURL": "https://merchant.test/payment/success", "Token": expected_token, } assert http_client.calls[0]["timeout"] == 6.25 def test_create_payment_link_maps_4xx_to_request_error() -> None: response = httpx.Response( 400, json={ "Success": False, "ErrorCode": "101", "Message": "bad request", "Details": "invalid token", }, request=httpx.Request("POST", "https://securepay.tinkoff.ru/v2/Init"), ) http_client = SequenceHTTPClient([response]) adapter = _make_adapter(http_client) with pytest.raises(TBankPaymentRequestError) as exc_info: asyncio.run(adapter.create_payment_link("order-uuid-1", 125000)) assert exc_info.value.status_code == 400 assert exc_info.value.error_code == "101" assert exc_info.value.provider_message == "bad request" assert exc_info.value.details == "invalid token" assert str(exc_info.value) == ( "TBank payment init request was rejected. " "status_code=400 error_code=101 message=bad request details=invalid token" ) def test_create_payment_link_maps_unsuccessful_payload_to_request_error() -> None: response = httpx.Response( 200, json={ "Success": False, "ErrorCode": "102", "Message": "duplicate order id", "Details": "OrderId must be unique", }, request=httpx.Request("POST", "https://securepay.tinkoff.ru/v2/Init"), ) http_client = SequenceHTTPClient([response]) adapter = _make_adapter(http_client) with pytest.raises(TBankPaymentRequestError) as exc_info: asyncio.run(adapter.create_payment_link("order-uuid-1", 125000)) assert exc_info.value.status_code is None assert exc_info.value.error_code == "102" assert exc_info.value.provider_message == "duplicate order id" assert exc_info.value.details == "OrderId must be unique" assert str(exc_info.value) == ( "TBank payment init request was rejected. " "error_code=102 message=duplicate order id details=OrderId must be unique" ) 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]