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, ) 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_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", } 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)