gendesign/tradein-mvp/backend/tests/test_payments_router.py

"""Роутер платежей (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 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:
            for payment in self.payments:
                if payment.order_id == params["order_id"] and "status" in params:
                    payment.status = params["status"]
                    if "payment_url" in params:
                        payment.payment_url = params["payment_url"]
            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_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 == []