967 lines
36 KiB
Python
967 lines
36 KiB
Python
from io import StringIO
|
|
from unittest.mock import patch
|
|
|
|
from django.conf import settings
|
|
from django.core.cache import cache
|
|
from django.core.management import call_command
|
|
from django.db import IntegrityError
|
|
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
|
|
from apps.users.services.google_oauth import GoogleProfile
|
|
|
|
|
|
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):
|
|
generate_and_send_otp.return_value = {
|
|
"detail": "OTP sent successfully",
|
|
"expires_in_seconds": 120,
|
|
"expires_at": "2026-05-12T10:00:00+03:30",
|
|
}
|
|
response = self.client.post(
|
|
"/api/users/otp/send/",
|
|
{"mobile": "09123330009", "mode": "login"},
|
|
format="json",
|
|
)
|
|
|
|
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
|
self.assertEqual(response.data["expires_in_seconds"], 120)
|
|
generate_and_send_otp.assert_called_once_with(
|
|
mobile="09123330009",
|
|
mode="login",
|
|
)
|
|
|
|
@patch("apps.users.api.views.generate_and_send_otp")
|
|
def test_send_otp_view_supports_forget_password_mode(self, generate_and_send_otp):
|
|
generate_and_send_otp.return_value = {
|
|
"detail": "OTP sent successfully",
|
|
"expires_in_seconds": 120,
|
|
"expires_at": "2026-05-12T10:00:00+03:30",
|
|
}
|
|
response = self.client.post(
|
|
"/api/users/otp/send/",
|
|
{"mobile": "09123330001", "mode": "forget_password"},
|
|
format="json",
|
|
)
|
|
|
|
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
|
generate_and_send_otp.assert_called_once_with(
|
|
mobile="09123330001",
|
|
mode="forget_password",
|
|
)
|
|
|
|
@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": "NewSecret1!",
|
|
"re_password": "NewSecret1!",
|
|
},
|
|
format="json",
|
|
)
|
|
|
|
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
|
reset_password_with_otp.assert_called_once_with(
|
|
mobile="09123330001",
|
|
code="123456",
|
|
password="NewSecret1!",
|
|
)
|
|
|
|
def test_reset_password_view_rejects_invalid_mobile_format(self):
|
|
response = self.client.post(
|
|
"/api/users/password/reset/",
|
|
{
|
|
"mobile": "9123330001",
|
|
"code": "123456",
|
|
"password": "NewSecret1!",
|
|
"re_password": "NewSecret1!",
|
|
},
|
|
format="json",
|
|
)
|
|
|
|
self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
|
|
self.assertIn("error", response.data)
|
|
|
|
def test_reset_password_view_rejects_weak_password(self):
|
|
response = self.client.post(
|
|
"/api/users/password/reset/",
|
|
{
|
|
"mobile": "09123330001",
|
|
"code": "123456",
|
|
"password": "Short1!",
|
|
"re_password": "Short1!",
|
|
},
|
|
format="json",
|
|
)
|
|
|
|
self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
|
|
self.assertIn("too short", response.data["error"])
|
|
|
|
@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": "NewSecret1!",
|
|
"re_password": "NewSecret1!",
|
|
},
|
|
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="NewSecret1!",
|
|
)
|
|
|
|
def test_change_password_view_rejects_reused_password(self):
|
|
self.client.force_authenticate(user=self.user)
|
|
|
|
response = self.client.patch(
|
|
"/api/users/password/change/",
|
|
{
|
|
"old_password": "secret123",
|
|
"new_password": "secret123",
|
|
"re_password": "secret123",
|
|
},
|
|
format="json",
|
|
)
|
|
|
|
self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
|
|
self.assertIn("رمز عبور جدید", response.data["error"])
|
|
|
|
@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_email_is_normalized_on_save(self):
|
|
user = User.objects.create_user(
|
|
mobile="09123330099",
|
|
password="Secret123!",
|
|
email=" MixedCase@Example.COM ",
|
|
)
|
|
|
|
self.assertEqual(user.email, "mixedcase@example.com")
|
|
|
|
def test_user_email_is_case_insensitively_unique(self):
|
|
User.objects.create_user(
|
|
mobile="09123330100",
|
|
password="Secret123!",
|
|
email="duplicate@example.com",
|
|
)
|
|
|
|
with self.assertRaises(IntegrityError):
|
|
User.objects.create_user(
|
|
mobile="09123330101",
|
|
password="Secret123!",
|
|
email=" DUPLICATE@example.com ",
|
|
)
|
|
|
|
def test_user_profile_patch_rejects_duplicate_normalized_email(self):
|
|
self.client.force_authenticate(user=self.user)
|
|
self.other_user.email = "other@example.com"
|
|
self.other_user.save(update_fields=["email"])
|
|
|
|
response = self.client.patch(
|
|
"/api/users/me/",
|
|
{"email": " OTHER@example.com "},
|
|
format="json",
|
|
)
|
|
|
|
self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
|
|
self.assertIn("already exists", response.data["error"])
|
|
|
|
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.linked_user = User.objects.create_user(
|
|
mobile="09125550001",
|
|
password="secret123",
|
|
first_name="Google",
|
|
last_name="Linked",
|
|
)
|
|
cls.email_matched_user = User.objects.create_user(
|
|
mobile="09125550002",
|
|
password="secret123",
|
|
first_name="Email",
|
|
last_name="Matched",
|
|
email="matched@example.com",
|
|
)
|
|
cls.blank_email_user = User.objects.create_user(
|
|
mobile="09125550003",
|
|
password="secret123",
|
|
first_name="Blank",
|
|
last_name="Email",
|
|
email=None,
|
|
)
|
|
cls.other_email_user = User.objects.create_user(
|
|
mobile="09125550004",
|
|
password="secret123",
|
|
first_name="Other",
|
|
last_name="Email",
|
|
email="other@example.com",
|
|
)
|
|
|
|
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 = GoogleProfile(
|
|
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.linked_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 = GoogleProfile(
|
|
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")
|
|
self.assertEqual(flow_response.data["resolution"], "new_account")
|
|
self.assertIsNone(flow_response.data["mobile_hint"])
|
|
|
|
@patch("apps.users.api.views.exchange_code_for_google_profile")
|
|
def test_google_callback_redirects_with_email_claim_flow_for_matching_email(
|
|
self,
|
|
exchange_code_for_google_profile,
|
|
):
|
|
exchange_code_for_google_profile.return_value = GoogleProfile(
|
|
provider_user_id="google-sub-email-match",
|
|
email="matched@example.com",
|
|
email_verified=True,
|
|
first_name="Email",
|
|
last_name="Matched",
|
|
avatar_url="https://example.com/matched.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["resolution"], "existing_email_claim")
|
|
self.assertEqual(flow_response.data["email"], "matched@example.com")
|
|
self.assertTrue(flow_response.data["mobile_hint"].startswith("09"))
|
|
|
|
@patch("apps.users.services.google_oauth.generate_and_send_otp")
|
|
def test_google_complete_matching_mobile_on_email_claim_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": "matched@example.com",
|
|
"email_verified": True,
|
|
"first_name": "Email",
|
|
"last_name": "Matched",
|
|
"avatar_url": "https://example.com/existing.png",
|
|
},
|
|
"email": "matched@example.com",
|
|
"first_name": "Email",
|
|
"last_name": "Matched",
|
|
"avatar_url": "https://example.com/existing.png",
|
|
"resolution": "existing_email_claim",
|
|
"target_user_id": str(self.email_matched_user.id),
|
|
"mobile_hint": "09*****0002",
|
|
},
|
|
900,
|
|
)
|
|
|
|
response = self.client.post(
|
|
"/api/users/oauth/google/complete/",
|
|
{"flow": "test-flow", "mobile": self.email_matched_user.mobile},
|
|
format="json",
|
|
)
|
|
|
|
self.assertEqual(response.status_code, 200)
|
|
self.assertEqual(response.data["status"], "claim_required")
|
|
self.assertEqual(response.data["resolution"], "existing_email_claim")
|
|
generate_and_send_otp.assert_called_once_with(self.email_matched_user.mobile, "login")
|
|
|
|
def test_google_complete_wrong_mobile_on_email_claim_returns_conflict(self):
|
|
cache.set(
|
|
"google_oauth_flow:email-conflict-flow",
|
|
{
|
|
"status": "collect_mobile",
|
|
"google_profile": {
|
|
"provider_user_id": "google-sub-email-conflict",
|
|
"email": "matched@example.com",
|
|
"email_verified": True,
|
|
"first_name": "Email",
|
|
"last_name": "Matched",
|
|
"avatar_url": "https://example.com/existing.png",
|
|
},
|
|
"email": "matched@example.com",
|
|
"first_name": "Email",
|
|
"last_name": "Matched",
|
|
"avatar_url": "https://example.com/existing.png",
|
|
"resolution": "existing_email_claim",
|
|
"target_user_id": str(self.email_matched_user.id),
|
|
"mobile_hint": "09*****0002",
|
|
},
|
|
900,
|
|
)
|
|
|
|
response = self.client.post(
|
|
"/api/users/oauth/google/complete/",
|
|
{"flow": "email-conflict-flow", "mobile": self.other_email_user.mobile},
|
|
format="json",
|
|
)
|
|
|
|
self.assertEqual(response.status_code, 409)
|
|
self.assertEqual(response.data["code"], "google_email_mobile_conflict")
|
|
self.assertEqual(response.data["mobile_hint"], "09*****0002")
|
|
|
|
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.services.google_oauth.generate_and_send_otp")
|
|
def test_google_complete_existing_blank_email_mobile_moves_flow_to_claim_required(
|
|
self,
|
|
generate_and_send_otp,
|
|
):
|
|
cache.set(
|
|
"google_oauth_flow:blank-email-flow",
|
|
{
|
|
"status": "collect_mobile",
|
|
"google_profile": {
|
|
"provider_user_id": "google-sub-blank",
|
|
"email": "blank-claim@example.com",
|
|
"email_verified": True,
|
|
"first_name": "Blank",
|
|
"last_name": "Claim",
|
|
"avatar_url": "https://example.com/blank.png",
|
|
},
|
|
"email": "blank-claim@example.com",
|
|
"first_name": "Blank",
|
|
"last_name": "Claim",
|
|
"avatar_url": "https://example.com/blank.png",
|
|
"resolution": "new_account",
|
|
"target_user_id": None,
|
|
"mobile_hint": None,
|
|
},
|
|
900,
|
|
)
|
|
|
|
response = self.client.post(
|
|
"/api/users/oauth/google/complete/",
|
|
{"flow": "blank-email-flow", "mobile": self.blank_email_user.mobile},
|
|
format="json",
|
|
)
|
|
|
|
self.assertEqual(response.status_code, 200)
|
|
self.assertEqual(response.data["status"], "claim_required")
|
|
self.assertEqual(response.data["resolution"], "existing_mobile_claim")
|
|
generate_and_send_otp.assert_called_once_with(self.blank_email_user.mobile, "login")
|
|
|
|
def test_google_complete_existing_non_empty_email_mobile_returns_conflict(self):
|
|
cache.set(
|
|
"google_oauth_flow:mobile-conflict-flow",
|
|
{
|
|
"status": "collect_mobile",
|
|
"google_profile": {
|
|
"provider_user_id": "google-sub-mobile-conflict",
|
|
"email": "new-google@example.com",
|
|
"email_verified": True,
|
|
"first_name": "New",
|
|
"last_name": "Google",
|
|
"avatar_url": "https://example.com/conflict.png",
|
|
},
|
|
"email": "new-google@example.com",
|
|
"first_name": "New",
|
|
"last_name": "Google",
|
|
"avatar_url": "https://example.com/conflict.png",
|
|
"resolution": "new_account",
|
|
"target_user_id": None,
|
|
"mobile_hint": None,
|
|
},
|
|
900,
|
|
)
|
|
|
|
response = self.client.post(
|
|
"/api/users/oauth/google/complete/",
|
|
{"flow": "mobile-conflict-flow", "mobile": self.other_email_user.mobile},
|
|
format="json",
|
|
)
|
|
|
|
self.assertEqual(response.status_code, 409)
|
|
self.assertEqual(response.data["code"], "google_mobile_belongs_to_other_email")
|
|
|
|
@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")
|
|
|
|
@patch("apps.users.services.google_oauth.get_tokens_for_user")
|
|
def test_google_claim_verify_sets_blank_user_email_from_google(self, get_tokens_for_user):
|
|
get_tokens_for_user.return_value = {"access": "a", "refresh": "r"}
|
|
cache.set(
|
|
"google_oauth_flow:claim-verify-flow",
|
|
{
|
|
"status": "claim_required",
|
|
"google_profile": {
|
|
"provider_user_id": "google-sub-verify",
|
|
"email": "claimed@example.com",
|
|
"email_verified": True,
|
|
"first_name": "Claimed",
|
|
"last_name": "User",
|
|
"avatar_url": "https://example.com/claim.png",
|
|
},
|
|
"mobile": self.blank_email_user.mobile,
|
|
"user_id": str(self.blank_email_user.id),
|
|
"resolution": "existing_mobile_claim",
|
|
"email": "claimed@example.com",
|
|
"mobile_hint": "09*****0003",
|
|
"detail": "claim",
|
|
},
|
|
900,
|
|
)
|
|
|
|
with patch("django_redis.get_redis_connection") as get_redis_connection:
|
|
redis_mock = get_redis_connection.return_value
|
|
redis_mock.get.return_value = b"12345"
|
|
|
|
response = self.client.post(
|
|
"/api/users/oauth/google/claim/verify/",
|
|
{"flow": "claim-verify-flow", "code": "12345"},
|
|
format="json",
|
|
)
|
|
|
|
self.assertEqual(response.status_code, 200)
|
|
self.assertEqual(response.data["status"], "authenticated")
|
|
self.blank_email_user.refresh_from_db()
|
|
self.assertEqual(self.blank_email_user.email, "claimed@example.com")
|
|
self.assertTrue(
|
|
UserSocialAccount.objects.filter(
|
|
user=self.blank_email_user,
|
|
provider=UserSocialAccount.ProviderType.GOOGLE,
|
|
provider_user_id="google-sub-verify",
|
|
).exists()
|
|
)
|
|
|
|
|
|
class GoogleOAuthAuditCommandTests(APITestCase):
|
|
def test_audit_google_social_links_reports_suspicious_links(self):
|
|
linked_user = User.objects.create_user(
|
|
mobile="09126660001",
|
|
password="secret123",
|
|
email="owner@example.com",
|
|
)
|
|
other_user = User.objects.create_user(
|
|
mobile="09126660002",
|
|
password="secret123",
|
|
email="shared@example.com",
|
|
)
|
|
third_user = User.objects.create_user(
|
|
mobile="09126660003",
|
|
password="secret123",
|
|
email=None,
|
|
)
|
|
|
|
UserSocialAccount.objects.create(
|
|
user=linked_user,
|
|
provider=UserSocialAccount.ProviderType.GOOGLE,
|
|
provider_user_id="google-audit-1",
|
|
email="different@example.com",
|
|
email_verified=True,
|
|
avatar_url="https://example.com/audit-1.png",
|
|
)
|
|
UserSocialAccount.objects.create(
|
|
user=third_user,
|
|
provider=UserSocialAccount.ProviderType.GOOGLE,
|
|
provider_user_id="google-audit-2",
|
|
email="shared@example.com",
|
|
email_verified=True,
|
|
avatar_url="https://example.com/audit-2.png",
|
|
)
|
|
UserSocialAccount.objects.create(
|
|
user=linked_user,
|
|
provider=UserSocialAccount.ProviderType.GOOGLE,
|
|
provider_user_id="google-audit-3",
|
|
email="duplicate@example.com",
|
|
email_verified=True,
|
|
avatar_url="https://example.com/audit-3.png",
|
|
)
|
|
UserSocialAccount.objects.create(
|
|
user=third_user,
|
|
provider=UserSocialAccount.ProviderType.GOOGLE,
|
|
provider_user_id="google-audit-4",
|
|
email="duplicate@example.com",
|
|
email_verified=True,
|
|
avatar_url="https://example.com/audit-4.png",
|
|
)
|
|
|
|
stdout = StringIO()
|
|
call_command("audit_google_social_links", stdout=stdout)
|
|
output = stdout.getvalue()
|
|
|
|
self.assertIn("linked_user_email_mismatch", output)
|
|
self.assertIn("provider_email_matches_other_user", output)
|
|
self.assertIn("duplicate_provider_email_across_users", output)
|