import asyncio import json import unittest from datetime import datetime, timedelta, timezone from types import SimpleNamespace from unittest.mock import AsyncMock, patch from bot.app.web import subscription_webapp as guides from config.subscription_guides_config import default_subscription_guides_config_text class _AsyncSessionFactory: def __call__(self): return self async def __aenter__(self): return object() async def __aexit__(self, exc_type, exc, tb): return False class SubscriptionGuidesRouteTests(unittest.IsolatedAsyncioTestCase): def _request(self, settings, panel_service, match_info=None): return SimpleNamespace( app={ "settings": settings, "async_session_factory": _AsyncSessionFactory(), "panel_service": panel_service, "subscription_guides_config_cache": {"fingerprint": None, "status": None}, "subscription_guides_config_lock": asyncio.Lock(), }, match_info=match_info or {}, headers={"User-Agent": "Mozilla/5.0", "Host": "app.example.test"}, host="app.example.test", scheme="https", ) def _settings(self, **overrides): values = { "SUBSCRIPTION_GUIDES_ENABLED": True, "SUBSCRIPTION_PAGE_CONFIG_PANEL_ENABLED": True, "SUBSCRIPTION_PAGE_CONFIG_JSON_OVERRIDE_ENABLED": False, "SUBSCRIPTION_PAGE_CONFIG_JSON": "", "SUBSCRIPTION_PAGE_CONFIG_PATH": "data/subpage-config/multiapp.json", "SUBSCRIPTION_MINI_APP_URL": "https://app.example.test", "CRYPT4_ENABLED": False, "CRYPT4_REDIRECT_URL": "", "CRYPT4_LINK_CACHE_TTL_SECONDS": 3600, } values.update(overrides) return SimpleNamespace(**values) def _auth_patch(self): return patch.dict( guides.subscription_guides_route.__globals__, {"_require_user_id": lambda _: 42}, ) async def test_uses_panel_config_when_admin_json_is_empty(self): default_uuid = "00000000-0000-0000-0000-000000000000" panel_service = SimpleNamespace( get_subscription_page_config_list=AsyncMock( return_value={"configs": [{"uuid": default_uuid, "viewPosition": 1}]} ), get_subscription_page_config_by_uuid=AsyncMock( return_value={ "uuid": default_uuid, "config": json.loads(default_subscription_guides_config_text()), } ) ) request = self._request(self._settings(), panel_service) with self._auth_patch(): response = await guides.subscription_guides_route(request) body = json.loads(response.text) self.assertTrue(body["enabled"]) self.assertEqual(body["source"], "panel") self.assertEqual(body["config"]["version"], "1") panel_service.get_subscription_page_config_list.assert_awaited_once() panel_service.get_subscription_page_config_by_uuid.assert_awaited_once_with(default_uuid) async def test_admin_json_override_takes_priority_over_panel(self): admin_config = json.loads(default_subscription_guides_config_text()) panel_service = SimpleNamespace(get_subscription_page_config_by_uuid=AsyncMock()) request = self._request( self._settings( SUBSCRIPTION_PAGE_CONFIG_JSON_OVERRIDE_ENABLED=True, SUBSCRIPTION_PAGE_CONFIG_JSON=json.dumps(admin_config), ), panel_service, ) with self._auth_patch(): response = await guides.subscription_guides_route(request) body = json.loads(response.text) self.assertTrue(body["enabled"]) self.assertEqual(body["source"], "admin_json") panel_service.get_subscription_page_config_by_uuid.assert_not_called() async def test_admin_json_is_ignored_until_override_switch_is_enabled(self): default_uuid = "00000000-0000-0000-0000-000000000000" admin_config = json.loads(default_subscription_guides_config_text()) panel_service = SimpleNamespace( get_subscription_page_config_list=AsyncMock( return_value={"configs": [{"uuid": default_uuid, "viewPosition": 1}]} ), get_subscription_page_config_by_uuid=AsyncMock( return_value={ "uuid": default_uuid, "config": json.loads(default_subscription_guides_config_text()), } ) ) request = self._request( self._settings(SUBSCRIPTION_PAGE_CONFIG_JSON=json.dumps(admin_config)), panel_service, ) with self._auth_patch(): response = await guides.subscription_guides_route(request) body = json.loads(response.text) self.assertTrue(body["enabled"]) self.assertEqual(body["source"], "panel") panel_service.get_subscription_page_config_by_uuid.assert_awaited_once_with(default_uuid) async def test_panel_config_is_cached_for_multiple_users(self): default_uuid = "00000000-0000-0000-0000-000000000000" panel_config = json.loads(default_subscription_guides_config_text()) panel_config["platforms"]["windows"]["apps"][0]["name"] = "Throne" panel_service = SimpleNamespace( get_subscription_page_config_list=AsyncMock( return_value={"configs": [{"uuid": default_uuid, "viewPosition": 1}]} ), get_subscription_page_config_by_uuid=AsyncMock( return_value={"uuid": default_uuid, "config": panel_config} ), ) request = self._request(self._settings(), panel_service) with self._auth_patch(): response = await guides.subscription_guides_route(request) second_response = await guides.subscription_guides_route(request) body = json.loads(response.text) second_body = json.loads(second_response.text) self.assertTrue(body["enabled"]) self.assertTrue(second_body["enabled"]) self.assertEqual(body["source"], "panel") self.assertEqual(body["config"]["version"], "1") self.assertIn("windows", body["config"]["platforms"]) windows_apps = [app["name"] for app in body["config"]["platforms"]["windows"]["apps"]] self.assertIn("Throne", windows_apps) panel_service.get_subscription_page_config_list.assert_awaited_once() panel_service.get_subscription_page_config_by_uuid.assert_awaited_once_with(default_uuid) async def test_public_route_returns_shared_config_and_subscription_payload(self): default_uuid = "00000000-0000-0000-0000-000000000000" share_token = "8f559061460e8fede78ef18dce887236" panel_config = json.loads(default_subscription_guides_config_text()) panel_service = SimpleNamespace( get_subscription_page_config_list=AsyncMock( return_value={"configs": [{"uuid": default_uuid, "viewPosition": 1}]} ), get_subscription_page_config_by_uuid=AsyncMock( return_value={"uuid": default_uuid, "config": panel_config} ), get_user_by_uuid=AsyncMock( return_value={ "shortUuid": "share-short", "subscriptionUrl": "https://sb.example.test/share-short", "username": "demo", } ), ) request = self._request( self._settings(SUBSCRIPTION_MINI_APP_URL="https://app.example.test/app"), panel_service, match_info={"share_token": share_token}, ) local_sub = SimpleNamespace( panel_user_uuid="panel-user", install_share_token=share_token, is_active=True, end_date=datetime.now(timezone.utc) + timedelta(days=3), ) with patch.object( guides.subscription_dal, "get_subscription_by_install_share_token", AsyncMock(return_value=local_sub), ): response = await guides.public_subscription_guides_route(request) body = json.loads(response.text) self.assertTrue(body["enabled"]) self.assertEqual(body["subscription"]["config_link"], "https://sb.example.test/share-short") self.assertEqual( body["subscription"]["share_url"], f"https://app.example.test/s/{share_token}", ) self.assertEqual(body["subscription"]["install_share_token"], share_token) panel_service.get_user_by_uuid.assert_awaited_once_with("panel-user") async def test_public_route_rejects_unknown_share_token_without_loading_config(self): share_token = "8f559061460e8fede78ef18dce887236" panel_service = SimpleNamespace( get_subscription_page_config_list=AsyncMock(), get_subscription_page_config_by_uuid=AsyncMock(), get_user_by_uuid=AsyncMock(), ) request = self._request( self._settings(SUBSCRIPTION_MINI_APP_URL="https://app.example.test/app"), panel_service, match_info={"share_token": share_token}, ) with patch.object( guides.subscription_dal, "get_subscription_by_install_share_token", AsyncMock(return_value=None), ): response = await guides.public_subscription_guides_route(request) body = json.loads(response.text) self.assertEqual(response.status, 404) self.assertFalse(body["ok"]) self.assertEqual(body["error"], "subscription_unavailable") self.assertFalse(body["enabled"]) self.assertIsNone(body["config"]) self.assertEqual(body["subscription"]["install_share_token"], share_token) self.assertFalse(body["subscription"]["active"]) panel_service.get_subscription_page_config_list.assert_not_called() panel_service.get_subscription_page_config_by_uuid.assert_not_called() panel_service.get_user_by_uuid.assert_not_called() async def test_public_route_rejects_inactive_share_token_without_panel_user_lookup(self): share_token = "8f559061460e8fede78ef18dce887236" panel_service = SimpleNamespace( get_subscription_page_config_list=AsyncMock(), get_subscription_page_config_by_uuid=AsyncMock(), get_user_by_uuid=AsyncMock(), ) request = self._request( self._settings(SUBSCRIPTION_MINI_APP_URL="https://app.example.test/app"), panel_service, match_info={"share_token": share_token}, ) local_sub = SimpleNamespace( panel_user_uuid="panel-user", install_share_token=share_token, is_active=False, end_date=datetime.now(timezone.utc) + timedelta(days=3), ) with patch.object( guides.subscription_dal, "get_subscription_by_install_share_token", AsyncMock(return_value=local_sub), ): response = await guides.public_subscription_guides_route(request) body = json.loads(response.text) self.assertEqual(response.status, 404) self.assertFalse(body["ok"]) self.assertEqual(body["error"], "subscription_unavailable") self.assertFalse(body["subscription"]["active"]) panel_service.get_user_by_uuid.assert_not_called() if __name__ == "__main__": unittest.main()