"""Tests for Telegram authentication and webhook flows.""" from __future__ import annotations from unittest.mock import patch from django.test import TestCase, override_settings from django.urls import reverse from apps.users.models import User from utils.telegram import TelegramDataValidationError, generate_signed_init_data @override_settings(TELEGRAM_BOT_TOKEN="test-bot-token") class AuthenticateTelegramTests(TestCase): def _post_auth(self, payload: dict) -> dict: response = self.client.post( reverse("telegram-auth"), data=payload, content_type="application/json", ) self.assertEqual(response.status_code, 200) return response.json() @patch("apps.users.views.validate_telegram_init_data") def test_creates_user_and_returns_tokens(self, mock_validate): """Creates a new user and returns JWTs and profile data.""" mock_validate.return_value = { "id": 12345, "username": "alice", "first_name": "Alice", "last_name": "Tester", "language_code": "en", "photo_url": "https://img.test/alice.jpg", } body = self._post_auth({"init_data": "signed-init-data"}) self.assertIn("access", body) self.assertIn("refresh", body) user = User.objects.get(telegram_user_id=12345) self.assertEqual(user.username, "alice") self.assertEqual(user.first_name, "Alice") self.assertEqual(user.last_name, "Tester") self.assertEqual(user.language_code, "en") self.assertEqual(user.photo_url, "https://img.test/alice.jpg") self.assertEqual(body["user"]["referral_code"], user.referral_code) def test_rejects_missing_init_data(self): """init_data must be provided.""" response = self.client.post( reverse("telegram-auth"), data={}, content_type="application/json", ) self.assertEqual(response.status_code, 400) self.assertIn("init_data is required", response.json()["error"]) @patch("apps.users.views.validate_telegram_init_data") def test_rejects_invalid_init_data_signature(self, mock_validate): """Invalid init data returns 401 with the validation error message.""" mock_validate.side_effect = TelegramDataValidationError("bad signature") response = self.client.post( reverse("telegram-auth"), data={"init_data": "invalid"}, content_type="application/json", ) self.assertEqual(response.status_code, 401) self.assertIn("bad signature", response.json()["error"]) @patch("apps.users.views.validate_telegram_init_data") def test_rejects_missing_user_id(self, mock_validate): """Validation output without an id is treated as bad input.""" mock_validate.return_value = {"username": "noid"} response = self.client.post( reverse("telegram-auth"), data={"init_data": "signed"}, content_type="application/json", ) self.assertEqual(response.status_code, 400) self.assertIn("Invalid user data", response.json()["error"]) @patch("apps.users.views.validate_telegram_init_data") def test_assigns_referral_when_start_param_present(self, mock_validate): """Links referrer by referral_code when provided.""" referrer = User.objects.create_user( telegram_user_id=1, username="referrer", first_name="Ref", referral_code="refcode1234", ) mock_validate.return_value = { "id": 22222, "username": "bob", "first_name": "Bob", } body = self._post_auth( {"init_data": "signed", "start_param": referrer.referral_code} ) new_user = User.objects.get(telegram_user_id=22222) self.assertEqual(new_user.referred_by_id, referrer.id) self.assertEqual(body["user"]["referred_by"], referrer.id) @patch("apps.users.views.validate_telegram_init_data") def test_does_not_overwrite_existing_referral(self, mock_validate): """Existing referred_by is preserved even if a new start_param arrives.""" original_referrer = User.objects.create_user( telegram_user_id=10, username="ref1", first_name="Ref1" ) other_referrer = User.objects.create_user( telegram_user_id=11, username="ref2", first_name="Ref2", referral_code="OTHERCODE" ) user = User.objects.create_user( telegram_user_id=44444, username="existing", first_name="Existing", referred_by=original_referrer, ) mock_validate.return_value = {"id": 44444, "username": "existing"} body = self._post_auth({"init_data": "signed", "start_param": other_referrer.referral_code}) user.refresh_from_db() self.assertEqual(user.referred_by_id, original_referrer.id) self.assertEqual(body["user"]["referred_by"], original_referrer.id) @patch("apps.users.views.validate_telegram_init_data") def test_ignores_invalid_referral_code(self, mock_validate): """Unknown referral codes are ignored without failing auth.""" mock_validate.return_value = {"id": 55555, "username": "nolink"} body = self._post_auth({"init_data": "signed", "start_param": "DOESNOTEXIST"}) user = User.objects.get(telegram_user_id=55555) self.assertIsNone(user.referred_by_id) self.assertIsNone(body["user"]["referred_by"]) @patch("apps.users.views.validate_telegram_init_data") def test_updates_existing_profile_fields(self, mock_validate): """Existing user gets profile fields refreshed on login.""" user = User.objects.create_user( telegram_user_id=33333, username="oldname", first_name="Old", last_name=None, language_code=None, ) mock_validate.return_value = { "id": 33333, "username": "newname", "first_name": "New", "last_name": "Name", "language_code": "id", } self._post_auth({"init_data": "signed"}) user.refresh_from_db() self.assertEqual(user.username, "newname") self.assertEqual(user.first_name, "New") self.assertEqual(user.last_name, "Name") self.assertEqual(user.language_code, "id") class TelegramWebhookTests(TestCase): @override_settings( TELEGRAM_BOT_TOKEN="test-bot-token", TELEGRAM_WEBHOOK_SECRET="secret-token", TELEGRAM_WEBAPP_URL="https://example.com/app", ) @patch("apps.users.views.send_telegram_message") def test_webhook_replies_with_webapp_button_and_valid_secret(self, mock_send): """Webhook responds with a WebApp button including start param.""" mock_send.return_value = True payload = { "message": { "chat": {"id": 98765}, "text": "/start ABCDEF", } } response = self.client.post( reverse("telegram-webhook"), data=payload, content_type="application/json", HTTP_X_TELEGRAM_BOT_API_SECRET_TOKEN="secret-token", ) self.assertEqual(response.status_code, 200) mock_send.assert_called_once() args, kwargs = mock_send.call_args self.assertEqual(kwargs["chat_id"], 98765) self.assertIn("reply_markup", kwargs) button_url = kwargs["reply_markup"]["keyboard"][0][0]["web_app"]["url"] self.assertEqual(button_url, "https://example.com/app?startapp=ABCDEF") @override_settings(TELEGRAM_WEBHOOK_SECRET="secret-token") def test_webhook_rejects_invalid_secret(self): """Webhook returns 403 when the secret token does not match.""" payload = { "message": { "chat": {"id": 1111}, "text": "/start", } } response = self.client.post( reverse("telegram-webhook"), data=payload, content_type="application/json", HTTP_X_TELEGRAM_BOT_API_SECRET_TOKEN="wrong-secret", ) self.assertEqual(response.status_code, 403) @override_settings(TELEGRAM_WEBHOOK_SECRET="secret-token") def test_webhook_rejects_missing_secret_header(self): """Webhook returns 403 when secret is configured but header absent.""" response = self.client.post( reverse("telegram-webhook"), data={"message": {"chat": {"id": 2222}, "text": "/start"}}, content_type="application/json", ) self.assertEqual(response.status_code, 403) @override_settings( TELEGRAM_BOT_TOKEN="test-bot-token", TELEGRAM_WEBHOOK_SECRET="secret-token", TELEGRAM_WEBAPP_URL="https://example.com/app", ) @patch("apps.users.views.send_telegram_message") def test_webhook_handles_non_start_message(self, mock_send): """Non-/start messages still reply without appending startapp param.""" mock_send.return_value = True payload = {"message": {"chat": {"id": 999}, "text": "hello"}} response = self.client.post( reverse("telegram-webhook"), data=payload, content_type="application/json", HTTP_X_TELEGRAM_BOT_API_SECRET_TOKEN="secret-token", ) self.assertEqual(response.status_code, 200) mock_send.assert_called_once() args, kwargs = mock_send.call_args self.assertEqual(kwargs["chat_id"], 999) button_url = kwargs["reply_markup"]["keyboard"][0][0]["web_app"]["url"] self.assertEqual(button_url, "https://example.com/app") @override_settings( TELEGRAM_BOT_TOKEN="test-bot-token", TELEGRAM_WEBHOOK_SECRET="secret-token", TELEGRAM_WEBAPP_URL="https://example.com/app", ) @patch("apps.users.views.send_telegram_message") def test_webhook_handles_callback_data(self, mock_send): """Callback queries use data as start param.""" mock_send.return_value = True payload = { "callback_query": { "data": "CALLBACKCODE", "message": {"chat": {"id": 777}}, } } response = self.client.post( reverse("telegram-webhook"), data=payload, content_type="application/json", HTTP_X_TELEGRAM_BOT_API_SECRET_TOKEN="secret-token", ) self.assertEqual(response.status_code, 200) button_url = mock_send.call_args.kwargs["reply_markup"]["keyboard"][0][0]["web_app"]["url"] self.assertEqual(button_url, "https://example.com/app?startapp=CALLBACKCODE") @override_settings( TELEGRAM_BOT_TOKEN="test-bot-token", TELEGRAM_WEBHOOK_SECRET="secret-token", TELEGRAM_WEBAPP_URL="https://example.com/app", ) @patch("apps.users.views.send_telegram_message") def test_webhook_acknowledges_missing_chat_without_sending(self, mock_send): """Gracefully ack when chat info is missing to avoid retries.""" mock_send.return_value = True response = self.client.post( reverse("telegram-webhook"), data={"message": {}}, content_type="application/json", HTTP_X_TELEGRAM_BOT_API_SECRET_TOKEN="secret-token", ) self.assertEqual(response.status_code, 200) self.assertEqual(response.json(), {"ok": True}) mock_send.assert_not_called() class DevAuthenticateTelegramTests(TestCase): @override_settings( DEBUG=True, ENABLE_DEV_TELEGRAM_AUTH=True, TELEGRAM_BOT_TOKEN="dev-bot-token", ) def test_dev_auth_validates_signed_payload_and_returns_debug(self): """Dev endpoint returns debug info and authenticates signed init_data.""" signed = generate_signed_init_data( user={"id": 99999, "first_name": "Dev", "username": "devuser"}, bot_token="dev-bot-token", query_id="dev-query", ) response = self.client.post( reverse("telegram-auth-dev"), data={"init_data": signed["init_data"]}, content_type="application/json", ) self.assertEqual(response.status_code, 200) body = response.json() self.assertEqual(body["user"]["telegram_user_id"], 99999) self.assertIn("debug", body) self.assertEqual(body["debug"]["provided_hash"], signed["hash"]) self.assertEqual(body["debug"]["calculated_hash"], signed["hash"]) self.assertIn("data_check_string", body["debug"]) @override_settings( DEBUG=False, ENABLE_DEV_TELEGRAM_AUTH=True, TELEGRAM_BOT_TOKEN="dev-bot-token", ) def test_dev_auth_disabled_when_not_debug(self): """Dev endpoint returns 404 outside DEBUG mode.""" response = self.client.post( reverse("telegram-auth-dev"), data={"init_data": "anything"}, content_type="application/json", ) self.assertEqual(response.status_code, 404)