austnv/remnawave-minishop
1
0
Fork 0
Code Issues 1 Pull Requests Activity

refactor: move cryptopay env-config into module

Browse Source
This commit is contained in:
3252a8
2026-05-18 16:32:19 +03:00
parent 1e93d25a40
commit cceba2e98e
12 changed files with 226 additions and 99 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/bot/app/web/admin_settings_manifest.py
-25
View File
@@ -279,31 +279,6 @@ SETTINGS_MANIFEST: List[SettingField] = [
SettingField("PLATEGA_FAILED_URL", "url", "payments", "Failed URL", subsection="Platega"),
*_payment_presentation_fields("PLATEGA_SBP", "Platega SBP", default_icon="CreditCard"),
*_payment_presentation_fields("PLATEGA_CRYPTO", "Platega Crypto", default_icon="Bitcoin"),
# CryptoPay
SettingField("CRYPTOPAY_ENABLED", "bool", "payments", "Включена", subsection="CryptoPay"),
SettingField(
"CRYPTOPAY_TOKEN", "string", "payments", "Token", subsection="CryptoPay", secret=True
),
SettingField(
"CRYPTOPAY_NETWORK",
"string",
"payments",
"Network",
"mainnet или testnet",
subsection="CryptoPay",
),
SettingField(
"CRYPTOPAY_CURRENCY_TYPE",
"string",
"payments",
"Currency type",
"fiat или crypto",
subsection="CryptoPay",
),
SettingField(
"CRYPTOPAY_ASSET", "string", "payments", "Asset", placeholder="RUB", subsection="CryptoPay"
),
*_payment_presentation_fields("CRYPTOPAY", "CryptoPay", default_icon="Bitcoin"),
# ─── Trial ─────────────────────────────────────────────────────
SettingField("TRIAL_ENABLED", "bool", "trial", "Триал включён"),
SettingField("TRIAL_DURATION_DAYS", "int", "trial", "Длительность триала (дней)", min=0),
backend/bot/payment_providers/base.py
+14
View File
@@ -1,11 +1,25 @@
from __future__ import annotations
import os
from dataclasses import dataclass, field
from typing import Any, Awaitable, Callable, List, Mapping, Optional, Sequence, Type
from pydantic_settings import BaseSettings, SettingsConfigDict
def provider_env_file() -> Optional[str]:
"""Resolve the env file every provider config should read.
Tests set ``PROVIDER_ENV_FILE=""`` via conftest so per-provider
BaseSettings models don't pick up real credentials from the project's
.env. Production reads from ``.env`` as usual.
"""
value = os.environ.get("PROVIDER_ENV_FILE")
if value is None:
return ".env"
return value or None
class ProviderEnvConfig(BaseSettings):
"""Base class for per-provider env-config models.
backend/bot/payment_providers/cryptopay.py
+116 -27
View File
@@ -8,6 +8,8 @@ from aiocryptopay import AioCryptoPay, Networks
from aiocryptopay.models.update import Update
from aiogram import Bot, F, Router, types
from aiohttp import web
from pydantic import Field, field_validator
from pydantic_settings import SettingsConfigDict
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.orm import sessionmaker
@@ -17,7 +19,14 @@ from bot.services.subscription_service import SubscriptionService
from config.settings import Settings
from db.dal import payment_dal
from .base import PaymentProviderSpec, ServiceFactoryContext, WebAppPaymentContext
from .base import (
PaymentProviderSpec,
ProviderEnvConfig,
ProviderManifestField,
ServiceFactoryContext,
WebAppPaymentContext,
provider_env_file,
)
from .shared import (
PaymentSuccessRequest,
describe_payment,
@@ -38,13 +47,54 @@ logger = logging.getLogger(__name__)
_LOG = "cryptopay"
class CryptoPayConfig(ProviderEnvConfig):
model_config = SettingsConfigDict(
env_file=provider_env_file(),
env_file_encoding="utf-8",
env_prefix="CRYPTOPAY_",
extra="ignore",
)
ENABLED: bool = Field(default=True)
TOKEN: Optional[str] = None
NETWORK: str = Field(default="mainnet")
CURRENCY_TYPE: str = Field(default="fiat")
ASSET: str = Field(default="RUB")
@field_validator("TOKEN", mode="before")
@classmethod
def _strip_optional(cls, v):
if isinstance(v, str) and not v.strip():
return None
return v
@property
def webhook_path(self) -> str:
return "/webhook/cryptopay"
class CryptoPayPresentation(ProviderEnvConfig):
model_config = SettingsConfigDict(
env_file=provider_env_file(),
env_file_encoding="utf-8",
env_prefix="PAYMENT_CRYPTOPAY_",
extra="ignore",
)
WEBAPP_LABEL_RU: Optional[str] = None
WEBAPP_LABEL_EN: Optional[str] = None
WEBAPP_ICON: Optional[str] = None
TELEGRAM_LABEL_RU: Optional[str] = None
TELEGRAM_LABEL_EN: Optional[str] = None
TELEGRAM_EMOJI: Optional[str] = None
class CryptoPayService:
def __init__(
self,
token: Optional[str],
network: str,
bot: Bot,
settings: Settings,
config: CryptoPayConfig,
i18n: JsonI18n,
async_session_factory: sessionmaker,
subscription_service: SubscriptionService,
@@ -52,14 +102,15 @@ class CryptoPayService:
):
self.bot = bot
self.settings = settings
self.config = config
self.i18n = i18n
self.async_session_factory = async_session_factory
self.subscription_service = subscription_service
self.referral_service = referral_service
self.token = token
if token:
net = Networks.TEST_NET if str(network).lower() == "testnet" else Networks.MAIN_NET
self.client = AioCryptoPay(token=token, network=net)
self.token = config.TOKEN
if self.token:
net = Networks.TEST_NET if str(config.NETWORK).lower() == "testnet" else Networks.MAIN_NET
self.client = AioCryptoPay(token=self.token, network=net)
self.client.register_pay_handler(self._invoice_paid_handler)
self.configured = True
else:
@@ -97,7 +148,7 @@ class CryptoPayService:
{
"user_id": user_id,
"amount": float(amount),
"currency": self.settings.CRYPTOPAY_ASSET,
"currency": self.config.ASSET,
"status": "pending_cryptopay",
"description": description,
"subscription_duration_months": (
@@ -127,13 +178,9 @@ class CryptoPayService:
try:
invoice = await self.client.create_invoice(
amount=amount,
currency_type=self.settings.CRYPTOPAY_CURRENCY_TYPE,
fiat=self.settings.CRYPTOPAY_ASSET
if self.settings.CRYPTOPAY_CURRENCY_TYPE == "fiat"
else None,
asset=self.settings.CRYPTOPAY_ASSET
if self.settings.CRYPTOPAY_CURRENCY_TYPE == "crypto"
else None,
currency_type=self.config.CURRENCY_TYPE,
fiat=self.config.ASSET if self.config.CURRENCY_TYPE == "fiat" else None,
asset=self.config.ASSET if self.config.CURRENCY_TYPE == "crypto" else None,
description=description,
payload=payload,
)
@@ -284,8 +331,7 @@ async def pay_crypto_callback_handler(
return
if (
not settings.CRYPTOPAY_ENABLED
or not cryptopay_service
not cryptopay_service
or not getattr(cryptopay_service, "configured", False)
):
await notify_service_unavailable(callback, translator)
@@ -324,15 +370,16 @@ async def pay_crypto_callback_handler(
def create_service(ctx: ServiceFactoryContext) -> CryptoPayService:
bundle = ctx.config_for("cryptopay_service")
config = bundle.config if bundle and isinstance(bundle.config, CryptoPayConfig) else CryptoPayConfig()
return CryptoPayService(
ctx.settings.CRYPTOPAY_TOKEN,
ctx.settings.CRYPTOPAY_NETWORK,
ctx.bot,
ctx.settings,
ctx.i18n,
ctx.async_session_factory,
ctx.subscription_service,
ctx.referral_service,
bot=ctx.bot,
settings=ctx.settings,
config=config,
i18n=ctx.i18n,
async_session_factory=ctx.async_session_factory,
subscription_service=ctx.subscription_service,
referral_service=ctx.referral_service,
)
@@ -354,6 +401,45 @@ async def create_webapp_payment(ctx: WebAppPaymentContext) -> web.Response:
return payment_link_response(payment_url=url, payment_id=None)
_PRESENTATION_MANIFEST = tuple(
ProviderManifestField(
key=key, type=type_, label=label, description=description,
placeholder=placeholder, subsection="CryptoPay",
target="presentation", attr=attr,
)
for key, type_, label, description, placeholder, attr in (
("PAYMENT_CRYPTOPAY_WEBAPP_LABEL_RU", "string", "WebApp button text (RU)",
"Custom Russian text shown in the Web App payment method button.", "", "WEBAPP_LABEL_RU"),
("PAYMENT_CRYPTOPAY_WEBAPP_LABEL_EN", "string", "WebApp button text (EN)",
"Custom English text shown in the Web App payment method button.", "", "WEBAPP_LABEL_EN"),
("PAYMENT_CRYPTOPAY_WEBAPP_ICON", "icon", "WebApp button icon",
"Lucide icon name rendered inside the Web App payment method button.",
"Bitcoin", "WEBAPP_ICON"),
("PAYMENT_CRYPTOPAY_TELEGRAM_LABEL_RU", "string", "Telegram button text (RU)",
"Custom Russian text shown in Telegram bot payment buttons.", "", "TELEGRAM_LABEL_RU"),
("PAYMENT_CRYPTOPAY_TELEGRAM_LABEL_EN", "string", "Telegram button text (EN)",
"Custom English text shown in Telegram bot payment buttons.", "", "TELEGRAM_LABEL_EN"),
("PAYMENT_CRYPTOPAY_TELEGRAM_EMOJI", "string", "Telegram button emoji",
"Emoji prepended to the Telegram bot payment button when customized.",
"₿", "TELEGRAM_EMOJI"),
)
)
_CONFIG_MANIFEST = (
ProviderManifestField("CRYPTOPAY_ENABLED", "bool", "Включена",
subsection="CryptoPay", attr="ENABLED"),
ProviderManifestField("CRYPTOPAY_TOKEN", "string", "Token",
subsection="CryptoPay", secret=True, attr="TOKEN"),
ProviderManifestField("CRYPTOPAY_NETWORK", "string", "Network",
placeholder="mainnet", subsection="CryptoPay", attr="NETWORK"),
ProviderManifestField("CRYPTOPAY_CURRENCY_TYPE", "string", "Currency type",
description="fiat or crypto.",
placeholder="fiat", subsection="CryptoPay", attr="CURRENCY_TYPE"),
ProviderManifestField("CRYPTOPAY_ASSET", "string", "Asset",
placeholder="RUB", subsection="CryptoPay", attr="ASSET"),
)
SPEC = PaymentProviderSpec(
id="cryptopay",
provider_key="cryptopay",
@@ -363,14 +449,17 @@ SPEC = PaymentProviderSpec(
webapp_icon="Bitcoin",
telegram_labels={"ru": "CryptoBot", "en": "CryptoBot"},
pending_status="pending_cryptopay",
enabled=lambda settings: settings.CRYPTOPAY_ENABLED,
enabled=lambda config: bool(getattr(config, "ENABLED", False)),
service_key="cryptopay_service",
callback_prefix="pay_crypto",
router=router,
create_service=create_service,
webhook_path=lambda settings: settings.cryptopay_webhook_path,
webhook_path=lambda source: "/webhook/cryptopay",
webhook_route=cryptopay_webhook_route,
create_webapp_payment=create_webapp_payment,
emoji="₿",
telegram_emoji="₿",
config_class=CryptoPayConfig,
presentation_class=CryptoPayPresentation,
manifest_fields=_CONFIG_MANIFEST + _PRESENTATION_MANIFEST,
)
backend/bot/payment_providers/freekassa.py
+3 -2
View File
@@ -28,6 +28,7 @@ from .base import (
ProviderManifestField,
ServiceFactoryContext,
WebAppPaymentContext,
provider_env_file,
)
from .shared import (
HttpClientMixin,
@@ -56,7 +57,7 @@ _LOG = "freekassa"
class FreeKassaConfig(ProviderEnvConfig):
model_config = SettingsConfigDict(
env_file=".env",
env_file=provider_env_file(),
env_file_encoding="utf-8",
env_prefix="FREEKASSA_",
extra="ignore",
@@ -104,7 +105,7 @@ class FreeKassaConfig(ProviderEnvConfig):
class FreeKassaPresentation(ProviderEnvConfig):
model_config = SettingsConfigDict(
env_file=".env",
env_file=provider_env_file(),
env_file_encoding="utf-8",
env_prefix="PAYMENT_FREEKASSA_",
extra="ignore",
backend/bot/payment_providers/heleket.py
+3 -2
View File
@@ -25,6 +25,7 @@ from .base import (
ProviderManifestField,
ServiceFactoryContext,
WebAppPaymentContext,
provider_env_file,
)
from .shared import (
HttpClientMixin,
@@ -60,7 +61,7 @@ class HeleketConfig(ProviderEnvConfig):
"""All Heleket-specific env vars. Lives inside the provider module."""
model_config = SettingsConfigDict(
env_file=".env",
env_file=provider_env_file(),
env_file_encoding="utf-8",
env_prefix="HELEKET_",
extra="ignore",
@@ -118,7 +119,7 @@ class HeleketPresentation(ProviderEnvConfig):
"""Admin-tunable button text/icon overrides for Heleket."""
model_config = SettingsConfigDict(
env_file=".env",
env_file=provider_env_file(),
env_file_encoding="utf-8",
env_prefix="PAYMENT_HELEKET_",
extra="ignore",
backend/bot/payment_providers/registry.py
+7 -2
View File
@@ -50,6 +50,11 @@ def build_provider_configs() -> Dict[str, ProviderConfigBundle]:
Specs with neither ``config_class`` nor ``presentation_class`` are skipped.
The result is cached as the process-wide bundle.
"""
from .base import provider_env_file
env_file = provider_env_file()
init_kwargs = {"_env_file": env_file} if env_file is not None else {"_env_file": None}
bundles: Dict[str, ProviderConfigBundle] = {}
seen: set[str] = set()
for spec in PAYMENT_PROVIDER_SPECS:
@@ -59,8 +64,8 @@ def build_provider_configs() -> Dict[str, ProviderConfigBundle]:
continue
seen.add(spec.service_key)
bundles[spec.service_key] = ProviderConfigBundle(
config=spec.config_class() if spec.config_class else None,
presentation=spec.presentation_class() if spec.presentation_class else None,
config=spec.config_class(**init_kwargs) if spec.config_class else None,
presentation=spec.presentation_class(**init_kwargs) if spec.presentation_class else None,
)
_provider_configs.clear()
_provider_configs.update(bundles)
backend/bot/payment_providers/severpay.py
+3 -2
View File
@@ -24,6 +24,7 @@ from .base import (
ProviderManifestField,
ServiceFactoryContext,
WebAppPaymentContext,
provider_env_file,
)
from .shared import (
HttpClientMixin,
@@ -53,7 +54,7 @@ _LOG = "severpay"
class SeverPayConfig(ProviderEnvConfig):
model_config = SettingsConfigDict(
env_file=".env",
env_file=provider_env_file(),
env_file_encoding="utf-8",
env_prefix="SEVERPAY_",
extra="ignore",
@@ -89,7 +90,7 @@ class SeverPayConfig(ProviderEnvConfig):
class SeverPayPresentation(ProviderEnvConfig):
model_config = SettingsConfigDict(
env_file=".env",
env_file=provider_env_file(),
env_file_encoding="utf-8",
env_prefix="PAYMENT_SEVERPAY_",
extra="ignore",
backend/bot/payment_providers/wata.py
+3 -2
View File
@@ -27,6 +27,7 @@ from .base import (
ProviderManifestField,
ServiceFactoryContext,
WebAppPaymentContext,
provider_env_file,
)
from .shared import (
HttpClientMixin,
@@ -58,7 +59,7 @@ _LOG = "wata"
class WataConfig(ProviderEnvConfig):
model_config = SettingsConfigDict(
env_file=".env",
env_file=provider_env_file(),
env_file_encoding="utf-8",
env_prefix="WATA_",
extra="ignore",
@@ -103,7 +104,7 @@ class WataConfig(ProviderEnvConfig):
class WataPresentation(ProviderEnvConfig):
model_config = SettingsConfigDict(
env_file=".env",
env_file=provider_env_file(),
env_file_encoding="utf-8",
env_prefix="PAYMENT_WATA_",
extra="ignore",
backend/config/settings.py
-33
View File
@@ -49,11 +49,6 @@ class PaymentSettings(BaseModel):
platega_crypto_method: int
platega_return_url: Optional[str]
platega_failed_url: Optional[str]
cryptopay_enabled: bool
cryptopay_token: Optional[str]
cryptopay_network: str
cryptopay_currency_type: str
cryptopay_asset: str
class EmailSettings(BaseModel):
@@ -187,11 +182,6 @@ class Settings(BaseSettings):
description="Comma-separated list of reverse proxy IPs or CIDRs trusted to forward X-Forwarded-For.", # noqa: E501
)
CRYPTOPAY_TOKEN: Optional[str] = None
CRYPTOPAY_NETWORK: str = Field(default="mainnet")
CRYPTOPAY_CURRENCY_TYPE: str = Field(default="fiat")
CRYPTOPAY_ASSET: str = Field(default="RUB")
CRYPTOPAY_ENABLED: bool = Field(default=True)
PLATEGA_ENABLED: bool = Field(default=False)
PLATEGA_BASE_URL: str = Field(default="https://app.platega.io")
PLATEGA_MERCHANT_ID: Optional[str] = None
@@ -249,12 +239,6 @@ class Settings(BaseSettings):
PAYMENT_STARS_TELEGRAM_LABEL_RU: Optional[str] = None
PAYMENT_STARS_TELEGRAM_LABEL_EN: Optional[str] = None
PAYMENT_STARS_TELEGRAM_EMOJI: Optional[str] = None
PAYMENT_CRYPTOPAY_WEBAPP_LABEL_RU: Optional[str] = None
PAYMENT_CRYPTOPAY_WEBAPP_LABEL_EN: Optional[str] = None
PAYMENT_CRYPTOPAY_WEBAPP_ICON: Optional[str] = None
PAYMENT_CRYPTOPAY_TELEGRAM_LABEL_RU: Optional[str] = None
PAYMENT_CRYPTOPAY_TELEGRAM_LABEL_EN: Optional[str] = None
PAYMENT_CRYPTOPAY_TELEGRAM_EMOJI: Optional[str] = None
MONTH_1_ENABLED: bool = Field(default=True, alias="1_MONTH_ENABLED")
MONTH_3_ENABLED: bool = Field(default=True, alias="3_MONTHS_ENABLED")
@@ -513,11 +497,6 @@ class Settings(BaseSettings):
platega_crypto_method=self.PLATEGA_CRYPTO_METHOD,
platega_return_url=self.PLATEGA_RETURN_URL,
platega_failed_url=self.PLATEGA_FAILED_URL,
cryptopay_enabled=self.CRYPTOPAY_ENABLED,
cryptopay_token=self.CRYPTOPAY_TOKEN,
cryptopay_network=self.CRYPTOPAY_NETWORK,
cryptopay_currency_type=self.CRYPTOPAY_CURRENCY_TYPE,
cryptopay_asset=self.CRYPTOPAY_ASSET,
)
@computed_field
@@ -656,18 +635,6 @@ class Settings(BaseSettings):
return f"{base.rstrip('/')}{self.panel_webhook_path}"
return None
@computed_field
@property
def cryptopay_webhook_path(self) -> str:
return "/webhook/cryptopay"
@computed_field
@property
def cryptopay_full_webhook_url(self) -> Optional[str]:
base = self.WEBHOOK_BASE_URL
if base:
return f"{base.rstrip('/')}{self.cryptopay_webhook_path}"
return None
@computed_field
tests/conftest.py
+44
View File
@@ -0,0 +1,44 @@
"""Test fixtures that isolate the suite from the developer's local .env.
Provider configs now declare their own ``BaseSettings`` with ``env_file=".env"``,
so a developer who runs ``pytest`` from a project that has real credentials in
``.env`` would otherwise see provider services try to connect (e.g. CryptoPay
spinning up an aiohttp session in __init__).
We set ``PROVIDER_ENV_FILE=""`` (consumed by each provider's
``ProviderEnvConfig.model_config["env_file"]`` factory) and strip real
provider env vars from the test process.
"""
from __future__ import annotations
import os
import pytest
@pytest.fixture(autouse=True)
def _isolate_provider_env(monkeypatch):
monkeypatch.setenv("PROVIDER_ENV_FILE", "")
for key in list(os.environ.keys()):
if any(
key.startswith(prefix)
for prefix in (
"FREEKASSA_",
"PLATEGA_",
"SEVERPAY_",
"WATA_",
"HELEKET_",
"CRYPTOPAY_",
"YOOKASSA_",
"PAYMENT_FREEKASSA_",
"PAYMENT_PLATEGA_",
"PAYMENT_SEVERPAY_",
"PAYMENT_WATA_",
"PAYMENT_HELEKET_",
"PAYMENT_CRYPTOPAY_",
"PAYMENT_YOOKASSA_",
"PAYMENT_STARS_",
)
):
monkeypatch.delenv(key, raising=False)
tests/test_build_services_wiring.py
+18 -1
View File
@@ -13,11 +13,12 @@ silently swallows the wiring step.
"""
import json
import os
import tempfile
import unittest
from pathlib import Path
from typing import Any
from unittest.mock import MagicMock
from unittest.mock import MagicMock, patch
from bot.app.factories.build_services import build_core_services
from bot.payment_providers.yookassa import YooKassaService
@@ -26,6 +27,22 @@ from bot.services.subscription_service import SubscriptionService
from config.settings import Settings
# Strip all provider env so per-provider BaseSettings models don't pick up
# real credentials from the local .env file during tests.
_PROVIDER_ENV_PREFIXES = (
"FREEKASSA_", "PLATEGA_", "SEVERPAY_", "WATA_", "HELEKET_",
"CRYPTOPAY_", "YOOKASSA_", "STARS_",
)
def _clean_env() -> dict[str, str]:
return {
k: v for k, v in os.environ.items()
if not any(k.startswith(p) for p in _PROVIDER_ENV_PREFIXES)
and not k.startswith("PAYMENT_")
}
def _make_settings(tmpdir: str, **overrides: Any) -> Settings:
config_path = Path(tmpdir) / "tariffs.json"
config_path.write_text(
tests/test_webapp_assets.py
+15 -3
View File
@@ -332,15 +332,27 @@ class WebAppAssetTests(unittest.IsolatedAsyncioTestCase):
self.assertEqual(plans[1]["stars_price"], 2500)
def test_serialize_payment_methods_respects_runtime_provider_toggles(self):
# Provider toggles now live in per-provider BaseSettings models. Disable
# all providers by giving each one an empty bundle (default ENABLED is
# False for everything except cryptopay/yookassa, so override those too).
from bot.payment_providers import (
build_provider_configs,
current_provider_configs,
)
build_provider_configs()
configs = current_provider_configs()
for service_key in ("cryptopay_service", "yookassa_service"):
bundle = configs.get(service_key)
if bundle and bundle.config is not None and hasattr(bundle.config, "ENABLED"):
bundle.config.ENABLED = False
settings = Settings(
_env_file=None,
BOT_TOKEN="token",
POSTGRES_USER="app_user",
POSTGRES_PASSWORD="app_password",
TARIFFS_CONFIG_PATH="missing-tariffs.json",
CRYPTOPAY_ENABLED=False,
FREEKASSA_ENABLED=False,
SEVERPAY_ENABLED=False,
YOOKASSA_ENABLED=False,
PLATEGA_ENABLED=False,
STARS_ENABLED=False,