fix: cover email telegram account linking

tests/test_account_linking_panel.py

@@ -1,7 +1,10 @@
+import json
 import unittest
 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,
@@ -9,6 +12,23 @@ from bot.app.web.webapp.auth import (
 
 
 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 = []
 
@@ -131,3 +151,140 @@ class AccountLinkingPanelTests(unittest.IsolatedAsyncioTestCase):
         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",
+        }
+
+        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=SimpleNamespace(notify_account_telegram_linked=AsyncMock()),
+            ),
+        ):
+            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)
+        self.assertIn("rw_webapp_session", response.cookies)