diff --git a/tests/test_hwid_device_topup.py b/tests/test_hwid_device_topup.py new file mode 100644 index 0000000..74c0f91 --- /dev/null +++ b/tests/test_hwid_device_topup.py @@ -0,0 +1,266 @@ +"""Behavioural tests for ``HwidDeviceMixin.activate_hwid_device_topup``. + +The HWID device top-up flow already checks the panel update result correctly +(unlike the regular/premium traffic top-ups, which were the subject of a +recent fix). These tests pin that: + +* an unknown package count for the tariff is rejected (anti-tamper); +* unlimited subscribers (``hwid_device_limit == 0``) are a no-op that still + returns a meaningful payload; +* the panel call uses ``hwidDeviceLimit = base + extra``, not just ``base``; +* a panel failure returns ``None`` and DOES NOT write the audit row; +* the happy path persists ``extra_hwid_devices = old + purchased`` and + records the device purchase. +""" + +import json +import tempfile +import unittest +from datetime import datetime, 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 + + +def _tariffs_config_payload(**overrides: Any) -> dict: + tariff = { + "key": "standard", + "names": {"en": "Standard"}, + "descriptions": {"en": "Base"}, + "squad_uuids": ["main-squad"], + "billing_model": "period", + "monthly_gb": 100, + "prices_rub": {"1": 100}, + "prices_stars": {"1": 0}, + "enabled_periods": [1], + "hwid_device_limit": 3, + "hwid_device_packages": { + "rub": [{"count": 1, "price": 50}, {"count": 3, "price": 120}], + "stars": [], + }, + "enabled": True, + } + tariff.update(overrides) + return {"default_tariff": "standard", "tariffs": [tariff]} + + +def _make_settings(tmpdir: str, payload: dict, **overrides: Any) -> Settings: + config_path = Path(tmpdir) / "tariffs.json" + config_path.write_text(json.dumps(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: + return SubscriptionService(settings, AsyncMock(spec=PanelApiService)) + + +def _make_sub(*, hwid_device_limit=3, extra_hwid_devices=0): + return SimpleNamespace( + subscription_id=11, + user_id=42, + panel_user_uuid="panel-uuid", + panel_subscription_uuid="panel-sub", + tariff_key="standard", + end_date=datetime(2099, 1, 1, tzinfo=timezone.utc), + hwid_device_limit=hwid_device_limit, + extra_hwid_devices=extra_hwid_devices, + ) + + +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 HwidDeviceTopupInputTests(unittest.IsolatedAsyncioTestCase): + async def test_rejects_non_positive_device_count(self): + with tempfile.TemporaryDirectory() as tmpdir: + settings = _make_settings(tmpdir, _tariffs_config_payload()) + service = _make_service(settings) + for bad in (0, -1, "abc"): + with self.subTest(bad=bad): + result = await service.activate_hwid_device_topup( + session=AsyncMock(), + user_id=42, + device_count=bad, + payment_amount=50, + payment_db_id=1, + ) + self.assertIsNone(result) + + async def test_rejects_count_not_in_tariff_packages(self): + # Tariff only offers 1-device and 3-device packages; 2 should be rejected. + with tempfile.TemporaryDirectory() as tmpdir: + settings = _make_settings(tmpdir, _tariffs_config_payload()) + service = _make_service(settings) + sub = _make_sub() + user = _make_user() + with ( + patch( + "bot.services.subscription_service_impl.devices.user_dal.get_user_by_id", + AsyncMock(return_value=user), + ), + patch( + "bot.services.subscription_service_impl.devices.subscription_dal.get_active_subscription_by_user_id", + AsyncMock(return_value=sub), + ), + ): + result = await service.activate_hwid_device_topup( + session=AsyncMock(), + user_id=42, + device_count=2, + payment_amount=70, + payment_db_id=1, + ) + self.assertIsNone(result) + + +class HwidDeviceTopupBehaviourTests(unittest.IsolatedAsyncioTestCase): + async def test_unlimited_subscriber_returns_noop_payload(self): + # hwid_device_limit == 0 means unlimited — top-up makes no sense and must skip. + with tempfile.TemporaryDirectory() as tmpdir: + settings = _make_settings(tmpdir, _tariffs_config_payload()) + service = _make_service(settings) + sub = _make_sub(hwid_device_limit=0, extra_hwid_devices=0) + user = _make_user() + with ( + patch( + "bot.services.subscription_service_impl.devices.user_dal.get_user_by_id", + AsyncMock(return_value=user), + ), + patch( + "bot.services.subscription_service_impl.devices.subscription_dal.get_active_subscription_by_user_id", + AsyncMock(return_value=sub), + ), + patch( + "bot.services.subscription_service_impl.devices.tariff_dal.create_hwid_device_purchase", + AsyncMock(), + ) as create_purchase, + ): + result = await service.activate_hwid_device_topup( + session=AsyncMock(), + user_id=42, + device_count=1, + payment_amount=50, + payment_db_id=1, + ) + self.assertEqual(result["hwid_device_limit"], 0) + self.assertEqual(result["purchased_hwid_devices"], 0) + # Unlimited subscriber: no audit row, no panel touch. + create_purchase.assert_not_awaited() + + async def test_panel_payload_uses_effective_limit_with_extras(self): + with tempfile.TemporaryDirectory() as tmpdir: + settings = _make_settings(tmpdir, _tariffs_config_payload()) + service = _make_service(settings) + sub = _make_sub(hwid_device_limit=3, extra_hwid_devices=2) + user = _make_user() + updated_sub = SimpleNamespace( + subscription_id=11, + end_date=sub.end_date, + ) + service.panel_service.update_user_details_on_panel = AsyncMock( + return_value={"ok": True} + ) + with ( + patch( + "bot.services.subscription_service_impl.devices.user_dal.get_user_by_id", + AsyncMock(return_value=user), + ), + patch( + "bot.services.subscription_service_impl.devices.subscription_dal.get_active_subscription_by_user_id", + AsyncMock(return_value=sub), + ), + patch( + "bot.services.subscription_service_impl.devices.subscription_dal.update_subscription", + AsyncMock(return_value=updated_sub), + ), + patch( + "bot.services.subscription_service_impl.payments.payment_dal.get_payment_by_db_id", + AsyncMock(return_value=SimpleNamespace()), + ), + patch( + "bot.services.subscription_service_impl.devices.tariff_dal.create_hwid_device_purchase", + AsyncMock(), + ), + ): + result = await service.activate_hwid_device_topup( + session=AsyncMock(), + user_id=42, + device_count=1, + payment_amount=50, + payment_db_id=1, + ) + + # Caller-facing payload: base=3, new extras = 2 + 1 purchased = 3 → effective = 6. + self.assertEqual(result["hwid_device_limit"], 6) + self.assertEqual(result["extra_hwid_devices"], 3) + self.assertEqual(result["purchased_hwid_devices"], 1) + # Panel must see the full effective limit, not just the base. + panel_payload = service.panel_service.update_user_details_on_panel.await_args.args[1] + self.assertEqual(panel_payload["hwidDeviceLimit"], 6) + + async def test_panel_failure_returns_none_and_skips_audit(self): + with tempfile.TemporaryDirectory() as tmpdir: + settings = _make_settings(tmpdir, _tariffs_config_payload()) + service = _make_service(settings) + sub = _make_sub() + user = _make_user() + updated_sub = SimpleNamespace(subscription_id=11, end_date=sub.end_date) + service.panel_service.update_user_details_on_panel = AsyncMock(return_value=None) + with ( + patch( + "bot.services.subscription_service_impl.devices.user_dal.get_user_by_id", + AsyncMock(return_value=user), + ), + patch( + "bot.services.subscription_service_impl.devices.subscription_dal.get_active_subscription_by_user_id", + AsyncMock(return_value=sub), + ), + patch( + "bot.services.subscription_service_impl.devices.subscription_dal.update_subscription", + AsyncMock(return_value=updated_sub), + ), + patch( + "bot.services.subscription_service_impl.payments.payment_dal.get_payment_by_db_id", + AsyncMock(return_value=SimpleNamespace()), + ), + patch( + "bot.services.subscription_service_impl.devices.tariff_dal.create_hwid_device_purchase", + AsyncMock(), + ) as create_purchase, + ): + result = await service.activate_hwid_device_topup( + session=AsyncMock(), + user_id=42, + device_count=1, + payment_amount=50, + payment_db_id=1, + ) + self.assertIsNone(result) + create_purchase.assert_not_awaited() + + +if __name__ == "__main__": # pragma: no cover + unittest.main()