from unittest.mock import patch from django.conf import settings from django.core.cache import cache from django.test import override_settings from rest_framework.test import APIRequestFactory from rest_framework import status from rest_framework.test import APITestCase from apps.users.api.views import RegisterWithPasswordView from apps.users.models import User, UserSocialAccount class UserApiViewTests(APITestCase): @classmethod def setUpTestData(cls): cls.user = User.objects.create_user( mobile="09123330001", password="secret123", first_name="Ali", last_name="Test", ) cls.other_user = User.objects.create_user( mobile="09123330002", password="secret123", first_name="Sara", last_name="Search", ) @patch("apps.users.api.views.register_user_with_password") def test_register_with_password_view_returns_tokens(self, register_user_with_password): register_user_with_password.return_value = { "access": "access-token", "refresh": "refresh-token", } request = APIRequestFactory().post( "/api/users/register/password/", {"mobile": "09123330009", "password": "secret123"}, format="json", ) response = RegisterWithPasswordView.as_view()(request) self.assertEqual(response.status_code, status.HTTP_201_CREATED) self.assertEqual(response.data["access"], "access-token") register_user_with_password.assert_called_once_with("09123330009", "secret123") def test_register_with_password_requires_mobile_and_password(self): request = APIRequestFactory().post("/api/users/register/password/", {}, format="json") response = RegisterWithPasswordView.as_view()(request) self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST) @patch("apps.users.api.views.generate_and_send_otp") def test_send_otp_view_validates_and_dispatches(self, generate_and_send_otp): response = self.client.post( "/api/users/otp/send/", {"mobile": "09123330009", "mode": "login"}, format="json", ) self.assertEqual(response.status_code, status.HTTP_200_OK) generate_and_send_otp.assert_called_once_with( mobile="09123330009", mode="login", ) @patch("apps.users.api.views.login_with_password") def test_login_view_returns_tokens(self, login_with_password): login_with_password.return_value = {"access": "a", "refresh": "r"} response = self.client.post( "/api/users/login/", {"mobile": "09123330001", "password": "secret123"}, format="json", ) self.assertEqual(response.status_code, status.HTTP_200_OK) self.assertEqual(response.data["refresh"], "r") login_with_password.assert_called_once() @patch("apps.users.api.views.login_with_otp") def test_login_otp_view_returns_tokens(self, login_with_otp): login_with_otp.return_value = {"access": "a", "refresh": "r"} response = self.client.post( "/api/users/otp/login/", {"mobile": "09123330001", "code": "123456"}, format="json", ) self.assertEqual(response.status_code, status.HTTP_200_OK) self.assertEqual(response.data["access"], "a") @patch("apps.users.api.views.reset_password_with_otp") def test_reset_password_view_calls_service(self, reset_password_with_otp): response = self.client.post( "/api/users/password/reset/", { "mobile": "09123330001", "code": "123456", "password": "new-secret123", "re_password": "new-secret123", }, format="json", ) self.assertEqual(response.status_code, status.HTTP_200_OK) reset_password_with_otp.assert_called_once_with( mobile="09123330001", code="123456", password="new-secret123", ) @patch("apps.users.api.views.change_password") def test_change_password_view_requires_auth_and_calls_service(self, change_password): self.client.force_authenticate(user=self.user) response = self.client.patch( "/api/users/password/change/", { "old_password": "secret123", "new_password": "new-secret123", "re_password": "new-secret123", }, format="json", ) self.assertEqual(response.status_code, status.HTTP_200_OK) change_password.assert_called_once_with( user=self.user, old_password="secret123", new_password="new-secret123", ) @patch("apps.users.api.views.logout_user") def test_logout_view_calls_service(self, logout_user): self.client.force_authenticate(user=self.user) response = self.client.post( "/api/users/logout/", {"refresh": "refresh-token"}, format="json", ) self.assertEqual(response.status_code, status.HTTP_205_RESET_CONTENT) logout_user.assert_called_once_with("refresh-token") def test_user_list_and_profile_views_require_authentication(self): list_response = self.client.get("/api/users/list/") me_response = self.client.get("/api/users/me/") self.assertEqual(list_response.status_code, status.HTTP_401_UNAUTHORIZED) self.assertEqual(me_response.status_code, status.HTTP_401_UNAUTHORIZED) def test_user_list_returns_users_for_authenticated_request(self): self.client.force_authenticate(user=self.user) response = self.client.get("/api/users/list/") self.assertEqual(response.status_code, status.HTTP_200_OK) items = ( response.data if isinstance(response.data, list) else response.data.get("results") or response.data.get("items") or [] ) mobiles = {item["mobile"] for item in items} self.assertIn(self.user.mobile, mobiles) self.assertIn(self.other_user.mobile, mobiles) def test_user_me_retrieve_and_patch_work(self): self.client.force_authenticate(user=self.user) retrieve_response = self.client.get("/api/users/me/") self.assertEqual(retrieve_response.status_code, status.HTTP_200_OK) self.assertEqual(retrieve_response.data["mobile"], self.user.mobile) patch_response = self.client.patch( "/api/users/me/", {"first_name": "Updated", "description": "Bio"}, format="json", ) self.assertEqual(patch_response.status_code, status.HTTP_200_OK) self.assertEqual(patch_response.data["first_name"], "Updated") self.user.refresh_from_db() self.assertEqual(self.user.description, "Bio") def test_user_search_handles_missing_mobile_not_found_and_success(self): self.client.force_authenticate(user=self.user) missing = self.client.get("/api/users/search/") self.assertEqual(missing.status_code, status.HTTP_400_BAD_REQUEST) not_found = self.client.get("/api/users/search/?mobile=09129999999") self.assertEqual(not_found.status_code, status.HTTP_404_NOT_FOUND) success = self.client.get(f"/api/users/search/?mobile={self.other_user.mobile}") self.assertEqual(success.status_code, status.HTTP_200_OK) self.assertEqual(success.data["mobile"], self.other_user.mobile) class UserThrottleTests(APITestCase): @classmethod def setUpTestData(cls): cls.user = User.objects.create_user( mobile="09124440001", password="secret123", ) def setUp(self): cache.clear() def tearDown(self): cache.clear() @override_settings( REST_FRAMEWORK={ **settings.REST_FRAMEWORK, "DEFAULT_THROTTLE_RATES": { **settings.REST_FRAMEWORK["DEFAULT_THROTTLE_RATES"], "login_password": "2/min", }, } ) @patch("apps.users.api.views.login_with_password") def test_password_login_returns_structured_429_with_retry_after(self, login_with_password): login_with_password.return_value = {"access": "a", "refresh": "r"} first = self.client.post( "/api/users/login/", {"mobile": "09124440001", "password": "secret123"}, format="json", REMOTE_ADDR="10.0.0.1", ) second = self.client.post( "/api/users/login/", {"mobile": "09124440001", "password": "secret123"}, format="json", REMOTE_ADDR="10.0.0.1", ) throttled = self.client.post( "/api/users/login/", {"mobile": "09124440001", "password": "secret123"}, format="json", REMOTE_ADDR="10.0.0.1", ) self.assertEqual(first.status_code, 200) self.assertEqual(second.status_code, 200) self.assertEqual(throttled.status_code, 429) self.assertEqual(throttled.data["code"], "throttled") self.assertEqual(throttled.data["scope"], "login_password") self.assertIsInstance(throttled.data["retry_after_seconds"], int) self.assertTrue(throttled.data["throttled_until"]) self.assertIn("Retry-After", throttled.headers) @override_settings( REST_FRAMEWORK={ **settings.REST_FRAMEWORK, "DEFAULT_THROTTLE_RATES": { **settings.REST_FRAMEWORK["DEFAULT_THROTTLE_RATES"], "otp_send_burst": "1/min", "otp_send_sustained": "10/day", }, } ) @patch("apps.users.api.views.generate_and_send_otp") def test_otp_send_throttle_is_keyed_by_mobile_and_ip(self, generate_and_send_otp): first_mobile_first = self.client.post( "/api/users/otp/send/", {"mobile": "09124440011", "mode": "login"}, format="json", REMOTE_ADDR="10.0.0.2", ) second_mobile_first = self.client.post( "/api/users/otp/send/", {"mobile": "09124440012", "mode": "login"}, format="json", REMOTE_ADDR="10.0.0.2", ) first_mobile_second = self.client.post( "/api/users/otp/send/", {"mobile": "09124440011", "mode": "login"}, format="json", REMOTE_ADDR="10.0.0.2", ) self.assertEqual(first_mobile_first.status_code, 200) self.assertEqual(second_mobile_first.status_code, 200) self.assertEqual(first_mobile_second.status_code, 429) self.assertEqual(first_mobile_second.data["scope"], "otp_send_burst") @override_settings( REST_FRAMEWORK={ **settings.REST_FRAMEWORK, "DEFAULT_THROTTLE_RATES": { **settings.REST_FRAMEWORK["DEFAULT_THROTTLE_RATES"], "login_otp": "1/min", }, } ) @patch("apps.users.api.views.login_with_otp") def test_otp_login_throttle_blocks_after_limit(self, login_with_otp): login_with_otp.return_value = {"access": "a", "refresh": "r"} allowed = self.client.post( "/api/users/otp/login/", {"mobile": "09124440021", "code": "123456"}, format="json", REMOTE_ADDR="10.0.0.3", ) throttled = self.client.post( "/api/users/otp/login/", {"mobile": "09124440021", "code": "123456"}, format="json", REMOTE_ADDR="10.0.0.3", ) self.assertEqual(allowed.status_code, 200) self.assertEqual(throttled.status_code, 429) self.assertEqual(throttled.data["scope"], "login_otp") @patch.dict("rest_framework.throttling.AnonRateThrottle.THROTTLE_RATES", {"anon": "1/min"}, clear=False) @patch("apps.users.api.views.register_user_with_otp") def test_global_anon_throttle_applies_to_unrestricted_anonymous_endpoint( self, register_user_with_otp, ): register_user_with_otp.return_value = {"access": "a", "refresh": "r"} first = self.client.post( "/api/users/register/", { "mobile": "09124440031", "code": "12345", "password": "secret123", "re_password": "secret123", }, format="json", REMOTE_ADDR="10.0.0.4", ) throttled = self.client.post( "/api/users/register/", { "mobile": "09124440032", "code": "12345", "password": "secret123", "re_password": "secret123", }, format="json", REMOTE_ADDR="10.0.0.4", ) self.assertEqual(first.status_code, 201) self.assertEqual(throttled.status_code, 429) self.assertEqual(throttled.data["code"], "throttled") @override_settings( REST_FRAMEWORK={ **settings.REST_FRAMEWORK, "DEFAULT_THROTTLE_RATES": { **settings.REST_FRAMEWORK["DEFAULT_THROTTLE_RATES"], "login_password": "1/min", }, } ) @patch("apps.users.api.views.login_with_password") def test_throttle_falls_back_to_ip_when_mobile_is_missing(self, login_with_password): login_with_password.return_value = {"access": "a", "refresh": "r"} first = self.client.post( "/api/users/login/", {"password": "secret123"}, format="json", REMOTE_ADDR="10.0.0.5", ) second = self.client.post( "/api/users/login/", {"password": "secret123"}, format="json", REMOTE_ADDR="10.0.0.5", ) self.assertEqual(first.status_code, 400) self.assertEqual(second.status_code, 429) @override_settings( GOOGLE_OAUTH_CLIENT_ID="google-client-id", GOOGLE_OAUTH_CLIENT_SECRET="google-client-secret", GOOGLE_OAUTH_REDIRECT_URI="http://testserver/api/users/oauth/google/callback/", GOOGLE_OAUTH_FRONTEND_CALLBACK_URL="http://localhost:5173/auth/google/callback", ) class GoogleOAuthApiTests(APITestCase): @classmethod def setUpTestData(cls): cls.user = User.objects.create_user( mobile="09125550001", password="secret123", first_name="Google", last_name="Linked", ) def setUp(self): cache.clear() def tearDown(self): cache.clear() def test_google_start_redirects_to_google_authorization_url(self): response = self.client.get("/api/users/oauth/google/start/") self.assertEqual(response.status_code, 302) self.assertIn("accounts.google.com", response["Location"]) self.assertIn("state=", response["Location"]) @patch("apps.users.api.views.exchange_code_for_google_profile") def test_google_callback_redirects_with_authenticated_flow_for_linked_account( self, exchange_code_for_google_profile, ): exchange_code_for_google_profile.return_value = type( "Profile", (), { "provider_user_id": "google-sub-1", "email": "linked@example.com", "email_verified": True, "first_name": "Google", "last_name": "Linked", "avatar_url": "https://example.com/avatar.png", }, )() UserSocialAccount.objects.create( user=self.user, provider=UserSocialAccount.ProviderType.GOOGLE, provider_user_id="google-sub-1", email="linked@example.com", email_verified=True, avatar_url="https://example.com/avatar.png", ) start_response = self.client.get("/api/users/oauth/google/start/") state = start_response["Location"].split("state=", 1)[1].split("&", 1)[0] response = self.client.get( f"/api/users/oauth/google/callback/?state={state}&code=google-code", ) self.assertEqual(response.status_code, 302) self.assertIn("/auth/google/callback?flow=", response["Location"]) flow = response["Location"].split("flow=", 1)[1] flow_response = self.client.get(f"/api/users/oauth/google/flow/?flow={flow}") self.assertEqual(flow_response.status_code, 200) self.assertEqual(flow_response.data["status"], "authenticated") self.assertIn("access", flow_response.data) self.assertIn("refresh", flow_response.data) @patch("apps.users.api.views.exchange_code_for_google_profile") def test_google_callback_redirects_with_mobile_collection_flow_for_new_account( self, exchange_code_for_google_profile, ): exchange_code_for_google_profile.return_value = type( "Profile", (), { "provider_user_id": "google-sub-2", "email": "new@example.com", "email_verified": True, "first_name": "New", "last_name": "User", "avatar_url": "https://example.com/new-avatar.png", }, )() start_response = self.client.get("/api/users/oauth/google/start/") state = start_response["Location"].split("state=", 1)[1].split("&", 1)[0] response = self.client.get( f"/api/users/oauth/google/callback/?state={state}&code=google-code", ) flow = response["Location"].split("flow=", 1)[1] flow_response = self.client.get(f"/api/users/oauth/google/flow/?flow={flow}") self.assertEqual(flow_response.status_code, 200) self.assertEqual(flow_response.data["status"], "collect_mobile") self.assertEqual(flow_response.data["email"], "new@example.com") @patch("apps.users.services.google_oauth.generate_and_send_otp") def test_google_complete_existing_mobile_moves_flow_to_claim_required(self, generate_and_send_otp): cache.set( "google_oauth_flow:test-flow", { "status": "collect_mobile", "google_profile": { "provider_user_id": "google-sub-3", "email": "existing@example.com", "email_verified": True, "first_name": "Existing", "last_name": "User", "avatar_url": "https://example.com/existing.png", }, "email": "existing@example.com", "first_name": "Existing", "last_name": "User", "avatar_url": "https://example.com/existing.png", }, 900, ) response = self.client.post( "/api/users/oauth/google/complete/", {"flow": "test-flow", "mobile": self.user.mobile}, format="json", ) self.assertEqual(response.status_code, 200) self.assertEqual(response.data["status"], "claim_required") generate_and_send_otp.assert_called_once_with(self.user.mobile, "login") def test_google_complete_new_mobile_creates_user_and_link(self): cache.set( "google_oauth_flow:new-flow", { "status": "collect_mobile", "google_profile": { "provider_user_id": "google-sub-4", "email": "created@example.com", "email_verified": True, "first_name": "Created", "last_name": "User", "avatar_url": "https://example.com/created.png", }, "email": "created@example.com", "first_name": "Created", "last_name": "User", "avatar_url": "https://example.com/created.png", }, 900, ) response = self.client.post( "/api/users/oauth/google/complete/", {"flow": "new-flow", "mobile": "09125550009"}, format="json", ) self.assertEqual(response.status_code, 200) self.assertEqual(response.data["status"], "authenticated") created_user = User.objects.get(mobile="09125550009") self.assertFalse(created_user.has_usable_password()) self.assertTrue( UserSocialAccount.objects.filter( user=created_user, provider=UserSocialAccount.ProviderType.GOOGLE, provider_user_id="google-sub-4", ).exists() ) @patch("apps.users.api.views.send_google_claim_otp") def test_google_claim_send_otp_endpoint_dispatches(self, send_google_claim_otp): send_google_claim_otp.return_value = {"detail": "Verification code sent successfully."} response = self.client.post( "/api/users/oauth/google/claim/send-otp/", {"flow": "claim-flow"}, format="json", ) self.assertEqual(response.status_code, 200) send_google_claim_otp.assert_called_once_with("claim-flow") @patch("apps.users.api.views.verify_google_claim") def test_google_claim_verify_returns_tokens(self, verify_google_claim): verify_google_claim.return_value = {"status": "authenticated", "access": "a", "refresh": "r"} response = self.client.post( "/api/users/oauth/google/claim/verify/", {"flow": "claim-flow", "code": "12345"}, format="json", ) self.assertEqual(response.status_code, 200) self.assertEqual(response.data["access"], "a") verify_google_claim.assert_called_once_with(flow="claim-flow", code="12345")