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

chore: cover topup, premium-topup and tariff-switch panel failure paths

Browse Source
This commit is contained in:
3252a8
2026-05-17 10:22:53 +03:00
parent 7d1c9ee373
commit 6886a90ee4
1 changed files with 347 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_topup_panel_failures.py
+347
View File
@@ -0,0 +1,347 @@
"""Regression tests for silent panel-update failures in top-up / tariff-switch flows.
Three sibling methods — ``activate_topup``, ``activate_premium_topup``, and
``switch_tariff_without_payment`` — used to ignore the result of
``panel_service.update_user_details_on_panel``. When the panel update failed,
the local DB still recorded the top-up / tariff change and the function
reported success. The user paid, saw the new traffic / tariff in the app,
but Remnawave never received the change, so the new entitlement was unusable.
Other activation paths (``activate_subscription``, ``_activate_traffic_package``,
``activate_hwid_device_topup``) all check the panel response. These tests pin
the same contract on the three previously-broken paths.
"""
import json
import tempfile
import unittest
from datetime import datetime, timedelta, timezone
from pathlib import Path
from types import SimpleNamespace
from typing import Any
from unittest.mock import AsyncMock, patch
from bot.services.panel_api_service import PanelApiService
from bot.services.subscription_service import SubscriptionService
from config.settings import Settings
GIB = 1024**3
def _tariffs_config_payload() -> dict:
return {
"default_tariff": "standard",
"tariffs": [
{
"key": "standard",
"names": {"en": "Standard"},
"descriptions": {"en": "Base"},
"squad_uuids": ["main-squad"],
"premium_squad_uuids": ["premium-squad"],
"premium_monthly_gb": 25,
"billing_model": "period",
"monthly_gb": 100,
"prices_rub": {"1": 150, "3": 400},
"prices_stars": {"1": 0},
"enabled_periods": [1, 3],
"hwid_device_limit": 3,
"topup_packages": {
"rub": [{"gb": 10, "price": 50}],
"stars": [],
},
"premium_topup_packages": {
"rub": [{"gb": 10, "price": 100}],
"stars": [],
},
"enabled": True,
},
{
"key": "premium",
"names": {"en": "Premium"},
"descriptions": {"en": "Premium plan"},
"squad_uuids": ["main-squad", "premium-extra"],
"premium_squad_uuids": ["premium-squad"],
"premium_monthly_gb": 50,
"billing_model": "period",
"monthly_gb": 200,
"prices_rub": {"1": 300, "3": 800},
"prices_stars": {"1": 0},
"enabled_periods": [1, 3],
"enabled": True,
},
],
}
def _make_settings(tmpdir: str, **overrides: Any) -> Settings:
config_path = Path(tmpdir) / "tariffs.json"
config_path.write_text(json.dumps(_tariffs_config_payload()), encoding="utf-8")
values: dict[str, Any] = {
"_env_file": None,
"BOT_TOKEN": "token",
"POSTGRES_USER": "u",
"POSTGRES_PASSWORD": "p",
"TARIFFS_CONFIG_PATH": str(config_path),
}
values.update(overrides)
return Settings(**values)
def _make_service(settings: Settings) -> SubscriptionService:
panel_service = AsyncMock(spec=PanelApiService)
return SubscriptionService(settings, panel_service)
def _make_sub(**overrides):
base = dict(
subscription_id=11,
user_id=42,
panel_user_uuid="panel-uuid",
panel_subscription_uuid="panel-sub",
tariff_key="standard",
traffic_limit_bytes=100 * GIB,
traffic_used_bytes=10 * GIB,
topup_balance_bytes=0,
tier_baseline_bytes=100 * GIB,
regular_bonus_bytes=0,
regular_unlimited_override=False,
extra_hwid_devices=0,
hwid_device_limit=3,
premium_baseline_bytes=25 * GIB,
premium_topup_balance_bytes=0,
premium_topup_used_bytes=0,
premium_used_bytes=0,
premium_bonus_bytes=0,
premium_unlimited_override=False,
premium_is_limited=False,
premium_period_start_at=None,
effective_monthly_price_rub=150,
end_date=datetime(2099, 1, 1, tzinfo=timezone.utc),
is_active=True,
)
base.update(overrides)
return SimpleNamespace(**base)
def _make_user():
return SimpleNamespace(
user_id=42,
telegram_id=42,
email=None,
username="u",
first_name="U",
last_name="L",
language_code="en",
panel_user_uuid="panel-uuid",
)
class ActivateTopupPanelFailureTests(unittest.IsolatedAsyncioTestCase):
"""Regular (non-premium) period top-up. The bug was traffic.py:252 ignoring panel result."""
async def _run(self, *, panel_response):
with tempfile.TemporaryDirectory() as tmpdir:
settings = _make_settings(tmpdir)
service = _make_service(settings)
sub = _make_sub()
user = _make_user()
service.panel_service.update_user_details_on_panel = AsyncMock(
return_value=panel_response
)
updated_sub = SimpleNamespace(end_date=sub.end_date, subscription_id=11)
with (
patch(
"bot.services.subscription_service_impl.payments.payment_dal.get_payment_by_db_id",
AsyncMock(return_value=SimpleNamespace()),
),
patch(
"bot.services.subscription_service_impl.traffic.user_dal.get_user_by_id",
AsyncMock(return_value=user),
),
patch(
"bot.services.subscription_service_impl.traffic.subscription_dal.get_active_subscription_by_user_id",
AsyncMock(return_value=sub),
),
patch(
"bot.services.subscription_service_impl.traffic.subscription_dal.update_subscription",
AsyncMock(return_value=updated_sub),
),
patch(
"bot.services.subscription_service_impl.traffic.tariff_dal.create_traffic_topup",
AsyncMock(),
) as create_topup,
):
result = await service.activate_topup(
session=AsyncMock(),
user_id=42,
tariff_key="standard",
traffic_gb=10,
payment_amount=50,
payment_db_id=1,
)
return result, create_topup
async def test_returns_none_when_panel_returns_none(self):
result, create_topup = await self._run(panel_response=None)
self.assertIsNone(result)
# The local audit row must NOT be written when the panel rejected the change.
create_topup.assert_not_awaited()
async def test_returns_none_when_panel_returns_error_dict(self):
result, create_topup = await self._run(panel_response={"error": True})
self.assertIsNone(result)
create_topup.assert_not_awaited()
async def test_returns_payload_when_panel_succeeds(self):
result, create_topup = await self._run(panel_response={"ok": True, "uuid": "panel-uuid"})
self.assertIsNotNone(result)
self.assertEqual(result["tariff_key"], "standard")
self.assertEqual(result["topup_balance_bytes"], 10 * GIB)
create_topup.assert_awaited_once()
class ActivatePremiumTopupPanelFailureTests(unittest.IsolatedAsyncioTestCase):
"""Premium top-up. The bug was traffic.py:350 ignoring panel result — meant a user could
pay to lift the premium quota but the panel never re-granted premium squad access."""
async def _run(self, *, panel_response):
with tempfile.TemporaryDirectory() as tmpdir:
settings = _make_settings(tmpdir)
service = _make_service(settings)
# Start with premium_is_limited=True so the recompute path is exercised.
sub = _make_sub(
premium_topup_balance_bytes=0,
premium_used_bytes=30 * GIB, # already over the 25 GB baseline
premium_is_limited=True,
)
user = _make_user()
service.panel_service.update_user_details_on_panel = AsyncMock(
return_value=panel_response
)
with (
patch(
"bot.services.subscription_service_impl.payments.payment_dal.get_payment_by_db_id",
AsyncMock(return_value=SimpleNamespace()),
),
patch(
"bot.services.subscription_service_impl.traffic.user_dal.get_user_by_id",
AsyncMock(return_value=user),
),
patch(
"bot.services.subscription_service_impl.traffic.subscription_dal.get_active_subscription_by_user_id",
AsyncMock(return_value=sub),
),
patch(
"bot.services.subscription_service_impl.traffic.subscription_dal.update_subscription",
AsyncMock(),
),
patch(
"bot.services.subscription_service_impl.traffic.tariff_dal.create_traffic_topup",
AsyncMock(),
) as create_topup,
):
result = await service.activate_premium_topup(
session=AsyncMock(),
user_id=42,
tariff_key="standard",
traffic_gb=10,
payment_amount=100,
payment_db_id=2,
)
return result, create_topup
async def test_returns_none_when_panel_returns_none(self):
result, create_topup = await self._run(panel_response=None)
self.assertIsNone(result)
create_topup.assert_not_awaited()
async def test_returns_none_when_panel_returns_error_dict(self):
result, create_topup = await self._run(panel_response={"error": True})
self.assertIsNone(result)
create_topup.assert_not_awaited()
async def test_returns_payload_when_panel_succeeds(self):
result, create_topup = await self._run(panel_response={"ok": True, "uuid": "panel-uuid"})
self.assertIsNotNone(result)
self.assertEqual(result["tariff_key"], "standard")
create_topup.assert_awaited_once()
class SwitchTariffPanelFailureTests(unittest.IsolatedAsyncioTestCase):
"""Free tariff switch. The bug was lifecycle.py:121 ignoring panel result — the tariff_key
flipped in the local DB while the panel kept the user on the old squad set."""
async def _run(self, *, panel_response, mode="recalc_days"):
with tempfile.TemporaryDirectory() as tmpdir:
settings = _make_settings(tmpdir)
service = _make_service(settings)
sub = _make_sub(
tariff_key="standard",
end_date=datetime.now(timezone.utc) + timedelta(days=20),
)
user = _make_user()
updated = _make_sub(
tariff_key="premium",
end_date=datetime.now(timezone.utc) + timedelta(days=10),
premium_is_limited=False,
effective_monthly_price_rub=300,
)
service.panel_service.update_user_details_on_panel = AsyncMock(
return_value=panel_response
)
with (
patch(
"bot.services.subscription_service_impl.lifecycle.user_dal.get_user_by_id",
AsyncMock(return_value=user),
),
patch(
"bot.services.subscription_service_impl.lifecycle.subscription_dal.get_active_subscription_by_user_id",
AsyncMock(return_value=sub),
),
patch(
"bot.services.subscription_service_impl.lifecycle.subscription_dal.update_subscription",
AsyncMock(return_value=updated),
),
patch(
"bot.services.subscription_service_impl.lifecycle.tariff_dal.create_tariff_change",
AsyncMock(),
) as create_change,
patch(
"bot.services.subscription_service_impl.lifecycle.tariff_dal.create_traffic_topup",
AsyncMock(),
),
):
result = await service.switch_tariff_without_payment(
session=AsyncMock(),
user_id=42,
target_tariff_key="premium",
mode=mode,
)
return result, create_change
async def test_returns_none_when_panel_returns_none(self):
result, create_change = await self._run(panel_response=None)
self.assertIsNone(result)
# tariff_changes audit row must NOT be inserted on panel failure.
create_change.assert_not_awaited()
async def test_returns_none_when_panel_returns_error_dict(self):
result, create_change = await self._run(panel_response={"error": True})
self.assertIsNone(result)
create_change.assert_not_awaited()
async def test_returns_payload_when_panel_succeeds(self):
result, create_change = await self._run(panel_response={"ok": True})
self.assertIsNotNone(result)
self.assertEqual(result["tariff_key"], "premium")
create_change.assert_awaited_once()
if __name__ == "__main__": # pragma: no cover
unittest.main()