bot-backend
a32ccabd0d
All checks were successful
Deploy Trade-In / changes (push) Successful in 12s
Deploy Trade-In / build-frontend (push) Has been skipped
Deploy Trade-In / build-browser (push) Has been skipped
Deploy Trade-In / test (push) Successful in 3m3s
Deploy Trade-In / build-backend (push) Successful in 1m9s
Deploy Trade-In / deploy (push) Successful in 1m24s
PR-C платёжного контура: token.py (sign + verify_notification_token, оба эталонных вектора Т-Банка перепроверены независимо), receipt.py (54-ФЗ ФФД 1.05, целые копейки), tbank_client.py (Init/GetState/CheckOrder/Confirm/Cancel, таймаут 15с, 4xx не ретраится). Слой инертный: 0 импортёров, роутеров нет. Co-authored-by: bot-backend <bot-backend@gendsgn.local> Co-committed-by: bot-backend <bot-backend@gendsgn.local>
293 lines
10 KiB
Python
293 lines
10 KiB
Python
"""Unit-тесты `app.services.payments.tbank_client.TBankClient`.
|
||
|
||
NEVER calls real T-Bank API — только `httpx.MockTransport` (тот же паттерн,
|
||
что `tests/services/tgbot/test_client.py` и `tests/services/test_dadata.py`).
|
||
`asyncio.sleep` патчится no-op'ом, чтобы retry-тесты шли мгновенно
|
||
независимо от реального backoff.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
from collections.abc import Callable
|
||
from unittest import mock
|
||
|
||
import httpx
|
||
import pytest
|
||
|
||
from app.services.payments.tbank_client import TBankApiError, TBankClient
|
||
|
||
_REAL_ASYNC_CLIENT = httpx.AsyncClient
|
||
|
||
|
||
def _install_transport(handler: Callable[[httpx.Request], httpx.Response]) -> None:
|
||
transport = httpx.MockTransport(handler)
|
||
|
||
def factory(*_: object, **__: object) -> httpx.AsyncClient:
|
||
return _REAL_ASYNC_CLIENT(transport=transport)
|
||
|
||
mock.patch("app.services.payments.tbank_client.httpx.AsyncClient", factory).start()
|
||
|
||
|
||
@pytest.fixture(autouse=True)
|
||
def _stop_patches_and_noop_sleep():
|
||
sleep_patcher = mock.patch(
|
||
"app.services.payments.tbank_client.asyncio.sleep", return_value=None
|
||
)
|
||
sleep_patcher.start()
|
||
yield
|
||
mock.patch.stopall()
|
||
|
||
|
||
def _client(**kwargs: object) -> TBankClient:
|
||
defaults: dict[str, object] = {
|
||
"terminal_key": "MerchantTerminalKey",
|
||
"password": "test-password",
|
||
"base_url": "https://rest-api-test.tinkoff.ru",
|
||
}
|
||
defaults.update(kwargs)
|
||
return TBankClient(**defaults) # type: ignore[arg-type]
|
||
|
||
|
||
# ── happy path ────────────────────────────────────────────────────────────────
|
||
|
||
|
||
async def test_init_payment_happy_path_returns_payment_url() -> None:
|
||
def handler(request: httpx.Request) -> httpx.Response:
|
||
assert request.url.path.endswith("/v2/Init")
|
||
return httpx.Response(
|
||
200,
|
||
json={
|
||
"Success": True,
|
||
"TerminalKey": "MerchantTerminalKey",
|
||
"Status": "NEW",
|
||
"PaymentId": "12345",
|
||
"OrderId": "order-1",
|
||
"Amount": 10000,
|
||
"PaymentURL": "https://securepay.tinkoff.ru/abc",
|
||
},
|
||
)
|
||
|
||
_install_transport(handler)
|
||
client = _client()
|
||
result = await client.init_payment(order_id="order-1", amount_kopecks=10000)
|
||
|
||
assert result["PaymentId"] == "12345"
|
||
assert result["PaymentURL"] == "https://securepay.tinkoff.ru/abc"
|
||
|
||
|
||
async def test_init_payment_signs_request_with_token() -> None:
|
||
"""Запрос обязан содержать TerminalKey + Token в теле."""
|
||
captured: dict[str, object] = {}
|
||
|
||
def handler(request: httpx.Request) -> httpx.Response:
|
||
import json
|
||
|
||
captured["body"] = json.loads(request.content)
|
||
return httpx.Response(200, json={"Success": True, "PaymentId": "1"})
|
||
|
||
_install_transport(handler)
|
||
client = _client()
|
||
await client.init_payment(order_id="00000", amount_kopecks=19200)
|
||
|
||
body = captured["body"]
|
||
assert isinstance(body, dict)
|
||
assert body["TerminalKey"] == "MerchantTerminalKey"
|
||
assert isinstance(body.get("Token"), str) and len(body["Token"]) == 64
|
||
|
||
|
||
async def test_init_payment_receipt_excluded_from_signed_token_but_present_in_body() -> None:
|
||
"""`Receipt` уходит в тело запроса, но не участвует в Token (см. token.py)."""
|
||
import json
|
||
|
||
from app.services.payments.token import sign
|
||
|
||
captured: dict[str, object] = {}
|
||
|
||
def handler(request: httpx.Request) -> httpx.Response:
|
||
captured["body"] = json.loads(request.content)
|
||
return httpx.Response(200, json={"Success": True, "PaymentId": "1"})
|
||
|
||
_install_transport(handler)
|
||
client = _client()
|
||
receipt = {"Email": "a@test.ru", "Taxation": "osn", "Items": []}
|
||
await client.init_payment(order_id="00000", amount_kopecks=19200, receipt=receipt)
|
||
|
||
body = captured["body"]
|
||
assert isinstance(body, dict)
|
||
assert body["Receipt"] == receipt
|
||
|
||
# Token, реально ушедший в теле, обязан совпадать с sign() тела БЕЗ Receipt
|
||
# (Receipt — dict, sign() сам его игнорирует) — пересчитаем и сверим.
|
||
without_token = {k: v for k, v in body.items() if k != "Token"}
|
||
assert body["Token"] == sign(without_token, "test-password")
|
||
|
||
|
||
async def test_get_state_posts_to_correct_path() -> None:
|
||
def handler(request: httpx.Request) -> httpx.Response:
|
||
assert request.url.path.endswith("/v2/GetState")
|
||
return httpx.Response(200, json={"Success": True, "Status": "CONFIRMED"})
|
||
|
||
_install_transport(handler)
|
||
client = _client()
|
||
result = await client.get_state(payment_id="12345")
|
||
assert result["Status"] == "CONFIRMED"
|
||
|
||
|
||
async def test_check_order_posts_to_correct_path() -> None:
|
||
def handler(request: httpx.Request) -> httpx.Response:
|
||
assert request.url.path.endswith("/v2/CheckOrder")
|
||
return httpx.Response(200, json={"Success": True, "Payments": []})
|
||
|
||
_install_transport(handler)
|
||
client = _client()
|
||
result = await client.check_order(order_id="order-1")
|
||
assert result["Payments"] == []
|
||
|
||
|
||
async def test_confirm_posts_to_correct_path() -> None:
|
||
def handler(request: httpx.Request) -> httpx.Response:
|
||
assert request.url.path.endswith("/v2/Confirm")
|
||
return httpx.Response(200, json={"Success": True, "Status": "CONFIRMED"})
|
||
|
||
_install_transport(handler)
|
||
client = _client()
|
||
result = await client.confirm(payment_id="12345")
|
||
assert result["Status"] == "CONFIRMED"
|
||
|
||
|
||
async def test_cancel_posts_to_correct_path() -> None:
|
||
def handler(request: httpx.Request) -> httpx.Response:
|
||
assert request.url.path.endswith("/v2/Cancel")
|
||
return httpx.Response(200, json={"Success": True, "Status": "REFUNDED"})
|
||
|
||
_install_transport(handler)
|
||
client = _client()
|
||
result = await client.cancel(payment_id="12345", amount_kopecks=5000)
|
||
assert result["Status"] == "REFUNDED"
|
||
|
||
|
||
# ── retry policy ─────────────────────────────────────────────────────────────
|
||
|
||
|
||
async def test_retries_on_5xx_then_succeeds() -> None:
|
||
calls = {"n": 0}
|
||
|
||
def handler(request: httpx.Request) -> httpx.Response:
|
||
calls["n"] += 1
|
||
if calls["n"] < 3:
|
||
return httpx.Response(502, json={"ErrorCode": "502", "Message": "bad gw"})
|
||
return httpx.Response(200, json={"Success": True, "PaymentId": "1"})
|
||
|
||
_install_transport(handler)
|
||
client = _client()
|
||
result = await client.init_payment(order_id="1", amount_kopecks=100)
|
||
|
||
assert result["PaymentId"] == "1"
|
||
assert calls["n"] == 3
|
||
|
||
|
||
async def test_retries_on_network_error_then_succeeds() -> None:
|
||
calls = {"n": 0}
|
||
|
||
def handler(request: httpx.Request) -> httpx.Response:
|
||
calls["n"] += 1
|
||
if calls["n"] < 2:
|
||
raise httpx.ConnectError("connection refused", request=request)
|
||
return httpx.Response(200, json={"Success": True, "PaymentId": "1"})
|
||
|
||
_install_transport(handler)
|
||
client = _client()
|
||
result = await client.init_payment(order_id="1", amount_kopecks=100)
|
||
|
||
assert result["PaymentId"] == "1"
|
||
assert calls["n"] == 2
|
||
|
||
|
||
async def test_gives_up_after_max_retries_on_persistent_5xx() -> None:
|
||
def handler(request: httpx.Request) -> httpx.Response:
|
||
return httpx.Response(500, json={"ErrorCode": "500", "Message": "boom"})
|
||
|
||
_install_transport(handler)
|
||
client = _client()
|
||
|
||
with pytest.raises(TBankApiError) as exc_info:
|
||
await client.get_state(payment_id="1")
|
||
|
||
assert exc_info.value.error_code == "500"
|
||
|
||
|
||
async def test_does_not_retry_on_4xx() -> None:
|
||
calls = {"n": 0}
|
||
|
||
def handler(request: httpx.Request) -> httpx.Response:
|
||
calls["n"] += 1
|
||
return httpx.Response(401, json={"ErrorCode": "401", "Message": "Terminal not found"})
|
||
|
||
_install_transport(handler)
|
||
client = _client()
|
||
|
||
with pytest.raises(TBankApiError) as exc_info:
|
||
await client.init_payment(order_id="1", amount_kopecks=100)
|
||
|
||
assert exc_info.value.error_code == "401"
|
||
assert calls["n"] == 1 # НЕ ретраится
|
||
|
||
|
||
async def test_business_failure_success_false_raises_without_retry() -> None:
|
||
"""HTTP 200, но `Success: false` — бизнес-отказ банка, не сбой транспорта."""
|
||
calls = {"n": 0}
|
||
|
||
def handler(request: httpx.Request) -> httpx.Response:
|
||
calls["n"] += 1
|
||
return httpx.Response(
|
||
200,
|
||
json={
|
||
"Success": False,
|
||
"ErrorCode": "9999",
|
||
"Message": "Неверные параметры запроса",
|
||
"Details": "Amount must be positive",
|
||
},
|
||
)
|
||
|
||
_install_transport(handler)
|
||
client = _client()
|
||
|
||
with pytest.raises(TBankApiError) as exc_info:
|
||
await client.init_payment(order_id="1", amount_kopecks=100)
|
||
|
||
assert exc_info.value.error_code == "9999"
|
||
assert exc_info.value.message == "Неверные параметры запроса"
|
||
assert calls["n"] == 1 # НЕ ретраится
|
||
|
||
|
||
async def test_malformed_json_response_raises_tbank_api_error() -> None:
|
||
def handler(request: httpx.Request) -> httpx.Response:
|
||
return httpx.Response(200, content=b"not json at all")
|
||
|
||
_install_transport(handler)
|
||
client = _client()
|
||
|
||
with pytest.raises(TBankApiError):
|
||
await client.get_state(payment_id="1")
|
||
|
||
|
||
# ── безопасность: пароль не попадает в тело запроса ─────────────────────────
|
||
|
||
|
||
async def test_password_never_sent_in_request_body() -> None:
|
||
import json
|
||
|
||
captured: dict[str, object] = {}
|
||
|
||
def handler(request: httpx.Request) -> httpx.Response:
|
||
captured["body"] = json.loads(request.content)
|
||
return httpx.Response(200, json={"Success": True, "PaymentId": "1"})
|
||
|
||
_install_transport(handler)
|
||
client = _client(password="super-secret-password")
|
||
await client.init_payment(order_id="1", amount_kopecks=100)
|
||
|
||
body = captured["body"]
|
||
assert isinstance(body, dict)
|
||
assert "Password" not in body
|
||
assert "super-secret-password" not in json.dumps(body)
|