Покрытие кода тестами!
This commit is contained in:
gy9vin
@@ -17,6 +17,8 @@
|
||||
!app/**
|
||||
!locales/
|
||||
!locales/**
|
||||
!tests/
|
||||
!tests/**
|
||||
|
||||
# Дополнительно разрешаем README и лицензию (опционально)
|
||||
!README.md
|
||||
|
||||
@@ -0,0 +1,151 @@
|
||||
"""Глобальные фикстуры и настройки окружения для тестов."""
|
||||
|
||||
import os
|
||||
import sys
|
||||
import types
|
||||
from datetime import datetime, timezone
|
||||
|
||||
import pytest
|
||||
|
||||
# Подменяем параметры подключения к БД, чтобы SQLAlchemy не требовал aiosqlite.
|
||||
os.environ.setdefault("DATABASE_MODE", "postgresql")
|
||||
os.environ.setdefault("DATABASE_URL", "postgresql+asyncpg://user:pass@localhost/test_db")
|
||||
os.environ.setdefault("BOT_TOKEN", "test-token")
|
||||
|
||||
# Создаём заглушки для драйверов, которых может не быть в окружении тестов.
|
||||
sys.modules.setdefault("asyncpg", types.ModuleType("asyncpg"))
|
||||
sys.modules.setdefault("aiosqlite", types.ModuleType("aiosqlite"))
|
||||
|
||||
# Эмуляция redis.asyncio, чтобы модуль кеша мог импортироваться.
|
||||
if "redis.asyncio" not in sys.modules:
|
||||
redis_module = types.ModuleType("redis")
|
||||
redis_async_module = types.ModuleType("redis.asyncio")
|
||||
|
||||
class _FakeRedisClient:
|
||||
async def ping(self):
|
||||
"""Имитируем успешный ответ ping."""
|
||||
return True
|
||||
|
||||
async def close(self):
|
||||
"""Закрытие соединения ничего не делает."""
|
||||
|
||||
async def get(self, key): # noqa: ANN001
|
||||
return None
|
||||
|
||||
async def set(self, key, value, ex=None): # noqa: ANN001
|
||||
return True
|
||||
|
||||
async def delete(self, *keys): # noqa: ANN001
|
||||
return 0
|
||||
|
||||
async def keys(self, pattern="*"): # noqa: ANN001
|
||||
return []
|
||||
|
||||
async def exists(self, key): # noqa: ANN001
|
||||
return False
|
||||
|
||||
async def expire(self, key, seconds): # noqa: ANN001
|
||||
return True
|
||||
|
||||
async def incr(self, key): # noqa: ANN001
|
||||
return 1
|
||||
|
||||
def _from_url(url): # noqa: ANN001
|
||||
return _FakeRedisClient()
|
||||
|
||||
redis_async_module.from_url = _from_url
|
||||
redis_async_module.Redis = _FakeRedisClient
|
||||
sys.modules["redis"] = redis_module
|
||||
sys.modules["redis.asyncio"] = redis_async_module
|
||||
|
||||
# Минимальная реализация SDK YooKassa, чтобы импорт сервисов не падал.
|
||||
if "yookassa" not in sys.modules:
|
||||
fake_yookassa = types.ModuleType("yookassa")
|
||||
|
||||
class _FakeConfiguration:
|
||||
@staticmethod
|
||||
def configure(*args, **kwargs):
|
||||
"""Конфигурация заглушки ничего не делает."""
|
||||
|
||||
class _FakePayment:
|
||||
@staticmethod
|
||||
def create(*args, **kwargs):
|
||||
"""Возвращает объект с минимально необходимыми атрибутами."""
|
||||
|
||||
class _Response:
|
||||
id = "yk_fake"
|
||||
status = "pending"
|
||||
paid = False
|
||||
refundable = False
|
||||
metadata = {}
|
||||
amount = types.SimpleNamespace(value="0.00", currency="RUB")
|
||||
confirmation = types.SimpleNamespace(confirmation_url="https://example.com")
|
||||
created_at = datetime.utcnow()
|
||||
description = ""
|
||||
test = False
|
||||
|
||||
return _Response()
|
||||
|
||||
fake_yookassa.Configuration = _FakeConfiguration
|
||||
fake_yookassa.Payment = _FakePayment
|
||||
sys.modules["yookassa"] = fake_yookassa
|
||||
|
||||
# Подготавливаем вложенные пакеты, используемые сервисом.
|
||||
domain_module = types.ModuleType("yookassa.domain")
|
||||
request_module = types.ModuleType("yookassa.domain.request")
|
||||
payment_builder_module = types.ModuleType("yookassa.domain.request.payment_request_builder")
|
||||
common_module = types.ModuleType("yookassa.domain.common")
|
||||
confirmation_module = types.ModuleType("yookassa.domain.common.confirmation_type")
|
||||
|
||||
class _FakePaymentRequestBuilder:
|
||||
def __init__(self):
|
||||
self.data: dict = {}
|
||||
|
||||
def set_amount(self, value): # noqa: ANN001 - упрощённая заглушка
|
||||
self.data["amount"] = value
|
||||
return self
|
||||
|
||||
def set_capture(self, value): # noqa: ANN001
|
||||
self.data["capture"] = value
|
||||
return self
|
||||
|
||||
def set_confirmation(self, value): # noqa: ANN001
|
||||
self.data["confirmation"] = value
|
||||
return self
|
||||
|
||||
def set_description(self, value): # noqa: ANN001
|
||||
self.data["description"] = value
|
||||
return self
|
||||
|
||||
def set_metadata(self, value): # noqa: ANN001
|
||||
self.data["metadata"] = value
|
||||
return self
|
||||
|
||||
def set_receipt(self, value): # noqa: ANN001
|
||||
self.data["receipt"] = value
|
||||
return self
|
||||
|
||||
def set_payment_method_data(self, value): # noqa: ANN001
|
||||
self.data["payment_method_data"] = value
|
||||
return self
|
||||
|
||||
def build(self):
|
||||
return self.data
|
||||
|
||||
class _FakeConfirmationType:
|
||||
REDIRECT = "redirect"
|
||||
|
||||
payment_builder_module.PaymentRequestBuilder = _FakePaymentRequestBuilder
|
||||
confirmation_module.ConfirmationType = _FakeConfirmationType
|
||||
|
||||
sys.modules["yookassa.domain"] = domain_module
|
||||
sys.modules["yookassa.domain.request"] = request_module
|
||||
sys.modules["yookassa.domain.request.payment_request_builder"] = payment_builder_module
|
||||
sys.modules["yookassa.domain.common"] = common_module
|
||||
sys.modules["yookassa.domain.common.confirmation_type"] = confirmation_module
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def fixed_datetime() -> datetime:
|
||||
"""Возвращает фиксированную отметку времени для воспроизводимых проверок."""
|
||||
return datetime(2024, 1, 1, 12, 0, 0, tzinfo=timezone.utc)
|
||||
Vendored
+1
@@ -0,0 +1 @@
|
||||
# Пакет для тестов внешних клиентов и вебхуков.
|
||||
+85
@@ -0,0 +1,85 @@
|
||||
"""Тесты для внешнего клиента CryptoBotService."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from pathlib import Path
|
||||
from typing import Any, Dict, Optional
|
||||
import sys
|
||||
import hashlib
|
||||
import hmac
|
||||
|
||||
import pytest
|
||||
|
||||
ROOT_DIR = Path(__file__).resolve().parents[2]
|
||||
if str(ROOT_DIR) not in sys.path:
|
||||
sys.path.insert(0, str(ROOT_DIR))
|
||||
|
||||
from app.config import settings # noqa: E402
|
||||
from app.external.cryptobot import CryptoBotService # noqa: E402
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def anyio_backend() -> str:
|
||||
return "asyncio"
|
||||
|
||||
|
||||
def _enable_token(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
monkeypatch.setattr(settings, "CRYPTOBOT_API_TOKEN", "token", raising=False)
|
||||
monkeypatch.setattr(type(settings), "get_cryptobot_base_url", lambda self: "https://cryptobot.test", raising=False)
|
||||
monkeypatch.setattr(settings, "CRYPTOBOT_WEBHOOK_SECRET", "secret", raising=False)
|
||||
|
||||
|
||||
@pytest.mark.anyio("asyncio")
|
||||
async def test_create_invoice_uses_make_request(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
_enable_token(monkeypatch)
|
||||
service = CryptoBotService()
|
||||
|
||||
captured: Dict[str, Any] = {}
|
||||
|
||||
async def fake_make_request(method: str, endpoint: str, data: Optional[Dict[str, Any]] = None):
|
||||
captured["method"] = method
|
||||
captured["endpoint"] = endpoint
|
||||
captured["data"] = data
|
||||
return {"invoice_id": 1}
|
||||
|
||||
monkeypatch.setattr(service, "_make_request", fake_make_request, raising=False)
|
||||
|
||||
result = await service.create_invoice(
|
||||
amount="10.00",
|
||||
asset="USDT",
|
||||
description="Пополнение",
|
||||
payload="payload",
|
||||
expires_in=600,
|
||||
)
|
||||
|
||||
assert result == {"invoice_id": 1}
|
||||
assert captured["method"] == "POST"
|
||||
assert captured["endpoint"] == "createInvoice"
|
||||
assert captured["data"]["amount"] == "10.00"
|
||||
assert captured["data"]["payload"] == "payload"
|
||||
|
||||
|
||||
@pytest.mark.anyio("asyncio")
|
||||
async def test_make_request_returns_none_without_token(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
monkeypatch.setattr(settings, "CRYPTOBOT_API_TOKEN", "", raising=False)
|
||||
service = CryptoBotService()
|
||||
result = await service._make_request("GET", "getMe")
|
||||
assert result is None
|
||||
|
||||
|
||||
def test_verify_webhook_signature(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
monkeypatch.setattr(settings, "CRYPTOBOT_WEBHOOK_SECRET", "supersecret", raising=False)
|
||||
service = CryptoBotService()
|
||||
|
||||
body = '{"invoice_id":1}'
|
||||
secret_hash = hashlib.sha256(b"supersecret").digest()
|
||||
signature = hmac.new(secret_hash, body.encode(), hashlib.sha256).hexdigest()
|
||||
|
||||
assert service.verify_webhook_signature(body, signature) is True
|
||||
assert service.verify_webhook_signature(body, "invalid") is False
|
||||
|
||||
|
||||
def test_verify_webhook_signature_without_secret(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
monkeypatch.setattr(settings, "CRYPTOBOT_WEBHOOK_SECRET", "", raising=False)
|
||||
service = CryptoBotService()
|
||||
assert service.verify_webhook_signature("{}", "anything") is True
|
||||
+137
@@ -0,0 +1,137 @@
|
||||
"""Тестирование хендлеров WebhookServer без запуска реального сервера."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
from pathlib import Path
|
||||
from typing import Any, Tuple
|
||||
import sys
|
||||
from unittest.mock import AsyncMock
|
||||
|
||||
import pytest
|
||||
from aiohttp.test_utils import make_mocked_request
|
||||
from aiohttp import web
|
||||
|
||||
ROOT_DIR = Path(__file__).resolve().parents[2]
|
||||
if str(ROOT_DIR) not in sys.path:
|
||||
sys.path.insert(0, str(ROOT_DIR))
|
||||
|
||||
from app.config import settings # noqa: E402
|
||||
from app.external.webhook_server import WebhookServer # noqa: E402
|
||||
|
||||
|
||||
class DummyBot:
|
||||
async def send_message(self, *args: Any, **kwargs: Any) -> None: # pragma: no cover - уведомления не проверяем
|
||||
return None
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def anyio_backend() -> str:
|
||||
return "asyncio"
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def webhook_server(monkeypatch: pytest.MonkeyPatch) -> Tuple[WebhookServer, AsyncMock, AsyncMock]:
|
||||
monkeypatch.setattr(settings, "TRIBUTE_WEBHOOK_PATH", "/tribute", raising=False)
|
||||
monkeypatch.setattr(settings, "MULENPAY_WEBHOOK_PATH", "/mulen", raising=False)
|
||||
monkeypatch.setattr(settings, "CRYPTOBOT_WEBHOOK_PATH", "/cryptobot", raising=False)
|
||||
monkeypatch.setattr(settings, "MULENPAY_SECRET_KEY", "mulen-secret", raising=False)
|
||||
monkeypatch.setattr(settings, "CRYPTOBOT_WEBHOOK_SECRET", "", raising=False)
|
||||
monkeypatch.setattr(type(settings), "is_mulenpay_enabled", lambda self: True, raising=False)
|
||||
monkeypatch.setattr(type(settings), "is_cryptobot_enabled", lambda self: True, raising=False)
|
||||
|
||||
server = WebhookServer(DummyBot())
|
||||
|
||||
tribute_mock = AsyncMock()
|
||||
tribute_mock.process_webhook = AsyncMock(return_value={"status": "ok"})
|
||||
server.tribute_service = tribute_mock
|
||||
|
||||
payment_mock = AsyncMock()
|
||||
payment_mock.process_mulenpay_callback = AsyncMock(return_value=True)
|
||||
payment_mock.process_cryptobot_webhook = AsyncMock(return_value=True)
|
||||
monkeypatch.setattr("app.external.webhook_server.PaymentService", lambda *args, **kwargs: payment_mock)
|
||||
monkeypatch.setattr("app.services.payment_service.PaymentService", lambda *args, **kwargs: payment_mock)
|
||||
|
||||
server._verify_mulenpay_signature = lambda request, raw: True # type: ignore[attr-defined]
|
||||
|
||||
class DummyDB:
|
||||
async def commit(self) -> None: # pragma: no cover - не проверяем транзакции
|
||||
return None
|
||||
|
||||
async def fake_get_db():
|
||||
yield DummyDB()
|
||||
|
||||
monkeypatch.setattr("app.external.webhook_server.get_db", fake_get_db)
|
||||
|
||||
class DummySessionManager:
|
||||
def __init__(self) -> None:
|
||||
self.session = DummyDB()
|
||||
|
||||
async def __aenter__(self) -> DummyDB:
|
||||
return self.session
|
||||
|
||||
async def __aexit__(self, exc_type, exc, tb) -> None:
|
||||
return None
|
||||
|
||||
monkeypatch.setattr("app.database.database.AsyncSessionLocal", lambda: DummySessionManager())
|
||||
|
||||
return server, tribute_mock, payment_mock
|
||||
|
||||
|
||||
def _mock_request(method: str, path: str, body: dict[str, Any], headers: dict[str, str] | None = None) -> AsyncMock:
|
||||
request = AsyncMock(spec=web.Request)
|
||||
request.method = method
|
||||
request.path = path
|
||||
request.headers = headers or {}
|
||||
request.read.return_value = json.dumps(body).encode("utf-8")
|
||||
return request
|
||||
|
||||
|
||||
@pytest.mark.anyio("asyncio")
|
||||
async def test_health_endpoint(webhook_server: Tuple[WebhookServer, AsyncMock, AsyncMock]) -> None:
|
||||
server, _, _ = webhook_server
|
||||
request = make_mocked_request("GET", "/health")
|
||||
response = await server._health_check(request)
|
||||
assert response.status == 200
|
||||
data = json.loads(response.text)
|
||||
assert data["status"] == "ok"
|
||||
assert data["service"] == "payment-webhooks"
|
||||
|
||||
|
||||
@pytest.mark.anyio("asyncio")
|
||||
async def test_tribute_webhook_success(monkeypatch: pytest.MonkeyPatch, webhook_server: Tuple[WebhookServer, AsyncMock, AsyncMock]) -> None:
|
||||
server, tribute_mock, _ = webhook_server
|
||||
monkeypatch.setattr(settings, "TRIBUTE_API_KEY", "key", raising=False)
|
||||
|
||||
class FakeTributeAPI:
|
||||
def verify_webhook_signature(self, payload: str, signature: str) -> bool:
|
||||
return True
|
||||
|
||||
monkeypatch.setattr("app.external.tribute.TributeService", FakeTributeAPI)
|
||||
|
||||
request = _mock_request("POST", "/tribute", {"event_type": "payment", "status": "paid"}, headers={"trbt-signature": "sig"})
|
||||
response = await server._tribute_webhook_handler(request)
|
||||
assert response.status == 200
|
||||
assert tribute_mock.process_webhook.await_count == 1
|
||||
|
||||
|
||||
@pytest.mark.anyio("asyncio")
|
||||
async def test_mulenpay_webhook_success(webhook_server: Tuple[WebhookServer, AsyncMock, AsyncMock]) -> None:
|
||||
server, _, payment_mock = webhook_server
|
||||
request = _mock_request("POST", "/mulen", {"uuid": "uuid", "payment_status": "success"})
|
||||
response = await server._mulenpay_webhook_handler(request)
|
||||
assert response.status == 200
|
||||
payment_mock.process_mulenpay_callback.assert_awaited_once()
|
||||
|
||||
|
||||
@pytest.mark.anyio("asyncio")
|
||||
async def test_cryptobot_webhook_success(webhook_server: Tuple[WebhookServer, AsyncMock, AsyncMock]) -> None:
|
||||
server, _, payment_mock = webhook_server
|
||||
request = _mock_request(
|
||||
"POST",
|
||||
"/cryptobot",
|
||||
{"update_type": "invoice_paid", "payload": {"invoice_id": 1}},
|
||||
)
|
||||
response = await server._cryptobot_webhook_handler(request)
|
||||
assert response.status == 200
|
||||
payment_mock.process_cryptobot_webhook.assert_awaited_once()
|
||||
@@ -0,0 +1 @@
|
||||
# Пакет для тестов сервисов бота.
|
||||
@@ -0,0 +1,107 @@
|
||||
"""Юнит-тесты MulenPayService."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from pathlib import Path
|
||||
from typing import Any, Dict, Optional
|
||||
import sys
|
||||
|
||||
import pytest
|
||||
|
||||
ROOT_DIR = Path(__file__).resolve().parents[2]
|
||||
if str(ROOT_DIR) not in sys.path:
|
||||
sys.path.insert(0, str(ROOT_DIR))
|
||||
|
||||
from app.config import settings # noqa: E402
|
||||
from app.services.mulenpay_service import MulenPayService # noqa: E402
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def anyio_backend() -> str:
|
||||
return "asyncio"
|
||||
|
||||
|
||||
def _enable_service(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
monkeypatch.setattr(type(settings), "is_mulenpay_enabled", lambda self: True, raising=False)
|
||||
monkeypatch.setattr(settings, "MULENPAY_API_KEY", "api", raising=False)
|
||||
monkeypatch.setattr(settings, "MULENPAY_SHOP_ID", "shop", raising=False)
|
||||
monkeypatch.setattr(settings, "MULENPAY_SECRET_KEY", "secret", raising=False)
|
||||
monkeypatch.setattr(settings, "MULENPAY_BASE_URL", "https://mulenpay.test", raising=False)
|
||||
|
||||
|
||||
def test_is_configured(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
service = MulenPayService()
|
||||
assert service.is_configured is False
|
||||
|
||||
_enable_service(monkeypatch)
|
||||
service = MulenPayService()
|
||||
assert service.is_configured is True
|
||||
|
||||
|
||||
def test_format_and_signature(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
_enable_service(monkeypatch)
|
||||
service = MulenPayService()
|
||||
assert service._format_amount(12345) == "123.45"
|
||||
signature = service._build_signature("rub", "100.00")
|
||||
assert isinstance(signature, str) and len(signature) == 40
|
||||
|
||||
|
||||
@pytest.mark.anyio("asyncio")
|
||||
async def test_create_payment_success(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
_enable_service(monkeypatch)
|
||||
|
||||
captured_payload: Dict[str, Any] = {}
|
||||
|
||||
async def fake_request(method: str, endpoint: str, **kwargs: Any) -> Dict[str, Any]:
|
||||
captured_payload.update({"method": method, "endpoint": endpoint, **kwargs})
|
||||
return {"success": True, "id": 101, "paymentUrl": "https://mulenpay/pay"}
|
||||
|
||||
service = MulenPayService()
|
||||
monkeypatch.setattr(service, "_request", fake_request, raising=False)
|
||||
|
||||
result = await service.create_payment(
|
||||
amount_kopeks=25000,
|
||||
description="Пополнение",
|
||||
uuid="uuid-1",
|
||||
items=[{"description": "item", "quantity": 1, "price": 250.0}],
|
||||
language="ru",
|
||||
website_url="https://example.com",
|
||||
)
|
||||
|
||||
assert result is not None
|
||||
assert result["id"] == 101
|
||||
assert captured_payload["method"] == "POST"
|
||||
assert captured_payload["endpoint"] == "/v2/payments"
|
||||
assert captured_payload["json_data"]["language"] == "ru"
|
||||
|
||||
|
||||
@pytest.mark.anyio("asyncio")
|
||||
async def test_create_payment_failure(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
_enable_service(monkeypatch)
|
||||
service = MulenPayService()
|
||||
|
||||
async def fake_request(*args: Any, **kwargs: Any) -> Optional[Dict[str, Any]]:
|
||||
return None
|
||||
|
||||
monkeypatch.setattr(service, "_request", fake_request, raising=False)
|
||||
|
||||
result = await service.create_payment(
|
||||
amount_kopeks=1000,
|
||||
description="desc",
|
||||
uuid="uuid",
|
||||
items=[],
|
||||
)
|
||||
assert result is None
|
||||
|
||||
|
||||
@pytest.mark.anyio("asyncio")
|
||||
async def test_get_payment(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
_enable_service(monkeypatch)
|
||||
service = MulenPayService()
|
||||
|
||||
async def fake_request(method: str, endpoint: str, **kwargs: Any) -> Dict[str, Any]:
|
||||
return {"id": 123, "status": "paid"}
|
||||
|
||||
monkeypatch.setattr(service, "_request", fake_request, raising=False)
|
||||
result = await service.get_payment(123)
|
||||
assert result == {"id": 123, "status": "paid"}
|
||||
@@ -0,0 +1,120 @@
|
||||
"""Тесты Pal24Service и вспомогательных функций."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from datetime import datetime, timedelta
|
||||
from decimal import Decimal
|
||||
from pathlib import Path
|
||||
from typing import Any, Dict, Optional
|
||||
import sys
|
||||
|
||||
import pytest
|
||||
|
||||
ROOT_DIR = Path(__file__).resolve().parents[2]
|
||||
if str(ROOT_DIR) not in sys.path:
|
||||
sys.path.insert(0, str(ROOT_DIR))
|
||||
|
||||
from app.config import settings # noqa: E402
|
||||
from app.external.pal24_client import Pal24Client, Pal24APIError # noqa: E402
|
||||
from app.services.pal24_service import Pal24Service # noqa: E402
|
||||
|
||||
|
||||
class StubPal24Client:
|
||||
def __init__(self, configured: bool = True, response: Optional[Dict[str, Any]] = None) -> None:
|
||||
self.is_configured = configured
|
||||
self.response = response or {
|
||||
"success": True,
|
||||
"bill_id": "BILL42",
|
||||
"status": "NEW",
|
||||
"transfer_url": "https://pal24/sbp",
|
||||
"link_url": "https://pal24/card",
|
||||
"currency": "RUB",
|
||||
}
|
||||
self.calls: list[Dict[str, Any]] = []
|
||||
|
||||
async def create_bill(self, **kwargs: Any) -> Dict[str, Any]:
|
||||
self.calls.append(kwargs)
|
||||
return self.response
|
||||
|
||||
async def get_bill_status(self, bill_id: str) -> Dict[str, Any]:
|
||||
return {"id": bill_id, "status": "NEW"}
|
||||
|
||||
async def get_payment_status(self, payment_id: str) -> Dict[str, Any]:
|
||||
return {"id": payment_id, "status": "SUCCESS"}
|
||||
|
||||
|
||||
def _enable_pal24(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
monkeypatch.setattr(type(settings), "is_pal24_enabled", lambda self: True, raising=False)
|
||||
monkeypatch.setattr(settings, "PAL24_SHOP_ID", "shop42", raising=False)
|
||||
monkeypatch.setattr(settings, "PAL24_SIGNATURE_TOKEN", "sigsecret", raising=False)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def anyio_backend() -> str:
|
||||
return "asyncio"
|
||||
|
||||
|
||||
@pytest.mark.anyio("asyncio")
|
||||
async def test_create_bill_success(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
_enable_pal24(monkeypatch)
|
||||
client = StubPal24Client()
|
||||
service = Pal24Service(client)
|
||||
|
||||
monkeypatch.setattr(Pal24Client, "normalize_amount", staticmethod(lambda amount: Decimal("500.00")), raising=False)
|
||||
|
||||
result = await service.create_bill(
|
||||
amount_kopeks=50000,
|
||||
user_id=7,
|
||||
order_id="order-7",
|
||||
description="Пополнение",
|
||||
ttl_seconds=600,
|
||||
custom_payload={"extra": "value"},
|
||||
payer_email="user@example.com",
|
||||
payment_method="CARD",
|
||||
)
|
||||
|
||||
assert result["bill_id"] == "BILL42"
|
||||
assert client.calls and client.calls[0]["amount"] == Decimal("500.00")
|
||||
assert client.calls[0]["shop_id"] == "shop42"
|
||||
assert client.calls[0]["description"] == "Пополнение"
|
||||
|
||||
|
||||
@pytest.mark.anyio("asyncio")
|
||||
async def test_create_bill_requires_configuration(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
_enable_pal24(monkeypatch)
|
||||
client = StubPal24Client(configured=False)
|
||||
service = Pal24Service(client)
|
||||
|
||||
with pytest.raises(Pal24APIError):
|
||||
await service.create_bill(
|
||||
amount_kopeks=1000,
|
||||
user_id=1,
|
||||
order_id="order",
|
||||
description="desc",
|
||||
)
|
||||
|
||||
|
||||
def test_parse_postback_success(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
_enable_pal24(monkeypatch)
|
||||
sig = Pal24Client.calculate_signature("100.00", "INV1", api_token="sigsecret")
|
||||
payload = {
|
||||
"InvId": "INV1",
|
||||
"OutSum": "100.00",
|
||||
"Status": "SUCCESS",
|
||||
"SignatureValue": sig,
|
||||
}
|
||||
result = Pal24Service.parse_postback(payload)
|
||||
assert result["InvId"] == "INV1"
|
||||
|
||||
|
||||
def test_parse_postback_missing_fields(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
_enable_pal24(monkeypatch)
|
||||
with pytest.raises(Pal24APIError):
|
||||
Pal24Service.parse_postback({"InvId": "1"})
|
||||
|
||||
|
||||
def test_convert_to_kopeks_and_expiration() -> None:
|
||||
assert Pal24Service.convert_to_kopeks("10.50") == 1050
|
||||
expiration = Pal24Service.get_expiration(60)
|
||||
assert isinstance(expiration, datetime)
|
||||
assert expiration - datetime.utcnow() <= timedelta(seconds=61)
|
||||
@@ -0,0 +1,152 @@
|
||||
"""Тесты сценариев CryptoBot в PaymentService."""
|
||||
|
||||
from pathlib import Path
|
||||
from typing import Any, Dict, Optional
|
||||
import sys
|
||||
from datetime import datetime
|
||||
|
||||
import pytest
|
||||
|
||||
ROOT_DIR = Path(__file__).resolve().parents[2]
|
||||
if str(ROOT_DIR) not in sys.path:
|
||||
sys.path.insert(0, str(ROOT_DIR))
|
||||
|
||||
from app.config import settings # noqa: E402
|
||||
from app.database.crud import cryptobot as cryptobot_crud # noqa: E402
|
||||
from app.services.payment_service import PaymentService # noqa: E402
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def anyio_backend() -> str:
|
||||
return "asyncio"
|
||||
|
||||
|
||||
class DummySession:
|
||||
def __init__(self) -> None:
|
||||
self.added_objects: list[Any] = []
|
||||
|
||||
async def commit(self) -> None: # pragma: no cover
|
||||
return None
|
||||
|
||||
def add(self, obj: Any) -> None: # pragma: no cover
|
||||
self.added_objects.append(obj)
|
||||
|
||||
async def flush(self) -> None: # pragma: no cover
|
||||
return None
|
||||
|
||||
|
||||
class DummyLocalPayment:
|
||||
def __init__(self, payment_id: int = 888) -> None:
|
||||
self.id = payment_id
|
||||
self.created_at = datetime(2024, 3, 1, 9, 0, 0)
|
||||
|
||||
|
||||
class StubCryptoBotService:
|
||||
def __init__(self, response: Optional[Dict[str, Any]]) -> None:
|
||||
self.response = response
|
||||
self.calls: list[Dict[str, Any]] = []
|
||||
|
||||
async def create_invoice(self, **kwargs: Any) -> Optional[Dict[str, Any]]:
|
||||
self.calls.append(kwargs)
|
||||
return self.response
|
||||
|
||||
|
||||
def _make_service(stub: Optional[StubCryptoBotService]) -> PaymentService:
|
||||
service = PaymentService.__new__(PaymentService) # type: ignore[call-arg]
|
||||
service.bot = None
|
||||
service.cryptobot_service = stub
|
||||
service.mulenpay_service = None
|
||||
service.pal24_service = None
|
||||
service.yookassa_service = None
|
||||
service.stars_service = None
|
||||
return service
|
||||
|
||||
|
||||
@pytest.mark.anyio("asyncio")
|
||||
async def test_create_cryptobot_payment_success(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
response = {
|
||||
"invoice_id": 12345,
|
||||
"bot_invoice_url": "https://t.me/invoice",
|
||||
"mini_app_invoice_url": "https://mini.app/invoice",
|
||||
"web_app_invoice_url": "https://web.app/invoice",
|
||||
}
|
||||
stub = StubCryptoBotService(response)
|
||||
service = _make_service(stub)
|
||||
db = DummySession()
|
||||
|
||||
captured_args: Dict[str, Any] = {}
|
||||
|
||||
async def fake_create_cryptobot_payment(**kwargs: Any) -> DummyLocalPayment:
|
||||
captured_args.update(kwargs)
|
||||
return DummyLocalPayment(payment_id=555)
|
||||
|
||||
monkeypatch.setattr(
|
||||
cryptobot_crud,
|
||||
"create_cryptobot_payment",
|
||||
fake_create_cryptobot_payment,
|
||||
raising=False,
|
||||
)
|
||||
monkeypatch.setattr(
|
||||
type(settings),
|
||||
"get_cryptobot_invoice_expires_seconds",
|
||||
lambda self: 600,
|
||||
raising=False,
|
||||
)
|
||||
|
||||
result = await service.create_cryptobot_payment(
|
||||
db=db,
|
||||
user_id=9,
|
||||
amount_usd=12.5,
|
||||
asset="USDT",
|
||||
description="Пополнение",
|
||||
payload="custom",
|
||||
)
|
||||
|
||||
assert result is not None
|
||||
assert result["local_payment_id"] == 555
|
||||
assert result["invoice_id"] == "12345"
|
||||
assert result["bot_invoice_url"] == "https://t.me/invoice"
|
||||
assert stub.calls and stub.calls[0]["expires_in"] == 600
|
||||
assert captured_args["invoice_id"] == "12345"
|
||||
assert captured_args["amount"] == "12.50"
|
||||
|
||||
|
||||
@pytest.mark.anyio("asyncio")
|
||||
async def test_create_cryptobot_payment_returns_none_when_service_missing() -> None:
|
||||
service = _make_service(None)
|
||||
db = DummySession()
|
||||
result = await service.create_cryptobot_payment(
|
||||
db=db,
|
||||
user_id=1,
|
||||
amount_usd=10,
|
||||
)
|
||||
assert result is None
|
||||
|
||||
|
||||
@pytest.mark.anyio("asyncio")
|
||||
async def test_create_cryptobot_payment_handles_empty_response(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
stub = StubCryptoBotService(response=None)
|
||||
service = _make_service(stub)
|
||||
db = DummySession()
|
||||
|
||||
called = False
|
||||
|
||||
async def fake_create_cryptobot_payment(**kwargs: Any) -> DummyLocalPayment:
|
||||
nonlocal called
|
||||
called = True
|
||||
return DummyLocalPayment()
|
||||
|
||||
monkeypatch.setattr(
|
||||
cryptobot_crud,
|
||||
"create_cryptobot_payment",
|
||||
fake_create_cryptobot_payment,
|
||||
raising=False,
|
||||
)
|
||||
|
||||
result = await service.create_cryptobot_payment(
|
||||
db=db,
|
||||
user_id=1,
|
||||
amount_usd=5,
|
||||
)
|
||||
assert result is None
|
||||
assert called is False
|
||||
@@ -0,0 +1,140 @@
|
||||
"""Тесты для сценариев MulenPay в PaymentService."""
|
||||
|
||||
from pathlib import Path
|
||||
from typing import Any, Dict, Optional
|
||||
import sys
|
||||
from datetime import datetime
|
||||
|
||||
import pytest
|
||||
|
||||
ROOT_DIR = Path(__file__).resolve().parents[2]
|
||||
if str(ROOT_DIR) not in sys.path:
|
||||
sys.path.insert(0, str(ROOT_DIR))
|
||||
|
||||
import app.services.payment_service as payment_service_module # noqa: E402
|
||||
from app.config import settings # noqa: E402
|
||||
from app.services.payment_service import PaymentService # noqa: E402
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def anyio_backend() -> str:
|
||||
return "asyncio"
|
||||
|
||||
|
||||
class DummySession:
|
||||
async def commit(self) -> None: # pragma: no cover - метод вызывается, но без логики
|
||||
return None
|
||||
|
||||
|
||||
class DummyLocalPayment:
|
||||
def __init__(self, payment_id: int = 501) -> None:
|
||||
self.id = payment_id
|
||||
self.created_at = datetime(2024, 1, 1, 12, 0, 0)
|
||||
|
||||
|
||||
class StubMulenPayService:
|
||||
def __init__(self, response: Optional[Dict[str, Any]]) -> None:
|
||||
self.response = response
|
||||
self.calls: list[Dict[str, Any]] = []
|
||||
|
||||
async def create_payment(self, **kwargs: Any) -> Optional[Dict[str, Any]]:
|
||||
self.calls.append(kwargs)
|
||||
return self.response
|
||||
|
||||
|
||||
def _make_service(stub: Optional[StubMulenPayService]) -> PaymentService:
|
||||
service = PaymentService.__new__(PaymentService) # type: ignore[call-arg]
|
||||
service.bot = None
|
||||
service.mulenpay_service = stub
|
||||
service.pal24_service = None
|
||||
service.yookassa_service = None
|
||||
service.stars_service = None
|
||||
service.cryptobot_service = None
|
||||
return service
|
||||
|
||||
|
||||
@pytest.mark.anyio("asyncio")
|
||||
async def test_create_mulenpay_payment_success(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
response = {"id": 123, "paymentUrl": "https://mulenpay/pay"}
|
||||
stub = StubMulenPayService(response)
|
||||
service = _make_service(stub)
|
||||
db = DummySession()
|
||||
|
||||
captured_args: Dict[str, Any] = {}
|
||||
|
||||
async def fake_create_mulenpay_payment(**kwargs: Any) -> DummyLocalPayment:
|
||||
captured_args.update(kwargs)
|
||||
return DummyLocalPayment(payment_id=999)
|
||||
|
||||
monkeypatch.setattr(
|
||||
payment_service_module,
|
||||
"create_mulenpay_payment",
|
||||
fake_create_mulenpay_payment,
|
||||
raising=False,
|
||||
)
|
||||
monkeypatch.setattr(settings, "MULENPAY_MIN_AMOUNT_KOPEKS", 1000, raising=False)
|
||||
monkeypatch.setattr(settings, "MULENPAY_MAX_AMOUNT_KOPEKS", 1_000_000, raising=False)
|
||||
monkeypatch.setattr(settings, "MULENPAY_VAT_CODE", 1, raising=False)
|
||||
monkeypatch.setattr(settings, "MULENPAY_PAYMENT_SUBJECT", "service", raising=False)
|
||||
monkeypatch.setattr(settings, "MULENPAY_PAYMENT_MODE", "full_payment", raising=False)
|
||||
monkeypatch.setattr(settings, "MULENPAY_LANGUAGE", "ru", raising=False)
|
||||
monkeypatch.setattr(settings, "WEBHOOK_URL", "https://example.com", raising=False)
|
||||
|
||||
result = await service.create_mulenpay_payment(
|
||||
db=db,
|
||||
user_id=77,
|
||||
amount_kopeks=25000,
|
||||
description="Пополнение",
|
||||
language="en",
|
||||
)
|
||||
|
||||
assert result is not None
|
||||
assert result["local_payment_id"] == 999
|
||||
assert result["mulen_payment_id"] == 123
|
||||
assert result["payment_url"] == "https://mulenpay/pay"
|
||||
assert result["status"] == "created"
|
||||
assert stub.calls and stub.calls[0]["language"] == "en"
|
||||
assert captured_args["user_id"] == 77
|
||||
assert captured_args["amount_kopeks"] == 25000
|
||||
assert captured_args["uuid"].startswith("mulen_77_")
|
||||
|
||||
|
||||
@pytest.mark.anyio("asyncio")
|
||||
async def test_create_mulenpay_payment_respects_amount_limits(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
stub = StubMulenPayService({"id": 1})
|
||||
service = _make_service(stub)
|
||||
db = DummySession()
|
||||
|
||||
monkeypatch.setattr(settings, "MULENPAY_MIN_AMOUNT_KOPEKS", 5000, raising=False)
|
||||
monkeypatch.setattr(settings, "MULENPAY_MAX_AMOUNT_KOPEKS", 10_000, raising=False)
|
||||
|
||||
result_low = await service.create_mulenpay_payment(
|
||||
db=db,
|
||||
user_id=1,
|
||||
amount_kopeks=1000,
|
||||
description="Пополнение",
|
||||
)
|
||||
assert result_low is None
|
||||
|
||||
result_high = await service.create_mulenpay_payment(
|
||||
db=db,
|
||||
user_id=1,
|
||||
amount_kopeks=20_000,
|
||||
description="Пополнение",
|
||||
)
|
||||
assert result_high is None
|
||||
assert not stub.calls
|
||||
|
||||
|
||||
@pytest.mark.anyio("asyncio")
|
||||
async def test_create_mulenpay_payment_returns_none_without_service() -> None:
|
||||
service = _make_service(None)
|
||||
db = DummySession()
|
||||
|
||||
result = await service.create_mulenpay_payment(
|
||||
db=db,
|
||||
user_id=1,
|
||||
amount_kopeks=5000,
|
||||
description="Пополнение",
|
||||
)
|
||||
assert result is None
|
||||
@@ -0,0 +1,166 @@
|
||||
"""Тесты Pal24 сценариев PaymentService."""
|
||||
|
||||
from pathlib import Path
|
||||
from typing import Any, Dict, Optional
|
||||
import sys
|
||||
from datetime import datetime
|
||||
|
||||
import pytest
|
||||
|
||||
ROOT_DIR = Path(__file__).resolve().parents[2]
|
||||
if str(ROOT_DIR) not in sys.path:
|
||||
sys.path.insert(0, str(ROOT_DIR))
|
||||
|
||||
import app.services.payment_service as payment_service_module # noqa: E402
|
||||
from app.config import settings # noqa: E402
|
||||
from app.services.payment_service import PaymentService # noqa: E402
|
||||
from app.services.pal24_service import Pal24APIError # noqa: E402
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def anyio_backend() -> str:
|
||||
return "asyncio"
|
||||
|
||||
|
||||
class DummySession:
|
||||
async def commit(self) -> None: # pragma: no cover
|
||||
return None
|
||||
|
||||
|
||||
class DummyLocalPayment:
|
||||
def __init__(self, payment_id: int = 404) -> None:
|
||||
self.id = payment_id
|
||||
self.created_at = datetime(2024, 1, 2, 10, 0, 0)
|
||||
|
||||
|
||||
class StubPal24Service:
|
||||
def __init__(self, *, configured: bool = True, response: Optional[Dict[str, Any]] = None) -> None:
|
||||
self.is_configured = configured
|
||||
self.response = response or {
|
||||
"success": True,
|
||||
"bill_id": "BILL-1",
|
||||
"transfer_url": "https://pal24/sbp",
|
||||
"link_url": "https://pal24/card",
|
||||
"status": "NEW",
|
||||
}
|
||||
self.calls: list[Dict[str, Any]] = []
|
||||
self.raise_error: Optional[Exception] = None
|
||||
|
||||
async def create_bill(self, **kwargs: Any) -> Dict[str, Any]:
|
||||
self.calls.append(kwargs)
|
||||
if self.raise_error:
|
||||
raise self.raise_error
|
||||
return self.response
|
||||
|
||||
|
||||
def _make_service(stub: Optional[StubPal24Service]) -> PaymentService:
|
||||
service = PaymentService.__new__(PaymentService) # type: ignore[call-arg]
|
||||
service.bot = None
|
||||
service.pal24_service = stub
|
||||
service.mulenpay_service = None
|
||||
service.yookassa_service = None
|
||||
service.cryptobot_service = None
|
||||
service.stars_service = None
|
||||
return service
|
||||
|
||||
|
||||
@pytest.mark.anyio("asyncio")
|
||||
async def test_create_pal24_payment_success(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
stub = StubPal24Service()
|
||||
service = _make_service(stub)
|
||||
db = DummySession()
|
||||
|
||||
captured_args: Dict[str, Any] = {}
|
||||
|
||||
async def fake_create_pal24_payment(*args: Any, **kwargs: Any) -> DummyLocalPayment:
|
||||
captured_args.update(kwargs)
|
||||
if args:
|
||||
captured_args["db_arg"] = args[0]
|
||||
return DummyLocalPayment(payment_id=321)
|
||||
|
||||
monkeypatch.setattr(
|
||||
payment_service_module,
|
||||
"create_pal24_payment",
|
||||
fake_create_pal24_payment,
|
||||
raising=False,
|
||||
)
|
||||
monkeypatch.setattr(settings, "PAL24_MIN_AMOUNT_KOPEKS", 1000, raising=False)
|
||||
monkeypatch.setattr(settings, "PAL24_MAX_AMOUNT_KOPEKS", 1_000_000, raising=False)
|
||||
|
||||
result = await service.create_pal24_payment(
|
||||
db=db,
|
||||
user_id=15,
|
||||
amount_kopeks=50000,
|
||||
description="Оплата подписки",
|
||||
language="ru",
|
||||
ttl_seconds=600,
|
||||
payer_email="user@example.com",
|
||||
payment_method="card",
|
||||
)
|
||||
|
||||
assert result is not None
|
||||
assert result["local_payment_id"] == 321
|
||||
assert result["bill_id"] == "BILL-1"
|
||||
assert result["payment_method"] == "CARD"
|
||||
assert result["link_url"] == "https://pal24/sbp"
|
||||
assert result["card_url"] == "https://pal24/card"
|
||||
assert stub.calls and stub.calls[0]["amount_kopeks"] == 50000
|
||||
assert "links" in captured_args["metadata"]
|
||||
|
||||
|
||||
@pytest.mark.anyio("asyncio")
|
||||
async def test_create_pal24_payment_limits_and_configuration(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
stub = StubPal24Service()
|
||||
service = _make_service(stub)
|
||||
db = DummySession()
|
||||
|
||||
monkeypatch.setattr(settings, "PAL24_MIN_AMOUNT_KOPEKS", 5000, raising=False)
|
||||
monkeypatch.setattr(settings, "PAL24_MAX_AMOUNT_KOPEKS", 20_000, raising=False)
|
||||
|
||||
result_low = await service.create_pal24_payment(
|
||||
db=db,
|
||||
user_id=1,
|
||||
amount_kopeks=1000,
|
||||
description="Пополнение",
|
||||
language="ru",
|
||||
)
|
||||
assert result_low is None
|
||||
|
||||
result_high = await service.create_pal24_payment(
|
||||
db=db,
|
||||
user_id=1,
|
||||
amount_kopeks=50_000,
|
||||
description="Пополнение",
|
||||
language="ru",
|
||||
)
|
||||
assert result_high is None
|
||||
|
||||
service_not_configured = _make_service(StubPal24Service(configured=False))
|
||||
result_config = await service_not_configured.create_pal24_payment(
|
||||
db=db,
|
||||
user_id=1,
|
||||
amount_kopeks=10_000,
|
||||
description="Пополнение",
|
||||
language="ru",
|
||||
)
|
||||
assert result_config is None
|
||||
|
||||
|
||||
@pytest.mark.anyio("asyncio")
|
||||
async def test_create_pal24_payment_handles_api_errors(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
stub = StubPal24Service()
|
||||
stub.raise_error = Pal24APIError("api failed")
|
||||
service = _make_service(stub)
|
||||
db = DummySession()
|
||||
|
||||
monkeypatch.setattr(settings, "PAL24_MIN_AMOUNT_KOPEKS", 1000, raising=False)
|
||||
monkeypatch.setattr(settings, "PAL24_MAX_AMOUNT_KOPEKS", 10_000, raising=False)
|
||||
|
||||
result = await service.create_pal24_payment(
|
||||
db=db,
|
||||
user_id=5,
|
||||
amount_kopeks=2000,
|
||||
description="Пополнение",
|
||||
language="ru",
|
||||
)
|
||||
assert result is None
|
||||
@@ -0,0 +1,142 @@
|
||||
"""Тесты для Telegram Stars-сценариев внутри PaymentService."""
|
||||
|
||||
from pathlib import Path
|
||||
from typing import Any, Dict, Optional
|
||||
import sys
|
||||
|
||||
import pytest
|
||||
|
||||
ROOT_DIR = Path(__file__).resolve().parents[2]
|
||||
if str(ROOT_DIR) not in sys.path:
|
||||
sys.path.insert(0, str(ROOT_DIR))
|
||||
|
||||
from app.services.payment_service import PaymentService # noqa: E402
|
||||
from app.config import settings # noqa: E402
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def anyio_backend() -> str:
|
||||
"""Ограничиваем anyio тесты только бэкендом asyncio."""
|
||||
return "asyncio"
|
||||
|
||||
|
||||
class DummyBot:
|
||||
"""Минимальная заглушка aiogram.Bot для тестов."""
|
||||
|
||||
def __init__(self) -> None:
|
||||
self.calls: list[Dict[str, Any]] = []
|
||||
|
||||
async def create_invoice_link(self, **kwargs: Any) -> str:
|
||||
"""Эмулируем создание платежной ссылки и сохраняем параметры вызова."""
|
||||
self.calls.append(kwargs)
|
||||
return "https://t.me/invoice/stars"
|
||||
|
||||
|
||||
def _make_service(bot: Optional[DummyBot]) -> PaymentService:
|
||||
"""Создаёт экземпляр PaymentService без выполнения полного конструктора."""
|
||||
service = PaymentService.__new__(PaymentService) # type: ignore[call-arg]
|
||||
service.bot = bot
|
||||
# Stars-сервис достаточно обозначить любым truthy-значением.
|
||||
service.stars_service = object() if bot else None
|
||||
return service
|
||||
|
||||
|
||||
@pytest.mark.anyio("asyncio")
|
||||
async def test_create_stars_invoice_calculates_stars(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
"""Количество звёзд должно рассчитываться по курсу с округлением вниз и нижним порогом 1."""
|
||||
bot = DummyBot()
|
||||
service = _make_service(bot)
|
||||
|
||||
monkeypatch.setattr(
|
||||
type(settings),
|
||||
"get_stars_rate",
|
||||
lambda self: 70,
|
||||
raising=False,
|
||||
)
|
||||
monkeypatch.setattr(
|
||||
type(settings),
|
||||
"format_price",
|
||||
lambda self, amount: f"{amount / 100:.0f}₽",
|
||||
raising=False,
|
||||
)
|
||||
|
||||
result = await service.create_stars_invoice(
|
||||
amount_kopeks=14000,
|
||||
description="Пополнение",
|
||||
payload="custom_payload",
|
||||
)
|
||||
|
||||
assert result == "https://t.me/invoice/stars"
|
||||
assert len(bot.calls) == 1
|
||||
call = bot.calls[0]
|
||||
assert call["title"] == "Пополнение баланса VPN"
|
||||
assert call["payload"] == "custom_payload"
|
||||
prices = call["prices"]
|
||||
assert len(prices) == 1
|
||||
assert prices[0].amount == 2 # 14000 коп. → 140 ₽ → 2 звезды при курсе 70
|
||||
assert "≈2 ⭐" in call["description"]
|
||||
|
||||
|
||||
@pytest.mark.anyio("asyncio")
|
||||
async def test_create_stars_invoice_enforces_minimum_star(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
"""При слишком маленькой сумме минимум должен составлять 1 звезду."""
|
||||
bot = DummyBot()
|
||||
service = _make_service(bot)
|
||||
|
||||
monkeypatch.setattr(type(settings), "get_stars_rate", lambda self: 500, raising=False)
|
||||
monkeypatch.setattr(type(settings), "format_price", lambda self, amount: amount, raising=False)
|
||||
|
||||
await service.create_stars_invoice(
|
||||
amount_kopeks=50, # 0.5 ₽ при курсе 500 => <1 звезды
|
||||
description="Микроплатёж",
|
||||
)
|
||||
|
||||
prices = bot.calls[0]["prices"]
|
||||
assert prices[0].amount == 1
|
||||
|
||||
|
||||
@pytest.mark.anyio("asyncio")
|
||||
async def test_create_stars_invoice_uses_explicit_stars(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
"""Если передано значение stars_amount, функция должна использовать его напрямую."""
|
||||
bot = DummyBot()
|
||||
service = _make_service(bot)
|
||||
|
||||
# При явном указании звёзд курс не запрашивается.
|
||||
monkeypatch.setattr(type(settings), "format_price", lambda self, amount: amount, raising=False)
|
||||
|
||||
await service.create_stars_invoice(
|
||||
amount_kopeks=1000,
|
||||
description="Оплата подписки",
|
||||
stars_amount=5,
|
||||
)
|
||||
|
||||
prices = bot.calls[0]["prices"]
|
||||
assert prices[0].amount == 5
|
||||
assert "≈5 ⭐" in bot.calls[0]["description"]
|
||||
|
||||
|
||||
@pytest.mark.anyio("asyncio")
|
||||
async def test_create_stars_invoice_rejects_invalid_rate(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
"""Отрицательный или нулевой курс должен приводить к исключению."""
|
||||
bot = DummyBot()
|
||||
service = _make_service(bot)
|
||||
|
||||
monkeypatch.setattr(type(settings), "get_stars_rate", lambda self: 0, raising=False)
|
||||
|
||||
with pytest.raises(ValueError, match="Stars rate must be positive"):
|
||||
await service.create_stars_invoice(
|
||||
amount_kopeks=1000,
|
||||
description="Пополнение",
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.anyio("asyncio")
|
||||
async def test_create_stars_invoice_requires_bot() -> None:
|
||||
"""Без экземпляра бота и stars_service функция должна отказывать."""
|
||||
service = _make_service(bot=None)
|
||||
|
||||
with pytest.raises(ValueError, match="Bot instance required"):
|
||||
await service.create_stars_invoice(
|
||||
amount_kopeks=1000,
|
||||
description="Пополнение",
|
||||
)
|
||||
@@ -0,0 +1,79 @@
|
||||
"""Тесты Tribute-платежей PaymentService."""
|
||||
|
||||
from pathlib import Path
|
||||
import sys
|
||||
import hmac
|
||||
import hashlib
|
||||
|
||||
import pytest
|
||||
|
||||
ROOT_DIR = Path(__file__).resolve().parents[2]
|
||||
if str(ROOT_DIR) not in sys.path:
|
||||
sys.path.insert(0, str(ROOT_DIR))
|
||||
|
||||
from app.services.payment_service import PaymentService # noqa: E402
|
||||
from app.config import settings # noqa: E402
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def anyio_backend() -> str:
|
||||
return "asyncio"
|
||||
|
||||
|
||||
def _make_service() -> PaymentService:
|
||||
service = PaymentService.__new__(PaymentService) # type: ignore[call-arg]
|
||||
service.bot = None
|
||||
service.yookassa_service = None
|
||||
service.mulenpay_service = None
|
||||
service.pal24_service = None
|
||||
service.cryptobot_service = None
|
||||
service.stars_service = None
|
||||
return service
|
||||
|
||||
|
||||
@pytest.mark.anyio("asyncio")
|
||||
async def test_create_tribute_payment_requires_enabled(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
service = _make_service()
|
||||
monkeypatch.setattr(settings, "TRIBUTE_ENABLED", False, raising=False)
|
||||
|
||||
with pytest.raises(ValueError):
|
||||
await service.create_tribute_payment(1000, 1, "Пополнение")
|
||||
|
||||
|
||||
@pytest.mark.anyio("asyncio")
|
||||
async def test_create_tribute_payment_success(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
service = _make_service()
|
||||
monkeypatch.setattr(settings, "TRIBUTE_ENABLED", True, raising=False)
|
||||
monkeypatch.setattr(settings, "WEBHOOK_URL", "https://example.com", raising=False)
|
||||
|
||||
result = await service.create_tribute_payment(
|
||||
amount_kopeks=15000,
|
||||
user_id=5,
|
||||
description="Оплата подписки",
|
||||
)
|
||||
|
||||
assert "https://tribute.ru/pay" in result
|
||||
assert "amount=15000" in result
|
||||
assert "user=5" in result
|
||||
|
||||
|
||||
def test_verify_tribute_webhook_signature(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
service = _make_service()
|
||||
monkeypatch.setattr(settings, "TRIBUTE_API_KEY", "secret", raising=False)
|
||||
|
||||
payload = {"payment": "ok"}
|
||||
signature = hmac.new(
|
||||
b"secret",
|
||||
str(payload).encode(),
|
||||
hashlib.sha256,
|
||||
).hexdigest()
|
||||
|
||||
assert service.verify_tribute_webhook(payload, signature) is True
|
||||
assert service.verify_tribute_webhook(payload, "invalid") is False
|
||||
|
||||
|
||||
def test_verify_tribute_webhook_returns_false_without_key(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
service = _make_service()
|
||||
monkeypatch.setattr(settings, "TRIBUTE_API_KEY", "", raising=False)
|
||||
|
||||
assert service.verify_tribute_webhook({}, "signature") is False
|
||||
@@ -0,0 +1,490 @@
|
||||
"""Интеграционные проверки обработки вебхуков PaymentService."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from pathlib import Path
|
||||
from types import SimpleNamespace, ModuleType
|
||||
from typing import Any, Dict
|
||||
import sys
|
||||
|
||||
import pytest
|
||||
from unittest.mock import AsyncMock
|
||||
|
||||
ROOT_DIR = Path(__file__).resolve().parents[2]
|
||||
if str(ROOT_DIR) not in sys.path:
|
||||
sys.path.insert(0, str(ROOT_DIR))
|
||||
|
||||
import app.services.payment_service as payment_service_module # noqa: E402
|
||||
from app.services.payment_service import PaymentService # noqa: E402
|
||||
from app.config import settings # noqa: E402
|
||||
|
||||
|
||||
class DummyBot:
|
||||
def __init__(self) -> None:
|
||||
self.sent_messages: list[Dict[str, Any]] = []
|
||||
|
||||
async def send_message(self, *args: Any, **kwargs: Any) -> None: # pragma: no cover - бизнес-логика тестируется через вызов
|
||||
self.sent_messages.append({"args": args, "kwargs": kwargs})
|
||||
|
||||
|
||||
class FakeSession:
|
||||
def __init__(self) -> None:
|
||||
self.commits = 0
|
||||
self.refreshed: list[Any] = []
|
||||
self.added: list[Any] = []
|
||||
|
||||
async def commit(self) -> None:
|
||||
self.commits += 1
|
||||
|
||||
async def rollback(self) -> None: # pragma: no cover
|
||||
return None
|
||||
|
||||
async def refresh(self, obj: Any) -> None:
|
||||
self.refreshed.append(obj)
|
||||
|
||||
def add(self, obj: Any) -> None: # pragma: no cover - используется при создании транзакций
|
||||
self.added.append(obj)
|
||||
|
||||
|
||||
def _make_service(bot: DummyBot) -> PaymentService:
|
||||
service = PaymentService.__new__(PaymentService) # type: ignore[call-arg]
|
||||
service.bot = bot
|
||||
service.yookassa_service = None
|
||||
service.stars_service = None
|
||||
service.mulenpay_service = None
|
||||
service.pal24_service = None
|
||||
service.cryptobot_service = None
|
||||
return service
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def anyio_backend() -> str:
|
||||
return "asyncio"
|
||||
|
||||
|
||||
@pytest.mark.anyio("asyncio")
|
||||
async def test_process_mulenpay_callback_success(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
bot = DummyBot()
|
||||
service = _make_service(bot)
|
||||
fake_session = FakeSession()
|
||||
payment = SimpleNamespace(
|
||||
uuid="mulen_uuid",
|
||||
mulen_payment_id=123,
|
||||
amount_kopeks=5000,
|
||||
user_id=42,
|
||||
transaction_id=None,
|
||||
is_paid=False,
|
||||
)
|
||||
|
||||
async def fake_get_by_uuid(db, uuid):
|
||||
return payment
|
||||
|
||||
async def fake_get_by_id(db, mid):
|
||||
return None
|
||||
|
||||
monkeypatch.setattr(payment_service_module, "get_mulenpay_payment_by_uuid", fake_get_by_uuid)
|
||||
monkeypatch.setattr(payment_service_module, "get_mulenpay_payment_by_mulen_id", fake_get_by_id)
|
||||
|
||||
transactions: list[Dict[str, Any]] = []
|
||||
|
||||
async def fake_create_transaction(db, **kwargs):
|
||||
transactions.append(kwargs)
|
||||
return SimpleNamespace(id=777, **kwargs)
|
||||
|
||||
monkeypatch.setattr(payment_service_module, "create_transaction", fake_create_transaction)
|
||||
|
||||
updated_status: dict[str, Any] = {}
|
||||
|
||||
async def fake_update_status(db, payment=None, status=None, **kwargs):
|
||||
payment.status = status
|
||||
payment.is_paid = status == "success"
|
||||
updated_status.update({"status": status, "kwargs": kwargs})
|
||||
|
||||
monkeypatch.setattr(payment_service_module, "update_mulenpay_payment_status", fake_update_status)
|
||||
|
||||
async def fake_link(db, payment=None, transaction_id=None):
|
||||
payment.transaction_id = transaction_id
|
||||
|
||||
monkeypatch.setattr(payment_service_module, "link_mulenpay_payment_to_transaction", fake_link)
|
||||
|
||||
user = SimpleNamespace(
|
||||
id=42,
|
||||
telegram_id=100500,
|
||||
balance_kopeks=0,
|
||||
has_made_first_topup=False,
|
||||
promo_group=None,
|
||||
subscription=None,
|
||||
referred_by_id=None,
|
||||
referrer=None,
|
||||
)
|
||||
|
||||
async def fake_get_user(db, user_id):
|
||||
return user
|
||||
|
||||
monkeypatch.setattr(payment_service_module, "get_user_by_id", fake_get_user)
|
||||
monkeypatch.setattr(type(settings), "format_price", lambda self, amount: f"{amount / 100:.2f}₽", raising=False)
|
||||
|
||||
referral_mock = SimpleNamespace(process_referral_topup=AsyncMock())
|
||||
monkeypatch.setitem(sys.modules, "app.services.referral_service", referral_mock)
|
||||
|
||||
class DummyAdminService:
|
||||
def __init__(self, bot):
|
||||
self.bot = bot
|
||||
self.calls: list[Any] = []
|
||||
|
||||
async def send_balance_topup_notification(self, *args, **kwargs):
|
||||
self.calls.append((args, kwargs))
|
||||
|
||||
admin_service = DummyAdminService(bot)
|
||||
monkeypatch.setitem(sys.modules, "app.services.admin_notification_service", SimpleNamespace(AdminNotificationService=lambda bot: admin_service))
|
||||
|
||||
service.build_topup_success_keyboard = AsyncMock(return_value=None)
|
||||
|
||||
payload = {
|
||||
"uuid": "mulen_uuid",
|
||||
"payment_status": "success",
|
||||
"id": 123,
|
||||
"amount": "50.00",
|
||||
}
|
||||
|
||||
result = await service.process_mulenpay_callback(fake_session, payload)
|
||||
|
||||
assert result is True
|
||||
assert transactions and transactions[0]["user_id"] == 42
|
||||
assert payment.transaction_id == 777
|
||||
assert updated_status["status"] == "success"
|
||||
assert user.balance_kopeks == 5000
|
||||
assert fake_session.commits >= 1
|
||||
assert bot.sent_messages # сообщение пользователю отправлено
|
||||
|
||||
|
||||
@pytest.mark.anyio("asyncio")
|
||||
async def test_process_cryptobot_webhook_success(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
bot = DummyBot()
|
||||
service = _make_service(bot)
|
||||
fake_session = FakeSession()
|
||||
payment = SimpleNamespace(
|
||||
invoice_id="inv_1",
|
||||
user_id=7,
|
||||
status="pending",
|
||||
transaction_id=None,
|
||||
amount="12.50",
|
||||
asset="USDT",
|
||||
amount_float=12.5,
|
||||
)
|
||||
|
||||
async def fake_get_crypto(db, invoice_id):
|
||||
return payment
|
||||
|
||||
async def fake_update_status(db, invoice_id, status, paid_at):
|
||||
payment.status = status
|
||||
payment.paid_at = paid_at
|
||||
return payment
|
||||
|
||||
async def fake_link(db, invoice_id, transaction_id):
|
||||
payment.transaction_id = transaction_id
|
||||
|
||||
fake_cryptobot_module = ModuleType("app.database.crud.cryptobot")
|
||||
fake_cryptobot_module.get_cryptobot_payment_by_invoice_id = fake_get_crypto
|
||||
fake_cryptobot_module.update_cryptobot_payment_status = fake_update_status
|
||||
fake_cryptobot_module.link_cryptobot_payment_to_transaction = fake_link
|
||||
monkeypatch.setitem(sys.modules, "app.database.crud.cryptobot", fake_cryptobot_module)
|
||||
|
||||
transactions: list[Dict[str, Any]] = []
|
||||
|
||||
async def fake_create_transaction(db, **kwargs):
|
||||
transactions.append(kwargs)
|
||||
return SimpleNamespace(id=888, **kwargs)
|
||||
|
||||
fake_transaction_module = ModuleType("app.database.crud.transaction")
|
||||
fake_transaction_module.create_transaction = fake_create_transaction
|
||||
monkeypatch.setitem(sys.modules, "app.database.crud.transaction", fake_transaction_module)
|
||||
monkeypatch.setattr(payment_service_module, "create_transaction", fake_create_transaction)
|
||||
|
||||
user = SimpleNamespace(
|
||||
id=7,
|
||||
telegram_id=700,
|
||||
balance_kopeks=0,
|
||||
has_made_first_topup=False,
|
||||
promo_group=None,
|
||||
subscription=None,
|
||||
referred_by_id=None,
|
||||
referrer=None,
|
||||
)
|
||||
|
||||
async def fake_get_user_crypto(db, user_id):
|
||||
return user
|
||||
|
||||
monkeypatch.setattr(payment_service_module, "get_user_by_id", fake_get_user_crypto)
|
||||
|
||||
referral_crypto = SimpleNamespace(process_referral_topup=AsyncMock())
|
||||
monkeypatch.setitem(sys.modules, "app.services.referral_service", referral_crypto)
|
||||
|
||||
admin_calls: list[Any] = []
|
||||
|
||||
class DummyAdminService2:
|
||||
def __init__(self, bot):
|
||||
self.bot = bot
|
||||
|
||||
async def send_balance_topup_notification(self, *args, **kwargs):
|
||||
admin_calls.append((args, kwargs))
|
||||
|
||||
monkeypatch.setitem(sys.modules, "app.services.admin_notification_service", SimpleNamespace(AdminNotificationService=lambda bot: DummyAdminService2(bot)))
|
||||
monkeypatch.setattr(payment_service_module.currency_converter, "usd_to_rub", AsyncMock(return_value=140.0))
|
||||
monkeypatch.setattr(type(settings), "format_price", lambda self, amount: f"{amount / 100:.2f}₽", raising=False)
|
||||
service.build_topup_success_keyboard = AsyncMock(return_value=None)
|
||||
|
||||
payload = {
|
||||
"update_type": "invoice_paid",
|
||||
"payload": {
|
||||
"invoice_id": "inv_1",
|
||||
"paid_at": "2024-01-01T12:00:00Z",
|
||||
},
|
||||
}
|
||||
|
||||
result = await service.process_cryptobot_webhook(fake_session, payload)
|
||||
|
||||
assert result is True
|
||||
assert transactions and transactions[0]["amount_kopeks"] == 14000
|
||||
assert user.balance_kopeks == 14000
|
||||
assert payment.transaction_id == 888
|
||||
assert bot.sent_messages
|
||||
assert admin_calls
|
||||
|
||||
|
||||
@pytest.mark.anyio("asyncio")
|
||||
async def test_process_yookassa_webhook_success(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
bot = DummyBot()
|
||||
service = _make_service(bot)
|
||||
fake_session = FakeSession()
|
||||
payment = SimpleNamespace(
|
||||
yookassa_payment_id="yk_123",
|
||||
user_id=21,
|
||||
amount_kopeks=10000,
|
||||
transaction_id=None,
|
||||
status="pending",
|
||||
is_paid=False,
|
||||
)
|
||||
|
||||
async def fake_get_payment(db, payment_id):
|
||||
return payment
|
||||
|
||||
async def fake_update(db, payment_id, status, is_paid, is_captured, captured_at, payment_method_type):
|
||||
payment.status = status
|
||||
payment.is_paid = is_paid
|
||||
payment.captured_at = captured_at
|
||||
return payment
|
||||
|
||||
async def fake_link(db, payment_id, transaction_id):
|
||||
payment.transaction_id = transaction_id
|
||||
|
||||
yk_module = ModuleType("app.database.crud.yookassa")
|
||||
yk_module.get_yookassa_payment_by_id = fake_get_payment
|
||||
yk_module.update_yookassa_payment_status = fake_update
|
||||
yk_module.link_yookassa_payment_to_transaction = fake_link
|
||||
monkeypatch.setitem(sys.modules, "app.database.crud.yookassa", yk_module)
|
||||
|
||||
transactions: list[Dict[str, Any]] = []
|
||||
|
||||
async def fake_create_transaction(db, **kwargs):
|
||||
transactions.append(kwargs)
|
||||
return SimpleNamespace(id=999, **kwargs)
|
||||
|
||||
trx_module = ModuleType("app.database.crud.transaction")
|
||||
trx_module.create_transaction = fake_create_transaction
|
||||
monkeypatch.setitem(sys.modules, "app.database.crud.transaction", trx_module)
|
||||
monkeypatch.setattr(payment_service_module, "create_transaction", fake_create_transaction)
|
||||
monkeypatch.setattr(payment_service_module, "create_transaction", fake_create_transaction)
|
||||
monkeypatch.setattr(payment_service_module, "create_transaction", fake_create_transaction)
|
||||
|
||||
user = SimpleNamespace(
|
||||
id=21,
|
||||
telegram_id=2100,
|
||||
balance_kopeks=0,
|
||||
has_made_first_topup=False,
|
||||
promo_group=None,
|
||||
subscription=None,
|
||||
referred_by_id=None,
|
||||
referrer=None,
|
||||
)
|
||||
|
||||
async def fake_get_user(db, user_id):
|
||||
return user
|
||||
|
||||
monkeypatch.setattr(payment_service_module, "get_user_by_id", fake_get_user)
|
||||
monkeypatch.setattr(type(settings), "format_price", lambda self, amount: f"{amount / 100:.2f}₽", raising=False)
|
||||
|
||||
referral_mock = SimpleNamespace(process_referral_topup=AsyncMock())
|
||||
monkeypatch.setitem(sys.modules, "app.services.referral_service", referral_mock)
|
||||
|
||||
admin_calls: list[Any] = []
|
||||
|
||||
class DummyAdminService:
|
||||
def __init__(self, bot):
|
||||
self.bot = bot
|
||||
|
||||
async def send_balance_topup_notification(self, *args, **kwargs):
|
||||
admin_calls.append((args, kwargs))
|
||||
|
||||
monkeypatch.setitem(sys.modules, "app.services.admin_notification_service", SimpleNamespace(AdminNotificationService=lambda bot: DummyAdminService(bot)))
|
||||
service.build_topup_success_keyboard = AsyncMock(return_value=None)
|
||||
|
||||
payload = {
|
||||
"object": {
|
||||
"id": "yk_123",
|
||||
"status": "succeeded",
|
||||
"paid": True,
|
||||
"payment_method": {"type": "bank_card"},
|
||||
}
|
||||
}
|
||||
|
||||
result = await service.process_yookassa_webhook(fake_session, payload)
|
||||
|
||||
assert result is True
|
||||
assert transactions and transactions[0]["amount_kopeks"] == 10000
|
||||
assert payment.transaction_id == 999
|
||||
assert user.balance_kopeks == 10000
|
||||
assert bot.sent_messages
|
||||
assert admin_calls
|
||||
|
||||
|
||||
@pytest.mark.anyio("asyncio")
|
||||
async def test_process_yookassa_webhook_missing_id(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
bot = DummyBot()
|
||||
service = _make_service(bot)
|
||||
db = FakeSession()
|
||||
|
||||
result = await service.process_yookassa_webhook(db, {"object": {}})
|
||||
assert result is False
|
||||
|
||||
|
||||
@pytest.mark.anyio("asyncio")
|
||||
async def test_process_pal24_postback_success(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
bot = DummyBot()
|
||||
service = _make_service(bot)
|
||||
service.pal24_service = SimpleNamespace(is_configured=True)
|
||||
fake_session = FakeSession()
|
||||
payment = SimpleNamespace(
|
||||
bill_id="BILL-1",
|
||||
order_id="order-1",
|
||||
amount_kopeks=5000,
|
||||
user_id=33,
|
||||
transaction_id=None,
|
||||
is_paid=False,
|
||||
status="NEW",
|
||||
)
|
||||
|
||||
async def fake_get_by_order(db, order_id):
|
||||
return payment
|
||||
|
||||
async def fake_get_by_bill(db, bill_id):
|
||||
return payment
|
||||
|
||||
async def fake_update(db, payment_obj, **kwargs):
|
||||
payment.status = kwargs.get("status", payment.status)
|
||||
payment.is_paid = kwargs.get("is_paid", payment.is_paid)
|
||||
payment.payment_status = kwargs.get("payment_status", payment.status)
|
||||
payment.callback_payload = kwargs.get("callback_payload")
|
||||
return payment
|
||||
|
||||
async def fake_link(db, payment_obj, transaction_id):
|
||||
payment.transaction_id = transaction_id
|
||||
|
||||
pal_module = ModuleType("app.database.crud.pal24")
|
||||
pal_module.get_pal24_payment_by_order_id = fake_get_by_order
|
||||
pal_module.get_pal24_payment_by_bill_id = fake_get_by_bill
|
||||
pal_module.update_pal24_payment_status = fake_update
|
||||
pal_module.link_pal24_payment_to_transaction = fake_link
|
||||
monkeypatch.setitem(sys.modules, "app.database.crud.pal24", pal_module)
|
||||
monkeypatch.setattr(payment_service_module, "get_pal24_payment_by_order_id", fake_get_by_order)
|
||||
monkeypatch.setattr(payment_service_module, "get_pal24_payment_by_bill_id", fake_get_by_bill)
|
||||
monkeypatch.setattr(payment_service_module, "update_pal24_payment_status", fake_update)
|
||||
monkeypatch.setattr(payment_service_module, "link_pal24_payment_to_transaction", fake_link)
|
||||
|
||||
async def fake_create_transaction(db, **kwargs):
|
||||
payment.transaction_id = 654
|
||||
return SimpleNamespace(id=654, **kwargs)
|
||||
|
||||
trx_module = ModuleType("app.database.crud.transaction")
|
||||
trx_module.create_transaction = fake_create_transaction
|
||||
monkeypatch.setitem(sys.modules, "app.database.crud.transaction", trx_module)
|
||||
monkeypatch.setattr(payment_service_module, "create_transaction", fake_create_transaction)
|
||||
|
||||
user = SimpleNamespace(
|
||||
id=33,
|
||||
telegram_id=3300,
|
||||
balance_kopeks=0,
|
||||
has_made_first_topup=False,
|
||||
promo_group=None,
|
||||
subscription=None,
|
||||
referred_by_id=None,
|
||||
referrer=None,
|
||||
)
|
||||
|
||||
async def fake_get_user(db, user_id):
|
||||
return user
|
||||
|
||||
monkeypatch.setattr(payment_service_module, "get_user_by_id", fake_get_user)
|
||||
monkeypatch.setattr(type(settings), "format_price", lambda self, amount: f"{amount / 100:.2f}₽", raising=False)
|
||||
|
||||
referral_pal = SimpleNamespace(process_referral_topup=AsyncMock())
|
||||
monkeypatch.setitem(sys.modules, "app.services.referral_service", referral_pal)
|
||||
|
||||
admin_calls: list[Any] = []
|
||||
|
||||
class DummyAdminServicePal:
|
||||
def __init__(self, bot):
|
||||
self.bot = bot
|
||||
|
||||
async def send_balance_topup_notification(self, *args, **kwargs):
|
||||
admin_calls.append((args, kwargs))
|
||||
|
||||
monkeypatch.setitem(sys.modules, "app.services.admin_notification_service", SimpleNamespace(AdminNotificationService=lambda bot: DummyAdminServicePal(bot)))
|
||||
service.build_topup_success_keyboard = AsyncMock(return_value=None)
|
||||
|
||||
payload = {
|
||||
"InvId": "order-1",
|
||||
"OutSum": "50.00",
|
||||
"Status": "SUCCESS",
|
||||
"TrsId": "trs-1",
|
||||
}
|
||||
|
||||
result = await service.process_pal24_postback(fake_session, payload)
|
||||
|
||||
assert result is True
|
||||
assert payment.transaction_id == 654
|
||||
assert user.balance_kopeks == 5000
|
||||
assert bot.sent_messages
|
||||
assert admin_calls
|
||||
|
||||
|
||||
@pytest.mark.anyio("asyncio")
|
||||
async def test_process_pal24_postback_payment_not_found(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
bot = DummyBot()
|
||||
service = _make_service(bot)
|
||||
service.pal24_service = SimpleNamespace(is_configured=True)
|
||||
db = FakeSession()
|
||||
|
||||
async def fake_get_by_order(db, order_id):
|
||||
return None
|
||||
|
||||
async def fake_get_by_bill(db, bill_id):
|
||||
return None
|
||||
|
||||
pal_module = ModuleType("app.database.crud.pal24")
|
||||
pal_module.get_pal24_payment_by_order_id = fake_get_by_order
|
||||
pal_module.get_pal24_payment_by_bill_id = fake_get_by_bill
|
||||
pal_module.update_pal24_payment_status = AsyncMock()
|
||||
pal_module.link_pal24_payment_to_transaction = AsyncMock()
|
||||
monkeypatch.setitem(sys.modules, "app.database.crud.pal24", pal_module)
|
||||
monkeypatch.setattr(payment_service_module, "get_pal24_payment_by_order_id", fake_get_by_order)
|
||||
monkeypatch.setattr(payment_service_module, "get_pal24_payment_by_bill_id", fake_get_by_bill)
|
||||
|
||||
payload = {
|
||||
"InvId": "order-unknown",
|
||||
"OutSum": "10.00",
|
||||
"Status": "SUCCESS",
|
||||
}
|
||||
|
||||
result = await service.process_pal24_postback(db, payload)
|
||||
assert result is False
|
||||
@@ -0,0 +1,245 @@
|
||||
"""Тесты для YooKassa-сценариев PaymentService."""
|
||||
|
||||
import sys
|
||||
from datetime import datetime
|
||||
from pathlib import Path
|
||||
from typing import Any, Dict, Optional
|
||||
|
||||
import pytest
|
||||
|
||||
ROOT_DIR = Path(__file__).resolve().parents[2]
|
||||
if str(ROOT_DIR) not in sys.path:
|
||||
sys.path.insert(0, str(ROOT_DIR))
|
||||
|
||||
import app.services.payment_service as payment_service_module # noqa: E402
|
||||
from app.config import settings # noqa: E402
|
||||
from app.services.payment_service import PaymentService # noqa: E402
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def anyio_backend() -> str:
|
||||
"""Запускаем async-тесты на asyncio, чтобы избежать зависимостей trio."""
|
||||
return "asyncio"
|
||||
|
||||
|
||||
class DummySession:
|
||||
"""Простейшая заглушка AsyncSession."""
|
||||
|
||||
def __init__(self) -> None:
|
||||
self.committed = False
|
||||
|
||||
async def commit(self) -> None:
|
||||
self.committed = True
|
||||
|
||||
async def rollback(self) -> None:
|
||||
self.rolled_back = True # type: ignore[attr-defined]
|
||||
|
||||
|
||||
class DummyLocalPayment:
|
||||
"""Объект, имитирующий локальную запись платежа."""
|
||||
|
||||
def __init__(self, payment_id: int = 101) -> None:
|
||||
self.id = payment_id
|
||||
self.created_at = datetime(2024, 1, 1, 12, 0, 0)
|
||||
|
||||
|
||||
class StubYooKassaService:
|
||||
"""Заглушка для SDK, сохраняющая вызовы."""
|
||||
|
||||
def __init__(self, response: Dict[str, Any]) -> None:
|
||||
self.response = response
|
||||
self.calls: list[Dict[str, Any]] = []
|
||||
|
||||
async def create_payment(self, **kwargs: Any) -> Dict[str, Any]:
|
||||
self.calls.append(kwargs)
|
||||
return self.response
|
||||
|
||||
async def create_sbp_payment(self, **kwargs: Any) -> Dict[str, Any]:
|
||||
self.calls.append(kwargs)
|
||||
return self.response
|
||||
|
||||
|
||||
def _make_service(yookassa_service: Optional[StubYooKassaService]) -> PaymentService:
|
||||
service = PaymentService.__new__(PaymentService) # type: ignore[call-arg]
|
||||
service.bot = None
|
||||
service.yookassa_service = yookassa_service
|
||||
service.stars_service = None
|
||||
service.mulenpay_service = None
|
||||
service.pal24_service = None
|
||||
service.mulenpay_service = None
|
||||
service.cryptobot_service = None
|
||||
return service
|
||||
|
||||
|
||||
@pytest.mark.anyio("asyncio")
|
||||
async def test_create_yookassa_payment_success(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
"""Успешное создание платежа формирует корректные метаданные и локальную запись."""
|
||||
|
||||
response = {
|
||||
"id": "yk_123",
|
||||
"status": "pending",
|
||||
"confirmation_url": "https://yookassa.ru/confirm",
|
||||
"amount": {"value": "140.00", "currency": "RUB"},
|
||||
"metadata": {"existing": "value"},
|
||||
"created_at": "2024-01-01T12:00:00Z",
|
||||
"test_mode": False,
|
||||
}
|
||||
service = _make_service(StubYooKassaService(response))
|
||||
db = DummySession()
|
||||
|
||||
captured_args: Dict[str, Any] = {}
|
||||
|
||||
async def fake_create_yookassa_payment(**kwargs: Any) -> DummyLocalPayment:
|
||||
captured_args.update(kwargs)
|
||||
return DummyLocalPayment(payment_id=555)
|
||||
|
||||
monkeypatch.setattr(
|
||||
payment_service_module,
|
||||
"create_yookassa_payment",
|
||||
fake_create_yookassa_payment,
|
||||
raising=False,
|
||||
)
|
||||
monkeypatch.setattr(
|
||||
type(settings),
|
||||
"format_price",
|
||||
lambda self, amount: f"{amount / 100:.0f}₽",
|
||||
raising=False,
|
||||
)
|
||||
|
||||
result = await service.create_yookassa_payment(
|
||||
db=db,
|
||||
user_id=42,
|
||||
amount_kopeks=14000,
|
||||
description="Пополнение",
|
||||
receipt_email="user@example.com",
|
||||
metadata={"custom": "data"},
|
||||
)
|
||||
|
||||
assert result is not None
|
||||
assert result["local_payment_id"] == 555
|
||||
assert result["yookassa_payment_id"] == "yk_123"
|
||||
assert result["amount_kopeks"] == 14000
|
||||
assert result["amount_rubles"] == 140
|
||||
assert result["status"] == "pending"
|
||||
|
||||
assert captured_args["user_id"] == 42
|
||||
assert captured_args["metadata_json"]["custom"] == "data"
|
||||
assert captured_args["metadata_json"]["user_id"] == "42"
|
||||
assert captured_args["metadata_json"]["amount_kopeks"] == "14000"
|
||||
assert isinstance(captured_args["yookassa_created_at"], datetime)
|
||||
|
||||
|
||||
@pytest.mark.anyio("asyncio")
|
||||
async def test_create_yookassa_payment_returns_none_when_service_missing() -> None:
|
||||
"""Если сервис не настроен, метод должен вернуть None."""
|
||||
service = _make_service(None)
|
||||
db = DummySession()
|
||||
result = await service.create_yookassa_payment(
|
||||
db=db,
|
||||
user_id=1,
|
||||
amount_kopeks=1000,
|
||||
description="Пополнение",
|
||||
)
|
||||
assert result is None
|
||||
|
||||
|
||||
@pytest.mark.anyio("asyncio")
|
||||
async def test_create_yookassa_payment_handles_error_response(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
"""Ответ с ключом error должен приводить к None без записи в БД."""
|
||||
response = {"error": True}
|
||||
service = _make_service(StubYooKassaService(response))
|
||||
db = DummySession()
|
||||
|
||||
called = False
|
||||
|
||||
async def fake_create_yookassa_payment(**kwargs: Any) -> DummyLocalPayment:
|
||||
nonlocal called
|
||||
called = True
|
||||
return DummyLocalPayment()
|
||||
|
||||
monkeypatch.setattr(
|
||||
payment_service_module,
|
||||
"create_yookassa_payment",
|
||||
fake_create_yookassa_payment,
|
||||
raising=False,
|
||||
)
|
||||
|
||||
result = await service.create_yookassa_payment(
|
||||
db=db,
|
||||
user_id=1,
|
||||
amount_kopeks=5000,
|
||||
description="Пополнение",
|
||||
)
|
||||
assert result is None
|
||||
assert called is False
|
||||
|
||||
|
||||
@pytest.mark.anyio("asyncio")
|
||||
async def test_create_yookassa_sbp_payment_success(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
"""Проверяем SBP-сценарий, включая передачу confirmation_token."""
|
||||
|
||||
response = {
|
||||
"id": "yk_sbp_001",
|
||||
"status": "pending",
|
||||
"confirmation_url": "https://yookassa.ru/confirm",
|
||||
"confirmation": {"confirmation_token": "token123"},
|
||||
"created_at": "2024-02-01T10:00:00Z",
|
||||
}
|
||||
service = _make_service(StubYooKassaService(response))
|
||||
db = DummySession()
|
||||
|
||||
captured_args: Dict[str, Any] = {}
|
||||
|
||||
async def fake_create_yookassa_payment(**kwargs: Any) -> DummyLocalPayment:
|
||||
captured_args.update(kwargs)
|
||||
return DummyLocalPayment(payment_id=777)
|
||||
|
||||
monkeypatch.setattr(
|
||||
payment_service_module,
|
||||
"create_yookassa_payment",
|
||||
fake_create_yookassa_payment,
|
||||
raising=False,
|
||||
)
|
||||
|
||||
result = await service.create_yookassa_sbp_payment(
|
||||
db=db,
|
||||
user_id=7,
|
||||
amount_kopeks=25000,
|
||||
description="СБП пополнение",
|
||||
)
|
||||
|
||||
assert result is not None
|
||||
assert result["confirmation_token"] == "token123"
|
||||
assert captured_args["payment_method_type"] == "bank_card"
|
||||
assert captured_args["metadata_json"]["type"] == "balance_topup_sbp"
|
||||
|
||||
|
||||
@pytest.mark.anyio("asyncio")
|
||||
async def test_create_yookassa_sbp_payment_returns_none_on_error(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
"""Ошибочный ответ СБП не должен создавать запись."""
|
||||
response = {"error": "invalid"}
|
||||
service = _make_service(StubYooKassaService(response))
|
||||
db = DummySession()
|
||||
|
||||
called = False
|
||||
|
||||
async def fake_create_yookassa_payment(**kwargs: Any) -> DummyLocalPayment:
|
||||
nonlocal called
|
||||
called = True
|
||||
return DummyLocalPayment()
|
||||
|
||||
monkeypatch.setattr(
|
||||
payment_service_module,
|
||||
"create_yookassa_payment",
|
||||
fake_create_yookassa_payment,
|
||||
raising=False,
|
||||
)
|
||||
|
||||
result = await service.create_yookassa_sbp_payment(
|
||||
db=db,
|
||||
user_id=1,
|
||||
amount_kopeks=1000,
|
||||
description="СБП пополнение",
|
||||
)
|
||||
assert result is None
|
||||
assert called is False
|
||||
@@ -0,0 +1,179 @@
|
||||
"""Тесты низкоуровневого сервиса YooKassaService."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
from datetime import datetime
|
||||
from pathlib import Path
|
||||
import sys
|
||||
from types import SimpleNamespace
|
||||
|
||||
import pytest
|
||||
|
||||
ROOT_DIR = Path(__file__).resolve().parents[2]
|
||||
if str(ROOT_DIR) not in sys.path:
|
||||
sys.path.insert(0, str(ROOT_DIR))
|
||||
|
||||
from yookassa import Configuration, Payment as YooKassaPayment # type: ignore # noqa: E402
|
||||
from app.config import settings # noqa: E402
|
||||
from app.services.yookassa_service import YooKassaService # noqa: E402
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def anyio_backend() -> str:
|
||||
return "asyncio"
|
||||
|
||||
|
||||
class DummyLoop:
|
||||
async def run_in_executor(self, _executor, func):
|
||||
return func()
|
||||
|
||||
|
||||
def _prepare_config(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
monkeypatch.setattr(settings, "YOOKASSA_SHOP_ID", "shop123", raising=False)
|
||||
monkeypatch.setattr(settings, "YOOKASSA_SECRET_KEY", "secret123", raising=False)
|
||||
monkeypatch.setattr(settings, "YOOKASSA_RETURN_URL", "https://example.com/return", raising=False)
|
||||
monkeypatch.setattr(settings, "YOOKASSA_VAT_CODE", 1, raising=False)
|
||||
monkeypatch.setattr(settings, "YOOKASSA_PAYMENT_MODE", "full_payment", raising=False)
|
||||
monkeypatch.setattr(settings, "YOOKASSA_PAYMENT_SUBJECT", "service", raising=False)
|
||||
|
||||
|
||||
def test_init_without_credentials(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
monkeypatch.setattr(settings, "YOOKASSA_SHOP_ID", "", raising=False)
|
||||
monkeypatch.setattr(settings, "YOOKASSA_SECRET_KEY", "", raising=False)
|
||||
service = YooKassaService()
|
||||
assert service.configured is False
|
||||
assert service.return_url == "https://t.me/"
|
||||
|
||||
|
||||
@pytest.mark.anyio("asyncio")
|
||||
async def test_create_payment_success(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
_prepare_config(monkeypatch)
|
||||
monkeypatch.setattr(settings, "YOOKASSA_DEFAULT_RECEIPT_EMAIL", None, raising=False)
|
||||
monkeypatch.setattr(asyncio, "get_running_loop", lambda: DummyLoop(), raising=False)
|
||||
|
||||
captured_config: dict[str, tuple[str, str]] = {}
|
||||
|
||||
def fake_configure(shop_id: str, secret_key: str) -> None:
|
||||
captured_config["values"] = (shop_id, secret_key)
|
||||
|
||||
monkeypatch.setattr(Configuration, "configure", fake_configure, raising=False)
|
||||
|
||||
response_obj = SimpleNamespace(
|
||||
id="yk_1",
|
||||
status="pending",
|
||||
paid=False,
|
||||
confirmation=SimpleNamespace(confirmation_url="https://yk/confirm"),
|
||||
metadata={"meta": "value"},
|
||||
amount=SimpleNamespace(value="140.00", currency="RUB"),
|
||||
refundable=True,
|
||||
created_at=datetime(2024, 1, 1, 12, 0, 0),
|
||||
description="Desc",
|
||||
test=False,
|
||||
)
|
||||
|
||||
monkeypatch.setattr(
|
||||
YooKassaPayment,
|
||||
"create",
|
||||
staticmethod(lambda payload, key: response_obj),
|
||||
raising=False,
|
||||
)
|
||||
|
||||
service = YooKassaService()
|
||||
monkeypatch.setattr(settings, "YOOKASSA_DEFAULT_RECEIPT_EMAIL", "fallback@example.com", raising=False)
|
||||
|
||||
result = await service.create_payment(
|
||||
amount=140.0,
|
||||
currency="RUB",
|
||||
description="Пополнение",
|
||||
metadata={"order": "1"},
|
||||
receipt_email="user@example.com",
|
||||
)
|
||||
|
||||
assert service.configured is True
|
||||
assert captured_config["values"] == ("shop123", "secret123")
|
||||
assert result is not None
|
||||
assert result["id"] == "yk_1"
|
||||
assert result["confirmation_url"] == "https://yk/confirm"
|
||||
assert result["amount_value"] == 140.0
|
||||
assert result["status"] == "pending"
|
||||
|
||||
|
||||
@pytest.mark.anyio("asyncio")
|
||||
async def test_create_payment_without_contacts(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
_prepare_config(monkeypatch)
|
||||
monkeypatch.setattr(settings, "YOOKASSA_DEFAULT_RECEIPT_EMAIL", None, raising=False)
|
||||
monkeypatch.setattr(Configuration, "configure", lambda *args, **kwargs: None, raising=False)
|
||||
monkeypatch.setattr(asyncio, "get_running_loop", lambda: DummyLoop(), raising=False)
|
||||
monkeypatch.setattr(
|
||||
YooKassaPayment,
|
||||
"create",
|
||||
staticmethod(lambda payload, key: SimpleNamespace()),
|
||||
raising=False,
|
||||
)
|
||||
|
||||
service = YooKassaService()
|
||||
result = await service.create_payment(
|
||||
amount=10,
|
||||
currency="RUB",
|
||||
description="desc",
|
||||
metadata={},
|
||||
)
|
||||
assert result is not None
|
||||
assert result.get("error") is True
|
||||
|
||||
|
||||
@pytest.mark.anyio("asyncio")
|
||||
async def test_create_payment_returns_none_when_not_configured(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
monkeypatch.setattr(settings, "YOOKASSA_SHOP_ID", "", raising=False)
|
||||
monkeypatch.setattr(settings, "YOOKASSA_SECRET_KEY", "", raising=False)
|
||||
service = YooKassaService()
|
||||
result = await service.create_payment(
|
||||
amount=10,
|
||||
currency="RUB",
|
||||
description="desc",
|
||||
metadata={},
|
||||
)
|
||||
assert result is None
|
||||
|
||||
|
||||
@pytest.mark.anyio("asyncio")
|
||||
async def test_create_sbp_payment_success(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
_prepare_config(monkeypatch)
|
||||
monkeypatch.setattr(asyncio, "get_running_loop", lambda: DummyLoop(), raising=False)
|
||||
monkeypatch.setattr(Configuration, "configure", lambda *args, **kwargs: None, raising=False)
|
||||
monkeypatch.setattr(settings, "YOOKASSA_DEFAULT_RECEIPT_EMAIL", "fallback@example.com", raising=False)
|
||||
|
||||
response_obj = SimpleNamespace(
|
||||
id="sbp_001",
|
||||
status="pending",
|
||||
paid=False,
|
||||
confirmation=SimpleNamespace(confirmation_url="https://sbp/confirm"),
|
||||
metadata={"meta": "value"},
|
||||
amount=SimpleNamespace(value="200.00", currency="RUB"),
|
||||
refundable=False,
|
||||
created_at=datetime(2024, 2, 1, 9, 0, 0),
|
||||
description="SBP payment",
|
||||
test=True,
|
||||
)
|
||||
|
||||
monkeypatch.setattr(
|
||||
YooKassaPayment,
|
||||
"create",
|
||||
staticmethod(lambda payload, key: response_obj),
|
||||
raising=False,
|
||||
)
|
||||
|
||||
service = YooKassaService()
|
||||
result = await service.create_sbp_payment(
|
||||
amount=200.0,
|
||||
currency="rub",
|
||||
description="Оплата",
|
||||
metadata={"type": "sbp"},
|
||||
receipt_phone="+70000000000",
|
||||
)
|
||||
|
||||
assert result is not None
|
||||
assert result["id"] == "sbp_001"
|
||||
assert result["confirmation_url"] == "https://sbp/confirm"
|
||||
assert result["status"] == "pending"
|
||||
@@ -0,0 +1 @@
|
||||
# В этом пакете будут жить модульные тесты для вспомогательных утилит приложения.
|
||||
@@ -0,0 +1,104 @@
|
||||
"""Тесты для базовых форматтеров из app.utils.formatters."""
|
||||
|
||||
from datetime import datetime, timedelta
|
||||
|
||||
from app.utils import formatters
|
||||
|
||||
|
||||
def test_format_datetime_handles_iso_strings(fixed_datetime: datetime) -> None:
|
||||
"""ISO-строка должна корректно преобразовываться в отформатированный текст."""
|
||||
iso_value = fixed_datetime.isoformat()
|
||||
assert formatters.format_datetime(iso_value) == fixed_datetime.strftime("%d.%m.%Y %H:%M")
|
||||
|
||||
|
||||
def test_format_date_uses_custom_format(fixed_datetime: datetime) -> None:
|
||||
"""Можно задавать собственный шаблон вывода."""
|
||||
iso_value = fixed_datetime.isoformat()
|
||||
assert formatters.format_date(iso_value, format_str="%Y/%m/%d") == fixed_datetime.strftime("%Y/%m/%d")
|
||||
|
||||
|
||||
def test_format_time_ago_returns_human_readable_text() -> None:
|
||||
"""Разница во времени должна переводиться в человеко-понятную строку."""
|
||||
point_in_time = datetime.utcnow() - timedelta(minutes=5)
|
||||
assert formatters.format_time_ago(point_in_time, language="ru") == "5 мин. назад"
|
||||
assert formatters.format_time_ago(point_in_time, language="en") == "5 minutes ago"
|
||||
|
||||
|
||||
def test_format_days_declension_handles_russian_rules() -> None:
|
||||
"""Склонение дней в русском языке зависит от числа."""
|
||||
assert formatters.format_days_declension(1) == "1 день"
|
||||
assert formatters.format_days_declension(3) == "3 дня"
|
||||
assert formatters.format_days_declension(10) == "10 дней"
|
||||
|
||||
|
||||
def test_format_duration_switches_units() -> None:
|
||||
"""В зависимости от длины интервала выбирается подходящая единица измерения."""
|
||||
assert formatters.format_duration(45) == "45 сек."
|
||||
assert formatters.format_duration(120) == "2 мин."
|
||||
assert formatters.format_duration(7200) == "2 ч."
|
||||
assert formatters.format_duration(172800) == "2 дн."
|
||||
|
||||
|
||||
def test_format_bytes_scales_value() -> None:
|
||||
"""Размер должен выражаться в наиболее подходящей единице."""
|
||||
assert formatters.format_bytes(0) == "0 B"
|
||||
assert formatters.format_bytes(1024) == "1 KB"
|
||||
assert formatters.format_bytes(1024 * 1024) == "1 MB"
|
||||
|
||||
|
||||
def test_format_percentage_respects_precision() -> None:
|
||||
"""Проценты форматируются с нужным количеством знаков."""
|
||||
assert formatters.format_percentage(12.3456, decimals=2) == "12.35%"
|
||||
|
||||
|
||||
def test_format_number_inserts_separators() -> None:
|
||||
"""Разделители тысяч должны расставляться корректно как для int, так и для float."""
|
||||
assert formatters.format_number(1234567) == "1 234 567"
|
||||
assert formatters.format_number(1234.56) == "1 234.55"
|
||||
|
||||
|
||||
def test_truncate_text_appends_suffix() -> None:
|
||||
"""Строки, превышающие лимит, должны обрезаться и дополняться суффиксом."""
|
||||
source = "a" * 10
|
||||
assert formatters.truncate_text(source, max_length=5) == "aa..."
|
||||
|
||||
|
||||
def test_format_username_prefers_full_name() -> None:
|
||||
"""Полное имя имеет приоритет, затем username, затем ID."""
|
||||
assert formatters.format_username("nickname", 1, full_name="Имя") == "Имя"
|
||||
assert formatters.format_username("nickname", 1, full_name=None) == "@nickname"
|
||||
assert formatters.format_username(None, 42, full_name=None) == "ID42"
|
||||
|
||||
|
||||
def test_format_subscription_status_handles_active_and_expired() -> None:
|
||||
"""Статус подписки различается для активных/просроченных случаев."""
|
||||
future = datetime.utcnow() + timedelta(days=2)
|
||||
active = formatters.format_subscription_status(
|
||||
is_active=True,
|
||||
is_trial=False,
|
||||
end_date=future,
|
||||
language="ru",
|
||||
)
|
||||
assert active.startswith("✅ Активна")
|
||||
assert "(" in active and ")" in active
|
||||
|
||||
past = datetime.utcnow() - timedelta(days=1)
|
||||
expired = formatters.format_subscription_status(
|
||||
is_active=True,
|
||||
is_trial=False,
|
||||
end_date=past,
|
||||
language="ru",
|
||||
)
|
||||
assert expired == "⏰ Истекла"
|
||||
|
||||
|
||||
def test_format_traffic_usage_supports_unlimited() -> None:
|
||||
"""При безлимитном тарифе в строке должна появляться бесконечность."""
|
||||
assert formatters.format_traffic_usage(50.0, 0, language="ru") == "50.0 ГБ / ∞"
|
||||
assert formatters.format_traffic_usage(10.0, 100, language="ru") == "10.0 ГБ / 100 ГБ (10.0%)"
|
||||
|
||||
|
||||
def test_format_boolean_localises_output() -> None:
|
||||
"""Булевые значения отображаются локализованными словами."""
|
||||
assert formatters.format_boolean(True, language="ru") == "✅ Да"
|
||||
assert formatters.format_boolean(False, language="en") == "❌ No"
|
||||
@@ -0,0 +1,54 @@
|
||||
"""Тесты для функций безопасности из app.utils.security."""
|
||||
|
||||
import hashlib
|
||||
|
||||
import pytest
|
||||
|
||||
from app.utils.security import generate_api_token, hash_api_token
|
||||
|
||||
|
||||
def test_hash_api_token_default_algorithm_matches_hashlib() -> None:
|
||||
"""Проверяем, что алгоритм по умолчанию совпадает с hashlib.sha256."""
|
||||
sample = "secret-token"
|
||||
# Самостоятельно считаем эталонное значение.
|
||||
expected = hashlib.sha256(sample.encode("utf-8")).hexdigest()
|
||||
# Сравниваем с функцией проекта.
|
||||
assert hash_api_token(sample) == expected
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"algorithm,hash_factory",
|
||||
[
|
||||
("sha256", hashlib.sha256),
|
||||
("sha384", hashlib.sha384),
|
||||
("sha512", hashlib.sha512),
|
||||
],
|
||||
)
|
||||
def test_hash_api_token_accepts_supported_algorithms(algorithm, hash_factory) -> None:
|
||||
"""Каждый поддерживаемый алгоритм должен выдавать корректный результат."""
|
||||
sample = "token-value"
|
||||
expected = hash_factory(sample.encode("utf-8")).hexdigest()
|
||||
assert hash_api_token(sample, algorithm=algorithm) == expected
|
||||
|
||||
|
||||
def test_hash_api_token_rejects_unknown_algorithm() -> None:
|
||||
"""Некорректное имя алгоритма должно приводить к ValueError."""
|
||||
with pytest.raises(ValueError):
|
||||
hash_api_token("value", algorithm="md5") # type: ignore[arg-type]
|
||||
|
||||
|
||||
@pytest.mark.parametrize("length", [8, 24, 48, 256])
|
||||
def test_generate_api_token_respects_length_bounds(length: int) -> None:
|
||||
"""Функция должна ограничивать длину токена безопасным диапазоном."""
|
||||
token = generate_api_token(length)
|
||||
clamped = max(24, min(length, 128))
|
||||
assert len(token) >= clamped
|
||||
# token_urlsafe расширяет строку, поэтому добавляем запас по длине.
|
||||
assert len(token) <= clamped * 2
|
||||
|
||||
|
||||
def test_generate_api_token_produces_random_values() -> None:
|
||||
"""Два последовательных вызова должны выдавать разные токены."""
|
||||
first = generate_api_token(48)
|
||||
second = generate_api_token(48)
|
||||
assert first != second
|
||||
@@ -0,0 +1,141 @@
|
||||
"""Базовые тесты для валидаторов из app.utils.validators."""
|
||||
|
||||
import pytest
|
||||
|
||||
from app.utils import validators
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"email,is_valid",
|
||||
[
|
||||
("user@example.com", True),
|
||||
("user.name+tag@sub.domain.ru", True),
|
||||
("plain-address", False),
|
||||
("missing-at.example.com", False),
|
||||
("user@invalid", False),
|
||||
],
|
||||
)
|
||||
def test_validate_email_handles_expected_patterns(email: str, is_valid: bool) -> None:
|
||||
"""Проверяем типичные корректные и некорректные адреса."""
|
||||
assert validators.validate_email(email) is is_valid
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"phone,is_valid",
|
||||
[
|
||||
("+71234567890", True),
|
||||
("+1 (202) 555-0101", True),
|
||||
("12345", True),
|
||||
("+0 123456789", False),
|
||||
("abc", False),
|
||||
],
|
||||
)
|
||||
def test_validate_phone_strips_formatting_and_checks_pattern(phone: str, is_valid: bool) -> None:
|
||||
"""Телефон должен соответствовать стандарту E.164 после очистки."""
|
||||
assert validators.validate_phone(phone) is is_valid
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"username,is_valid",
|
||||
[
|
||||
("@valid_name", True),
|
||||
("simpleUser", True),
|
||||
("bad", False),
|
||||
("toolongusername_more_than32_chars", False),
|
||||
("", False),
|
||||
],
|
||||
)
|
||||
def test_validate_telegram_username_enforces_length(username: str, is_valid: bool) -> None:
|
||||
"""Telegram-логин должен быть 5-32 символов и содержать допустимые символы."""
|
||||
assert validators.validate_telegram_username(username) is is_valid
|
||||
|
||||
|
||||
def test_validate_amount_returns_float_within_bounds() -> None:
|
||||
"""Числа должны конвертироваться с уважением к диапазону."""
|
||||
assert validators.validate_amount("10.5", min_amount=5, max_amount=20) == pytest.approx(10.5)
|
||||
assert validators.validate_amount("2", min_amount=5, max_amount=20) is None
|
||||
assert validators.validate_amount("abc", min_amount=0, max_amount=10) is None
|
||||
|
||||
|
||||
def test_validate_positive_integer_enforces_upper_bound() -> None:
|
||||
"""Положительное целое число выходит за пределы — возвращаем None."""
|
||||
assert validators.validate_positive_integer("12", max_value=20) == 12
|
||||
assert validators.validate_positive_integer("0", max_value=20) is None
|
||||
assert validators.validate_positive_integer("50", max_value=20) is None
|
||||
assert validators.validate_positive_integer("NaN") is None
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"value,expected",
|
||||
[
|
||||
("500", 500),
|
||||
("10gb", 10240),
|
||||
("2 TB", 2097152),
|
||||
("безлимит", 0),
|
||||
("invalid", None),
|
||||
],
|
||||
)
|
||||
def test_validate_traffic_amount_supports_units(value: str, expected: int | None) -> None:
|
||||
"""Валидатор трафика распознаёт разные единицы измерения и особые значения."""
|
||||
assert validators.validate_traffic_amount(value) == expected
|
||||
|
||||
|
||||
def test_validate_subscription_period_accepts_reasonable_range() -> None:
|
||||
"""Диапазон допустимой длительности от 1 до 3650 дней."""
|
||||
assert validators.validate_subscription_period("30") == 30
|
||||
assert validators.validate_subscription_period(0) is None
|
||||
assert validators.validate_subscription_period(4000) is None
|
||||
|
||||
|
||||
def test_validate_uuid_detects_standard_format() -> None:
|
||||
"""UUID должен соответствовать HEX шаблону версии 4/5."""
|
||||
sample = "123e4567-e89b-12d3-a456-426614174000"
|
||||
assert validators.validate_uuid(sample) is True
|
||||
assert validators.validate_uuid("not-a-uuid") is False
|
||||
|
||||
|
||||
def test_validate_url_recognises_https_links() -> None:
|
||||
"""Валидатор URL допускает http/https ссылки и отклоняет произвольные строки."""
|
||||
assert validators.validate_url("https://example.com/path?query=1")
|
||||
assert not validators.validate_url("ftp://example.com")
|
||||
|
||||
|
||||
def test_validate_html_tags_rejects_unknown_tags() -> None:
|
||||
"""Неизвестные HTML теги должны приводить к отказу."""
|
||||
ok, message = validators.validate_html_tags("<b>bold</b>")
|
||||
assert ok is True
|
||||
bad, error = validators.validate_html_tags("<marquee>run</marquee>")
|
||||
assert bad is False
|
||||
assert "Неподдерживаемый тег" in error
|
||||
|
||||
|
||||
def test_validate_html_structure_detects_wrong_nesting() -> None:
|
||||
"""Неправильная вложенность тегов должна сообщаться пользователю."""
|
||||
ok, message = validators.validate_html_structure("<b><i>text</i></b>")
|
||||
assert ok is True
|
||||
bad, error = validators.validate_html_structure("<b><i>text</b></i>")
|
||||
assert bad is False
|
||||
assert "Неправильная вложенность" in error
|
||||
|
||||
|
||||
def test_fix_html_tags_repairs_missing_quotes() -> None:
|
||||
"""Автоисправление должно добавлять кавычки у ссылок."""
|
||||
broken = '<a href=https://example.com>link</a>'
|
||||
fixed = validators.fix_html_tags(broken)
|
||||
assert 'href="https://example.com"' in fixed
|
||||
|
||||
|
||||
def test_validate_rules_content_detects_structure_error() -> None:
|
||||
"""При нарушении структуры должны вернуться сообщение и отсутствие подсказки."""
|
||||
is_valid, message, suggestion = validators.validate_rules_content("<b>text</i>")
|
||||
assert is_valid is False
|
||||
assert "Неправильная вложенность" in message
|
||||
assert suggestion is None
|
||||
|
||||
|
||||
def test_validate_rules_content_accepts_supported_markup() -> None:
|
||||
"""Корректный HTML должен проходить проверку без сообщений."""
|
||||
is_valid, message, suggestion = validators.validate_rules_content("<b>Добро пожаловать!</b>")
|
||||
assert is_valid is True
|
||||
assert message == ""
|
||||
assert suggestion is None
|
||||
Reference in New Issue
Block a user