Files
qlockify-backend-deployment/apps/users/tests/test_api_views.py

389 lines
14 KiB
Python

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
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)