Harden auth, AI tutor, payments, and subscriptions
This commit is contained in:
@@ -1,3 +1,341 @@
|
||||
from django.test import TestCase
|
||||
from unittest.mock import Mock, patch
|
||||
|
||||
# Create your tests here.
|
||||
from django.conf import settings
|
||||
from django.core.cache import cache
|
||||
from django.test import TestCase, override_settings
|
||||
from django.utils import timezone
|
||||
from rest_framework.test import APIClient
|
||||
from rest_framework.throttling import ScopedRateThrottle
|
||||
|
||||
from .models import OTPCode, Session, User
|
||||
from .services import OTPDeliveryError, send_otp_sms
|
||||
|
||||
|
||||
class SendOtpSmsTests(TestCase):
|
||||
"""
|
||||
تست‌های واحد برای accounts.services.send_otp_sms — تمرکز روی همون رفتاری
|
||||
که این fix اضافه می‌کنه: در dev فقط لاگ، در پروداکشن یا واقعاً ارسال یا
|
||||
خطای صریح (هیچ‌وقت موفقیتِ خاموش/ساختگی).
|
||||
"""
|
||||
|
||||
@override_settings(DEBUG=True)
|
||||
def test_dev_mode_does_not_call_provider(self):
|
||||
with patch('accounts.services.requests.post') as mock_post:
|
||||
send_otp_sms(phone='+989120000000', code='1234')
|
||||
mock_post.assert_not_called()
|
||||
|
||||
@override_settings(DEBUG=False, KAVENEGAR_API_KEY='')
|
||||
def test_production_without_api_key_raises_instead_of_silent_success(self):
|
||||
with patch('accounts.services.requests.post') as mock_post:
|
||||
with self.assertRaises(OTPDeliveryError):
|
||||
send_otp_sms(phone='+989120000000', code='1234')
|
||||
mock_post.assert_not_called()
|
||||
|
||||
@override_settings(DEBUG=False, KAVENEGAR_API_KEY='test-key', KAVENEGAR_SENDER='')
|
||||
def test_production_success_calls_provider(self):
|
||||
# پاسخِ موفقِ کاوه‌نگار: HTTP 200 + return.status == 200 در بدنه.
|
||||
mock_response = Mock(
|
||||
status_code=200,
|
||||
text='{"return": {"status": 200, "message": "OK"}, "entries": []}',
|
||||
)
|
||||
mock_response.json.return_value = {
|
||||
'return': {'status': 200, 'message': 'OK'},
|
||||
'entries': [{'messageid': 1, 'status': 1}],
|
||||
}
|
||||
with patch('accounts.services.requests.post', return_value=mock_response) as mock_post:
|
||||
send_otp_sms(phone='+989120000000', code='1234')
|
||||
mock_post.assert_called_once()
|
||||
args, kwargs = mock_post.call_args
|
||||
self.assertIn('test-key', args[0])
|
||||
self.assertEqual(kwargs['data']['receptor'], '+989120000000')
|
||||
self.assertIn('1234', kwargs['data']['message'])
|
||||
|
||||
@override_settings(DEBUG=False, KAVENEGAR_API_KEY='test-key')
|
||||
def test_production_provider_error_status_raises(self):
|
||||
mock_response = Mock(status_code=400, text='bad request')
|
||||
with patch('accounts.services.requests.post', return_value=mock_response):
|
||||
with self.assertRaises(OTPDeliveryError):
|
||||
send_otp_sms(phone='+989120000000', code='1234')
|
||||
|
||||
@override_settings(DEBUG=False, KAVENEGAR_API_KEY='test-key')
|
||||
def test_production_http_200_with_application_level_rejection_raises(self):
|
||||
# کاوه‌نگار حتی رد کردنِ اپلیکیشنی (مثلاً کلیدِ نامعتبر) رو با
|
||||
# HTTP 200 برمی‌گردونه؛ return.status غیر از 200 باید شکست باشه.
|
||||
mock_response = Mock(
|
||||
status_code=200,
|
||||
text='{"return": {"status": 401, "message": "Invalid API key"}, "entries": []}',
|
||||
)
|
||||
mock_response.json.return_value = {
|
||||
'return': {'status': 401, 'message': 'Invalid API key'},
|
||||
'entries': [],
|
||||
}
|
||||
with patch('accounts.services.requests.post', return_value=mock_response):
|
||||
with self.assertRaises(OTPDeliveryError):
|
||||
send_otp_sms(phone='+989120000000', code='1234')
|
||||
|
||||
@override_settings(DEBUG=False, KAVENEGAR_API_KEY='test-key')
|
||||
def test_production_malformed_provider_body_raises(self):
|
||||
# بدنه‌ی نامعتبر/غیرمنتظره (نه JSON، یا بدونِ ساختارِ return.status)
|
||||
# باید شکستِ تحویل در نظر گرفته بشه، نه موفقیتِ خاموش.
|
||||
mock_response = Mock(status_code=200, text='not json')
|
||||
mock_response.json.side_effect = ValueError('No JSON object could be decoded')
|
||||
with patch('accounts.services.requests.post', return_value=mock_response):
|
||||
with self.assertRaises(OTPDeliveryError):
|
||||
send_otp_sms(phone='+989120000000', code='1234')
|
||||
|
||||
@override_settings(DEBUG=False, KAVENEGAR_API_KEY='test-key')
|
||||
def test_production_network_failure_raises(self):
|
||||
import requests as requests_module
|
||||
with patch('accounts.services.requests.post', side_effect=requests_module.ConnectionError('boom')):
|
||||
with self.assertRaises(OTPDeliveryError):
|
||||
send_otp_sms(phone='+989120000000', code='1234')
|
||||
|
||||
|
||||
class RequestOTPViewTests(TestCase):
|
||||
"""
|
||||
تست‌های سطح API برای RequestOTPView: مطمئن می‌شیم رفتارِ dev قبلی حفظ
|
||||
شده و در پروداکشن، شکستِ تحویلِ پیامک باعثِ یک خطای صریح می‌شه (نه
|
||||
success=True دروغین) در حالی که ردیفِ OTPCode همچنان ساخته شده.
|
||||
"""
|
||||
|
||||
def setUp(self):
|
||||
self.client = APIClient()
|
||||
|
||||
@override_settings(DEBUG=True)
|
||||
def test_dev_mode_returns_otp_in_response_without_provider_call(self):
|
||||
with patch('accounts.views.send_otp_sms') as mock_send:
|
||||
response = self.client.post(
|
||||
'/api/auth/otp/request', {'phone': '+989120000001'}, format='json',
|
||||
)
|
||||
self.assertEqual(response.status_code, 200)
|
||||
self.assertTrue(response.data['success'])
|
||||
self.assertTrue(response.data['devOtp'])
|
||||
mock_send.assert_called_once()
|
||||
self.assertTrue(OTPCode.objects.filter(phone='+989120000001').exists())
|
||||
|
||||
@override_settings(DEBUG=False, KAVENEGAR_API_KEY='')
|
||||
def test_production_delivery_failure_does_not_report_success(self):
|
||||
response = self.client.post(
|
||||
'/api/auth/otp/request', {'phone': '+989120000002'}, format='json',
|
||||
)
|
||||
self.assertEqual(response.status_code, 503)
|
||||
# OTP row is created before delivery is attempted; only the SMS leg fails.
|
||||
self.assertTrue(OTPCode.objects.filter(phone='+989120000002').exists())
|
||||
|
||||
@override_settings(DEBUG=True)
|
||||
def test_generated_otp_is_always_a_4_digit_string_with_leading_zeros(self):
|
||||
# ⚠️ فیکسِ این تست: تولیدِ OTP از random.randint به secrets.randbelow
|
||||
# تغییر کرده؛ این تست مطمئن می‌شه فرمتِ ۴ رقمی (با صفرهای ابتدایی
|
||||
# حفظ‌شده، نه یک عددِ کوتاه‌تر) دست‌نخورده مونده.
|
||||
with patch('accounts.views.secrets.randbelow', return_value=7):
|
||||
with patch('accounts.views.send_otp_sms'):
|
||||
response = self.client.post(
|
||||
'/api/auth/otp/request', {'phone': '+989120000003'}, format='json',
|
||||
)
|
||||
self.assertEqual(response.status_code, 200)
|
||||
self.assertEqual(response.data['devOtp'], '0007')
|
||||
otp = OTPCode.objects.get(phone='+989120000003')
|
||||
self.assertEqual(otp.code, '0007')
|
||||
self.assertEqual(len(otp.code), 4)
|
||||
self.assertTrue(otp.code.isdigit())
|
||||
|
||||
|
||||
class RequestOTPStaleInvalidationTests(TestCase):
|
||||
"""
|
||||
تست‌های متمرکز روی فیکسِ stale OTP: قبل از ساختِ OTPCode جدید، تمامِ
|
||||
OTPCode های is_used=False قبلیِ همون شماره باید invalid (is_used=True)
|
||||
بشن — طوری که فقط آخرین کدِ صادرشده برای همون شماره برای ورود کار کنه.
|
||||
"""
|
||||
|
||||
def setUp(self):
|
||||
self.client = APIClient()
|
||||
# throttle (ScopedRateThrottle) از همون cache پیش‌فرضِ جنگو استفاده
|
||||
# می‌کنه؛ بین تست‌ها پاکش می‌کنیم تا شمارنده‌ی یک تست روی تستِ بعدی
|
||||
# اثر نذاره.
|
||||
cache.clear()
|
||||
|
||||
def _request_otp(self, phone, code):
|
||||
# با patch کردنِ secrets.randbelow، کدِ هر درخواست معلوم و
|
||||
# قابل‌پیش‌بینیه (به‌جای اتکا به شانسِ برخورد نکردنِ دو کدِ تصادفی).
|
||||
with patch('accounts.views.secrets.randbelow', return_value=int(code)):
|
||||
with patch('accounts.views.send_otp_sms'):
|
||||
response = self.client.post(
|
||||
'/api/auth/otp/request', {'phone': phone}, format='json',
|
||||
)
|
||||
self.assertEqual(response.status_code, 200)
|
||||
self.assertEqual(response.data['devOtp'], code)
|
||||
return code
|
||||
|
||||
@override_settings(DEBUG=True)
|
||||
def test_second_request_invalidates_first_otp(self):
|
||||
phone = '+989121110001'
|
||||
first_code = self._request_otp(phone, '1111')
|
||||
first_otp = OTPCode.objects.get(phone=phone, code=first_code)
|
||||
self.assertFalse(first_otp.is_used)
|
||||
|
||||
self._request_otp(phone, '2222')
|
||||
|
||||
first_otp.refresh_from_db()
|
||||
self.assertTrue(first_otp.is_used)
|
||||
|
||||
@override_settings(DEBUG=True)
|
||||
def test_newest_otp_still_works_for_login(self):
|
||||
phone = '+989121110002'
|
||||
self._request_otp(phone, '1111')
|
||||
second_code = self._request_otp(phone, '2222')
|
||||
|
||||
response = self.client.post(
|
||||
'/api/auth/login', {'phone': phone, 'otp': second_code}, format='json',
|
||||
)
|
||||
self.assertEqual(response.status_code, 200)
|
||||
|
||||
@override_settings(DEBUG=True)
|
||||
def test_older_otp_is_rejected_after_new_request(self):
|
||||
phone = '+989121110003'
|
||||
first_code = self._request_otp(phone, '1111')
|
||||
self._request_otp(phone, '2222')
|
||||
|
||||
response = self.client.post(
|
||||
'/api/auth/login', {'phone': phone, 'otp': first_code}, format='json',
|
||||
)
|
||||
self.assertEqual(response.status_code, 400)
|
||||
|
||||
|
||||
class RefreshTokenReplayTests(TestCase):
|
||||
"""
|
||||
🔒 اثبات فیکسِ replay: یک refresh token معتبر باید فقط یک‌بار قابل
|
||||
استفاده باشه. بعد از یک رفرشِ موفق، همون توکن دیگه نباید کار کنه —
|
||||
حتی بدون rotation، مصرفِ همون توکن دوباره باید 401 بگیره.
|
||||
"""
|
||||
|
||||
def setUp(self):
|
||||
self.client = APIClient()
|
||||
self.user = User.objects.create(is_guest=True, device_id='device-refresh-test')
|
||||
self.session = Session.objects.create(
|
||||
user=self.user,
|
||||
refresh_token='test-refresh-token-abc123',
|
||||
device_id='device-refresh-test',
|
||||
expires_at=timezone.now() + timezone.timedelta(days=30),
|
||||
revoked=False,
|
||||
)
|
||||
|
||||
def test_valid_refresh_token_succeeds(self):
|
||||
response = self.client.post(
|
||||
'/api/auth/refresh', {'refreshToken': self.session.refresh_token}, format='json',
|
||||
)
|
||||
self.assertEqual(response.status_code, 200)
|
||||
self.assertTrue(response.data['token'])
|
||||
|
||||
def test_same_refresh_token_rejected_on_second_use(self):
|
||||
first_response = self.client.post(
|
||||
'/api/auth/refresh', {'refreshToken': self.session.refresh_token}, format='json',
|
||||
)
|
||||
self.assertEqual(first_response.status_code, 200)
|
||||
|
||||
second_response = self.client.post(
|
||||
'/api/auth/refresh', {'refreshToken': self.session.refresh_token}, format='json',
|
||||
)
|
||||
self.assertEqual(second_response.status_code, 401)
|
||||
|
||||
self.session.refresh_from_db()
|
||||
self.assertTrue(self.session.revoked)
|
||||
|
||||
def test_successful_refresh_returns_new_refresh_token(self):
|
||||
response = self.client.post(
|
||||
'/api/auth/refresh', {'refreshToken': self.session.refresh_token}, format='json',
|
||||
)
|
||||
self.assertEqual(response.status_code, 200)
|
||||
new_refresh_token = response.data['refreshToken']
|
||||
self.assertTrue(new_refresh_token)
|
||||
self.assertNotEqual(new_refresh_token, self.session.refresh_token)
|
||||
self.assertTrue(
|
||||
Session.objects.filter(
|
||||
refresh_token=new_refresh_token, user=self.user, revoked=False,
|
||||
).exists()
|
||||
)
|
||||
|
||||
|
||||
class LoginPhoneRateLimitTests(TestCase):
|
||||
"""
|
||||
🔒 اثبات محدودیتِ جدیدِ بر اساسِ شماره‌تلفن روی LoginView: تلاش‌های
|
||||
تکراریِ ورود برای همون شماره باید بعد از حدِ مجاز بلاک بشن، حتی اگه
|
||||
throttle_scope='otp_login' (IP-محور) رو دور بزنیم.
|
||||
"""
|
||||
|
||||
def setUp(self):
|
||||
self.client = APIClient()
|
||||
cache.clear()
|
||||
|
||||
@override_settings(DEBUG=True, OTP_LOGIN_PHONE_MAX_ATTEMPTS=3)
|
||||
def test_repeated_attempts_for_same_phone_are_blocked(self):
|
||||
phone = '+989123334444'
|
||||
|
||||
for _ in range(3):
|
||||
response = self.client.post(
|
||||
'/api/auth/login', {'phone': phone, 'otp': 'wrong'}, format='json',
|
||||
)
|
||||
self.assertEqual(response.status_code, 400)
|
||||
|
||||
blocked_response = self.client.post(
|
||||
'/api/auth/login', {'phone': phone, 'otp': 'wrong'}, format='json',
|
||||
)
|
||||
self.assertEqual(blocked_response.status_code, 429)
|
||||
|
||||
|
||||
class GuestSessionThrottleTests(TestCase):
|
||||
"""
|
||||
🔒 اثبات محدودیتِ جدیدِ ScopedRateThrottle (scope='guest_session') روی
|
||||
GuestSessionView: درخواست‌های تکراری از همون کلاینت (IP) باید بعد از
|
||||
حدِ مجاز با ۴۲۹ رد بشن.
|
||||
"""
|
||||
|
||||
def setUp(self):
|
||||
self.client = APIClient()
|
||||
cache.clear()
|
||||
|
||||
@override_settings(
|
||||
REST_FRAMEWORK={
|
||||
**settings.REST_FRAMEWORK,
|
||||
'DEFAULT_THROTTLE_RATES': {
|
||||
**settings.REST_FRAMEWORK['DEFAULT_THROTTLE_RATES'],
|
||||
'guest_session': '3/hour',
|
||||
},
|
||||
}
|
||||
)
|
||||
def test_repeated_guest_session_requests_are_eventually_throttled(self):
|
||||
# ⚠️ اصلاح (باگِ خودِ تست، نه اپلیکیشن): این تست دو مشکلِ جدا
|
||||
# داشت:
|
||||
# ۱) `type` رو نمی‌فرستاد (پایین‌تر فیکس شده) — قراردادِ واقعیِ
|
||||
# API طبقِ README نیازش داره، پس هر درخواست قبل از throttle
|
||||
# با 400 رد می‌شد.
|
||||
# ۲) override_settings(REST_FRAMEWORK=...) به‌تنهایی رفتارِ
|
||||
# ScopedRateThrottle رو تغییر نمی‌ده: DRF مقدارِ
|
||||
# THROTTLE_RATES رو یک‌بار موقعِ import ماژولِ throttling
|
||||
# به‌عنوانِ یک class attribute از api_settings.
|
||||
# DEFAULT_THROTTLE_RATES می‌گیره (همون دیکشنریِ اصلیِ
|
||||
# settings.py، با reference). override_settings یک دیکشنریِ
|
||||
# *جدید* برای settings.REST_FRAMEWORK می‌سازه، اما این
|
||||
# class attribute هنوز به همون دیکشنریِ قدیمی اشاره می‌کنه —
|
||||
# سیگنالِ setting_changed فقط api_settings رو reload می‌کنه،
|
||||
# نه این class attribute رو. نتیجه: نرخِ واقعیِ اعمال‌شده
|
||||
# همیشه همون دیفالتِ 20/hour در settings.py می‌موند، نه
|
||||
# 3/hour، پس ۵ درخواست هیچ‌وقت به سقف نمی‌رسیدن. برای اینکه
|
||||
# این تست واقعاً نرخِ محدودشده رو بسنجه، مستقیماً همون
|
||||
# دیکشنریِ زنده‌ی ScopedRateThrottle.THROTTLE_RATES رو
|
||||
# patch.dict می‌کنیم (که override_settings بالا هم برای
|
||||
# سازگاری/مستندسازیِ نیت نگه داشته می‌شه).
|
||||
with patch.dict(ScopedRateThrottle.THROTTLE_RATES, {'guest_session': '3/hour'}):
|
||||
for i in range(3):
|
||||
response = self.client.post(
|
||||
'/api/auth/session',
|
||||
{'type': 'guest', 'deviceId': f'device-{i}'},
|
||||
format='json',
|
||||
)
|
||||
self.assertEqual(response.status_code, 200)
|
||||
|
||||
blocked_response = self.client.post(
|
||||
'/api/auth/session',
|
||||
{'type': 'guest', 'deviceId': 'device-blocked'},
|
||||
format='json',
|
||||
)
|
||||
self.assertEqual(blocked_response.status_code, 429)
|
||||
|
||||
Reference in New Issue
Block a user