feat(tradein/payments): подпись Token, клиент Т-Банка и сборка чека (#2733)
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>
This commit is contained in:
bot-backend 2026-08-06 12:36:57 +00:00 committed by lekss361
parent 2a1577738a
commit a32ccabd0d
8 changed files with 1206 additions and 0 deletions

View file

@ -0,0 +1,17 @@
"""Т-Банк интернет-эквайринг — чистый интеграционный слой (PR-C).
Модули здесь НЕ импортируют `app.core.config` и не пишут в БД: все секреты
(`terminal_key`, `password`, `base_url`) принимаются аргументами функций/
конструктора. Причина параллельный PR-B вводит эти поля в `config.py`,
а проводку (роутер, `_PUBLIC_PATHS`, `payments`-таблицы, статус-машина)
делает следующий PR-D. См. `mera-tbank-acquiring-recon.md` (корень репо)
§3/§9 для полной схемы разбивки.
- `token.py` подпись `Token` запросов + проверка подписи нотификаций.
- `receipt.py` сборка `Receipt` (54-ФЗ, ФФД 1.05) для услуги.
- `tbank_client.py` httpx-клиент `Init/GetState/CheckOrder/Confirm/Cancel`.
Docs: https://developer.tbank.ru/eacq/intro
"""
from __future__ import annotations

View file

@ -0,0 +1,143 @@
"""Сборка объекта `Receipt` (54-ФЗ, ФФД 1.05) для чека Т-Банк эквайринга.
Продукт продаёт УСЛУГУ (не товар) везде фиксированы `PaymentObject="service"`
и `PaymentMethod="full_payment"` (одномоментная оплата за уже готовую услугу,
без предоплат/кредита/частичных расчётов).
Схема (`Receipt` в `Init`, ФФД 1.05) источник, снят живым запросом
2026-08-06: https://developer.tbank.ru/eacq/api/init
- `Email` ИЛИ `Phone` обязательно хотя бы одно (перекрёстный required).
- `Taxation` обязателен: `osn|usn_income|usn_income_outcome|esn|patent`.
- `Items[].Name` <=128 символов, обязателен.
- `Items[].Price`/`Quantity`/`Amount` числа, В КОПЕЙКАХ; `Amount` это
произведение `Price * Quantity` (дословно из API-reference).
- `Items[].Tax` ставка НДС. Актуальный список 2026 (Init API reference):
`none|vat0|vat5|vat7|vat10|vat22|vat105|vat107|vat110|vat122`.
`vat20`/`vat120` В СПИСКЕ НЕТ сняты, не использовать (см. recon §6/§11
в `mera-tbank-acquiring-recon.md`, корень репо).
ВАЖНО: `Receipt` НЕ участвует в расчёте `Token` (`token.py` отсекает любые
вложенные `dict`/`list` из подписи) это архитектурно гарантировано самой
функцией `token.sign`, а не соглашением здесь.
"""
from __future__ import annotations
from dataclasses import dataclass
from typing import Any, Literal
TaxRate = Literal[
"none", "vat0", "vat5", "vat7", "vat10", "vat22", "vat105", "vat107", "vat110", "vat122"
]
Taxation = Literal["osn", "usn_income", "usn_income_outcome", "esn", "patent"]
_ALLOWED_TAX_RATES: frozenset[str] = frozenset(
{"none", "vat0", "vat5", "vat7", "vat10", "vat22", "vat105", "vat107", "vat110", "vat122"}
)
_ALLOWED_TAXATION: frozenset[str] = frozenset(
{"osn", "usn_income", "usn_income_outcome", "esn", "patent"}
)
_MAX_ITEM_NAME_LEN = 128
_MAX_ITEMS = 100 # "Количество товаров в чеке — не больше 100" (API reference)
class ReceiptBuildError(ValueError):
"""Невалидные данные для сборки Receipt — не пройдёт валидацию Т-Банка."""
@dataclass(frozen=True, slots=True)
class ReceiptItem:
"""Одна позиция чека — услуга. `price_kopecks`/`quantity` — целые копейки/штуки."""
name: str
price_kopecks: int
quantity: int = 1
tax: TaxRate = "none"
@property
def amount_kopecks(self) -> int:
"""Items[].Amount = Price * Quantity (дословно из API reference)."""
return self.price_kopecks * self.quantity
def to_payload(self) -> dict[str, Any]:
if not self.name or len(self.name) > _MAX_ITEM_NAME_LEN:
raise ReceiptBuildError(
f"Items[].Name должен быть 1..{_MAX_ITEM_NAME_LEN} символов, "
f"получено {len(self.name)}"
)
if self.price_kopecks <= 0:
raise ReceiptBuildError("Items[].Price должен быть > 0 (в копейках)")
if self.quantity <= 0:
raise ReceiptBuildError("Items[].Quantity должен быть > 0")
if self.tax not in _ALLOWED_TAX_RATES:
raise ReceiptBuildError(
f"Items[].Tax={self.tax!r} не входит в актуальный список Т-Банка "
f"({sorted(_ALLOWED_TAX_RATES)}) — vat20/vat120 сняты, не используются"
)
return {
"Name": self.name,
"Price": self.price_kopecks,
"Quantity": self.quantity,
"Amount": self.amount_kopecks,
"Tax": self.tax,
"PaymentMethod": "full_payment",
"PaymentObject": "service",
}
def build_receipt(
*,
items: list[ReceiptItem],
taxation: Taxation,
email: str | None = None,
phone: str | None = None,
) -> dict[str, Any]:
"""Собирает `Receipt` (ФФД 1.05) для одного заказа (может быть >1 позиции).
Инвариант «сумма Items[].Amount == Init.Amount» здесь НЕ проверяется
`Receipt` строится независимо от `Init`-payload заказа. Сверка на
вызывающей стороне (`service.py`, следующий PR) через
`receipt_total_kopecks(receipt) == init_amount_kopecks`. См. тест
`test_receipt_total_matches_order_amount_invariant` в
`tests/test_payments_receipt.py`, который проверяет именно эту сверку.
"""
if not items:
raise ReceiptBuildError("Receipt.Items не может быть пустым")
if len(items) > _MAX_ITEMS:
raise ReceiptBuildError(f"Receipt.Items — не больше {_MAX_ITEMS} позиций")
if taxation not in _ALLOWED_TAXATION:
raise ReceiptBuildError(
f"Taxation={taxation!r} не входит в допустимый список ({sorted(_ALLOWED_TAXATION)})"
)
email_norm = (email or "").strip() or None
phone_norm = (phone or "").strip() or None
if not email_norm and not phone_norm:
raise ReceiptBuildError("Нужно указать Email или Phone (хотя бы одно)")
payload: dict[str, Any] = {
"Taxation": taxation,
"Items": [item.to_payload() for item in items],
}
if email_norm:
payload["Email"] = email_norm
if phone_norm:
payload["Phone"] = phone_norm
return payload
def receipt_total_kopecks(receipt: dict[str, Any]) -> int:
"""Сумма `Items[].Amount` — для сверки вызывающей стороной с `Init.Amount`."""
items = receipt.get("Items")
if not isinstance(items, list):
return 0
total = 0
for item in items:
if isinstance(item, dict):
amount = item.get("Amount")
if isinstance(amount, int):
total += amount
return total

View file

@ -0,0 +1,249 @@
"""httpx-клиент Т-Банк эквайринга (Init/GetState/CheckOrder/Confirm/Cancel).
Стиль и обработка ошибок по образцу
`app.services.tgbot.client.TelegramClient`: единственные нужные методы,
не тянем отдельный SDK ради пяти HTTP-вызовов.
Модуль НЕ импортирует `app.core.config` все параметры (`terminal_key`,
`password`, `base_url`) передаются в конструктор явно аргументами.
Архитектурное ограничение PR-C (см. `app/services/payments/__init__.py`):
параллельный PR-B вводит эти поля в `config.py`, проводку делает PR-D.
Docs: https://developer.tbank.ru/eacq/api
Ретраи:
- Сетевые ошибки (timeout/connect) и HTTP 5xx экспоненциальный backoff,
capped на `_MAX_BACKOFF_S`.
- Любая 4xx НЕ ретраится (запрос некорректен / права не те повтор
транспортного вызова не поможет), сразу `TBankApiError`.
- Бизнес-отказ (HTTP 200, но `Success: false` в теле) тоже НЕ
ретраится: это содержательный ответ банка, а не сбой транспорта.
БЕЗОПАСНОСТЬ: `password` и `Token` НИКОГДА не попадают в `logger.*`
логируем только имя метода, HTTP-статус, `ErrorCode`/`Message`/`Details`
из ответа банка.
"""
from __future__ import annotations
import asyncio
import logging
from typing import Any
import httpx
from app.services.payments.token import sign
logger = logging.getLogger(__name__)
_DEFAULT_TIMEOUT_S = 15.0
_MAX_BACKOFF_S = 30.0
_DEFAULT_MAX_RETRIES = 3
DEFAULT_BASE_URL = "https://securepay.tinkoff.ru"
class TBankApiError(Exception):
"""T-Bank Acquiring API ответил ошибкой (HTTP-ошибка или `Success: false`)."""
def __init__(self, method: str, error_code: str, message: str, details: str = "") -> None:
self.method = method
self.error_code = error_code
self.message = message
self.details = details
text = f"T-Bank API {method} failed: [{error_code}] {message}"
if details:
text += f"{details}"
super().__init__(text)
def _error_from_body(response: httpx.Response) -> tuple[str, str, str]:
"""Парсит (ErrorCode, Message, Details) из тела ответа; fallback на HTTP-статус."""
try:
data = response.json()
except ValueError:
return str(response.status_code), (response.text or "")[:200], ""
if not isinstance(data, dict):
return str(response.status_code), str(data)[:200], ""
error_code = str(data.get("ErrorCode", response.status_code))
message = str(data.get("Message", ""))
details = str(data.get("Details", ""))
return error_code, message, details
class TBankClient:
"""Клиент Т-Банк эквайринга на `httpx.AsyncClient`.
Каждый вызов отдельное короткоживущее соединение (без общего
connection-pool между вызовами; частота вызовов в checkout-потоке
низкая, держать долгоживущий клиент не нужно тот же паттерн, что
`TelegramClient`).
"""
def __init__(
self,
*,
terminal_key: str,
password: str,
base_url: str = DEFAULT_BASE_URL,
timeout: float = _DEFAULT_TIMEOUT_S,
) -> None:
self._terminal_key = terminal_key
self._password = password
self._base = f"{base_url.rstrip('/')}/v2"
self._timeout = timeout
def _signed_payload(self, payload: dict[str, Any]) -> dict[str, Any]:
"""Добавляет `TerminalKey` + `Token`. Сам `password` в тело не уходит."""
body: dict[str, Any] = {"TerminalKey": self._terminal_key, **payload}
body["Token"] = sign(body, self._password)
return body
async def _request(
self,
method: str,
payload: dict[str, Any],
*,
max_retries: int = _DEFAULT_MAX_RETRIES,
) -> dict[str, Any]:
"""POST `method` с подписанным JSON-телом. Ретраит network/5xx, иначе raise сразу."""
body = self._signed_payload(payload)
url = f"{self._base}/{method}"
attempt = 0
while True:
attempt += 1
try:
async with httpx.AsyncClient(timeout=self._timeout) as client:
response = await client.post(url, json=body)
except (httpx.TimeoutException, httpx.NetworkError) as exc:
if attempt > max_retries:
logger.error(
"tbank client: %s — network error после %d попыток: %s",
method,
attempt,
exc,
)
raise TBankApiError(method, "network_error", str(exc)) from exc
backoff = min(2.0**attempt, _MAX_BACKOFF_S)
logger.warning(
"tbank client: %s — network error (попытка %d/%d) — retry через %.0fs",
method,
attempt,
max_retries,
backoff,
)
await asyncio.sleep(backoff)
continue
if response.status_code >= 500:
if attempt > max_retries:
error_code, message, details = _error_from_body(response)
logger.error(
"tbank client: %s — HTTP %d после %d попыток, сдаёмся",
method,
response.status_code,
attempt,
)
raise TBankApiError(method, error_code, message, details)
backoff = min(2.0**attempt, _MAX_BACKOFF_S)
logger.warning(
"tbank client: %s — HTTP %d (попытка %d/%d) — retry через %.0fs",
method,
response.status_code,
attempt,
max_retries,
backoff,
)
await asyncio.sleep(backoff)
continue
if response.status_code >= 400:
# 4xx кроме сетевых сценариев выше — запрос некорректен, повтор не поможет.
error_code, message, details = _error_from_body(response)
raise TBankApiError(method, error_code, message, details)
try:
data = response.json()
except ValueError as exc:
raise TBankApiError(method, "invalid_json", str(exc)) from exc
if not isinstance(data, dict):
raise TBankApiError(method, "invalid_response", "тело ответа — не JSON-объект")
if not data.get("Success"):
error_code = str(data.get("ErrorCode", response.status_code))
message = str(data.get("Message", ""))
details = str(data.get("Details", ""))
raise TBankApiError(method, error_code, message, details)
return data
async def init_payment(
self,
*,
order_id: str,
amount_kopecks: int,
description: str = "",
notification_url: str | None = None,
success_url: str | None = None,
fail_url: str | None = None,
receipt: dict[str, Any] | None = None,
pay_type: str | None = None,
data: dict[str, str] | None = None,
) -> dict[str, Any]:
"""`POST /v2/Init` — инициирует платёж, возвращает `PaymentId` + `PaymentURL`."""
payload: dict[str, Any] = {"OrderId": order_id, "Amount": amount_kopecks}
if description:
payload["Description"] = description
if notification_url:
payload["NotificationURL"] = notification_url
if success_url:
payload["SuccessURL"] = success_url
if fail_url:
payload["FailURL"] = fail_url
if receipt:
payload["Receipt"] = receipt
if pay_type:
payload["PayType"] = pay_type
if data:
payload["DATA"] = data
return await self._request("Init", payload)
async def get_state(self, *, payment_id: str) -> dict[str, Any]:
"""`POST /v2/GetState` — статус платежа по `PaymentId`."""
return await self._request("GetState", {"PaymentId": payment_id})
async def check_order(self, *, order_id: str) -> dict[str, Any]:
"""`POST /v2/CheckOrder` — список платежей по `OrderId` (для реконсиляции)."""
return await self._request("CheckOrder", {"OrderId": order_id})
async def confirm(
self,
*,
payment_id: str,
amount_kopecks: int | None = None,
receipt: dict[str, Any] | None = None,
) -> dict[str, Any]:
"""`POST /v2/Confirm` — подтверждение холда (двухстадийная оплата, `PayType=T`)."""
payload: dict[str, Any] = {"PaymentId": payment_id}
if amount_kopecks is not None:
payload["Amount"] = amount_kopecks
if receipt:
payload["Receipt"] = receipt
return await self._request("Confirm", payload)
async def cancel(
self,
*,
payment_id: str,
amount_kopecks: int | None = None,
receipt: dict[str, Any] | None = None,
) -> dict[str, Any]:
"""`POST /v2/Cancel` — отмена/возврат (полный, если `amount_kopecks` не передан)."""
payload: dict[str, Any] = {"PaymentId": payment_id}
if amount_kopecks is not None:
payload["Amount"] = amount_kopecks
if receipt:
payload["Receipt"] = receipt
return await self._request("Cancel", payload)

View file

@ -0,0 +1,98 @@
"""Подпись `Token` запросов Т-Банк эквайринга и проверка подписи нотификаций.
Docs (проверено живым запросом к doc-порталу, 2026-08-06):
- https://developer.tbank.ru/eacq/intro/developer/token формирование Token.
- https://developer.tbank.ru/eacq/intro/developer/notification
(раздел «Проверить токен уведомлений») тот же алгоритм для входящих
нотификаций.
Алгоритм (идентичен для исходящего запроса и для проверки нотификации):
1. Берём ТОЛЬКО плоские поля payload: исключаем ключ `Token`, исключаем
`None`, исключаем значения-`dict`/`list` (документация формулирует это
как «кроме параметра Token и вложенных объектов (Data, Receipt)»
здесь обобщено до правила по ТИПУ значения, а не по имени ключа: любые
вложенные объекты/массивы, будь то `Receipt`, `DATA`, `Data`, `Items`
или `Shops`, отсекаются одинаково, потому что все они не примитивы).
2. `bool` `"true"`/`"false"` (нижний регистр); `int`/`float` строка без
экспоненциальной записи; `str` как есть.
3. Добавляем пару `Password: <пароль_терминала>`.
4. Сортируем пары по имени ключа (лексикографически по строке ключа),
конкатенируем ТОЛЬКО значения (не ключи и не имена) в одну строку.
5. SHA-256 (UTF-8) от строки, hex-digest в нижнем регистре.
Эталонные векторы (см. `tests/test_payments_token.py`) сняты дословно с
doc-портала оба подтверждены живым запросом, не выдуманы.
"""
from __future__ import annotations
import hashlib
import hmac
from typing import Any
_EXCLUDED_KEYS = frozenset({"Token"})
def _stringify_value(value: bool | int | float | str) -> str:
"""Приводит плоское значение к строке по правилам Т-Банка.
`bool` проверяем ДО `int`: в Python `bool` подкласс `int`
(`isinstance(True, int) is True`), поэтому порядок веток важен
иначе `True` попал бы в ветку int и дал `"1"` вместо `"true"`.
"""
if isinstance(value, bool):
return "true" if value else "false"
if isinstance(value, int):
return str(value)
if isinstance(value, float):
# `format(..., "f")` — фиксированная нотация, Python никогда не
# добавляет экспоненту при presentation type 'f' (в отличие от
# str()/repr(), которые для очень больших/малых float дают "1e+21").
text = format(value, "f")
if "." in text:
text = text.rstrip("0").rstrip(".")
return text
return str(value)
def _flatten_signable_fields(payload: dict[str, Any]) -> dict[str, str]:
"""Плоские поля payload, готовые к конкатенации: без Token/None/dict/list."""
result: dict[str, str] = {}
for key, value in payload.items():
if key in _EXCLUDED_KEYS or value is None:
continue
if isinstance(value, dict | list):
continue
result[key] = _stringify_value(value)
return result
def sign(payload: dict[str, Any], password: str) -> str:
"""Считает `Token` для исходящего запроса (Init/GetState/CheckOrder/...).
`payload` тело запроса ДО добавления поля `Token` (поле `Password`
самому передавать не нужно функция добавляет его сама и удаляет
участие любых вложенных объектов автоматически).
"""
fields = _flatten_signable_fields(payload)
fields["Password"] = password
raw = "".join(fields[key] for key in sorted(fields))
return hashlib.sha256(raw.encode("utf-8")).hexdigest()
def verify_notification_token(payload: dict[str, Any], password: str) -> bool:
"""Проверяет `Token` входящей нотификации: пересчёт + `hmac.compare_digest`.
`payload` полное тело нотификации, включая присланный `Token` (сам
алгоритм сборки исключает ключ `Token` из подписи см. `_EXCLUDED_KEYS`).
Возвращает `False`, если в payload нет строкового непустого `Token`
(нечего сравнивать) вызывающая сторона обязана трактовать это как
отказ в обработке нотификации, а не как «пропустить проверку».
"""
received_token = payload.get("Token")
if not isinstance(received_token, str) or not received_token:
return False
expected_token = sign(payload, password)
return hmac.compare_digest(expected_token, received_token)

View file

@ -0,0 +1,293 @@
"""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)

View file

@ -0,0 +1,208 @@
"""Тесты `app.services.payments.receipt` — сборка Receipt (ФФД 1.05, услуга)."""
from __future__ import annotations
import pytest
from app.services.payments.receipt import (
ReceiptBuildError,
ReceiptItem,
build_receipt,
receipt_total_kopecks,
)
def _item(price_kopecks: int = 99000, quantity: int = 1, tax: str = "none") -> ReceiptItem:
return ReceiptItem(
name="Отчёт об оценке квартиры (электронный)",
price_kopecks=price_kopecks,
quantity=quantity,
tax=tax, # type: ignore[arg-type]
)
# ── happy path ────────────────────────────────────────────────────────────────
def test_build_receipt_happy_path_with_email() -> None:
receipt = build_receipt(items=[_item()], taxation="usn_income", email="buyer@example.com")
assert receipt["Taxation"] == "usn_income"
assert receipt["Email"] == "buyer@example.com"
assert "Phone" not in receipt
assert len(receipt["Items"]) == 1
item = receipt["Items"][0]
assert item["Name"] == "Отчёт об оценке квартиры (электронный)"
assert item["Price"] == 99000
assert item["Quantity"] == 1
assert item["Amount"] == 99000
assert item["Tax"] == "none"
assert item["PaymentMethod"] == "full_payment"
assert item["PaymentObject"] == "service"
def test_build_receipt_happy_path_with_phone_only() -> None:
receipt = build_receipt(items=[_item()], taxation="osn", phone="+79990000000")
assert receipt["Phone"] == "+79990000000"
assert "Email" not in receipt
def test_build_receipt_accepts_both_email_and_phone() -> None:
receipt = build_receipt(
items=[_item()], taxation="osn", email="buyer@example.com", phone="+79990000000"
)
assert receipt["Email"] == "buyer@example.com"
assert receipt["Phone"] == "+79990000000"
def test_amount_is_price_times_quantity() -> None:
item = _item(price_kopecks=10000, quantity=3)
assert item.amount_kopecks == 30000
payload = item.to_payload()
assert payload["Amount"] == 30000
# ── Email/Phone обязательность ──────────────────────────────────────────────
def test_build_receipt_requires_email_or_phone() -> None:
with pytest.raises(ReceiptBuildError, match=r"Email.*Phone|Phone.*Email"):
build_receipt(items=[_item()], taxation="osn")
def test_build_receipt_rejects_blank_email_and_phone() -> None:
with pytest.raises(ReceiptBuildError):
build_receipt(items=[_item()], taxation="osn", email=" ", phone="")
# ── Items[].Name длина ───────────────────────────────────────────────────────
def test_item_name_exactly_128_chars_is_ok() -> None:
item = ReceiptItem(name="A" * 128, price_kopecks=1000)
payload = item.to_payload()
assert payload["Name"] == "A" * 128
def test_item_name_over_128_chars_rejected() -> None:
item = ReceiptItem(name="A" * 129, price_kopecks=1000)
with pytest.raises(ReceiptBuildError, match="128"):
item.to_payload()
def test_item_name_empty_rejected() -> None:
item = ReceiptItem(name="", price_kopecks=1000)
with pytest.raises(ReceiptBuildError):
item.to_payload()
# ── Price / Quantity валидация ──────────────────────────────────────────────
def test_item_zero_price_rejected() -> None:
item = ReceiptItem(name="X", price_kopecks=0)
with pytest.raises(ReceiptBuildError):
item.to_payload()
def test_item_negative_price_rejected() -> None:
item = ReceiptItem(name="X", price_kopecks=-100)
with pytest.raises(ReceiptBuildError):
item.to_payload()
def test_item_zero_quantity_rejected() -> None:
item = ReceiptItem(name="X", price_kopecks=100, quantity=0)
with pytest.raises(ReceiptBuildError):
item.to_payload()
# ── Taxation / Tax enum ──────────────────────────────────────────────────────
@pytest.mark.parametrize("taxation", ["osn", "usn_income", "usn_income_outcome", "esn", "patent"])
def test_all_documented_taxation_values_accepted(taxation: str) -> None:
build_receipt(items=[_item()], taxation=taxation, email="a@test.ru") # type: ignore[arg-type]
def test_unknown_taxation_rejected() -> None:
with pytest.raises(ReceiptBuildError):
build_receipt(
items=[_item()],
taxation="usn",
email="a@test.ru", # type: ignore[arg-type]
)
@pytest.mark.parametrize(
"tax_rate",
["none", "vat0", "vat5", "vat7", "vat10", "vat22", "vat105", "vat107", "vat110", "vat122"],
)
def test_all_documented_2026_tax_rates_accepted(tax_rate: str) -> None:
"""Актуальный список 2026 года — все значения проходят валидацию."""
item = _item(tax=tax_rate)
payload = item.to_payload()
assert payload["Tax"] == tax_rate
@pytest.mark.parametrize("removed_rate", ["vat20", "vat120"])
def test_removed_vat20_vat120_rates_rejected(removed_rate: str) -> None:
"""vat20/vat120 сняты с актуального списка 2026 — не должны проходить."""
item = _item(tax=removed_rate)
with pytest.raises(ReceiptBuildError, match="vat20/vat120"):
item.to_payload()
# ── Items[] границы ──────────────────────────────────────────────────────────
def test_build_receipt_rejects_empty_items() -> None:
with pytest.raises(ReceiptBuildError):
build_receipt(items=[], taxation="osn", email="a@test.ru")
def test_build_receipt_rejects_more_than_100_items() -> None:
items = [_item() for _ in range(101)]
with pytest.raises(ReceiptBuildError, match="100"):
build_receipt(items=items, taxation="osn", email="a@test.ru")
def test_build_receipt_accepts_exactly_100_items() -> None:
items = [_item() for _ in range(100)]
receipt = build_receipt(items=items, taxation="osn", email="a@test.ru")
assert len(receipt["Items"]) == 100
# ── инвариант: сумма Items[].Amount == сумме заказа ─────────────────────────
def test_receipt_total_matches_order_amount_invariant() -> None:
"""Ключевой инвариант задачи: сумма Items[].Amount == общей сумме заказа."""
order_amount_kopecks = 148500
items = [
ReceiptItem(name="Отчёт об оценке", price_kopecks=99000, quantity=1, tax="none"),
ReceiptItem(name="Персональный оффер", price_kopecks=49500, quantity=1, tax="none"),
]
receipt = build_receipt(items=items, taxation="usn_income", email="a@test.ru")
assert receipt_total_kopecks(receipt) == order_amount_kopecks
def test_receipt_total_multi_quantity_item() -> None:
items = [ReceiptItem(name="Оценка", price_kopecks=5000, quantity=4, tax="vat22")]
receipt = build_receipt(items=items, taxation="osn", email="a@test.ru")
assert receipt_total_kopecks(receipt) == 20000
def test_receipt_total_kopecks_empty_items_key_returns_zero() -> None:
assert receipt_total_kopecks({"Taxation": "osn"}) == 0
def test_receipt_total_mismatch_detected_by_caller() -> None:
"""Демонстрирует, как вызывающая сторона обязана сверять сумму с Init.Amount."""
items = [ReceiptItem(name="Оценка", price_kopecks=10000, quantity=1, tax="none")]
receipt = build_receipt(items=items, taxation="osn", email="a@test.ru")
wrong_init_amount_kopecks = 99999
assert receipt_total_kopecks(receipt) != wrong_init_amount_kopecks

View file

@ -0,0 +1,198 @@
"""Тесты `app.services.payments.token` — подпись Token + проверка нотификаций.
Эталонные векторы (`test_sign_matches_official_init_vector`,
`test_sign_matches_official_notification_vector`) сняты ДОСЛОВНО живым
запросом (curl, 2026-08-06) с официального doc-портала Т-Банка:
- Init: https://developer.tbank.ru/eacq/intro/developer/token
(раздел «Сформировать токен»)
- Нотификация: https://developer.tbank.ru/eacq/intro/developer/notification
(раздел «Проверить токен уведомлений»)
Оба payload'а и оба итоговых hex-digest скопированы из HTML doc-портала
(не выдуманы) см. промежуточные шаги в комментариях у каждого теста.
Если хеш перестанет сходиться чинить `token.py`, НЕ тест.
"""
from __future__ import annotations
import hashlib
from app.services.payments.token import sign, verify_notification_token
# ── эталонный вектор №1: Init ────────────────────────────────────────────────
# Doc-портал, шаг за шагом (см. `token.py` docstring для полного описания):
# 1) [{"TerminalKey": "MerchantTerminalKey"}, {"Amount": 19200},
# {"OrderId": "00000"}, {"Description": "Подарочная карта на 1000 рублей"}]
# 2) + {"Password": "11111111111111"}
# 3) отсортировано по ключу: Amount, Description, OrderId, Password, TerminalKey
# 4) конкатенация значений:
# "19200Подарочная карта на 1000 рублей0000011111111111111MerchantTerminalKey"
# 5) SHA-256 → "72dd466f8ace0a37a1f740ce5fb78101712bc0665d91a8108c7c8a0ccd426db2"
_INIT_VECTOR_PAYLOAD = {
"TerminalKey": "MerchantTerminalKey",
"Amount": 19200,
"OrderId": "00000",
"Description": "Подарочная карта на 1000 рублей",
}
_INIT_VECTOR_PASSWORD = "11111111111111"
_INIT_VECTOR_TOKEN = "72dd466f8ace0a37a1f740ce5fb78101712bc0665d91a8108c7c8a0ccd426db2"
def test_sign_matches_official_init_vector() -> None:
"""Официальный вектор Init из документации Т-Банка."""
assert sign(_INIT_VECTOR_PAYLOAD, _INIT_VECTOR_PASSWORD) == _INIT_VECTOR_TOKEN
# ── эталонный вектор №2: нотификация ─────────────────────────────────────────
# Doc-портал, шаг за шагом:
# 1) [{"TerminalKey": "1234567890DEMO"}, {"OrderId": "000000"},
# {"Success": true}, {"Status": "AUTHORIZED"}, {"PaymentId": "0000000"},
# {"ErrorCode": "0"}, {"Amount": 1111}, {"CardId": "000000"},
# {"Pan": "200000******0000"}, {"ExpDate": "1111"}, {"RebillId": "000000"}]
# 2) + {"Password": "11111111111"}
# 3) отсортировано: Amount, CardId, ErrorCode, ExpDate, OrderId, Pan,
# Password, PaymentId, RebillId, Status, Success, TerminalKey
# 4) конкатенация значений:
# "111100000001111000000200000******0000111111111110000000000000AUTHORIZEDtrue1234567890DEMO"
# 5) SHA-256 → "1c0964277d0213349243065a0d5b838b8e90d2d25f740d0f2767836e710e80c8"
_NOTIFICATION_VECTOR_PAYLOAD = {
"TerminalKey": "1234567890DEMO",
"OrderId": "000000",
"Success": True,
"Status": "AUTHORIZED",
"PaymentId": "0000000",
"ErrorCode": "0",
"Amount": 1111,
"CardId": "000000",
"Pan": "200000******0000",
"ExpDate": "1111",
"RebillId": "000000",
}
_NOTIFICATION_VECTOR_PASSWORD = "11111111111"
_NOTIFICATION_VECTOR_TOKEN = "1c0964277d0213349243065a0d5b838b8e90d2d25f740d0f2767836e710e80c8"
def test_sign_matches_official_notification_vector() -> None:
"""Официальный вектор нотификации (Success/AUTHORIZED) из документации Т-Банка."""
assert (
sign(_NOTIFICATION_VECTOR_PAYLOAD, _NOTIFICATION_VECTOR_PASSWORD)
== _NOTIFICATION_VECTOR_TOKEN
)
def test_verify_notification_token_accepts_valid_official_vector() -> None:
"""`verify_notification_token` — тот же вектор, но с полем Token внутри payload."""
payload_with_token = {**_NOTIFICATION_VECTOR_PAYLOAD, "Token": _NOTIFICATION_VECTOR_TOKEN}
assert verify_notification_token(payload_with_token, _NOTIFICATION_VECTOR_PASSWORD) is True
def test_verify_notification_token_rejects_tampered_field() -> None:
"""Изменили Amount после подписи → Token больше не совпадает → False."""
tampered = {**_NOTIFICATION_VECTOR_PAYLOAD, "Token": _NOTIFICATION_VECTOR_TOKEN, "Amount": 9999}
assert verify_notification_token(tampered, _NOTIFICATION_VECTOR_PASSWORD) is False
def test_verify_notification_token_rejects_wrong_password() -> None:
payload_with_token = {**_NOTIFICATION_VECTOR_PAYLOAD, "Token": _NOTIFICATION_VECTOR_TOKEN}
assert verify_notification_token(payload_with_token, "wrong-password") is False
def test_verify_notification_token_rejects_missing_token() -> None:
assert verify_notification_token(dict(_NOTIFICATION_VECTOR_PAYLOAD), "any-password") is False
def test_verify_notification_token_rejects_empty_token() -> None:
payload = {**_NOTIFICATION_VECTOR_PAYLOAD, "Token": ""}
assert verify_notification_token(payload, _NOTIFICATION_VECTOR_PASSWORD) is False
# ── unit-детали алгоритма ─────────────────────────────────────────────────────
def test_bool_true_becomes_lowercase_string() -> None:
"""`Success: True` (Python bool) → строка "true" в конкатенации."""
with_bool = sign({"A": True}, "pw")
with_string = sign({"A": "true"}, "pw")
assert with_bool == with_string
def test_bool_false_becomes_lowercase_string() -> None:
with_bool = sign({"A": False}, "pw")
with_string = sign({"A": "false"}, "pw")
assert with_bool == with_string
def test_int_amount_stringified_without_quotes_semantics() -> None:
"""`Amount: 1111` (int) даёт тот же результат, что и `Amount: "1111"` (str)."""
with_int = sign({"Amount": 1111}, "pw")
with_str = sign({"Amount": "1111"}, "pw")
assert with_int == with_str
def test_float_without_leading_zero_loss_and_no_exponent() -> None:
"""Дробное число сериализуется без экспоненты и без хвостовых нулей.
Ключи после добавления Password: "A" < "Password" (лексикографически),
поэтому конкатенация значение A, затем значение Password.
"""
raw = "1234.5" + "pw"
expected = hashlib.sha256(raw.encode("utf-8")).hexdigest()
assert sign({"A": 1234.5}, "pw") == expected
def test_large_float_has_no_exponential_notation() -> None:
"""Очень большое число не сваливается в экспоненциальную запись (`1e+21`)."""
raw = "1000000000000000000000" + "pw"
expected = hashlib.sha256(raw.encode("utf-8")).hexdigest()
assert sign({"A": 1e21}, "pw") == expected
def test_none_values_are_skipped() -> None:
"""`None`-поля не участвуют в конкатенации вообще (не как пустая строка)."""
with_none = sign({"A": "x", "B": None}, "pw")
without_key = sign({"A": "x"}, "pw")
assert with_none == without_key
def test_nested_receipt_dict_is_ignored() -> None:
"""Вложенный `Receipt` (dict) не участвует в подписи."""
without_receipt = sign({"A": "x"}, "pw")
with_receipt = sign({"A": "x", "Receipt": {"Email": "a@test.ru", "Items": []}}, "pw")
assert without_receipt == with_receipt
def test_nested_data_dict_is_ignored() -> None:
"""Вложенный `DATA`/`Data` (dict) не участвует в подписи — оба варианта регистра ключа."""
baseline = sign({"A": "x"}, "pw")
assert sign({"A": "x", "DATA": {"Phone": "+70000000000"}}, "pw") == baseline
assert sign({"A": "x", "Data": {"Phone": "+70000000000"}}, "pw") == baseline
def test_nested_list_items_is_ignored() -> None:
"""Вложенный список (`Items`/`Shops` как root-ключ) не участвует в подписи."""
baseline = sign({"A": "x"}, "pw")
assert sign({"A": "x", "Items": [{"Name": "тест"}]}, "pw") == baseline
assert sign({"A": "x", "Shops": [{"ShopCode": "1"}]}, "pw") == baseline
def test_existing_token_field_in_payload_is_excluded() -> None:
"""Если в payload уже есть `Token` (например, переподписываем нотификацию) — игнорируется."""
without_token = sign({"A": "x"}, "pw")
with_token = sign({"A": "x", "Token": "stale-value-from-previous-signing"}, "pw")
assert without_token == with_token
def test_sort_is_by_key_name_not_insertion_order() -> None:
"""Порядок вставки ключей в payload не влияет на результат — сортировка по ключу."""
forward = sign({"Zeta": "1", "Alpha": "2", "Mid": "3"}, "pw")
reversed_order = sign({"Mid": "3", "Alpha": "2", "Zeta": "1"}, "pw")
assert forward == reversed_order
# sanity: строка действительно собрана в алфавитном порядке ключей.
# Ключи с Password: Alpha < Mid < Password < Zeta (лексикографически).
raw = "".join(["2", "3", "pw", "1"]) # Alpha->2, Mid->3, Password->pw, Zeta->1
expected = hashlib.sha256(raw.encode("utf-8")).hexdigest()
assert forward == expected