feat(tradein/payments): подпись Token, клиент Т-Банка и сборка чека #2733
8 changed files with 1206 additions and 0 deletions
17
tradein-mvp/backend/app/services/payments/__init__.py
Normal file
17
tradein-mvp/backend/app/services/payments/__init__.py
Normal 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
|
||||
143
tradein-mvp/backend/app/services/payments/receipt.py
Normal file
143
tradein-mvp/backend/app/services/payments/receipt.py
Normal 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
|
||||
249
tradein-mvp/backend/app/services/payments/tbank_client.py
Normal file
249
tradein-mvp/backend/app/services/payments/tbank_client.py
Normal 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)
|
||||
98
tradein-mvp/backend/app/services/payments/token.py
Normal file
98
tradein-mvp/backend/app/services/payments/token.py
Normal 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)
|
||||
0
tradein-mvp/backend/tests/services/payments/__init__.py
Normal file
0
tradein-mvp/backend/tests/services/payments/__init__.py
Normal file
293
tradein-mvp/backend/tests/services/payments/test_tbank_client.py
Normal file
293
tradein-mvp/backend/tests/services/payments/test_tbank_client.py
Normal 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)
|
||||
208
tradein-mvp/backend/tests/test_payments_receipt.py
Normal file
208
tradein-mvp/backend/tests/test_payments_receipt.py
Normal 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
|
||||
198
tradein-mvp/backend/tests/test_payments_token.py
Normal file
198
tradein-mvp/backend/tests/test_payments_token.py
Normal 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
|
||||
Loading…
Add table
Reference in a new issue