VPN/remnawave-bedolaga-telegram-bot
1
0
Fork 0
mirror of https://github.com/BEDOLAGA-DEV/remnawave-bedolaga-telegram-bot.git synced 2026-01-19 19:32:10 +00:00
Code Issues Packages Projects Releases Wiki Activity

Revert "Revert "feat: auto purchase subscription after top-up""

Browse Source
This commit is contained in:
Egor
2025-10-26 18:27:01 +03:00
committed by GitHub
parent a56e14c736
commit edf4132c2d
13 changed files with 681 additions and 13 deletions
Download Patch File Download Diff File Expand all files Collapse all files

13
app/config.py
View File

@@ -187,7 +187,9 @@ class Settings(BaseSettings):
DISABLE_TOPUP_BUTTONS: bool = False
PAYMENT_VERIFICATION_AUTO_CHECK_ENABLED: bool = False
PAYMENT_VERIFICATION_AUTO_CHECK_INTERVAL_MINUTES: int = 10
AUTO_PURCHASE_AFTER_TOPUP_ENABLED: bool = False
# Настройки простой покупки
SIMPLE_SUBSCRIPTION_ENABLED: bool = False
SIMPLE_SUBSCRIPTION_PERIOD_DAYS: int = 30
@@ -626,6 +628,15 @@ class Settings(BaseSettings):
return normalized in {"1", "true", "yes", "on"}
return bool(value)
def is_auto_purchase_after_topup_enabled(self) -> bool:
value = getattr(self, "AUTO_PURCHASE_AFTER_TOPUP_ENABLED", False)
if isinstance(value, str):
normalized = value.strip().lower()
return normalized in {"1", "true", "yes", "on"}
return bool(value)
def get_available_languages(self) -> List[str]:
try:

2
app/localization/locales/en.json
View File

@@ -780,6 +780,8 @@
"BALANCE_SUPPORT_REQUEST": "🛠️ Request via support",
"BALANCE_TOPUP": "💳 Top up balance",
"BALANCE_TOPUP_CART_REMINDER_DETAILED": "\\n💡 <b>Balance top-up required</b>\\n\\nYour cart contains items totaling {total_amount}, but your current balance is insufficient.\\n\\n💳 <b>Top up your balance</b> to complete the purchase.\\n\\nChoose a top-up method:",
"AUTO_PURCHASE_SUBSCRIPTION_SUCCESS": "✅ Your {period} subscription was purchased automatically after topping up your balance.",
"AUTO_PURCHASE_SUBSCRIPTION_HINT": "Open the ‘My subscription’ section to access your link and setup instructions.",
"BALANCE_TOP_UP": "💳 Top up",
"BLOCK_BY_TIME": "⏳ Temporary block",
"BLOCK_FOREVER": "🚫 Block permanently",

2
app/localization/locales/ru.json
View File

@@ -780,6 +780,8 @@
"BALANCE_SUPPORT_REQUEST": "🛠️ Запрос через поддержку",
"BALANCE_TOPUP": "💳 Пополнить баланс",
"BALANCE_TOPUP_CART_REMINDER_DETAILED": "\n💡 <b>Требуется пополнение баланса</b>\n\nВ вашей корзине находятся товары на общую сумму {total_amount}, но на балансе недостаточно средств.\n\n💳 <b>Пополните баланс</b>, чтобы завершить покупку.\n\nВыберите способ пополнения:",
"AUTO_PURCHASE_SUBSCRIPTION_SUCCESS": "✅ Подписка на {period} автоматически оформлена после пополнения баланса.",
"AUTO_PURCHASE_SUBSCRIPTION_HINT": "Перейдите в раздел «Моя подписка», чтобы получить ссылку и инструкции.",
"BALANCE_TOP_UP": "💳 Пополнить",
"BLOCK_BY_TIME": "⏳ Блокировка по времени",
"BLOCK_FOREVER": "🚫 Заблокировать",

30
app/services/payment/cryptobot.py
View File

@@ -11,6 +11,9 @@ from sqlalchemy.ext.asyncio import AsyncSession
from app.config import settings
from app.database.models import PaymentMethod, TransactionType
from app.services.subscription_auto_purchase_service import (
auto_purchase_saved_cart_after_topup,
)
from app.utils.currency_converter import currency_converter
from app.utils.user_utils import format_referrer_info
@@ -294,9 +297,29 @@ class CryptoBotPaymentMixin:
try:
from app.services.user_cart_service import user_cart_service
from aiogram import types
has_saved_cart = await user_cart_service.has_user_cart(user.id)
auto_purchase_success = False
if has_saved_cart:
try:
auto_purchase_success = await auto_purchase_saved_cart_after_topup(
db,
user,
bot=getattr(self, "bot", None),
)
except Exception as auto_error:
logger.error(
"Ошибка автоматической покупки подписки для пользователя %s: %s",
user.id,
auto_error,
exc_info=True,
)
if auto_purchase_success:
has_saved_cart = False
if has_saved_cart and getattr(self, "bot", None):
# Если у пользователя есть сохраненная корзина,
# Если у пользователя есть сохраненная корзина,
# отправляем ему уведомление с кнопкой вернуться к оформлению
from app.localization.texts import get_texts
@@ -326,7 +349,10 @@ class CryptoBotPaymentMixin:
text=f"✅ Баланс пополнен на {settings.format_price(payment.amount_kopeks)}!\n\n{cart_message}",
reply_markup=keyboard
)
logger.info(f"Отправлено уведомление с кнопкой возврата к оформлению подписки пользователю {user.id}")
logger.info(
"Отправлено уведомление с кнопкой возврата к оформлению подписки пользователю %s",
user.id,
)
except Exception as e:
logger.error(f"Ошибка при работе с сохраненной корзиной для пользователя {user.id}: {e}", exc_info=True)

30
app/services/payment/mulenpay.py
View File

@@ -11,6 +11,9 @@ from sqlalchemy.ext.asyncio import AsyncSession
from app.config import settings
from app.database.models import PaymentMethod, TransactionType
from app.services.subscription_auto_purchase_service import (
auto_purchase_saved_cart_after_topup,
)
from app.utils.user_utils import format_referrer_info
logger = logging.getLogger(__name__)
@@ -325,9 +328,29 @@ class MulenPayPaymentMixin:
try:
from app.services.user_cart_service import user_cart_service
from aiogram import types
has_saved_cart = await user_cart_service.has_user_cart(user.id)
auto_purchase_success = False
if has_saved_cart:
try:
auto_purchase_success = await auto_purchase_saved_cart_after_topup(
db,
user,
bot=getattr(self, "bot", None),
)
except Exception as auto_error:
logger.error(
"Ошибка автоматической покупки подписки для пользователя %s: %s",
user.id,
auto_error,
exc_info=True,
)
if auto_purchase_success:
has_saved_cart = False
if has_saved_cart and getattr(self, "bot", None):
# Если у пользователя есть сохраненная корзина,
# Если у пользователя есть сохраненная корзина,
# отправляем ему уведомление с кнопкой вернуться к оформлению
from app.localization.texts import get_texts
@@ -359,7 +382,10 @@ class MulenPayPaymentMixin:
text=f"✅ Баланс пополнен на {settings.format_price(payment.amount_kopeks)}!\n\n{cart_message}",
reply_markup=keyboard
)
logger.info(f"Отправлено уведомление с кнопкой возврата к оформлению подписки пользователю {user.id}")
logger.info(
"Отправлено уведомление с кнопкой возврата к оформлению подписки пользователю %s",
user.id,
)
except Exception as e:
logger.error(f"Ошибка при работе с сохраненной корзиной для пользователя {user.id}: {e}", exc_info=True)

27
app/services/payment/pal24.py
View File

@@ -13,6 +13,9 @@ from sqlalchemy.ext.asyncio import AsyncSession
from app.config import settings
from app.database.models import PaymentMethod, TransactionType
from app.services.pal24_service import Pal24APIError
from app.services.subscription_auto_purchase_service import (
auto_purchase_saved_cart_after_topup,
)
from app.utils.user_utils import format_referrer_info
logger = logging.getLogger(__name__)
@@ -434,6 +437,25 @@ class Pal24PaymentMixin:
from aiogram import types
has_saved_cart = await user_cart_service.has_user_cart(user.id)
auto_purchase_success = False
if has_saved_cart:
try:
auto_purchase_success = await auto_purchase_saved_cart_after_topup(
db,
user,
bot=getattr(self, "bot", None),
)
except Exception as auto_error:
logger.error(
"Ошибка автоматической покупки подписки для пользователя %s: %s",
user.id,
auto_error,
exc_info=True,
)
if auto_purchase_success:
has_saved_cart = False
if has_saved_cart and getattr(self, "bot", None):
from app.localization.texts import get_texts
@@ -475,6 +497,11 @@ class Pal24PaymentMixin:
"Отправлено уведомление с кнопкой возврата к оформлению подписки пользователю %s",
user.id,
)
else:
logger.info(
"У пользователя %s нет сохраненной корзины или автопокупка выполнена",
user.id,
)
except Exception as error:
logger.error(
"Ошибка при работе с сохраненной корзиной для пользователя %s: %s",

24
app/services/payment/stars.py
View File

@@ -19,6 +19,9 @@ from app.database.crud.transaction import create_transaction
from app.database.crud.user import get_user_by_id
from app.database.models import PaymentMethod, TransactionType
from app.external.telegram_stars import TelegramStarsService
from app.services.subscription_auto_purchase_service import (
auto_purchase_saved_cart_after_topup,
)
from app.utils.user_utils import format_referrer_info
logger = logging.getLogger(__name__)
@@ -242,8 +245,27 @@ class TelegramStarsMixin:
from app.services.user_cart_service import user_cart_service
from aiogram import types
has_saved_cart = await user_cart_service.has_user_cart(user.id)
auto_purchase_success = False
if has_saved_cart:
try:
auto_purchase_success = await auto_purchase_saved_cart_after_topup(
db,
user,
bot=getattr(self, "bot", None),
)
except Exception as auto_error:
logger.error(
"Ошибка автоматической покупки подписки для пользователя %s: %s",
user.id,
auto_error,
exc_info=True,
)
if auto_purchase_success:
has_saved_cart = False
if has_saved_cart and getattr(self, "bot", None):
# Если у пользователя есть сохраненная корзина,
# Если у пользователя есть сохраненная корзина,
# отправляем ему уведомление с кнопкой вернуться к оформлению
from app.localization.texts import get_texts

22
app/services/payment/wata.py
View File

@@ -12,6 +12,9 @@ from sqlalchemy.ext.asyncio import AsyncSession
from app.config import settings
from app.database.models import PaymentMethod, TransactionType
from app.services.subscription_auto_purchase_service import (
auto_purchase_saved_cart_after_topup,
)
from app.services.wata_service import WataAPIError, WataService
from app.utils.user_utils import format_referrer_info
@@ -524,6 +527,25 @@ class WataPaymentMixin:
from aiogram import types
has_saved_cart = await user_cart_service.has_user_cart(user.id)
auto_purchase_success = False
if has_saved_cart:
try:
auto_purchase_success = await auto_purchase_saved_cart_after_topup(
db,
user,
bot=getattr(self, "bot", None),
)
except Exception as auto_error:
logger.error(
"Ошибка автоматической покупки подписки для пользователя %s: %s",
user.id,
auto_error,
exc_info=True,
)
if auto_purchase_success:
has_saved_cart = False
if has_saved_cart and getattr(self, "bot", None):
from app.localization.texts import get_texts

34
app/services/payment/yookassa.py
View File

@@ -16,6 +16,9 @@ from sqlalchemy.ext.asyncio import AsyncSession
from app.config import settings
from app.database.models import PaymentMethod, TransactionType
from app.services.subscription_auto_purchase_service import (
auto_purchase_saved_cart_after_topup,
)
from app.utils.user_utils import format_referrer_info
logger = logging.getLogger(__name__)
@@ -465,7 +468,31 @@ class YooKassaPaymentMixin:
from app.services.user_cart_service import user_cart_service
try:
has_saved_cart = await user_cart_service.has_user_cart(user.id)
logger.info(f"Результат проверки корзины для пользователя {user.id}: {has_saved_cart}")
logger.info(
"Результат проверки корзины для пользователя %s: %s",
user.id,
has_saved_cart,
)
auto_purchase_success = False
if has_saved_cart:
try:
auto_purchase_success = await auto_purchase_saved_cart_after_topup(
db,
user,
bot=getattr(self, "bot", None),
)
except Exception as auto_error:
logger.error(
"Ошибка автоматической покупки подписки для пользователя %s: %s",
user.id,
auto_error,
exc_info=True,
)
if auto_purchase_success:
has_saved_cart = False
if has_saved_cart and getattr(self, "bot", None):
# Если у пользователя есть сохраненная корзина,
# отправляем ему уведомление с кнопкой вернуться к оформлению
@@ -510,7 +537,10 @@ class YooKassaPaymentMixin:
f"Отправлено уведомление с кнопкой возврата к оформлению подписки пользователю {user.id}"
)
else:
logger.info(f"У пользователя {user.id} нет сохраненной корзины или бот недоступен")
logger.info(
"У пользователя %s нет сохраненной корзины, бот недоступен или покупка уже выполнена",
user.id,
)
except Exception as e:
logger.error(
f"Критическая ошибка при работе с сохраненной корзиной для пользователя {user.id}: {e}",

275
app/services/subscription_auto_purchase_service.py Normal file
View File

@@ -0,0 +1,275 @@
"""Automatic subscription purchase from a saved cart after balance top-up."""
from __future__ import annotations
import logging
from dataclasses import dataclass
from typing import Optional
from aiogram import Bot
from aiogram.types import InlineKeyboardButton, InlineKeyboardMarkup
from sqlalchemy.ext.asyncio import AsyncSession
from app.config import settings
from app.database.models import User
from app.localization.texts import get_texts
from app.services.admin_notification_service import AdminNotificationService
from app.services.subscription_checkout_service import clear_subscription_checkout_draft
from app.services.subscription_purchase_service import (
MiniAppSubscriptionPurchaseService,
PurchaseOptionsContext,
PurchasePricingResult,
PurchaseSelection,
PurchaseValidationError,
PurchaseBalanceError,
SubscriptionPurchaseService,
)
from app.services.user_cart_service import user_cart_service
from app.utils.pricing_utils import format_period_description
logger = logging.getLogger(__name__)
@dataclass(slots=True)
class AutoPurchaseContext:
"""Aggregated data prepared for automatic checkout processing."""
context: PurchaseOptionsContext
pricing: PurchasePricingResult
selection: PurchaseSelection
async def _prepare_auto_purchase(
db: AsyncSession,
user: User,
cart_data: dict,
) -> Optional[AutoPurchaseContext]:
"""Builds purchase context and pricing for a saved cart."""
period_days = int(cart_data.get("period_days") or 0)
if period_days <= 0:
logger.info(
"🔁 Автопокупка: у пользователя %s нет корректного периода в сохранённой корзине",
user.telegram_id,
)
return None
miniapp_service = MiniAppSubscriptionPurchaseService()
context = await miniapp_service.build_options(db, user)
period_config = context.period_map.get(f"days:{period_days}")
if not period_config:
logger.warning(
"🔁 Автопокупка: период %s дней недоступен для пользователя %s",
period_days,
user.telegram_id,
)
return None
traffic_value = cart_data.get("traffic_gb")
if traffic_value is None:
traffic_value = (
period_config.traffic.current_value
if period_config.traffic.current_value is not None
else period_config.traffic.default_value or 0
)
else:
traffic_value = int(traffic_value)
devices = int(cart_data.get("devices") or period_config.devices.current or 1)
servers = list(cart_data.get("countries") or [])
if not servers:
servers = list(period_config.servers.default_selection)
selection = PurchaseSelection(
period=period_config,
traffic_value=traffic_value,
servers=servers,
devices=devices,
)
pricing = await miniapp_service.calculate_pricing(db, context, selection)
return AutoPurchaseContext(context=context, pricing=pricing, selection=selection)
async def auto_purchase_saved_cart_after_topup(
db: AsyncSession,
user: User,
*,
bot: Optional[Bot] = None,
) -> bool:
"""Attempts to automatically purchase a subscription from a saved cart."""
if not settings.is_auto_purchase_after_topup_enabled():
return False
if not user or not getattr(user, "id", None):
return False
cart_data = await user_cart_service.get_user_cart(user.id)
if not cart_data:
return False
logger.info(
"🔁 Автопокупка: обнаружена сохранённая корзина у пользователя %s", user.telegram_id
)
try:
prepared = await _prepare_auto_purchase(db, user, cart_data)
except PurchaseValidationError as error:
logger.error(
"❌ Автопокупка: ошибка валидации корзины пользователя %s: %s",
user.telegram_id,
error,
)
return False
except Exception as error: # pragma: no cover - defensive logging
logger.error(
"❌ Автопокупка: непредвиденная ошибка при подготовке корзины %s: %s",
user.telegram_id,
error,
exc_info=True,
)
return False
if prepared is None:
return False
pricing = prepared.pricing
selection = prepared.selection
if pricing.final_total <= 0:
logger.warning(
"❌ Автопокупка: итоговая сумма для пользователя %s некорректна (%s)",
user.telegram_id,
pricing.final_total,
)
return False
if user.balance_kopeks < pricing.final_total:
logger.info(
"🔁 Автопокупка: у пользователя %s недостаточно средств (%s < %s)",
user.telegram_id,
user.balance_kopeks,
pricing.final_total,
)
return False
purchase_service = SubscriptionPurchaseService()
try:
purchase_result = await purchase_service.submit_purchase(
db,
prepared.context,
pricing,
)
except PurchaseBalanceError:
logger.info(
"🔁 Автопокупка: баланс пользователя %s изменился и стал недостаточным",
user.telegram_id,
)
return False
except PurchaseValidationError as error:
logger.error(
"❌ Автопокупка: не удалось подтвердить корзину пользователя %s: %s",
user.telegram_id,
error,
)
return False
except Exception as error: # pragma: no cover - defensive logging
logger.error(
"❌ Автопокупка: ошибка оформления подписки для пользователя %s: %s",
user.telegram_id,
error,
exc_info=True,
)
return False
await user_cart_service.delete_user_cart(user.id)
await clear_subscription_checkout_draft(user.id)
subscription = purchase_result.get("subscription")
transaction = purchase_result.get("transaction")
was_trial_conversion = purchase_result.get("was_trial_conversion", False)
texts = get_texts(getattr(user, "language", "ru"))
if bot:
try:
notification_service = AdminNotificationService(bot)
await notification_service.send_subscription_purchase_notification(
db,
user,
subscription,
transaction,
selection.period.days,
was_trial_conversion,
)
except Exception as error: # pragma: no cover - defensive logging
logger.error(
"⚠️ Автопокупка: не удалось отправить уведомление админам (%s): %s",
user.telegram_id,
error,
)
try:
period_label = format_period_description(
selection.period.days,
getattr(user, "language", "ru"),
)
auto_message = texts.t(
"AUTO_PURCHASE_SUBSCRIPTION_SUCCESS",
"✅ Subscription purchased automatically after balance top-up ({period}).",
).format(period=period_label)
hint_message = texts.t(
"AUTO_PURCHASE_SUBSCRIPTION_HINT",
"Open the ‘My subscription’ section to access your link.",
)
purchase_message = purchase_result.get("message", "")
full_message = "\n\n".join(
part.strip()
for part in [auto_message, purchase_message, hint_message]
if part and part.strip()
)
keyboard = InlineKeyboardMarkup(
inline_keyboard=[
[
InlineKeyboardButton(
text=texts.t("MY_SUBSCRIPTION_BUTTON", "📱 My subscription"),
callback_data="menu_subscription",
)
],
[
InlineKeyboardButton(
text=texts.t("BACK_TO_MAIN_MENU_BUTTON", "🏠 Main menu"),
callback_data="back_to_menu",
)
],
]
)
await bot.send_message(
chat_id=user.telegram_id,
text=full_message,
reply_markup=keyboard,
parse_mode="HTML",
)
except Exception as error: # pragma: no cover - defensive logging
logger.error(
"⚠️ Автопокупка: не удалось уведомить пользователя %s: %s",
user.telegram_id,
error,
)
logger.info(
"✅ Автопокупка: подписка на %s дней оформлена для пользователя %s",
selection.period.days,
user.telegram_id,
)
return True
__all__ = ["auto_purchase_saved_cart_after_topup"]

11
app/services/system_settings_service.py
View File

@@ -236,6 +236,7 @@ class BotConfigurationService:
"PAYMENT_SUBSCRIPTION_DESCRIPTION": "PAYMENT",
"PAYMENT_BALANCE_TEMPLATE": "PAYMENT",
"PAYMENT_SUBSCRIPTION_TEMPLATE": "PAYMENT",
"AUTO_PURCHASE_AFTER_TOPUP_ENABLED": "PAYMENT",
"SIMPLE_SUBSCRIPTION_ENABLED": "SIMPLE_SUBSCRIPTION",
"SIMPLE_SUBSCRIPTION_PERIOD_DAYS": "SIMPLE_SUBSCRIPTION",
"SIMPLE_SUBSCRIPTION_DEVICE_LIMIT": "SIMPLE_SUBSCRIPTION",
@@ -474,6 +475,16 @@ class BotConfigurationService:
"warning": "Слишком малый интервал может привести к частым обращениям к платёжным API.",
"dependencies": "PAYMENT_VERIFICATION_AUTO_CHECK_ENABLED",
},
"AUTO_PURCHASE_AFTER_TOPUP_ENABLED": {
"description": (
"При достаточном балансе автоматически оформляет сохранённую подписку сразу после пополнения."
),
"format": "Булево значение.",
"example": "true",
"warning": (
"Используйте с осторожностью: средства будут списаны мгновенно, если корзина найдена."
),
},
"SUPPORT_TICKET_SLA_MINUTES": {
"description": "Лимит времени для ответа модераторов на тикет в минутах.",
"format": "Целое число от 1 до 1440.",

35
app/services/tribute_service.py
View File

@@ -14,6 +14,9 @@ from app.database.crud.transaction import (
from app.database.crud.user import get_user_by_telegram_id
from app.external.tribute import TributeService as TributeAPI
from app.services.payment_service import PaymentService
from app.services.subscription_auto_purchase_service import (
auto_purchase_saved_cart_after_topup,
)
from app.utils.user_utils import format_referrer_info
logger = logging.getLogger(__name__)
@@ -267,17 +270,36 @@ class TributeService:
# Проверяем наличие сохраненной корзины для возврата к оформлению подписки
from app.services.user_cart_service import user_cart_service
has_saved_cart = await user_cart_service.has_user_cart(user.id)
auto_purchase_success = False
if has_saved_cart:
try:
auto_purchase_success = await auto_purchase_saved_cart_after_topup(
session,
user,
bot=self.bot,
)
except Exception as auto_error:
logger.error(
"Ошибка автоматической покупки подписки для пользователя %s: %s",
user.id,
auto_error,
exc_info=True,
)
if auto_purchase_success:
has_saved_cart = False
if has_saved_cart and self.bot:
# Если у пользователя есть сохраненная корзина,
# Если у пользователя есть сохраненная корзина,
# отправляем ему уведомление с кнопкой вернуться к оформлению
from app.localization.texts import get_texts
from aiogram import types
texts = get_texts(user.language)
cart_message = texts.BALANCE_TOPUP_CART_REMINDER_DETAILED.format(
total_amount=settings.format_price(amount_kopeks)
)
# Создаем клавиатуру с кнопками
keyboard = types.InlineKeyboardMarkup(inline_keyboard=[
[types.InlineKeyboardButton(
@@ -293,13 +315,16 @@ class TributeService:
callback_data="back_to_menu"
)]
])
await self.bot.send_message(
chat_id=user_id,
text=f"✅ Баланс пополнен на {settings.format_price(amount_kopeks)}!\n\n{cart_message}",
reply_markup=keyboard
)
logger.info(f"Отправлено уведомление с кнопкой возврата к оформлению подписки пользователю {user_id}")
logger.info(
"Отправлено уведомление с кнопкой возврата к оформлению подписки пользователю %s",
user_id,
)
except Exception as e:
logger.error(f"Ошибка отправки уведомления об успешном платеже: {e}")

189
tests/services/test_subscription_auto_purchase_service.py Normal file
View File

@@ -0,0 +1,189 @@
import pytest
from unittest.mock import AsyncMock, MagicMock
from app.config import settings
from app.database.models import User
from app.services.subscription_auto_purchase_service import auto_purchase_saved_cart_after_topup
from app.services.subscription_purchase_service import (
PurchaseDevicesConfig,
PurchaseOptionsContext,
PurchasePeriodConfig,
PurchasePricingResult,
PurchaseSelection,
PurchaseServersConfig,
PurchaseTrafficConfig,
)
from sqlalchemy.ext.asyncio import AsyncSession
class DummyTexts:
def t(self, key: str, default: str):
return default
def format_price(self, value: int) -> str:
return f"{value / 100:.0f} ₽"
@pytest.mark.asyncio
async def test_auto_purchase_saved_cart_after_topup_success(monkeypatch):
monkeypatch.setattr(settings, "AUTO_PURCHASE_AFTER_TOPUP_ENABLED", True)
user = MagicMock(spec=User)
user.id = 42
user.telegram_id = 4242
user.balance_kopeks = 200_000
user.language = "ru"
user.subscription = None
cart_data = {
"period_days": 30,
"countries": ["ru"],
"traffic_gb": 0,
"devices": 1,
}
traffic_config = PurchaseTrafficConfig(
selectable=False,
mode="fixed",
options=[],
default_value=0,
current_value=0,
)
servers_config = PurchaseServersConfig(
options=[],
min_selectable=0,
max_selectable=0,
default_selection=["ru"],
)
devices_config = PurchaseDevicesConfig(
minimum=1,
maximum=5,
default=1,
current=1,
price_per_device=0,
discounted_price_per_device=0,
price_label="0 ₽",
)
period_config = PurchasePeriodConfig(
id="days:30",
days=30,
months=1,
label="30 дней",
base_price=100_000,
base_price_label="1000 ₽",
base_price_original=100_000,
base_price_original_label=None,
discount_percent=0,
per_month_price=100_000,
per_month_price_label="1000 ₽",
traffic=traffic_config,
servers=servers_config,
devices=devices_config,
)
context = PurchaseOptionsContext(
user=user,
subscription=None,
currency="RUB",
balance_kopeks=user.balance_kopeks,
periods=[period_config],
default_period=period_config,
period_map={"days:30": period_config},
server_uuid_to_id={"ru": 1},
payload={},
)
base_pricing = PurchasePricingResult(
selection=PurchaseSelection(
period=period_config,
traffic_value=0,
servers=["ru"],
devices=1,
),
server_ids=[1],
server_prices_for_period=[100_000],
base_original_total=100_000,
discounted_total=100_000,
promo_discount_value=0,
promo_discount_percent=0,
final_total=100_000,
months=1,
details={"servers_individual_prices": [100_000]},
)
class DummyMiniAppService:
async def build_options(self, db, user):
return context
async def calculate_pricing(self, db, ctx, selection):
return PurchasePricingResult(
selection=selection,
server_ids=base_pricing.server_ids,
server_prices_for_period=base_pricing.server_prices_for_period,
base_original_total=base_pricing.base_original_total,
discounted_total=base_pricing.discounted_total,
promo_discount_value=base_pricing.promo_discount_value,
promo_discount_percent=base_pricing.promo_discount_percent,
final_total=base_pricing.final_total,
months=base_pricing.months,
details=base_pricing.details,
)
class DummyPurchaseService:
async def submit_purchase(self, db, prepared_context, pricing):
return {
"subscription": MagicMock(),
"transaction": MagicMock(),
"was_trial_conversion": False,
"message": "🎉 Subscription purchased",
}
monkeypatch.setattr(
"app.services.subscription_auto_purchase_service.MiniAppSubscriptionPurchaseService",
lambda: DummyMiniAppService(),
)
monkeypatch.setattr(
"app.services.subscription_auto_purchase_service.SubscriptionPurchaseService",
lambda: DummyPurchaseService(),
)
monkeypatch.setattr(
"app.services.subscription_auto_purchase_service.user_cart_service.get_user_cart",
AsyncMock(return_value=cart_data),
)
delete_cart_mock = AsyncMock()
monkeypatch.setattr(
"app.services.subscription_auto_purchase_service.user_cart_service.delete_user_cart",
delete_cart_mock,
)
clear_draft_mock = AsyncMock()
monkeypatch.setattr(
"app.services.subscription_auto_purchase_service.clear_subscription_checkout_draft",
clear_draft_mock,
)
monkeypatch.setattr(
"app.services.subscription_auto_purchase_service.get_texts",
lambda lang: DummyTexts(),
)
monkeypatch.setattr(
"app.services.subscription_auto_purchase_service.format_period_description",
lambda days, lang: f"{days} дней",
)
admin_service_mock = MagicMock()
admin_service_mock.send_subscription_purchase_notification = AsyncMock()
monkeypatch.setattr(
"app.services.subscription_auto_purchase_service.AdminNotificationService",
lambda bot: admin_service_mock,
)
bot = AsyncMock()
db_session = AsyncMock(spec=AsyncSession)
result = await auto_purchase_saved_cart_after_topup(db_session, user, bot=bot)
assert result is True
delete_cart_mock.assert_awaited_once_with(user.id)
clear_draft_mock.assert_awaited_once_with(user.id)
bot.send_message.assert_awaited()
admin_service_mock.send_subscription_purchase_notification.assert_awaited()