bot-backend
ec8213c269
All checks were successful
CI Trade-In / changes (pull_request) Successful in 9s
CI Trade-In / browser-tests (pull_request) Has been skipped
CI Trade-In / frontend-checks (pull_request) Has been skipped
CI / changes (pull_request) Successful in 11s
CI / backend-tests (pull_request) Has been skipped
CI / frontend-tests (pull_request) Has been skipped
CI / openapi-codegen-check (pull_request) Has been skipped
CI Trade-In / backend-tests (pull_request) Successful in 5m11s
Три дыры в одном замке (строка без payment_url невидима для _find_live_payment, но видима предикату UNIQUE 279 → ложный 409 на 30 минут): - tbank_client ловил пару (TimeoutException, NetworkError): RemoteProtocolError, ProxyError и UnsupportedProtocol летели наружу голым httpx-типом мимо `except TBankApiError` в checkout. Ловим родителя — httpx.TransportError. - Ветка «Success:true без PaymentURL» отвечала 502, не трогая статус, — здесь банк заказ ПРИНЯЛ. Тот же терминальный статус, error_code=no_payment_url; UPDATE вынесен в общий _mark_init_failed. - _FakeDb в тестах применял статус по наличию ключа в params, а не по тексту SQL: мутант без `SET status = :status` оставался зелёным. Гейт по SQL — мутант краснит все три теста про замок. Комментарий про «возможный холд» на ветке отказа Init поправлен: Init холд не создаёт, авторизация идёт с оплаты формы, а форму покупателю не выдавали.
998 lines
46 KiB
Python
998 lines
46 KiB
Python
"""Роутер платежей (app/api/v1/payments.py): идемпотентность выдачи, подпись,
|
||
kill-switch, capability-ссылка.
|
||
|
||
ПОЧЕМУ здесь свой мини-эмулятор БД, а не MagicMock. Проверяемое свойство —
|
||
«повторная нотификация НЕ создаёт вторую выдачу» — целиком держится на UNIQUE
|
||
из миграции 233 плюс `ON CONFLICT DO NOTHING`. MagicMock отдаёт то, что ему
|
||
скажут, поэтому такой тест был бы зелёным по построению: он не покраснел бы,
|
||
если убрать `ON CONFLICT` или заменить его на «SELECT, потом INSERT».
|
||
`_FakeDb` ниже объявляет UNIQUE-ключи ОТДЕЛЬНО от проверяемого SQL — ровно
|
||
теми колонками, что записаны в миграции, — и ведёт себя как Postgres: дубль
|
||
без `ON CONFLICT` падает ошибкой, дубль с `ON CONFLICT DO NOTHING` не
|
||
возвращает строку.
|
||
|
||
Фальсификация проверена руками (каждый раз краснеет ИМЕННО тот тест, который
|
||
про это свойство, и по значению, а не по ImportError):
|
||
- убрать `ON CONFLICT DO NOTHING` из INSERT в `payment_entitlements` →
|
||
`test_retry_after_crash_between_issue_and_processed_does_not_double_issue`
|
||
падает 500 вместо "OK";
|
||
- убрать его же из INSERT в `payment_notifications` → падают все три теста
|
||
про идемпотентность;
|
||
- отключить проверку подписи → `test_notification_with_invalid_token_is_rejected`;
|
||
- отключить проверку срока → `test_report_link_rejects_expired_token`;
|
||
- убрать `ON CONFLICT DO NOTHING` из INSERT в `payments` →
|
||
`test_parallel_checkout_does_not_create_second_payment` краснеет
|
||
_UniqueViolationError вместо 409;
|
||
- растянуть `_ABANDONED_AFTER_MINUTES` до бесконечности (= убрать границу по
|
||
времени) → `test_abandoned_checkout_does_not_lock_the_buyer_out` получает в
|
||
ответе мёртвую ссылку — красное ПО ЗНАЧЕНИЮ;
|
||
- убрать `_assert_estimate_access` из checkout → `test_checkout_rejects_foreign_estimate`
|
||
видит 200 и созданный платёж вместо 404.
|
||
|
||
SQLite вместо этого не годится: NULLS NOT DISTINCT, jsonb, make_interval и
|
||
CAST(:x AS uuid) там не существуют, а настоящий Postgres в юнит-тестах этого
|
||
репозитория не поднимается (см. tests/conftest.py — DATABASE_URL заглушка).
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import os
|
||
import re
|
||
import sys
|
||
from datetime import UTC, datetime, timedelta
|
||
from pathlib import Path
|
||
from types import SimpleNamespace
|
||
from typing import Any
|
||
from unittest.mock import MagicMock
|
||
|
||
os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test")
|
||
|
||
_wp_mock = MagicMock()
|
||
sys.modules.setdefault("weasyprint", _wp_mock)
|
||
sys.modules.setdefault("weasyprint.CSS", _wp_mock)
|
||
sys.modules.setdefault("weasyprint.HTML", _wp_mock)
|
||
|
||
import httpx # noqa: E402
|
||
import pytest # noqa: E402
|
||
from fastapi import FastAPI # noqa: E402
|
||
from fastapi.testclient import TestClient # noqa: E402
|
||
from pydantic import SecretStr # noqa: E402
|
||
|
||
_PASSWORD = "test-terminal-password"
|
||
_ORDER_ID = "mera-0123456789abcdef0123456789abcdef"
|
||
_PAYMENT_ID = "3000000001"
|
||
_PAYMENT_UUID = "22222222-2222-2222-2222-222222222222"
|
||
_ESTIMATE_UUID = "33333333-3333-3333-3333-333333333333"
|
||
_AMOUNT = 15_000
|
||
|
||
_REPO_ROOT = Path(__file__).resolve().parents[1].parent
|
||
_MIGRATION = _REPO_ROOT / "backend" / "data" / "sql" / "233_payments.sql"
|
||
_CONTENT_TS = _REPO_ROOT / "frontend" / "src" / "app" / "mera-public" / "content.ts"
|
||
|
||
|
||
class _UniqueViolationError(RuntimeError):
|
||
"""Стенд-in для psycopg UniqueViolation — INSERT без ON CONFLICT в дубль."""
|
||
|
||
|
||
class _FakeDb:
|
||
"""Мини-Postgres на словарях: только те statement'ы, что шлёт роутер.
|
||
|
||
UNIQUE-ключи заданы ЗДЕСЬ, по миграции 233, а не выведены из проверяемого
|
||
SQL — иначе тест поедет вслед за дефектом вместо того, чтобы его поймать.
|
||
NULL считается равным NULL (NULLS NOT DISTINCT), как в миграции.
|
||
"""
|
||
|
||
_NOTIFICATION_KEY = ("tbank_payment_id", "status", "amount_kopecks", "token")
|
||
_ENTITLEMENT_KEY = ("payment_id", "kind", "ref_id")
|
||
# Частичный UNIQUE миграции 279: ключ (estimate_id, product_code), предикат —
|
||
# «живые» статусы. Переписан здесь ПО МИГРАЦИИ, а не импортирован из
|
||
# payments.py: иначе тест поехал бы вслед за дефектом. Совпадение списка с
|
||
# кодом отдельно гейтит test_live_status_predicate_matches_code.
|
||
_LIVE_PAYMENT_KEY = ("estimate_id", "product_code")
|
||
_LIVE_STATUSES = frozenset(
|
||
{
|
||
"NEW",
|
||
"FORM_SHOWED",
|
||
"PREAUTHORIZING",
|
||
"AUTHORIZING",
|
||
"AUTHORIZED",
|
||
"3DS_CHECKING",
|
||
"3DS_CHECKED",
|
||
"CHECKING",
|
||
"CHECKED",
|
||
"PROCESSING",
|
||
"CONFIRMING",
|
||
}
|
||
)
|
||
|
||
def __init__(self) -> None:
|
||
self.notifications: list[SimpleNamespace] = []
|
||
self.entitlements: list[SimpleNamespace] = []
|
||
self.payments: list[SimpleNamespace] = [
|
||
SimpleNamespace(
|
||
id=_PAYMENT_UUID,
|
||
order_id=_ORDER_ID,
|
||
status="NEW",
|
||
amount_kopecks=_AMOUNT,
|
||
estimate_id=_ESTIMATE_UUID,
|
||
created_by=None,
|
||
# product_code=None — эта строка обслуживает тесты нотификаций и
|
||
# не должна попадать в ключ живого платежа checkout-тестов.
|
||
product_code=None,
|
||
payment_url=None,
|
||
created_at=datetime.now(tz=UTC),
|
||
)
|
||
]
|
||
self.estimate_created_by: str | None = None
|
||
self.retain_until_updates: list[str] = []
|
||
self.commits = 0
|
||
|
||
# -- SQLAlchemy-совместимая поверхность -------------------------------
|
||
def execute(self, statement: Any, params: dict[str, Any] | None = None) -> Any:
|
||
sql = " ".join(str(statement).split())
|
||
params = params or {}
|
||
if "INSERT INTO payment_notifications" in sql:
|
||
return self._insert_notification(sql, params)
|
||
if "FROM payment_notifications" in sql and sql.startswith("SELECT"):
|
||
return _Result(self._find_notification(params))
|
||
if "UPDATE payment_notifications" in sql:
|
||
for row in self.notifications:
|
||
if row.id == params["id"] and row.processed_at is None:
|
||
row.processed_at = datetime.now(tz=UTC)
|
||
return _Result(None)
|
||
if "INSERT INTO payment_entitlements" in sql:
|
||
return self._insert_entitlement(sql, params)
|
||
if "FROM payment_entitlements" in sql and sql.startswith("SELECT"):
|
||
return _Result(
|
||
next(
|
||
(
|
||
row
|
||
for row in self.entitlements
|
||
if row.kind == params["kind"] and row.subject == params["token"]
|
||
),
|
||
None,
|
||
)
|
||
)
|
||
if "INSERT INTO payments" in sql:
|
||
return self._insert_payment(sql, params)
|
||
if "FROM payments" in sql and sql.startswith("SELECT"):
|
||
if "estimate_id" in params:
|
||
return _Result(self._find_live_payment(params))
|
||
return _Result(
|
||
next((p for p in self.payments if p.order_id == params.get("order_id")), None)
|
||
)
|
||
if "UPDATE payments" in sql and "DEADLINE_EXPIRED" in sql:
|
||
return _Result(self._expire_abandoned(params))
|
||
if "UPDATE payments" in sql:
|
||
# Колонки берутся из ТЕКСТА запроса, а не из наличия ключа в params:
|
||
# по params фейк применял бы статус и к запросу, из которого
|
||
# `SET status = :status` выкинули, — мутант оставался бы зелёным.
|
||
sets_status = "status = :status" in sql
|
||
sets_url = "payment_url = :payment_url" in sql
|
||
sets_error = "error_code = :code" in sql
|
||
for payment in self.payments:
|
||
if payment.order_id != params["order_id"]:
|
||
continue
|
||
if sets_status:
|
||
payment.status = params["status"]
|
||
if sets_url:
|
||
payment.payment_url = params["payment_url"]
|
||
if sets_error:
|
||
payment.error_code = params["code"]
|
||
return _Result(None)
|
||
if "FROM trade_in_estimates" in sql and sql.startswith("SELECT"):
|
||
if params.get("id") != _ESTIMATE_UUID:
|
||
return _Result(None)
|
||
return _Result(SimpleNamespace(id=_ESTIMATE_UUID, created_by=self.estimate_created_by))
|
||
if "UPDATE trade_in_estimates" in sql:
|
||
self.retain_until_updates.append(params["id"])
|
||
return _Result(None)
|
||
raise AssertionError(f"неожиданный SQL в тесте: {sql[:120]}")
|
||
|
||
def commit(self) -> None:
|
||
self.commits += 1
|
||
|
||
def close(self) -> None:
|
||
pass
|
||
|
||
# -- эмуляция UNIQUE ---------------------------------------------------
|
||
def _insert_notification(self, sql: str, params: dict[str, Any]) -> _Result:
|
||
key = tuple(
|
||
params[name]
|
||
for name in ("payment_id", "status", "amount", "token") # порядок = _NOTIFICATION_KEY
|
||
)
|
||
if any(self._key_of(row, self._NOTIFICATION_KEY) == key for row in self.notifications):
|
||
return self._conflict(sql)
|
||
row = SimpleNamespace(
|
||
id=len(self.notifications) + 1,
|
||
order_id=params["order_id"],
|
||
tbank_payment_id=params["payment_id"],
|
||
status=params["status"],
|
||
amount_kopecks=params["amount"],
|
||
token=params["token"],
|
||
token_valid=params["token_valid"],
|
||
processed_at=None,
|
||
)
|
||
self.notifications.append(row)
|
||
return _Result(row)
|
||
|
||
def _insert_payment(self, sql: str, params: dict[str, Any]) -> _Result:
|
||
"""Ведёт себя как Postgres с частичным UNIQUE миграции 279.
|
||
|
||
Конфликт наступает только когда УЖЕ есть строка с той же парой
|
||
(estimate_id, product_code) И статусом из предиката — ровно как у
|
||
частичного индекса. Без `ON CONFLICT DO NOTHING` в SQL — исключение.
|
||
"""
|
||
key = (params["estimate_id"], params["product_code"])
|
||
if key[0] is not None and any(
|
||
self._key_of(row, self._LIVE_PAYMENT_KEY) == key and row.status in self._LIVE_STATUSES
|
||
for row in self.payments
|
||
):
|
||
return self._conflict(sql)
|
||
row = SimpleNamespace(
|
||
id=f"pay-{len(self.payments) + 1}",
|
||
order_id=params["order_id"],
|
||
status="NEW",
|
||
amount_kopecks=params["amount"],
|
||
estimate_id=params["estimate_id"],
|
||
created_by=params["created_by"],
|
||
product_code=params["product_code"],
|
||
payment_url=None,
|
||
created_at=datetime.now(tz=UTC),
|
||
)
|
||
self.payments.append(row)
|
||
return _Result(row)
|
||
|
||
def _find_live_payment(self, params: dict[str, Any]) -> SimpleNamespace | None:
|
||
return next(
|
||
(
|
||
row
|
||
for row in self.payments
|
||
if self._key_of(row, self._LIVE_PAYMENT_KEY)
|
||
== (params["estimate_id"], params["product_code"])
|
||
and row.payment_url is not None
|
||
and row.status in set(params["reusable"])
|
||
),
|
||
None,
|
||
)
|
||
|
||
def _expire_abandoned(self, params: dict[str, Any]) -> None:
|
||
deadline = datetime.now(tz=UTC) - timedelta(minutes=int(params["mins"]))
|
||
for row in self.payments:
|
||
if (
|
||
row.estimate_id == params["estimate_id"]
|
||
and row.product_code == params["product_code"]
|
||
and row.status in set(params["abandonable"])
|
||
and row.created_at < deadline
|
||
):
|
||
row.status = "DEADLINE_EXPIRED"
|
||
return None
|
||
|
||
def _insert_entitlement(self, sql: str, params: dict[str, Any]) -> _Result:
|
||
key = (params["payment_id"], params["kind"], params["ref_id"])
|
||
if any(self._key_of(row, self._ENTITLEMENT_KEY) == key for row in self.entitlements):
|
||
return self._conflict(sql)
|
||
row = SimpleNamespace(
|
||
id=f"ent-{len(self.entitlements) + 1}",
|
||
payment_id=params["payment_id"],
|
||
subject=params["subject"],
|
||
kind=params["kind"],
|
||
ref_id=params["ref_id"],
|
||
expires_at=datetime.now(tz=UTC) + timedelta(days=int(params["days"])),
|
||
)
|
||
self.entitlements.append(row)
|
||
return _Result(row)
|
||
|
||
@staticmethod
|
||
def _key_of(row: SimpleNamespace, names: tuple[str, ...]) -> tuple[Any, ...]:
|
||
return tuple(getattr(row, name) for name in names)
|
||
|
||
@staticmethod
|
||
def _conflict(sql: str) -> _Result:
|
||
if "ON CONFLICT DO NOTHING" not in sql:
|
||
raise _UniqueViolationError("duplicate key value violates unique constraint")
|
||
return _Result(None)
|
||
|
||
def _find_notification(self, params: dict[str, Any]) -> SimpleNamespace | None:
|
||
key = (params["payment_id"], params["status"], params["amount"], params["token"])
|
||
return next(
|
||
(row for row in self.notifications if self._key_of(row, self._NOTIFICATION_KEY) == key),
|
||
None,
|
||
)
|
||
|
||
|
||
class _Result:
|
||
def __init__(self, row: Any) -> None:
|
||
self._row = row
|
||
|
||
def fetchone(self) -> Any:
|
||
return self._row
|
||
|
||
|
||
@pytest.fixture()
|
||
def db() -> _FakeDb:
|
||
return _FakeDb()
|
||
|
||
|
||
@pytest.fixture()
|
||
def client(db: _FakeDb, monkeypatch: pytest.MonkeyPatch) -> TestClient:
|
||
from app.api.v1 import payments as payments_module
|
||
from app.core.config import settings
|
||
from app.core.db import get_db
|
||
|
||
monkeypatch.setattr(settings, "payments_enabled", True)
|
||
monkeypatch.setattr(settings, "tbank_password", SecretStr(_PASSWORD))
|
||
monkeypatch.setattr(settings, "tbank_terminal_key", "TERM-TEST")
|
||
|
||
app = FastAPI()
|
||
app.include_router(payments_module.router, prefix="/api/v1/trade-in")
|
||
app.dependency_overrides[get_db] = lambda: db
|
||
return TestClient(app)
|
||
|
||
|
||
def _signed_notification(**overrides: Any) -> dict[str, Any]:
|
||
from app.services.payments.token import sign
|
||
|
||
body: dict[str, Any] = {
|
||
"TerminalKey": "TERM-TEST",
|
||
"OrderId": _ORDER_ID,
|
||
"PaymentId": _PAYMENT_ID,
|
||
"Status": "CONFIRMED",
|
||
"Success": True,
|
||
"Amount": _AMOUNT,
|
||
}
|
||
body.update(overrides)
|
||
body["Token"] = sign(body, _PASSWORD)
|
||
return body
|
||
|
||
|
||
# ── идемпотентность выдачи ───────────────────────────────────────────────────
|
||
|
||
|
||
def test_repeat_notification_issues_only_one_entitlement(client: TestClient, db: _FakeDb) -> None:
|
||
"""Ретрай банка (тот же Token) не выдаёт второй отчёт и не рвёт ответ "OK".
|
||
|
||
Фальсификация (проверено руками): убрать `ON CONFLICT DO NOTHING` из
|
||
INSERT в payment_entitlements → второй запрос падает _UniqueViolationError и
|
||
тест краснеет на status_code 500; заменить дедуп на «SELECT потом INSERT»
|
||
и снять UNIQUE → len(entitlements) == 2.
|
||
"""
|
||
body = _signed_notification()
|
||
|
||
first = client.post("/api/v1/trade-in/payments/notify", json=body)
|
||
second = client.post("/api/v1/trade-in/payments/notify", json=body)
|
||
|
||
assert first.status_code == 200, first.text
|
||
assert first.text == "OK"
|
||
assert second.status_code == 200, second.text
|
||
assert second.text == "OK"
|
||
assert len(db.entitlements) == 1, "повторная нотификация выдала второй отчёт"
|
||
assert len(db.notifications) == 1, "дубль нотификации записался второй строкой"
|
||
assert db.notifications[0].processed_at is not None
|
||
assert db.retain_until_updates == [_ESTIMATE_UUID]
|
||
|
||
|
||
def test_unprocessed_duplicate_is_fulfilled_on_retry(client: TestClient, db: _FakeDb) -> None:
|
||
"""Строка нотификации есть, а processed_at пуст → выдача ОБЯЗАНА состояться.
|
||
|
||
Это контракт processed_at из миграции 233: падение процесса между записью
|
||
нотификации и выдачей не должно оставить клиента без товара при списанных
|
||
деньгах. Красный вариант — трактовать существование строки как «уже
|
||
обработано» (тогда entitlements пуст).
|
||
"""
|
||
body = _signed_notification()
|
||
db.notifications.append(
|
||
SimpleNamespace(
|
||
id=1,
|
||
order_id=_ORDER_ID,
|
||
tbank_payment_id=_PAYMENT_ID,
|
||
status="CONFIRMED",
|
||
amount_kopecks=_AMOUNT,
|
||
token=body["Token"],
|
||
token_valid=True,
|
||
processed_at=None,
|
||
)
|
||
)
|
||
|
||
response = client.post("/api/v1/trade-in/payments/notify", json=body)
|
||
|
||
assert response.status_code == 200, response.text
|
||
assert len(db.entitlements) == 1
|
||
assert db.notifications[0].processed_at is not None
|
||
|
||
|
||
def test_retry_after_crash_between_issue_and_processed_does_not_double_issue(
|
||
client: TestClient, db: _FakeDb
|
||
) -> None:
|
||
"""Худший реальный случай: выдача прошла, а processed_at проставить не успели.
|
||
|
||
Ретрай банка ОБЯЗАН дойти до конца (иначе processed_at не проставится
|
||
никогда и так будет каждый час сутки) — и при этом не выдать второй отчёт.
|
||
Единственное, что здесь работает, — UNIQUE (payment_id, kind, ref_id) +
|
||
ON CONFLICT DO NOTHING: `_FakeDb` ведёт себя как Postgres и на INSERT без
|
||
ON CONFLICT кидает _UniqueViolationError (проверено руками: убрать ON CONFLICT
|
||
из INSERT в payment_entitlements → 500 вместо "OK", тест краснеет).
|
||
"""
|
||
body = _signed_notification()
|
||
db.notifications.append(
|
||
SimpleNamespace(
|
||
id=1,
|
||
order_id=_ORDER_ID,
|
||
tbank_payment_id=_PAYMENT_ID,
|
||
status="CONFIRMED",
|
||
amount_kopecks=_AMOUNT,
|
||
token=body["Token"],
|
||
token_valid=True,
|
||
processed_at=None,
|
||
)
|
||
)
|
||
_issue_entitlement(db, "already-issued-token", expires_at=None)
|
||
|
||
response = client.post("/api/v1/trade-in/payments/notify", json=body)
|
||
|
||
assert response.status_code == 200, response.text
|
||
assert response.text == "OK"
|
||
assert len(db.entitlements) == 1, "выдан второй отчёт по тому же платежу"
|
||
assert db.entitlements[0].subject == "already-issued-token", "токен подменён на новый"
|
||
assert db.notifications[0].processed_at is not None
|
||
|
||
|
||
# ── подпись и сумма ──────────────────────────────────────────────────────────
|
||
|
||
|
||
def test_notification_with_invalid_token_is_rejected(client: TestClient, db: _FakeDb) -> None:
|
||
body = _signed_notification()
|
||
body["Token"] = "deadbeef" * 8 # подпись не от нашего пароля
|
||
|
||
response = client.post("/api/v1/trade-in/payments/notify", json=body)
|
||
|
||
assert response.status_code == 403
|
||
assert db.entitlements == [], "выдача по неподписанной нотификации"
|
||
assert len(db.notifications) == 1, "факт попытки должен остаться в append-only логе"
|
||
assert db.notifications[0].token_valid is False
|
||
|
||
|
||
def test_amount_mismatch_is_rejected(client: TestClient, db: _FakeDb) -> None:
|
||
"""Подпись Т-Банка не гарантирует сумму (см. docstring notification.py) —
|
||
расхождение с суммой заказа обязано быть отказом, а не выдачей."""
|
||
response = client.post(
|
||
"/api/v1/trade-in/payments/notify", json=_signed_notification(Amount=_AMOUNT + 1)
|
||
)
|
||
|
||
assert response.status_code == 400
|
||
assert db.entitlements == []
|
||
|
||
|
||
def test_non_confirmed_status_does_not_fulfill(client: TestClient, db: _FakeDb) -> None:
|
||
response = client.post(
|
||
"/api/v1/trade-in/payments/notify", json=_signed_notification(Status="AUTHORIZED")
|
||
)
|
||
|
||
assert response.status_code == 200
|
||
assert db.entitlements == []
|
||
assert db.payments[0].status == "AUTHORIZED"
|
||
|
||
|
||
def test_pre_confirm_notification_does_not_downgrade_confirmed(
|
||
client: TestClient, db: _FakeDb
|
||
) -> None:
|
||
"""Отставший AUTHORIZED не имеет права откатить уже подтверждённый платёж."""
|
||
db.payments[0].status = "CONFIRMED"
|
||
|
||
client.post("/api/v1/trade-in/payments/notify", json=_signed_notification(Status="AUTHORIZED"))
|
||
|
||
assert db.payments[0].status == "CONFIRMED"
|
||
|
||
|
||
# ── capability-ссылка ────────────────────────────────────────────────────────
|
||
|
||
|
||
def _issue_entitlement(db: _FakeDb, token: str, *, expires_at: datetime | None) -> None:
|
||
db.entitlements.append(
|
||
SimpleNamespace(
|
||
id="ent-1",
|
||
payment_id=_PAYMENT_UUID,
|
||
subject=token,
|
||
kind="report_link",
|
||
ref_id=_ESTIMATE_UUID,
|
||
expires_at=expires_at,
|
||
)
|
||
)
|
||
|
||
|
||
def test_report_link_serves_estimate_for_valid_token(
|
||
client: TestClient, db: _FakeDb, monkeypatch: pytest.MonkeyPatch
|
||
) -> None:
|
||
from app.api.v1 import payments as payments_module
|
||
from app.schemas.trade_in import AggregatedEstimate
|
||
|
||
seen: dict[str, Any] = {}
|
||
|
||
def _fake_loader(_db: Any, estimate_id: Any, **kwargs: Any) -> AggregatedEstimate:
|
||
seen["estimate_id"] = str(estimate_id)
|
||
seen.update(kwargs)
|
||
return AggregatedEstimate(
|
||
estimate_id=_ESTIMATE_UUID,
|
||
median_price_rub=5_000_000,
|
||
range_low_rub=4_500_000,
|
||
range_high_rub=5_500_000,
|
||
median_price_per_m2=100_000,
|
||
confidence="medium",
|
||
n_analogs=7,
|
||
period_months=6,
|
||
analogs=[],
|
||
actual_deals=[],
|
||
expires_at=datetime.now(tz=UTC) + timedelta(hours=12),
|
||
)
|
||
|
||
monkeypatch.setattr(payments_module, "load_estimate", _fake_loader)
|
||
_issue_entitlement(db, "good-token", expires_at=datetime.now(tz=UTC) + timedelta(days=1))
|
||
|
||
response = client.get("/api/v1/trade-in/r/good-token")
|
||
|
||
assert response.status_code == 200, response.text
|
||
assert seen["estimate_id"] == _ESTIMATE_UUID
|
||
assert seen["capability_granted"] is True
|
||
assert seen["x_authenticated_user"] is None
|
||
|
||
|
||
def test_report_link_rejects_foreign_token(client: TestClient, db: _FakeDb) -> None:
|
||
_issue_entitlement(db, "good-token", expires_at=datetime.now(tz=UTC) + timedelta(days=1))
|
||
|
||
response = client.get("/api/v1/trade-in/r/someone-elses-token")
|
||
|
||
assert response.status_code == 404
|
||
|
||
|
||
def test_report_link_rejects_expired_token(client: TestClient, db: _FakeDb) -> None:
|
||
_issue_entitlement(db, "stale-token", expires_at=datetime.now(tz=UTC) - timedelta(seconds=1))
|
||
|
||
response = client.get("/api/v1/trade-in/r/stale-token")
|
||
|
||
assert response.status_code == 404
|
||
|
||
|
||
def test_issued_token_is_unpredictable(client: TestClient, db: _FakeDb) -> None:
|
||
"""Токен — это всё право доступа: он обязан быть случайным, а не производной
|
||
от order_id/payment_id (иначе выводится по данным, которые видит покупатель)."""
|
||
client.post("/api/v1/trade-in/payments/notify", json=_signed_notification())
|
||
|
||
token = db.entitlements[0].subject
|
||
assert len(token) >= 40
|
||
assert _ORDER_ID not in token
|
||
assert _PAYMENT_ID not in token
|
||
|
||
|
||
# ── kill-switch ──────────────────────────────────────────────────────────────
|
||
|
||
|
||
def test_endpoints_are_closed_when_payments_disabled(
|
||
db: _FakeDb, monkeypatch: pytest.MonkeyPatch
|
||
) -> None:
|
||
"""PAYMENTS_ENABLED=false — 503 на всех ручках, без падений и без записи в БД."""
|
||
from app.api.v1 import payments as payments_module
|
||
from app.core.config import settings
|
||
from app.core.db import get_db
|
||
|
||
monkeypatch.setattr(settings, "payments_enabled", False)
|
||
app = FastAPI()
|
||
app.include_router(payments_module.router, prefix="/api/v1/trade-in")
|
||
app.dependency_overrides[get_db] = lambda: db
|
||
disabled = TestClient(app)
|
||
|
||
assert disabled.get("/api/v1/trade-in/r/any-token").status_code == 503
|
||
assert disabled.get(f"/api/v1/trade-in/payments/status/{_ORDER_ID}").status_code == 503
|
||
assert disabled.post("/api/v1/trade-in/payments/notify", json={}).status_code == 503
|
||
assert (
|
||
disabled.post(
|
||
"/api/v1/trade-in/payments/checkout", json={"estimate_id": _ESTIMATE_UUID}
|
||
).status_code
|
||
== 503
|
||
)
|
||
assert db.notifications == []
|
||
assert db.entitlements == []
|
||
|
||
|
||
# ── периметр и синхронизация констант ────────────────────────────────────────
|
||
|
||
|
||
def test_notify_is_public_and_checkout_is_not() -> None:
|
||
from app.core.rbac import _PUBLIC_PATH_PREFIXES, _PUBLIC_PATHS
|
||
|
||
assert "/api/v1/trade-in/payments/notify" in _PUBLIC_PATHS
|
||
assert "/api/v1/trade-in/payments/checkout" not in _PUBLIC_PATHS
|
||
assert "/api/v1/trade-in/r/some-token".startswith(_PUBLIC_PATH_PREFIXES)
|
||
assert not "/api/v1/trade-in/history".startswith(_PUBLIC_PATH_PREFIXES)
|
||
|
||
|
||
def test_report_link_token_is_redacted_from_sentry_events() -> None:
|
||
from app.observability.sentry_scrub import scrub_pii_event
|
||
|
||
event = {
|
||
"request": {"url": "https://meraocenka.ru/api/v1/trade-in/r/s3cr3t-token-value"},
|
||
"transaction": "GET /api/v1/trade-in/r/s3cr3t-token-value",
|
||
}
|
||
scrubbed = scrub_pii_event(event, {}) # type: ignore[arg-type]
|
||
|
||
assert "s3cr3t-token-value" not in str(scrubbed)
|
||
assert "/api/v1/trade-in/r/[REDACTED]" in scrubbed["transaction"] # type: ignore[index]
|
||
|
||
|
||
def test_status_whitelist_matches_migration() -> None:
|
||
"""Свой список статусов не должен разъезжаться с CHECK миграции 233:
|
||
статус, которого нет в CHECK, уронит INSERT уже на проде."""
|
||
from app.api.v1.payments import _KNOWN_STATUSES
|
||
|
||
source = _MIGRATION.read_text(encoding="utf-8")
|
||
check = re.search(
|
||
r"ADD CONSTRAINT\s+payments_status_check\s+CHECK \(status IN \((.*?)\)\)", source, re.S
|
||
)
|
||
assert check is not None, "CHECK payments_status_check не найден в миграции 233"
|
||
migration_statuses = set(re.findall(r"'([^']+)'", check.group(1)))
|
||
|
||
assert _KNOWN_STATUSES == migration_statuses
|
||
|
||
|
||
def test_price_matches_published_offer() -> None:
|
||
"""Цена в коде обязана совпадать с ценой, названной в оферте (п. 4.1) —
|
||
единственный её источник для юр-текстов, mera-public/content.ts."""
|
||
from app.api.v1.payments import _PRODUCTS
|
||
|
||
match = re.search(r"SERVICE_PRICE_RUB\s*=\s*(\d+)", _CONTENT_TS.read_text(encoding="utf-8"))
|
||
assert match is not None, "SERVICE_PRICE_RUB не найден в mera-public/content.ts"
|
||
|
||
_name, price_kopecks = _PRODUCTS["paid_report"]
|
||
assert price_kopecks == int(match.group(1)) * 100
|
||
|
||
|
||
def test_live_status_predicate_matches_code() -> None:
|
||
"""Предикат частичного UNIQUE (279) и `_REUSABLE_STATUSES` — один список.
|
||
|
||
Индекс шире кода запрещает легитимную повторную попытку оплаты; индекс уже
|
||
кода пропускает второй холд на карте. И то и другое — про деньги, поэтому
|
||
дрейф ловит тест, а не внимательность читателя.
|
||
"""
|
||
from app.api.v1.payments import _REUSABLE_STATUSES
|
||
|
||
path = _REPO_ROOT / "backend" / "data" / "sql" / "279_payments_live_checkout_uidx.sql"
|
||
source = path.read_text(encoding="utf-8")
|
||
predicate = re.search(r"AND status IN \((.*?)\)", source, re.S)
|
||
assert predicate is not None, "предикат по статусам не найден в миграции 279"
|
||
|
||
assert set(re.findall(r"'([^']+)'", predicate.group(1))) == set(_REUSABLE_STATUSES)
|
||
|
||
|
||
# ── checkout: один живой платёж на оценку ────────────────────────────────────
|
||
|
||
|
||
@pytest.fixture()
|
||
def bank(monkeypatch: pytest.MonkeyPatch) -> list[dict[str, Any]]:
|
||
"""Подменяет Т-Банк: собирает вызовы Init и отдаёт предсказуемую ссылку.
|
||
|
||
Список вызовов — не украшение: «второй Init» и есть «второй холд на карте»,
|
||
поэтому проверяется именно его длина, а не только тело ответа.
|
||
"""
|
||
from app.api.v1 import payments as payments_module
|
||
|
||
calls: list[dict[str, Any]] = []
|
||
|
||
class _FakeClient:
|
||
async def init_payment(self, **kwargs: Any) -> dict[str, Any]:
|
||
calls.append(kwargs)
|
||
return {
|
||
"Success": True,
|
||
"Status": "NEW",
|
||
"PaymentId": f"300000000{len(calls)}",
|
||
"PaymentURL": f"https://securepayments.tinkoff.ru/{len(calls)}",
|
||
}
|
||
|
||
monkeypatch.setattr(payments_module, "_client", _FakeClient)
|
||
return calls
|
||
|
||
|
||
def _seed_live_payment(db: _FakeDb, *, payment_url: str | None, age: timedelta) -> SimpleNamespace:
|
||
row = SimpleNamespace(
|
||
id="pay-seed",
|
||
order_id="mera-seed",
|
||
status="NEW",
|
||
amount_kopecks=_AMOUNT,
|
||
estimate_id=_ESTIMATE_UUID,
|
||
created_by=None,
|
||
product_code="paid_report",
|
||
payment_url=payment_url,
|
||
created_at=datetime.now(tz=UTC) - age,
|
||
)
|
||
db.payments.append(row)
|
||
return row
|
||
|
||
|
||
def _paid_report_rows(db: _FakeDb) -> list[SimpleNamespace]:
|
||
return [p for p in db.payments if p.product_code == "paid_report"]
|
||
|
||
|
||
def test_parallel_checkout_does_not_create_second_payment(
|
||
client: TestClient, db: _FakeDb, bank: list[dict[str, Any]]
|
||
) -> None:
|
||
"""Двойной клик: соперник уже вставил строку, но ссылки от банка ещё нет.
|
||
|
||
Второй запрос обязан отбиться о частичный UNIQUE (миграция 279) и не пойти
|
||
в банк — второй Init это второй холд на карте покупателя. Ответ 409, а не
|
||
выдуманная ссылка.
|
||
|
||
Фальсификация (проверено): убрать `ON CONFLICT DO NOTHING` из INSERT в
|
||
payments → _FakeDb кидает _UniqueViolationError, как настоящий Postgres, и
|
||
тест краснеет вместо ответа 409; вернуть «SELECT, потом INSERT» без
|
||
обработки конфликта → вторая строка payments и второй вызов Init.
|
||
"""
|
||
_seed_live_payment(db, payment_url=None, age=timedelta(seconds=1))
|
||
|
||
response = client.post(
|
||
"/api/v1/trade-in/payments/checkout", json={"estimate_id": _ESTIMATE_UUID}
|
||
)
|
||
|
||
assert response.status_code == 409, response.text
|
||
assert len(_paid_report_rows(db)) == 1, "параллельный checkout создал второй платёж"
|
||
assert bank == [], "второй Init = второй холд на карте покупателя"
|
||
|
||
|
||
def test_repeat_checkout_reuses_live_link(
|
||
client: TestClient, db: _FakeDb, bank: list[dict[str, Any]]
|
||
) -> None:
|
||
"""Честный повтор по живому платежу возвращает ТУ ЖЕ ссылку и не зовёт Init."""
|
||
seeded = _seed_live_payment(
|
||
db, payment_url="https://securepayments/live", age=timedelta(minutes=5)
|
||
)
|
||
|
||
response = client.post(
|
||
"/api/v1/trade-in/payments/checkout", json={"estimate_id": _ESTIMATE_UUID}
|
||
)
|
||
|
||
assert response.status_code == 200, response.text
|
||
assert response.json()["payment_url"] == "https://securepayments/live"
|
||
assert response.json()["order_id"] == seeded.order_id
|
||
assert len(_paid_report_rows(db)) == 1
|
||
assert bank == []
|
||
|
||
|
||
def test_abandoned_checkout_does_not_lock_the_buyer_out(
|
||
client: TestClient, db: _FakeDb, bank: list[dict[str, Any]]
|
||
) -> None:
|
||
"""Брошенный NEW старше окна не переиспользуется: ссылка у банка протухла.
|
||
|
||
Без границы по времени покупатель навсегда получал бы одну и ту же мёртвую
|
||
ссылку и не мог начать оплату заново — нотификации по зависшему NEW может
|
||
не прийти вовсе, сам из этого статуса платёж не выйдет.
|
||
|
||
Фальсификация (проверено руками): убрать UPDATE ... DEADLINE_EXPIRED (или
|
||
условие по created_at в нём) → в ответе старая мёртвая ссылка
|
||
https://securepayments/dead, Init не вызывается: тест краснеет по значению,
|
||
а не по исключению.
|
||
"""
|
||
stale = _seed_live_payment(
|
||
db, payment_url="https://securepayments/dead", age=timedelta(hours=2)
|
||
)
|
||
|
||
response = client.post(
|
||
"/api/v1/trade-in/payments/checkout", json={"estimate_id": _ESTIMATE_UUID}
|
||
)
|
||
|
||
assert response.status_code == 200, response.text
|
||
assert response.json()["payment_url"] == "https://securepayments.tinkoff.ru/1"
|
||
assert stale.status == "DEADLINE_EXPIRED"
|
||
assert len(bank) == 1, "новая попытка оплаты обязана получить свежую ссылку банка"
|
||
assert len(_paid_report_rows(db)) == 2
|
||
|
||
|
||
def test_bank_refusal_on_init_does_not_lock_the_buyer_out(
|
||
client: TestClient, db: _FakeDb, monkeypatch: pytest.MonkeyPatch
|
||
) -> None:
|
||
"""Отказ банка на Init не запирает покупателя до истечения окна.
|
||
|
||
Строка после провального Init остаётся навсегда без `payment_url`:
|
||
`_find_live_payment` её не видит (там `payment_url IS NOT NULL`), а предикат
|
||
UNIQUE миграции 279 — видит. Если оставить ей статус NEW, следующий checkout
|
||
получит конфликт и ложный 409 «retry shortly» на все 30 минут
|
||
`_ABANDONED_AFTER_MINUTES` — из-за чужого сбоя, а не своего действия.
|
||
|
||
Фальсификация (проверено мутацией): убрать `SET status = :status` из UPDATE
|
||
в `_mark_init_failed` (params оставить как есть) → второй checkout отвечает
|
||
409 вместо 200, новой строки и второго Init нет. Тест краснеет по значению.
|
||
Мутант ловится потому, что `_FakeDb` применяет статус по тексту SQL, а не по
|
||
наличию ключа `status` в params.
|
||
"""
|
||
from app.api.v1 import payments as payments_module
|
||
from app.api.v1.payments import _REUSABLE_STATUSES
|
||
from app.services.payments.tbank_client import TBankApiError
|
||
|
||
calls: list[dict[str, Any]] = []
|
||
|
||
class _FlakyClient:
|
||
async def init_payment(self, **kwargs: Any) -> dict[str, Any]:
|
||
calls.append(kwargs)
|
||
if len(calls) == 1:
|
||
raise TBankApiError("Init", "9999", "терминал временно недоступен")
|
||
return {
|
||
"Success": True,
|
||
"Status": "NEW",
|
||
"PaymentId": "3000000042",
|
||
"PaymentURL": "https://securepayments.tinkoff.ru/after-retry",
|
||
}
|
||
|
||
monkeypatch.setattr(payments_module, "_client", _FlakyClient)
|
||
|
||
failed = client.post("/api/v1/trade-in/payments/checkout", json={"estimate_id": _ESTIMATE_UUID})
|
||
assert failed.status_code == 502, failed.text
|
||
|
||
dead = _paid_report_rows(db)[0]
|
||
assert dead.status not in _REUSABLE_STATUSES, "мёртвая строка осталась в предикате 279"
|
||
assert dead.payment_url is None
|
||
|
||
retry = client.post("/api/v1/trade-in/payments/checkout", json={"estimate_id": _ESTIMATE_UUID})
|
||
|
||
assert retry.status_code == 200, retry.text
|
||
assert retry.json()["payment_url"] == "https://securepayments.tinkoff.ru/after-retry"
|
||
assert len(_paid_report_rows(db)) == 2, "повтор после отказа банка не создал новую попытку"
|
||
assert len(calls) == 2
|
||
|
||
|
||
def test_broken_connection_to_bank_does_not_lock_the_buyer_out(
|
||
client: TestClient, db: _FakeDb, monkeypatch: pytest.MonkeyPatch
|
||
) -> None:
|
||
"""Обрыв соединения (`RemoteProtocolError`) — тот же замок, что и отказ банка.
|
||
|
||
`_request` ловил пару (TimeoutException, NetworkError), мимо которой летят
|
||
RemoteProtocolError (банк оборвал ответ), ProxyError и UnsupportedProtocol.
|
||
Такое исключение выходило наружу голым httpx-типом, `except TBankApiError` в
|
||
checkout его не видел, статус строки оставался NEW без `payment_url` — и
|
||
следующий checkout получал ложный 409 на все 30 минут.
|
||
|
||
Фальсификация (проверено мутацией): вернуть в `tbank_client._request`
|
||
`except (httpx.TimeoutException, httpx.NetworkError)` → первый checkout
|
||
отвечает 500 вместо 502, строка остаётся NEW, повтор отвечает 409. Красное
|
||
по значению.
|
||
"""
|
||
from app.api.v1 import payments as payments_module
|
||
from app.api.v1.payments import _REUSABLE_STATUSES
|
||
|
||
async def _no_sleep(_delay: float) -> None:
|
||
return None
|
||
|
||
async def _broken_post(*_args: Any, **_kwargs: Any) -> Any:
|
||
raise httpx.RemoteProtocolError("Server disconnected without sending a response")
|
||
|
||
# Ретраи `_request` спят 2+4+8 с — сон гасим, иначе тест стоит 14 секунд.
|
||
monkeypatch.setattr("app.services.payments.tbank_client.asyncio.sleep", _no_sleep)
|
||
monkeypatch.setattr(httpx.AsyncClient, "post", _broken_post)
|
||
|
||
failed = client.post("/api/v1/trade-in/payments/checkout", json={"estimate_id": _ESTIMATE_UUID})
|
||
assert failed.status_code == 502, failed.text
|
||
|
||
dead = _paid_report_rows(db)[0]
|
||
assert dead.status not in _REUSABLE_STATUSES, "оборванный Init оставил строку в предикате 279"
|
||
assert dead.status == "DEADLINE_EXPIRED"
|
||
assert dead.error_code == "network_error"
|
||
assert dead.payment_url is None
|
||
|
||
# Повтор идёт мимо httpx: проверяется, что пара (estimate_id, product_code)
|
||
# освободилась, а не то, как ведёт себя транспорт во второй раз.
|
||
class _WorkingClient:
|
||
async def init_payment(self, **_kwargs: Any) -> dict[str, Any]:
|
||
return {
|
||
"Success": True,
|
||
"Status": "NEW",
|
||
"PaymentId": "3000000043",
|
||
"PaymentURL": "https://securepayments.tinkoff.ru/after-disconnect",
|
||
}
|
||
|
||
monkeypatch.setattr(payments_module, "_client", _WorkingClient)
|
||
|
||
retry = client.post("/api/v1/trade-in/payments/checkout", json={"estimate_id": _ESTIMATE_UUID})
|
||
|
||
assert retry.status_code == 200, retry.text
|
||
assert retry.json()["payment_url"] == "https://securepayments.tinkoff.ru/after-disconnect"
|
||
assert len(_paid_report_rows(db)) == 2, "повтор после обрыва не создал новую попытку"
|
||
|
||
|
||
def test_init_without_payment_url_does_not_lock_the_buyer_out(
|
||
client: TestClient, db: _FakeDb, monkeypatch: pytest.MonkeyPatch
|
||
) -> None:
|
||
"""`Success:true` без `PaymentURL` — замок тот же, и здесь банк заказ ПРИНЯЛ.
|
||
|
||
Ветка отвечала 502, не тронув статус: строка навсегда оставалась NEW без
|
||
ссылки, то есть невидимой для `_find_live_payment` и видимой предикату
|
||
UNIQUE миграции 279. Помечаем её тем же терминальным статусом с говорящим
|
||
`error_code`.
|
||
|
||
Фальсификация (проверено мутацией): убрать `_mark_init_failed` из ветки
|
||
«нет PaymentURL» → повтор отвечает 409 вместо 200, второго Init нет.
|
||
"""
|
||
from app.api.v1 import payments as payments_module
|
||
from app.api.v1.payments import _REUSABLE_STATUSES
|
||
|
||
calls: list[dict[str, Any]] = []
|
||
|
||
class _UrllessClient:
|
||
async def init_payment(self, **kwargs: Any) -> dict[str, Any]:
|
||
calls.append(kwargs)
|
||
if len(calls) == 1:
|
||
# Контракт банка нарушен: Success есть, ссылки нет.
|
||
return {"Success": True, "Status": "NEW", "PaymentId": "3000000044"}
|
||
return {
|
||
"Success": True,
|
||
"Status": "NEW",
|
||
"PaymentId": "3000000045",
|
||
"PaymentURL": "https://securepayments.tinkoff.ru/after-urlless",
|
||
}
|
||
|
||
monkeypatch.setattr(payments_module, "_client", _UrllessClient)
|
||
|
||
failed = client.post("/api/v1/trade-in/payments/checkout", json={"estimate_id": _ESTIMATE_UUID})
|
||
assert failed.status_code == 502, failed.text
|
||
|
||
dead = _paid_report_rows(db)[0]
|
||
assert dead.status not in _REUSABLE_STATUSES, "строка без ссылки осталась в предикате 279"
|
||
assert dead.status == "DEADLINE_EXPIRED"
|
||
assert dead.error_code == "no_payment_url"
|
||
assert dead.payment_url is None
|
||
|
||
retry = client.post("/api/v1/trade-in/payments/checkout", json={"estimate_id": _ESTIMATE_UUID})
|
||
|
||
assert retry.status_code == 200, retry.text
|
||
assert retry.json()["payment_url"] == "https://securepayments.tinkoff.ru/after-urlless"
|
||
assert len(_paid_report_rows(db)) == 2, "повтор после пустого PaymentURL не создал попытку"
|
||
assert len(calls) == 2
|
||
|
||
|
||
def test_init_failed_status_is_outside_the_live_predicate() -> None:
|
||
"""`_INIT_FAILED_STATUS` обязан быть вне предиката 279 и внутри CHECK 233.
|
||
|
||
Сверка идёт с ТЕКСТОМ миграций, а не с константами того же модуля: статус,
|
||
попавший в предикат, вернул бы ложный 409, а статус вне CHECK уронил бы
|
||
UPDATE прямо в ветке обработки отказа банка.
|
||
"""
|
||
from app.api.v1.payments import _INIT_FAILED_STATUS
|
||
|
||
sql_dir = _REPO_ROOT / "backend" / "data" / "sql"
|
||
predicate = re.search(
|
||
r"AND status IN \((.*?)\)",
|
||
(sql_dir / "279_payments_live_checkout_uidx.sql").read_text(encoding="utf-8"),
|
||
re.S,
|
||
)
|
||
assert predicate is not None
|
||
assert _INIT_FAILED_STATUS not in set(re.findall(r"'([^']+)'", predicate.group(1)))
|
||
|
||
check = re.search(
|
||
r"payments_status_check\s+CHECK \(status IN \((.*?)\)\)",
|
||
(sql_dir / "233_payments.sql").read_text(encoding="utf-8"),
|
||
re.S,
|
||
)
|
||
assert check is not None, "CHECK статусов не найден в миграции 233"
|
||
assert _INIT_FAILED_STATUS in set(re.findall(r"'([^']+)'", check.group(1)))
|
||
|
||
|
||
def test_checkout_rejects_foreign_estimate(
|
||
client: TestClient, db: _FakeDb, bank: list[dict[str, Any]], monkeypatch: pytest.MonkeyPatch
|
||
) -> None:
|
||
"""IDOR: checkout по чужой оценке — 404, платёж не создаётся.
|
||
|
||
Дверь здесь не «показать чужой отчёт», а order_id: он же право доступа для
|
||
/payments/status/<order_id>, который отдаёт capability-ссылку на отчёт,
|
||
как только владелец заплатит.
|
||
|
||
Фальсификация (проверено руками): убрать вызов `_assert_estimate_access` в
|
||
checkout → 200 и строка в payments, тест краснеет по значению.
|
||
"""
|
||
from app.core import auth
|
||
|
||
monkeypatch.setattr(auth, "get_role", lambda username: "pilot")
|
||
db.estimate_created_by = "victim"
|
||
|
||
response = client.post(
|
||
"/api/v1/trade-in/payments/checkout",
|
||
json={"estimate_id": _ESTIMATE_UUID},
|
||
headers={"X-Authenticated-User": "attacker"},
|
||
)
|
||
|
||
assert response.status_code == 404, response.text
|
||
assert _paid_report_rows(db) == []
|
||
assert bank == []
|