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

fix: format premium traffic limits from bytes

Browse Source
This commit is contained in:
3252a8
2026-05-30 12:00:01 +03:00
parent 067d6fb59f
commit c6c5352813
4 changed files with 153 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/bot/handlers/user/subscription/core.py
+16 -2
View File
@@ -114,6 +114,20 @@ def _with_subscription_purchase_description(
return f"{description}\n\n{text}"
def _format_premium_bytes(value: object) -> str:
try:
bytes_value = max(0, int(value or 0))
except (TypeError, ValueError):
bytes_value = 0
return f"{bytes_value / 2**30:.2f} GB"
def _format_premium_usage_limit(active: dict[str, object]) -> str:
used = _format_premium_bytes(active.get("premium_used_bytes"))
limit = _format_premium_bytes(active.get("premium_limit_bytes"))
return f"{used} из {limit}"
async def display_subscription_options(
event: Union[types.Message, types.CallbackQuery],
i18n_data: dict,
@@ -456,7 +470,7 @@ async def tariff_topup_list_callback(
if len(labels) > len(visible):
premium_lines.append(f"• ... еще {len(labels) - len(visible)}")
premium_lines.append(
f"Premium использовано: {active.get('premium_used')} из {active.get('premium_limit')}. Осталось: {premium_left / 2**30:.2f} GB." # noqa: E501
f"Premium использовано: {_format_premium_usage_limit(active)}. Осталось: {premium_left / 2**30:.2f} GB." # noqa: E501
)
text = get_text("choose_payment_method_traffic")
if carryover_lines:
@@ -1065,7 +1079,7 @@ async def my_subscription_command_handler(
text += (
"\n\n🚀 <b>Premium-серверы</b>\n"
f"Статус: <b>{premium_status}</b>\n"
f"Лимит: <b>{active.get('premium_used')} из {active.get('premium_limit')}</b>\n"
f"Лимит: <b>{_format_premium_usage_limit(active)}</b>\n"
f"Осталось: <b>{premium_left / 2**30:.2f} GB</b>\n"
f"Докупленный остаток: <b>{premium_balance / 2**30:.2f} GB</b>\n"
"Отдельный лимит действует на:\n"
backend/bot/services/subscription_service_impl/lifecycle.py
+47 -2
View File
@@ -54,6 +54,37 @@ class SubscriptionLifecycleMixin:
reason = f"{reason} message={message}"
return reason
@staticmethod
def _display_datetime_text(value: Optional[Any]) -> Optional[str]:
if not value:
return None
if isinstance(value, datetime):
normalized = value if value.tzinfo else value.replace(tzinfo=timezone.utc)
return normalized.strftime("%d.%m.%Y %H:%M")
return str(value)
@staticmethod
def _device_topup_renewal_available(
extra_hwid_devices: int,
extra_hwid_valid_until: Optional[Any],
subscription_end_date: Optional[Any],
) -> bool:
if not isinstance(extra_hwid_valid_until, datetime) or not isinstance(
subscription_end_date, datetime
):
return False
valid_until = (
extra_hwid_valid_until
if extra_hwid_valid_until.tzinfo
else extra_hwid_valid_until.replace(tzinfo=timezone.utc)
)
end_date = (
subscription_end_date
if subscription_end_date.tzinfo
else subscription_end_date.replace(tzinfo=timezone.utc)
)
return bool(int(extra_hwid_devices or 0) > 0 and valid_until < end_date)
async def _local_active_subscription_details_fallback(
self,
db_user: User,
@@ -130,7 +161,9 @@ class SubscriptionLifecycleMixin:
"base_hwid_device_limit": local_active_sub.hwid_device_limit,
"extra_hwid_devices": int(local_active_sub.extra_hwid_devices or 0),
"extra_hwid_devices_valid_until": None,
"extra_hwid_devices_valid_until_text": None,
"extra_hwid_devices_next_valid_from": None,
"device_topup_renewal_available": False,
"user_bot_username": db_user.username,
"is_panel_data": False,
"max_devices": self._effective_hwid_limit(
@@ -1011,6 +1044,14 @@ class SubscriptionLifecycleMixin:
if expected_hwid_limit is not None:
hwid_limit = expected_hwid_limit
extra_hwid_valid_until = hwid_entitlement_summary.get("active_until")
extra_hwid_next_valid_from = hwid_entitlement_summary.get("next_valid_from")
device_topup_renewal_available = self._device_topup_renewal_available(
active_extra_hwid_devices,
extra_hwid_valid_until,
panel_end_date,
)
return {
"user_id": panel_user_data.get("uuid"),
"panel_subscription_uuid": panel_user_data.get("subscriptionUuid")
@@ -1071,8 +1112,12 @@ class SubscriptionLifecycleMixin:
if local_active_sub
else None,
"extra_hwid_devices": active_extra_hwid_devices,
"extra_hwid_devices_valid_until": hwid_entitlement_summary.get("active_until"),
"extra_hwid_devices_next_valid_from": hwid_entitlement_summary.get("next_valid_from"),
"extra_hwid_devices_valid_until": extra_hwid_valid_until,
"extra_hwid_devices_valid_until_text": self._display_datetime_text(
extra_hwid_valid_until
),
"extra_hwid_devices_next_valid_from": extra_hwid_next_valid_from,
"device_topup_renewal_available": device_topup_renewal_available,
"user_bot_username": db_user.username,
"is_panel_data": True,
"max_devices": hwid_limit,
tests/test_subscription_service_behavior.py
+72
View File
@@ -592,6 +592,78 @@ class SubscriptionServiceActiveDetailsTests(unittest.IsolatedAsyncioTestCase):
deactivate_all.assert_awaited_once_with(session, 42)
update_user.assert_awaited_once_with(session, 42, {"panel_user_uuid": None})
async def test_get_active_subscription_details_includes_device_topup_renewal_fields(
self,
):
with tempfile.TemporaryDirectory() as tmpdir:
settings = _make_settings(_tariffs_config_payload(), tmpdir)
service = _make_service(settings)
active_until = datetime(2099, 1, 2, 3, 4, tzinfo=timezone.utc)
service.panel_service.get_user_by_uuid_lookup = AsyncMock(
return_value={
"ok": True,
"user": {
"uuid": "panel-user",
"shortUuid": "short-uuid",
"status": "ACTIVE",
"expireAt": "2099-02-01T00:00:00Z",
"subscriptionUrl": "https://panel.example.test/sub/short-uuid",
"trafficLimitBytes": 1000,
"trafficLimitStrategy": "MONTH",
"userTraffic": {
"usedTrafficBytes": 100,
"lifetimeUsedTrafficBytes": 100,
},
},
}
)
service.premium_access_for_tariff = AsyncMock(
return_value={"squad_uuids": [], "squad_labels": [], "node_labels": []}
)
session = AsyncMock()
db_user = SimpleNamespace(
user_id=42,
panel_user_uuid="panel-user",
username="alice",
language_code="en",
lifetime_used_traffic_bytes=100,
)
local_sub = self._local_active_sub()
local_sub.tariff_key = "standard"
local_sub.extra_hwid_devices = 0
local_sub.hwid_device_limit = 3
with (
patch(
"bot.services.subscription_service_impl.lifecycle.user_dal.get_user_by_id",
AsyncMock(return_value=db_user),
),
patch(
"bot.services.subscription_service_impl.lifecycle.subscription_dal.get_active_subscription_by_user_id",
AsyncMock(return_value=local_sub),
),
patch(
"bot.services.subscription_service_impl.lifecycle.subscription_dal.update_subscription",
AsyncMock(),
),
patch(
"bot.services.subscription_service_impl.lifecycle.tariff_dal.get_hwid_device_entitlement_summary",
AsyncMock(
return_value={
"active_devices": 1,
"active_until": active_until,
"next_valid_from": None,
}
),
),
):
result = await service.get_active_subscription_details(session, user_id=42)
self.assertTrue(result["device_topup_renewal_available"])
self.assertEqual(result["extra_hwid_devices"], 1)
self.assertEqual(result["extra_hwid_devices_valid_until"], active_until)
self.assertEqual(result["extra_hwid_devices_valid_until_text"], "02.01.2099 03:04")
class SubscriptionDalPayloadTests(unittest.TestCase):
def test_subscription_model_payload_drops_panel_only_keys(self):
tests/test_user_bot_menu.py
+18 -1
View File
@@ -5,7 +5,10 @@ from types import SimpleNamespace
from bot.app.web import subscription_webapp
from bot.handlers.user import referral
from bot.handlers.user.subscription.core import _with_subscription_purchase_description
from bot.handlers.user.subscription.core import (
_format_premium_usage_limit,
_with_subscription_purchase_description,
)
from bot.keyboards.inline.user_keyboards import (
get_bot_interface_inline_keyboard,
get_connect_and_main_keyboard,
@@ -207,6 +210,20 @@ class UserBotMenuTests(unittest.TestCase):
"Choose traffic",
)
def test_premium_usage_limit_uses_byte_fields_when_display_fields_missing(self):
gib = 1024**3
active = {
"premium_used": None,
"premium_limit": None,
"premium_used_bytes": int(1.25 * gib),
"premium_limit_bytes": 26 * gib,
}
text = _format_premium_usage_limit(active)
self.assertEqual(text, "1.25 GB из 26.00 GB")
self.assertNotIn("None", text)
def test_payment_navigation_context_keeps_bot_menu_source(self):
settings = SimpleNamespace(
payment_methods_order=[],