{displayValue}
+ {user_id}\n📧 Email: {email}{referral_text}\n📅 Time: {timestamp}",
"log_account_email_linked": "📧 Email linked\n\n🆔 User ID: {user_id}\n📨 Telegram ID: {telegram_id}\n👤 User: {user_display}\n📧 Email: {email}\n🕐 Time: {timestamp}",
"log_account_telegram_linked": "📨 Telegram linked\n\n🆔 User ID: {user_id}\n📨 Telegram ID: {telegram_id}\n👤 User: {user_display}\n📧 Email: {email}\n🕐 Time: {timestamp}",
+ "log_account_merged": "🔗 Accounts merged\n\n🆔 Kept user ID: {primary_user_id}\n🗑 Removed user ID: {removed_user_id}\n📨 Telegram ID: {telegram_id}\n👤 User: {user_display}\n📧 Email: {email}\n⏰ New end date: {final_end_date}\n📋 Kept panel UUID: {primary_panel_user_uuid}\n📋 Removed panel UUID: {removed_panel_user_uuid}\n🕐 Time: {timestamp}",
"log_payment_received": "{provider_emoji} Payment Received\n\n👤 User: {user_display}\n💰 Amount: {amount} {currency}\n📅 Period: {months} mo.\n🏦 Provider: {payment_provider}\n🕐 Time: {timestamp}",
"log_payment_received_traffic": "{provider_emoji} Payment Received (traffic top-up)\n\n👤 User: {user_display}\n💰 Amount: {amount} {currency}\n🗂 {traffic_summary}\n{tariff_line}🏦 Provider: {payment_provider}\n🕐 Time: {timestamp}",
"log_payment_traffic_purchase_line": "{gb} GB · {kind}",
@@ -956,6 +957,11 @@
"admin_settings_subsection_cryptopay": "CryptoPay",
"admin_settings_subsection_wata": "Wata",
"admin_settings_subsection_heleket": "Heleket",
+ "admin_settings_provider_webhook_url": "Webhook URL",
+ "admin_settings_provider_webhook_url_hint": "Use this URL in the provider webhook settings.",
+ "admin_settings_provider_webhook_base_missing": "Set WEBHOOK_BASE_URL in .env to show the full URL for {path}.",
+ "admin_copy": "Copy",
+ "admin_copied": "Copied",
"admin_settings_validation_errors": "Errors: {errors}",
"admin_settings_save_error": "Error: {error}",
"admin_sync_started": "Synchronization started",
diff --git a/locales/ru.json b/locales/ru.json
index 2a480dc..f77c31f 100644
--- a/locales/ru.json
+++ b/locales/ru.json
@@ -363,6 +363,7 @@
"log_new_email_user_registration": "📧 Новый пользователь (email)\n\n🆔 ID: {user_id}\n📧 Email: {email}{referral_text}\n📅 Время: {timestamp}",
"log_account_email_linked": "📧 Привязана почта\n\n🆔 ID пользователя: {user_id}\n📨 Telegram ID: {telegram_id}\n👤 Пользователь: {user_display}\n📧 Email: {email}\n🕐 Время: {timestamp}",
"log_account_telegram_linked": "📨 Привязан Telegram\n\n🆔 ID пользователя: {user_id}\n📨 Telegram ID: {telegram_id}\n👤 Пользователь: {user_display}\n📧 Email: {email}\n🕐 Время: {timestamp}",
+ "log_account_merged": "🔗 Аккаунты объединены\n\n🆔 Оставлен ID: {primary_user_id}\n🗑 Удалён ID: {removed_user_id}\n📨 Telegram ID: {telegram_id}\n👤 Пользователь: {user_display}\n📧 Email: {email}\n⏰ Новая дата окончания: {final_end_date}\n📋 UUID оставленного в панели: {primary_panel_user_uuid}\n📋 UUID удалённого в панели: {removed_panel_user_uuid}\n🕐 Время: {timestamp}",
"log_payment_received": "{provider_emoji} Получен платеж\n\n👤 Пользователь: {user_display}\n💰 Сумма: {amount} {currency}\n📅 Период: {months} мес.\n🏦 Провайдер: {payment_provider}\n🕐 Время: {timestamp}",
"log_payment_received_traffic": "{provider_emoji} Получен платеж (докупка трафика)\n\n👤 Пользователь: {user_display}\n💰 Сумма: {amount} {currency}\n🗂 {traffic_summary}\n{tariff_line}🏦 Провайдер: {payment_provider}\n🕐 Время: {timestamp}",
"log_payment_traffic_purchase_line": "{gb} ГБ · {kind}",
@@ -956,6 +957,11 @@
"admin_settings_subsection_cryptopay": "CryptoPay",
"admin_settings_subsection_wata": "Wata",
"admin_settings_subsection_heleket": "Heleket",
+ "admin_settings_provider_webhook_url": "Webhook URL",
+ "admin_settings_provider_webhook_url_hint": "Укажите этот адрес в настройках вебхуков провайдера.",
+ "admin_settings_provider_webhook_base_missing": "Укажите WEBHOOK_BASE_URL в .env, чтобы увидеть полный адрес для {path}.",
+ "admin_copy": "Копировать",
+ "admin_copied": "Скопировано",
"admin_settings_validation_errors": "Ошибки: {errors}",
"admin_settings_save_error": "Ошибка: {error}",
"admin_sync_started": "Синхронизация запущена",
diff --git a/tests/test_account_linking_panel.py b/tests/test_account_linking_panel.py
new file mode 100644
index 0000000..46139fa
--- /dev/null
+++ b/tests/test_account_linking_panel.py
@@ -0,0 +1,343 @@
+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 # noqa: F401
+from bot.app.web.webapp import account as account_routes
+from bot.app.web.webapp.auth import (
+ _link_telegram_to_user,
+ _sync_merged_panel_identity_for_user,
+)
+
+
+class AccountLinkingPanelTests(unittest.IsolatedAsyncioTestCase):
+ class _AsyncSessionFactory:
+ def __init__(self):
+ self.session = SimpleNamespace(
+ commit=AsyncMock(),
+ rollback=AsyncMock(),
+ flush=AsyncMock(),
+ )
+
+ def __call__(self):
+ return self
+
+ async def __aenter__(self):
+ return self.session
+
+ async def __aexit__(self, exc_type, exc, tb):
+ return None
+
+ async def test_merged_panel_identity_deletes_source_before_updating_target(self):
+ calls = []
+
+ async def delete_source(*args, **kwargs):
+ calls.append("delete")
+ return True
+
+ async def update_target(*args, **kwargs):
+ calls.append("update")
+ return {"uuid": "panel-target"}
+
+ panel_service = SimpleNamespace(
+ delete_user_from_panel=AsyncMock(side_effect=delete_source),
+ update_user_details_on_panel=AsyncMock(side_effect=update_target),
+ )
+ request = SimpleNamespace(
+ app={"subscription_service": SimpleNamespace(panel_service=panel_service)}
+ )
+ user = SimpleNamespace(
+ user_id=42,
+ panel_user_uuid="panel-target",
+ telegram_id=42,
+ email="linked@example.com",
+ username="alice",
+ first_name="Alice",
+ last_name=None,
+ )
+
+ result = await _sync_merged_panel_identity_for_user(
+ request,
+ user,
+ source_panel_uuid="panel-source",
+ final_panel_uuid="panel-target",
+ )
+
+ self.assertTrue(result)
+ self.assertEqual(calls, ["delete", "update"])
+ panel_service.delete_user_from_panel.assert_awaited_once_with(
+ "panel-source",
+ log_response=False,
+ )
+ panel_service.update_user_details_on_panel.assert_awaited_once()
+ update_uuid, payload = panel_service.update_user_details_on_panel.await_args.args[:2]
+ self.assertEqual(update_uuid, "panel-target")
+ self.assertEqual(payload["email"], "linked@example.com")
+ self.assertEqual(payload["telegramId"], 42)
+
+ async def test_merged_panel_identity_reactivates_expired_target_with_transferred_time(self):
+ expire_at = datetime.now(timezone.utc) + timedelta(days=30)
+ panel_service = SimpleNamespace(
+ delete_user_from_panel=AsyncMock(return_value=True),
+ update_user_details_on_panel=AsyncMock(return_value={"uuid": "panel-target"}),
+ )
+ request = SimpleNamespace(
+ app={"subscription_service": SimpleNamespace(panel_service=panel_service)}
+ )
+ user = SimpleNamespace(
+ user_id=42,
+ panel_user_uuid="panel-target",
+ telegram_id=42,
+ email="linked@example.com",
+ username="alice",
+ first_name="Alice",
+ last_name=None,
+ )
+
+ result = await _sync_merged_panel_identity_for_user(
+ request,
+ user,
+ source_panel_uuid="panel-email",
+ final_panel_uuid="panel-target",
+ expire_at=expire_at,
+ )
+
+ self.assertTrue(result)
+ panel_service.delete_user_from_panel.assert_awaited_once_with(
+ "panel-email",
+ log_response=False,
+ )
+ _, payload = panel_service.update_user_details_on_panel.await_args.args[:2]
+ expected_expire_at = expire_at.isoformat(timespec="milliseconds").replace("+00:00", "Z")
+ self.assertEqual(payload["expireAt"], expected_expire_at)
+ self.assertEqual(payload["status"], "ACTIVE")
+
+ async def test_telegram_merge_defers_panel_sync_until_source_cleanup(self):
+ current_user = SimpleNamespace(
+ user_id=-100,
+ email="linked@example.com",
+ email_verified_at=None,
+ panel_user_uuid="panel-source",
+ telegram_id=None,
+ username=None,
+ first_name=None,
+ last_name=None,
+ language_code="ru",
+ telegram_photo_url=None,
+ )
+ existing_telegram_user = SimpleNamespace(
+ user_id=42,
+ email=None,
+ email_verified_at=None,
+ panel_user_uuid="panel-target",
+ telegram_id=42,
+ username="old",
+ first_name=None,
+ last_name=None,
+ language_code="ru",
+ telegram_photo_url=None,
+ )
+ merged_user = SimpleNamespace(
+ user_id=42,
+ email="linked@example.com",
+ email_verified_at=None,
+ panel_user_uuid="panel-target",
+ telegram_id=42,
+ username="old",
+ first_name=None,
+ last_name=None,
+ language_code="ru",
+ telegram_photo_url=None,
+ )
+ panel_service = SimpleNamespace(update_user_details_on_panel=AsyncMock())
+ request = SimpleNamespace(
+ app={"subscription_service": SimpleNamespace(panel_service=panel_service)}
+ )
+ session = SimpleNamespace(flush=AsyncMock())
+ telegram_user = {
+ "id": 42,
+ "username": "alice",
+ "first_name": "Alice",
+ "last_name": "",
+ "language_code": "ru",
+ }
+
+ with (
+ patch(
+ "bot.app.web.webapp.auth.user_dal.get_user_by_id",
+ AsyncMock(return_value=current_user),
+ ),
+ patch(
+ "bot.app.web.webapp.auth.user_dal.get_user_by_telegram_id",
+ AsyncMock(return_value=existing_telegram_user),
+ ),
+ patch(
+ "bot.app.web.webapp.auth.user_dal.merge_users",
+ AsyncMock(return_value=merged_user),
+ ),
+ ):
+ result = await _link_telegram_to_user(
+ request,
+ session,
+ current_user_id=-100,
+ telegram_user=telegram_user,
+ settings=SimpleNamespace(DEFAULT_LANGUAGE="ru"),
+ )
+
+ self.assertIs(result, merged_user)
+ panel_service.update_user_details_on_panel.assert_not_awaited()
+ self.assertEqual(merged_user.username, "alice")
+
+ async def test_email_only_session_can_link_existing_telegram_only_account(self):
+ email_user = SimpleNamespace(
+ user_id=-100,
+ email="linked@example.com",
+ email_verified_at=object(),
+ panel_user_uuid="panel-email",
+ telegram_id=None,
+ username=None,
+ first_name=None,
+ last_name=None,
+ language_code="ru",
+ telegram_photo_url=None,
+ is_banned=False,
+ )
+ telegram_user_record = SimpleNamespace(
+ user_id=42,
+ email=None,
+ email_verified_at=None,
+ panel_user_uuid="panel-telegram",
+ telegram_id=42,
+ username="old",
+ first_name=None,
+ last_name=None,
+ language_code="ru",
+ telegram_photo_url=None,
+ is_banned=False,
+ )
+ merged_user = SimpleNamespace(
+ user_id=42,
+ email="linked@example.com",
+ email_verified_at=object(),
+ panel_user_uuid="panel-telegram",
+ telegram_id=42,
+ username="old",
+ first_name=None,
+ last_name=None,
+ language_code="ru",
+ telegram_photo_url=None,
+ is_banned=False,
+ )
+ panel_calls = []
+
+ async def delete_source(*args, **kwargs):
+ panel_calls.append("delete")
+ return True
+
+ async def update_target(*args, **kwargs):
+ panel_calls.append("update")
+ return {"uuid": "panel-telegram"}
+
+ panel_service = SimpleNamespace(
+ delete_user_from_panel=AsyncMock(side_effect=delete_source),
+ update_user_details_on_panel=AsyncMock(side_effect=update_target),
+ )
+ settings = SimpleNamespace(
+ WEBAPP_SESSION_SECRET="session-secret",
+ WEBAPP_SESSION_TTL_SECONDS=3600,
+ REDIS_URL=None,
+ REDIS_KEY_PREFIX="test",
+ DEFAULT_LANGUAGE="ru",
+ )
+ request = SimpleNamespace(
+ app={
+ "settings": settings,
+ "async_session_factory": self._AsyncSessionFactory(),
+ "subscription_service": SimpleNamespace(panel_service=panel_service),
+ "email_auth_service": None,
+ "i18n": None,
+ "bot": SimpleNamespace(),
+ },
+ json=AsyncMock(return_value={"init_data": "telegram-init-data"}),
+ )
+ telegram_auth_payload = {
+ "id": 42,
+ "username": "alice",
+ "first_name": "Alice",
+ "last_name": "",
+ "language_code": "ru",
+ }
+ notification_service = SimpleNamespace(
+ notify_account_telegram_linked=AsyncMock(),
+ notify_account_merged=AsyncMock(),
+ )
+
+ with (
+ patch.object(account_routes, "_require_user_id", return_value=-100),
+ patch.object(
+ account_routes,
+ "_validate_telegram_auth_payload",
+ AsyncMock(return_value=telegram_auth_payload),
+ ),
+ patch.object(
+ account_routes.user_dal,
+ "get_user_by_id",
+ AsyncMock(return_value=email_user),
+ ),
+ patch.object(
+ account_routes.user_dal,
+ "get_user_by_telegram_id",
+ AsyncMock(return_value=telegram_user_record),
+ ),
+ patch.object(
+ account_routes.user_dal,
+ "merge_users",
+ AsyncMock(return_value=merged_user),
+ ) as merge_users,
+ patch.object(
+ account_routes.subscription_dal,
+ "get_active_subscription_by_user_id",
+ AsyncMock(return_value=None),
+ ),
+ patch(
+ "bot.services.notification_service.NotificationService",
+ return_value=notification_service,
+ ),
+ ):
+ response = await account_routes.account_telegram_link_route(request)
+
+ self.assertEqual(response.status, 200)
+ payload = json.loads(response.text)
+ self.assertTrue(payload["ok"])
+ self.assertEqual(payload["user_id"], 42)
+ self.assertEqual(payload["telegram_id"], 42)
+ self.assertEqual(payload["account_merge"]["removed_user_id"], -100)
+ self.assertEqual(payload["account_merge"]["primary_user_id"], 42)
+ merge_users.assert_awaited_once_with(
+ request.app["async_session_factory"].session,
+ source_user_id=-100,
+ target_user_id=42,
+ )
+ self.assertEqual(panel_calls, ["delete", "update"])
+ panel_service.delete_user_from_panel.assert_awaited_once_with(
+ "panel-email",
+ log_response=False,
+ )
+ update_uuid, update_payload = panel_service.update_user_details_on_panel.await_args.args[:2]
+ self.assertEqual(update_uuid, "panel-telegram")
+ self.assertEqual(update_payload["email"], "linked@example.com")
+ self.assertEqual(update_payload["telegramId"], 42)
+ notification_service.notify_account_merged.assert_awaited_once_with(
+ primary_user_id=42,
+ removed_user_id=-100,
+ email="linked@example.com",
+ telegram_id=42,
+ username="alice",
+ first_name="Alice",
+ final_end_date_text="",
+ primary_panel_user_uuid="panel-telegram",
+ removed_panel_user_uuid="panel-email",
+ )
+ self.assertIn("rw_webapp_session", response.cookies)
diff --git a/tests/test_admin_settings_manifest_i18n.py b/tests/test_admin_settings_manifest_i18n.py
index 8633a66..39d55e0 100644
--- a/tests/test_admin_settings_manifest_i18n.py
+++ b/tests/test_admin_settings_manifest_i18n.py
@@ -93,3 +93,13 @@ def test_subscription_guide_settings_i18n_keys_exist():
field = manifest[setting_key]
assert field["i18n_label_key"] in messages
assert field["i18n_description_key"] in messages
+
+
+def test_payment_provider_settings_include_webhook_metadata():
+ manifest = _manifest_by_key()
+
+ assert manifest["FREEKASSA_ENABLED"]["webhook_path"] == "/webhook/freekassa"
+ assert manifest["FREEKASSA_ENABLED"]["provider_id"] == "freekassa"
+ assert manifest["PAYMENT_PLATEGA_CRYPTO_WEBAPP_LABEL_RU"]["webhook_path"] == "/webhook/platega"
+ assert manifest["YOOKASSA_SHOP_ID"]["webhook_requires_base_url"] is True
+ assert "webhook_path" not in manifest["PAYMENT_STARS_WEBAPP_LABEL_RU"]
diff --git a/tests/test_admin_sync_performance.py b/tests/test_admin_sync_performance.py
index 68a7ecd..ce9e156 100644
--- a/tests/test_admin_sync_performance.py
+++ b/tests/test_admin_sync_performance.py
@@ -1,7 +1,10 @@
+import asyncio
from datetime import datetime, timedelta, timezone
from types import SimpleNamespace
+from unittest.mock import AsyncMock, patch
from bot.handlers.admin.sync_admin import (
+ _absorb_duplicate_panel_identity,
_coerce_panel_telegram_id,
_description_matches,
_should_update_lifetime_used_traffic,
@@ -14,6 +17,20 @@ def test_description_match_ignores_whitespace_shape():
assert _description_matches("email@example.com username", "email@example.com\nusername")
+def test_description_match_accepts_cp1251_mojibake_from_panel():
+ desired = "user@example.com\nalice\nАлексей\nЧерников"
+ panel_value = "user@example.com\nalice\nÀëåêñåé\n×åðíèêîâ"
+
+ assert _description_matches(panel_value, desired)
+
+
+def test_description_match_rejects_different_identity_after_mojibake_repair():
+ desired = "user@example.com\nalice\nАлексей"
+ panel_value = "other@example.com\nalice\nÀëåêñåé"
+
+ assert not _description_matches(panel_value, desired)
+
+
def test_panel_telegram_id_is_coerced_to_int():
assert _coerce_panel_telegram_id("12345") == 12345
assert _coerce_panel_telegram_id("") is None
@@ -122,3 +139,98 @@ def test_lifetime_traffic_update_allows_large_delta_and_skips_duplicate_panel_id
settings=settings,
is_duplicate_panel_identity=True,
)
+
+
+def test_absorb_duplicate_panel_identity_extends_kept_user_and_deletes_duplicate():
+ now = datetime.now(timezone.utc)
+ target_sub = SimpleNamespace(
+ subscription_id=10,
+ user_id=42,
+ panel_user_uuid="panel-keep",
+ panel_subscription_uuid="sub-keep",
+ end_date=now - timedelta(days=2),
+ is_active=False,
+ status_from_panel="EXPIRED",
+ )
+ duplicate_sub = SimpleNamespace(
+ subscription_id=11,
+ user_id=42,
+ panel_user_uuid="panel-duplicate",
+ panel_subscription_uuid="sub-duplicate",
+ end_date=now + timedelta(days=30),
+ is_active=True,
+ skip_notifications=False,
+ status_from_panel="ACTIVE",
+ )
+ panel_service = SimpleNamespace(
+ update_user_details_on_panel=AsyncMock(return_value={"uuid": "panel-keep"}),
+ delete_user_from_panel=AsyncMock(return_value=True),
+ )
+ session = SimpleNamespace(execute=AsyncMock())
+ settings = SimpleNamespace(user_traffic_limit_bytes=0)
+ user = SimpleNamespace(
+ user_id=42,
+ panel_user_uuid="panel-keep",
+ telegram_id=969808056,
+ email="paid@example.com",
+ username="alice",
+ first_name="Alice",
+ last_name=None,
+ )
+
+ async def update_subscription(_session, subscription_id, update_data):
+ sub = target_sub if subscription_id == target_sub.subscription_id else duplicate_sub
+ for key, value in update_data.items():
+ setattr(sub, key, value)
+ return sub
+
+ with patch(
+ "bot.handlers.admin.sync_admin.subscription_dal.update_subscription",
+ AsyncMock(side_effect=update_subscription),
+ ):
+ result = asyncio.run(
+ _absorb_duplicate_panel_identity(
+ session,
+ panel_service=panel_service,
+ existing_user=user,
+ keep_panel_uuid="panel-keep",
+ keep_panel_user={
+ "uuid": "panel-keep",
+ "subscriptionUuid": "sub-keep",
+ "status": "EXPIRED",
+ "expireAt": (now - timedelta(days=2)).isoformat(),
+ },
+ duplicate_panel_user={
+ "uuid": "panel-duplicate",
+ "subscriptionUuid": "sub-duplicate",
+ "telegramId": 969808056,
+ "status": "ACTIVE",
+ "expireAt": (now + timedelta(days=30)).isoformat(),
+ },
+ settings=settings,
+ subscriptions_by_panel_uuid={
+ "sub-keep": target_sub,
+ "sub-duplicate": duplicate_sub,
+ },
+ active_subscriptions_by_user_panel={},
+ )
+ )
+
+ assert result["resolved"]
+ assert result["subscriptions_updated"] == 2
+ assert target_sub.is_active
+ assert target_sub.status_from_panel == "ACTIVE_EXTENDED_BY_PANEL_DUPLICATE_MERGE"
+ assert target_sub.panel_user_uuid == "panel-keep"
+ assert target_sub.end_date > now + timedelta(days=29)
+ assert not duplicate_sub.is_active
+ assert duplicate_sub.skip_notifications
+ assert duplicate_sub.status_from_panel == "MERGED_PANEL_DUPLICATE"
+ panel_service.update_user_details_on_panel.assert_awaited_once()
+ update_uuid, update_payload = panel_service.update_user_details_on_panel.await_args.args[:2]
+ assert update_uuid == "panel-keep"
+ assert update_payload["status"] == "ACTIVE"
+ assert update_payload["telegramId"] == 969808056
+ panel_service.delete_user_from_panel.assert_awaited_once_with(
+ "panel-duplicate",
+ log_response=False,
+ )
diff --git a/tests/test_app_version_resolution.py b/tests/test_app_version_resolution.py
index 098792e..defb55d 100644
--- a/tests/test_app_version_resolution.py
+++ b/tests/test_app_version_resolution.py
@@ -1,20 +1,21 @@
-"""Tests for ``_resolve_app_version`` — the source of the admin sidebar footer.
+"""Tests for ``_resolve_app_version``: the admin sidebar footer source.
The admin sidebar shows ``{appVersion}`` next to a "remnawave-minishop" GitHub
-link. That string is rendered by
-``backend/bot/app/web/webapp/assets.py::_resolve_app_version`` through the
-following precedence chain:
+link. Release builds from ``main`` should render the latest reachable tag and
+commit sha, without a dirty suffix. Builds from other branches include the
+branch name. That string is rendered by
+``backend/bot/app/web/webapp/assets.py::_resolve_app_version`` through this
+precedence chain:
- 1. ``REMNAWAVE_MINISHOP_VERSION`` env var — manual override;
- 2. ``$APP_ROOT/.build-version`` file — baked at Docker build time by the
- ``version-builder`` stage in ``deploy/docker/Dockerfile`` (consumes .git
- in a throwaway stage and ships only this one tiny file);
- 3. live ``git describe`` / ``rev-parse`` — works in local dev where .git
- is present;
+ 1. ``REMNAWAVE_MINISHOP_VERSION`` env var: manual override;
+ 2. ``$APP_ROOT/.build-version`` file: baked at Docker build time by the
+ ``version-builder`` stage in ``deploy/docker/Dockerfile``;
+ 3. live ``git describe`` / ``rev-parse``: local dev fallback where .git is
+ present;
4. ``"dev+unknown"`` as the ultimate fallback.
-Before the build-time bake, the admin footer in production silently fell back
-to step 4 because the Docker image carries no .git tree and no git binary.
+The Docker image carries no git binary and no .git tree at runtime, so the
+baked ``.build-version`` file is the production source for the sidebar.
"""
import importlib
@@ -28,11 +29,20 @@ from unittest.mock import patch
import bot.app.web.subscription_webapp # noqa: F401
from bot.app.web.webapp import assets as assets_module
+_VERSION_ENV_NAMES = (
+ "REMNAWAVE_MINISHOP_VERSION",
+ "REMNAWAVE_MINISHOP_BRANCH",
+ "GIT_BRANCH",
+ "BRANCH_NAME",
+ "GITHUB_REF_NAME",
+ "CI_COMMIT_REF_NAME",
+)
+
def _reset_cache() -> None:
# The resolver memoizes the first result in a module-level global.
assets_module._APP_VERSION_CACHE = None # type: ignore[attr-defined]
- # Some callers reach through the facade re-export — clear that too.
+ # Some callers reach through the facade re-export; clear that too.
runtime = importlib.import_module("bot.app.web.webapp._runtime")
runtime._APP_VERSION_CACHE = None # type: ignore[attr-defined]
@@ -41,6 +51,13 @@ def _resolve():
return assets_module._resolve_app_version()
+def _clean_version_env() -> dict:
+ env = dict(os.environ)
+ for name in _VERSION_ENV_NAMES:
+ env.pop(name, None)
+ return env
+
+
class EnvOverrideTests(unittest.TestCase):
def setUp(self) -> None:
_reset_cache()
@@ -62,12 +79,13 @@ class EnvOverrideTests(unittest.TestCase):
self.assertEqual(called["git"], 0)
def test_blank_env_var_falls_through(self):
- env = {"REMNAWAVE_MINISHOP_VERSION": " "}
+ env = _clean_version_env()
+ env["REMNAWAVE_MINISHOP_VERSION"] = " "
with (
- patch.dict(os.environ, env),
+ patch.dict(os.environ, env, clear=True),
patch.object(assets_module, "_run_git_command", lambda *a: ""),
):
- # No env, no file, no git → fallback string.
+ # No env, no file, no git: fallback string.
self.assertEqual(_resolve(), "dev+unknown")
@@ -80,24 +98,22 @@ class BuildVersionFileTests(unittest.TestCase):
def test_reads_baked_version_file(self):
with tempfile.TemporaryDirectory() as tmp:
- (Path(tmp) / ".build-version").write_text("v3.4.5+12.gabcdef1", encoding="utf-8")
- env = dict(os.environ)
- env.pop("REMNAWAVE_MINISHOP_VERSION", None)
+ (Path(tmp) / ".build-version").write_text("v3.4.5+gabcdef1", encoding="utf-8")
+ env = _clean_version_env()
with (
patch.dict(os.environ, env, clear=True),
patch.object(assets_module, "APP_ROOT", Path(tmp)),
- # ensure live git doesn't accidentally win if file read fails:
+ # Ensure live git does not accidentally win if file read fails.
patch.object(assets_module, "_run_git_command", lambda *a: ""),
):
- self.assertEqual(_resolve(), "v3.4.5+12.gabcdef1")
+ self.assertEqual(_resolve(), "v3.4.5+gabcdef1")
def test_strips_trailing_whitespace_and_newlines_in_file(self):
- # Shell ``printf '%s'`` writes no newline, but earlier helpers used
- # ``echo`` which appends one. Both must produce the same result.
+ # Shell ``printf '%s'`` writes no newline, but older helpers may have
+ # used ``echo``. Both must produce the same result.
with tempfile.TemporaryDirectory() as tmp:
(Path(tmp) / ".build-version").write_text("v1.2.3\n\n", encoding="utf-8")
- env = dict(os.environ)
- env.pop("REMNAWAVE_MINISHOP_VERSION", None)
+ env = _clean_version_env()
with (
patch.dict(os.environ, env, clear=True),
patch.object(assets_module, "APP_ROOT", Path(tmp)),
@@ -108,14 +124,13 @@ class BuildVersionFileTests(unittest.TestCase):
def test_empty_file_falls_through_to_git_then_unknown(self):
with tempfile.TemporaryDirectory() as tmp:
(Path(tmp) / ".build-version").write_text("", encoding="utf-8")
- env = dict(os.environ)
- env.pop("REMNAWAVE_MINISHOP_VERSION", None)
+ env = _clean_version_env()
with (
patch.dict(os.environ, env, clear=True),
patch.object(assets_module, "APP_ROOT", Path(tmp)),
patch.object(assets_module, "_run_git_command", lambda *a: ""),
):
- # No env, empty file, no live git → ultimate fallback.
+ # No env, empty file, no live git: ultimate fallback.
self.assertEqual(_resolve(), "dev+unknown")
@@ -126,56 +141,83 @@ class LiveGitFallbackTests(unittest.TestCase):
_reset_cache()
self.addCleanup(_reset_cache)
- def _run_with_git(self, replies: dict, *, dirty: bool = False) -> str:
- # Map (subcommand, *args) tuples to canned stdout values.
- def fake_git(*args):
- return replies.get(args, "")
-
+ def _run_with_git(self, replies: dict) -> str:
with tempfile.TemporaryDirectory() as tmp:
# ``.build-version`` deliberately absent so we fall through.
- env = dict(os.environ)
- env.pop("REMNAWAVE_MINISHOP_VERSION", None)
+ env = _clean_version_env()
base = {
("describe", "--tags", "--abbrev=0"): replies.get("tag", ""),
("rev-parse", "--short", "HEAD"): replies.get("sha", ""),
- ("status", "--porcelain"): "M file\n" if dirty else "",
- ("rev-list", f"{replies.get('tag', '')}..HEAD", "--count"): replies.get(
- "commits_since_tag", "0"
- ),
+ ("branch", "--show-current"): replies.get("branch", ""),
}
- def real_fake_git(*args):
+ def fake_git(*args):
return base.get(args, "")
with (
patch.dict(os.environ, env, clear=True),
patch.object(assets_module, "APP_ROOT", Path(tmp)),
- patch.object(assets_module, "_run_git_command", real_fake_git),
+ patch.object(assets_module, "_run_git_command", fake_git),
):
return _resolve()
- def test_tag_with_zero_commits_since_returns_bare_tag(self):
- result = self._run_with_git({"tag": "v2.0.0", "sha": "abcdef1", "commits_since_tag": "0"})
- self.assertEqual(result, "v2.0.0")
+ def test_tag_with_sha_returns_tag_plus_sha(self):
+ result = self._run_with_git({"tag": "v2.0.0", "sha": "abcdef1"})
+ self.assertEqual(result, "v2.0.0+gabcdef1")
- def test_tag_plus_distance_plus_sha_format(self):
+ def test_main_branch_does_not_add_branch_suffix(self):
+ result = self._run_with_git({"tag": "v2.0.0", "sha": "abcdef1", "branch": "main"})
+ self.assertEqual(result, "v2.0.0+gabcdef1")
+
+ def test_non_main_branch_adds_branch_suffix(self):
+ result = self._run_with_git({"tag": "v2.0.0", "sha": "abcdef1", "branch": "dev"})
+ self.assertEqual(result, "v2.0.0-dev+gabcdef1")
+
+ def test_branch_name_is_sanitized_for_version(self):
+ result = self._run_with_git(
+ {"tag": "v2.0.0", "sha": "abcdef1", "branch": "feature/cool build"}
+ )
+ self.assertEqual(result, "v2.0.0-feature-cool-build+gabcdef1")
+
+ def test_commit_distance_is_not_included(self):
result = self._run_with_git({"tag": "v2.0.0", "sha": "abcdef1", "commits_since_tag": "7"})
- self.assertEqual(result, "v2.0.0+7.gabcdef1")
+ self.assertEqual(result, "v2.0.0+gabcdef1")
def test_sha_only_when_no_tag(self):
result = self._run_with_git({"sha": "abcdef1"})
self.assertEqual(result, "dev+gabcdef1")
+ def test_tag_only_when_no_sha(self):
+ result = self._run_with_git({"tag": "v2.0.0"})
+ self.assertEqual(result, "v2.0.0")
+
def test_neither_tag_nor_sha_is_unknown(self):
result = self._run_with_git({})
self.assertEqual(result, "dev+unknown")
- def test_dirty_suffix_is_appended(self):
- result = self._run_with_git(
- {"tag": "v2.0.0", "sha": "abcdef1", "commits_since_tag": "0"},
- dirty=True,
- )
- self.assertEqual(result, "v2.0.0-dirty")
+ def test_dirty_suffix_is_not_appended_or_queried(self):
+ calls = []
+
+ def fake_git(*args):
+ calls.append(args)
+ replies = {
+ ("describe", "--tags", "--abbrev=0"): "v2.0.0",
+ ("rev-parse", "--short", "HEAD"): "abcdef1",
+ ("status", "--porcelain"): "M file\n",
+ }
+ return replies.get(args, "")
+
+ with tempfile.TemporaryDirectory() as tmp:
+ env = _clean_version_env()
+ with (
+ patch.dict(os.environ, env, clear=True),
+ patch.object(assets_module, "APP_ROOT", Path(tmp)),
+ patch.object(assets_module, "_run_git_command", fake_git),
+ ):
+ result = _resolve()
+
+ self.assertEqual(result, "v2.0.0+gabcdef1")
+ self.assertNotIn(("status", "--porcelain"), calls)
class CacheBehaviourTests(unittest.TestCase):
@@ -191,8 +233,7 @@ class CacheBehaviourTests(unittest.TestCase):
return "abcdef1" if args == ("rev-parse", "--short", "HEAD") else ""
with tempfile.TemporaryDirectory() as tmp:
- env = dict(os.environ)
- env.pop("REMNAWAVE_MINISHOP_VERSION", None)
+ env = _clean_version_env()
with (
patch.dict(os.environ, env, clear=True),
patch.object(assets_module, "APP_ROOT", Path(tmp)),
diff --git a/tests/test_security.py b/tests/test_security.py
index 2b15d38..61cfb6c 100644
--- a/tests/test_security.py
+++ b/tests/test_security.py
@@ -722,6 +722,61 @@ class AdminSettingsSecurityTests(unittest.IsolatedAsyncioTestCase):
self.assertTrue(secret_field["has_value"])
self.assertNotIn("super-secret", response.text)
+ async def test_admin_settings_exposes_payment_webhook_urls(self):
+ class AsyncSessionFactory:
+ def __call__(self):
+ return self
+
+ async def __aenter__(self):
+ return object()
+
+ async def __aexit__(self, exc_type, exc, tb):
+ return False
+
+ settings = Settings(
+ _env_file=None,
+ BOT_TOKEN="token",
+ POSTGRES_USER="app_user",
+ POSTGRES_PASSWORD="app_password",
+ SHOP_NAME="Visible shop",
+ WEBHOOK_BASE_URL="https://web.tnnl.cc/",
+ )
+ request = SimpleNamespace(
+ app={"settings": settings, "async_session_factory": AsyncSessionFactory()},
+ headers={},
+ cookies={},
+ admin_telegram_id=1,
+ )
+ request.get = lambda key, default=None: getattr(request, key, default)
+
+ with (
+ patch.object(admin_settings_routes, "_require_admin_user_id", return_value=1),
+ patch.object(
+ admin_api.app_settings_dal,
+ "get_overrides_with_meta",
+ AsyncMock(return_value=[]),
+ ),
+ ):
+ response = await admin_api.admin_settings_get_route(request)
+
+ payload = json.loads(response.text)
+ fields = {
+ field["key"]: field
+ for section in payload["sections"]
+ for field in section["fields"]
+ }
+
+ self.assertEqual(
+ fields["FREEKASSA_ENABLED"]["webhook_url"],
+ "https://web.tnnl.cc/webhook/freekassa",
+ )
+ self.assertTrue(fields["FREEKASSA_ENABLED"]["webhook_base_url_configured"])
+ self.assertEqual(
+ fields["PAYMENT_PLATEGA_CRYPTO_WEBAPP_LABEL_RU"]["webhook_url"],
+ "https://web.tnnl.cc/webhook/platega",
+ )
+ self.assertNotIn("webhook_url", fields["PAYMENT_STARS_WEBAPP_LABEL_RU"])
+
class DatabaseLoggingSecurityTests(unittest.TestCase):
def test_database_url_redaction_hides_password(self):
diff --git a/tests/test_support_notifications.py b/tests/test_support_notifications.py
index 35d94eb..d73e4c1 100644
--- a/tests/test_support_notifications.py
+++ b/tests/test_support_notifications.py
@@ -364,3 +364,51 @@ def test_support_user_reply_can_send_email_without_telegram_channels():
assert channels == []
assert emails[0]["ticket_id"] == 7
+
+
+def test_account_merge_notification_goes_to_log_channel():
+ messages = []
+
+ class I18n:
+ def gettext(self, _language, key, **kwargs):
+ if key == "log_open_profile_link":
+ return "Open profile"
+ assert key == "log_account_merged"
+ return (
+ f"merged primary={kwargs['primary_user_id']} "
+ f"removed={kwargs['removed_user_id']} "
+ f"email={kwargs['email']} end={kwargs['final_end_date']}"
+ )
+
+ service = NotificationService(
+ bot=SimpleNamespace(),
+ settings=_settings(LOG_CHAT_ID=-100123, DEFAULT_LANGUAGE="en"),
+ i18n=I18n(),
+ )
+
+ async def send_to_log_channel(message, thread_id=None, reply_markup=None):
+ messages.append((message, thread_id, reply_markup))
+
+ service._send_to_log_channel = send_to_log_channel
+
+ asyncio.run(
+ service.notify_account_merged(
+ primary_user_id=42,
+ removed_user_id=-100,
+ email="paid@example.com",
+ telegram_id=100200300,
+ username="alice",
+ first_name="Alice",
+ final_end_date_text="2026-06-21 10:00",
+ primary_panel_user_uuid="panel-telegram",
+ removed_panel_user_uuid="panel-email",
+ )
+ )
+
+ assert len(messages) == 1
+ message, thread_id, reply_markup = messages[0]
+ assert "primary=42" in message
+ assert "removed=-100" in message
+ assert "paid@example.com" in message
+ assert thread_id is None
+ assert reply_markup.inline_keyboard[0][0].url == "tg://user?id=100200300"
diff --git a/tests/test_tariff_worker.py b/tests/test_tariff_worker.py
index ab65078..8e2245a 100644
--- a/tests/test_tariff_worker.py
+++ b/tests/test_tariff_worker.py
@@ -1,3 +1,4 @@
+import asyncio
import json
import tempfile
import unittest
@@ -617,3 +618,107 @@ class TariffWorkerTests(unittest.IsolatedAsyncioTestCase):
self.assertIsNone(result)
panel_service.get_all_panel_users.assert_not_awaited()
+
+ async def test_missing_panel_subscription_repairs_to_user_panel_uuid(self):
+ panel_service = AsyncMock(spec=PanelApiService)
+ worker = TariffTrafficWorker(
+ settings=SimpleNamespace(),
+ session_factory=SimpleNamespace(),
+ panel_service=panel_service,
+ subscription_service=SimpleNamespace(),
+ )
+ sub = SimpleNamespace(
+ subscription_id=10,
+ user_id=123,
+ panel_user_uuid="old-panel",
+ is_active=True,
+ status_from_panel="ACTIVE",
+ skip_notifications=False,
+ )
+ panel_user = {"uuid": "new-panel", "username": "tg_123"}
+
+ with patch(
+ "bot.services.tariff_worker.user_dal.get_user_by_id",
+ new=AsyncMock(return_value=SimpleNamespace(panel_user_uuid="new-panel")),
+ ):
+ result = await worker._repair_missing_panel_user_for_subscription(
+ AsyncMock(),
+ sub,
+ panel_users_by_uuid={"new-panel": panel_user},
+ semaphore=asyncio.Semaphore(1),
+ confirmed_missing=True,
+ )
+
+ self.assertEqual(result, panel_user)
+ self.assertEqual(sub.panel_user_uuid, "new-panel")
+ self.assertTrue(sub.is_active)
+ panel_service.get_user_by_uuid.assert_not_awaited()
+
+ async def test_missing_panel_subscription_deactivates_when_bulk_prefetch_confirms_absent(self):
+ panel_service = AsyncMock(spec=PanelApiService)
+ worker = TariffTrafficWorker(
+ settings=SimpleNamespace(),
+ session_factory=SimpleNamespace(),
+ panel_service=panel_service,
+ subscription_service=SimpleNamespace(),
+ )
+ sub = SimpleNamespace(
+ subscription_id=11,
+ user_id=123,
+ panel_user_uuid="missing-panel",
+ is_active=True,
+ status_from_panel="ACTIVE",
+ skip_notifications=False,
+ )
+
+ with patch(
+ "bot.services.tariff_worker.user_dal.get_user_by_id",
+ new=AsyncMock(return_value=SimpleNamespace(panel_user_uuid="missing-panel")),
+ ):
+ result = await worker._repair_missing_panel_user_for_subscription(
+ AsyncMock(),
+ sub,
+ panel_users_by_uuid={},
+ semaphore=asyncio.Semaphore(1),
+ confirmed_missing=True,
+ )
+
+ self.assertEqual(result, {})
+ self.assertFalse(sub.is_active)
+ self.assertTrue(sub.skip_notifications)
+ self.assertEqual(sub.status_from_panel, "PANEL_USER_NOT_FOUND")
+
+ async def test_missing_panel_subscription_only_skips_when_absence_is_not_confirmed(self):
+ panel_service = AsyncMock(spec=PanelApiService)
+ panel_service.get_user_by_uuid = AsyncMock(return_value=None)
+ worker = TariffTrafficWorker(
+ settings=SimpleNamespace(),
+ session_factory=SimpleNamespace(),
+ panel_service=panel_service,
+ subscription_service=SimpleNamespace(),
+ )
+ sub = SimpleNamespace(
+ subscription_id=12,
+ user_id=123,
+ panel_user_uuid="missing-panel",
+ is_active=True,
+ status_from_panel="ACTIVE",
+ skip_notifications=False,
+ )
+
+ with patch(
+ "bot.services.tariff_worker.user_dal.get_user_by_id",
+ new=AsyncMock(return_value=SimpleNamespace(panel_user_uuid="missing-panel")),
+ ):
+ result = await worker._repair_missing_panel_user_for_subscription(
+ AsyncMock(),
+ sub,
+ panel_users_by_uuid=None,
+ semaphore=asyncio.Semaphore(1),
+ confirmed_missing=False,
+ )
+
+ self.assertEqual(result, {})
+ self.assertTrue(sub.is_active)
+ self.assertFalse(sub.skip_notifications)
+ self.assertEqual(sub.status_from_panel, "ACTIVE")
diff --git a/tests/test_user_dal.py b/tests/test_user_dal.py
index 7f62f4d..3ff2223 100644
--- a/tests/test_user_dal.py
+++ b/tests/test_user_dal.py
@@ -1,5 +1,5 @@
import unittest
-from datetime import datetime, timezone
+from datetime import datetime, timedelta, timezone
from types import SimpleNamespace
from unittest.mock import AsyncMock, patch
@@ -131,3 +131,109 @@ class UserDalMergeTests(unittest.IsolatedAsyncioTestCase):
self.assertIn("user_payment_methods", delete_tables)
self.assertIn("promo_code_activations", delete_tables)
session.delete.assert_awaited_once_with(source)
+
+ async def test_merge_users_moves_active_email_subscription_onto_expired_telegram_account(self):
+ before = datetime.now(timezone.utc)
+ source = SimpleNamespace(
+ user_id=-100,
+ email="paid@example.com",
+ telegram_id=None,
+ panel_user_uuid="panel-email",
+ email_verified_at=before,
+ username=None,
+ first_name=None,
+ last_name=None,
+ language_code="ru",
+ telegram_photo_url=None,
+ channel_subscription_verified=False,
+ channel_subscription_checked_at=None,
+ channel_subscription_verified_for=None,
+ lifetime_used_traffic_bytes=0,
+ referred_by_id=None,
+ referral_code=None,
+ )
+ target = SimpleNamespace(
+ user_id=42,
+ email=None,
+ telegram_id=42,
+ panel_user_uuid="panel-telegram",
+ email_verified_at=None,
+ username="old",
+ first_name=None,
+ last_name=None,
+ language_code="ru",
+ telegram_photo_url=None,
+ channel_subscription_verified=False,
+ channel_subscription_checked_at=None,
+ channel_subscription_verified_for=None,
+ lifetime_used_traffic_bytes=0,
+ referred_by_id=None,
+ referral_code=None,
+ )
+ source_active_sub = SimpleNamespace(
+ end_date=before + timedelta(days=30),
+ is_active=True,
+ skip_notifications=False,
+ last_notification_sent=before,
+ status_from_panel="ACTIVE",
+ panel_user_uuid="panel-email",
+ )
+ expired_target_sub = SimpleNamespace(
+ end_date=before - timedelta(days=3),
+ is_active=False,
+ skip_notifications=False,
+ last_notification_sent=before,
+ status_from_panel="EXPIRED",
+ panel_user_uuid="panel-telegram",
+ )
+ session = SimpleNamespace(
+ execute=AsyncMock(side_effect=lambda stmt: FakeResult()),
+ delete=AsyncMock(),
+ flush=AsyncMock(),
+ refresh=AsyncMock(),
+ )
+
+ async def fake_get_user_by_id(_session, user_id):
+ if user_id == source.user_id:
+ return source
+ if user_id == target.user_id:
+ return target
+ return None
+
+ async def fake_get_active_subscription(_session, user_id, panel_user_uuid=None):
+ if user_id == source.user_id and panel_user_uuid == source.panel_user_uuid:
+ return source_active_sub
+ return None
+
+ async def fake_get_latest_subscription(_session, user_id, panel_user_uuid=None, **_kwargs):
+ if user_id == target.user_id and panel_user_uuid == target.panel_user_uuid:
+ return expired_target_sub
+ return None
+
+ with (
+ patch("db.dal.user_dal.get_user_by_id", side_effect=fake_get_user_by_id),
+ patch(
+ "db.dal.user_dal._get_active_subscription_for_user",
+ side_effect=fake_get_active_subscription,
+ ),
+ patch(
+ "db.dal.user_dal._get_latest_subscription_for_user",
+ side_effect=fake_get_latest_subscription,
+ ),
+ ):
+ merged = await user_dal.merge_users(
+ session,
+ source_user_id=source.user_id,
+ target_user_id=target.user_id,
+ )
+
+ self.assertIs(merged, target)
+ self.assertEqual(target.email, "paid@example.com")
+ self.assertTrue(expired_target_sub.is_active)
+ self.assertEqual(expired_target_sub.status_from_panel, "ACTIVE_EXTENDED_BY_MERGE")
+ self.assertIsNone(expired_target_sub.last_notification_sent)
+ self.assertGreater(expired_target_sub.end_date, before + timedelta(days=29))
+ self.assertLess(expired_target_sub.end_date, before + timedelta(days=31))
+ self.assertFalse(source_active_sub.is_active)
+ self.assertTrue(source_active_sub.skip_notifications)
+ self.assertEqual(source_active_sub.status_from_panel, "MERGED_INTO_ACCOUNT")
diff --git a/tests/test_webapp_assets.py b/tests/test_webapp_assets.py
index 83ab5b9..4879a11 100644
--- a/tests/test_webapp_assets.py
+++ b/tests/test_webapp_assets.py
@@ -15,6 +15,7 @@ from PIL import Image
from bot.app.web import subscription_webapp
from bot.app.web.admin_api_impl import themes as admin_themes
from bot.app.web.webapp import assets as webapp_assets
+from bot.app.web.webapp import cache_helpers
from config.settings import Settings
from config.webapp_themes_config import WebappThemesConfig, builtin_webapp_themes_config
@@ -449,6 +450,58 @@ class WebAppAssetTests(unittest.IsolatedAsyncioTestCase):
[{"id": "yookassa", "name": "Bank card", "icon": "WalletCards"}],
)
+ def test_serialize_payment_methods_includes_wata_from_provider_config(self):
+ from bot.payment_providers import build_provider_configs, get_provider_bundle
+
+ build_provider_configs(force=True)
+ bundle = get_provider_bundle("wata_service")
+ self.assertIsNotNone(bundle)
+ bundle.config.ENABLED = True
+ bundle.config.API_TOKEN = "wata-token"
+
+ settings = Settings(
+ _env_file=None,
+ BOT_TOKEN="token",
+ POSTGRES_USER="app_user",
+ POSTGRES_PASSWORD="app_password",
+ TARIFFS_CONFIG_PATH="missing-tariffs.json",
+ PAYMENT_METHODS_ORDER="wata",
+ STARS_ENABLED=False,
+ )
+ app = {"wata_service": SimpleNamespace(configured=True)}
+
+ methods = subscription_webapp._serialize_payment_methods(settings, app, "en")
+
+ self.assertEqual(methods, [{"id": "wata", "name": "Wata", "icon": "WalletCards"}])
+
+ async def test_invalidate_all_webapp_user_caches_clears_cached_me_payload(self):
+ settings = Settings(
+ _env_file=None,
+ BOT_TOKEN="token",
+ POSTGRES_USER="app_user",
+ POSTGRES_PASSWORD="app_password",
+ REDIS_URL=None,
+ )
+ calls = 0
+
+ async def loader():
+ nonlocal calls
+ calls += 1
+ return {"payment_methods": [{"id": f"method-{calls}"}]}
+
+ first = await cache_helpers.webapp_cached_user_payload(settings, "me", 42, 60, loader)
+ second = await cache_helpers.webapp_cached_user_payload(settings, "me", 42, 60, loader)
+
+ self.assertEqual(first, {"payment_methods": [{"id": "method-1"}]})
+ self.assertEqual(second, first)
+ self.assertEqual(calls, 1)
+
+ await cache_helpers.invalidate_all_webapp_user_caches(settings)
+ third = await cache_helpers.webapp_cached_user_payload(settings, "me", 42, 60, loader)
+
+ self.assertEqual(third, {"payment_methods": [{"id": "method-2"}]})
+ self.assertEqual(calls, 2)
+
def test_serialize_plans_includes_stars_only_subscription_options(self):
settings = Settings(
_env_file=None,
@@ -488,7 +541,7 @@ class WebAppAssetTests(unittest.IsolatedAsyncioTestCase):
"subscription_webapp.min.22222222.js",
)
- def test_resolve_webapp_admin_asset_names_prefer_latest_minified_builds(self):
+ def test_resolve_webapp_admin_asset_names_use_stable_runtime_builds(self):
with tempfile.TemporaryDirectory() as tmpdir:
asset_dir = Path(tmpdir)
(asset_dir / "subscription_webapp_admin.js").write_text(
@@ -513,11 +566,11 @@ class WebAppAssetTests(unittest.IsolatedAsyncioTestCase):
with patch.object(webapp_assets, "ASSET_DIR", asset_dir):
self.assertEqual(
subscription_webapp._resolve_webapp_admin_js_asset_name(),
- "subscription_webapp_admin.min.22222222.js",
+ "subscription_webapp_admin.js",
)
self.assertEqual(
subscription_webapp._resolve_webapp_admin_css_asset_name(),
- "subscription_webapp_admin.22222222.css",
+ "subscription_webapp_admin.css",
)
async def test_js_asset_route_sets_immutable_cache_control_for_minified_asset(self):
diff --git a/tests/test_webapp_telegram_logout.py b/tests/test_webapp_telegram_logout.py
new file mode 100644
index 0000000..eb5a262
--- /dev/null
+++ b/tests/test_webapp_telegram_logout.py
@@ -0,0 +1,35 @@
+from pathlib import Path
+
+REPO_ROOT = Path(__file__).resolve().parents[1]
+
+
+def _read(path: str) -> str:
+ return (REPO_ROOT / path).read_text(encoding="utf-8")
+
+
+def test_telegram_init_data_login_runs_before_manual_logout_gate():
+ source = _read("frontend/src/lib/webapp/webappBoot.js")
+
+ init_data_pos = source.index("const initData = getInitDataForBoot();")
+ manual_logout_pos = source.index("if (isManuallyLoggedOut())")
+
+ assert init_data_pos < manual_logout_pos
+
+
+def test_logout_button_is_controlled_by_telegram_context():
+ app_source = _read("frontend/src/App.svelte")
+ settings_source = _read("frontend/src/webapp/screens/SettingsScreen.svelte")
+
+ assert "$: telegramMiniAppContext = hasTelegramLaunchParams();" in app_source
+ assert "showLogout={!telegramMiniAppContext}" in app_source
+ assert "export let showLogout = true;" in settings_source
+ assert "{#if showLogout}" in settings_source
+
+
+def test_logout_handler_is_noop_inside_telegram_mini_app():
+ source = _read("frontend/src/lib/webapp/stores/accountStore.js")
+
+ guard_pos = source.index("if (telegramSdk.hasLaunchParams()) return;")
+ mark_logout_pos = source.index("markManualLogout();")
+
+ assert guard_pos < mark_logout_pos