from unittest.mock import Mock, patch from django.conf import settings from django.contrib import admin from django.contrib.auth import get_user_model from django.contrib.auth.models import Group, Permission from django.core.cache import cache from django.core.management import call_command from django.test import TestCase, TransactionTestCase, override_settings from django.urls import reverse from django.utils import timezone from rest_framework.test import APIClient from rest_framework.throttling import ScopedRateThrottle from .authentication import issue_access_token from .models import OTPCode, Profile, Session, User from .services import OTPDeliveryError, SMSDeliveryError, 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='09120000000', 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='09120000000', 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='09120000000', code='1234') mock_post.assert_called_once() args, kwargs = mock_post.call_args self.assertIn('test-key', args[0]) self.assertEqual(kwargs['data']['receptor'], '09120000000') 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='09120000000', 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='09120000000', 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='09120000000', 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='09120000000', code='1234') @override_settings(DEBUG=True) def test_dev_mode_log_never_contains_full_phone_only_safe_suffix(self): # 🔒 اثبات فیکسِ این دور: لاگِ OTP نباید هیچ‌وقت شماره‌تلفنِ کامل رو # نشون بده؛ فقط ۴ رقمِ آخر (شناسه‌ی امن) باید در پیامِ لاگ باشه. phone = '09120000000' with self.assertLogs('accounts.services', level='INFO') as captured: send_otp_sms(phone=phone, code='1234') self.assertEqual(len(captured.records), 1) message = captured.records[0].getMessage() self.assertNotIn(phone, message) self.assertIn('phone_suffix=0000', message) # رفتار/سطح/کدِ قبلی حفظ شده: کدِ OTP هیچ‌وقت لاگ نمی‌شه، فقط طولش. self.assertNotIn('1234', message) self.assertIn('code_len=4', message) @override_settings(DEBUG=False, KAVENEGAR_API_KEY='test-key') def test_provider_error_logs_never_contain_raw_response_body(self): # 🔒 اثبات فیکسِ این دور: لاگ‌های خطای provider دیگه response.text # خام رو شامل نمی‌شن؛ فقط context مفید و بی‌خطر (phone_suffix، # status، provider_status) باقی می‌مونه. # حالتِ ۱: HTTP غیرِ ۲۰۰. mock_response = Mock(status_code=400, text='bad request body secret-ish') with patch('accounts.services.requests.post', return_value=mock_response): with self.assertLogs('accounts.services', level='ERROR') as captured: with self.assertRaises(OTPDeliveryError): send_otp_sms(phone='09120000000', code='1234') message = captured.records[0].getMessage() self.assertNotIn('bad request body secret-ish', message) self.assertIn('phone_suffix=0000', message) self.assertIn('status=400', message) # حالتِ ۲: بدنه‌ی نامعتبر/غیرقابل‌parse. mock_response = Mock(status_code=200, text='not json body secret-ish') mock_response.json.side_effect = ValueError('No JSON object could be decoded') with patch('accounts.services.requests.post', return_value=mock_response): with self.assertLogs('accounts.services', level='ERROR') as captured: with self.assertRaises(OTPDeliveryError): send_otp_sms(phone='09120000000', code='1234') message = captured.records[0].getMessage() self.assertNotIn('not json body secret-ish', message) self.assertIn('phone_suffix=0000', message) self.assertIn('status=200', message) # حالتِ ۳: ردِ اپلیکیشنی (HTTP 200، return.status != 200). mock_response = Mock( status_code=200, text='{"return": {"status": 401, "message": "secret-ish detail"}, "entries": []}', ) mock_response.json.return_value = { 'return': {'status': 401, 'message': 'secret-ish detail'}, 'entries': [], } with patch('accounts.services.requests.post', return_value=mock_response): with self.assertLogs('accounts.services', level='ERROR') as captured: with self.assertRaises(OTPDeliveryError): send_otp_sms(phone='09120000000', code='1234') message = captured.records[0].getMessage() self.assertNotIn('secret-ish detail', message) self.assertIn('phone_suffix=0000', message) self.assertIn('provider_status=401', message) 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': '09120000001'}, 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='09120000001').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': '09120000002'}, format='json', ) self.assertEqual(response.status_code, 503) # ردیفِ کدی که هرگز نرسید حذف می‌شود تا در cooldown و سقفِ روزانه‌ی # دیتابیسی شمرده نشود (۲۰۲۶-۰۹-۲۵). self.assertFalse(OTPCode.objects.filter(phone='09120000002').exists()) @override_settings(DEBUG=True) def test_generated_otp_is_always_a_6_digit_string_with_leading_zeros(self): # ۶ رقم از ۲۰۲۶-۰۹-۲۵ (با ۴ رقم حدس زدن عملی بود)؛ صفرهای ابتدایی باید بمانند. 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': '09120000003'}, format='json', ) self.assertEqual(response.status_code, 200) self.assertEqual(response.data['devOtp'], '000007') otp = OTPCode.objects.get(phone='09120000003') self.assertEqual(otp.code, '000007') self.assertEqual(len(otp.code), 6) 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 # ⚠️ این کلاس چندین درخواست پشتِ‌سرِهم برای *همون* شماره می‌فرسته (بدونِ # فاصله)، چون داره رفتارِ invalidation رو می‌سنجه، نه cooldown رو. پس # cooldown جدیدِ بین‌درخواستی رو اینجا خاموش می‌کنیم تا این تست‌ها با # اون تداخل نکنن؛ cooldown خودش تستِ اختصاصیِ جدا داره # (RequestOTPPhoneCooldownTests). @override_settings(DEBUG=True, OTP_REQUEST_PHONE_COOLDOWN_SECONDS=0) def test_second_request_invalidates_first_otp(self): phone = '09121110001' first_code = self._request_otp(phone, '111111') first_otp = OTPCode.objects.get(phone=phone, code=first_code) self.assertFalse(first_otp.is_used) self._request_otp(phone, '222222') first_otp.refresh_from_db() self.assertTrue(first_otp.is_used) @override_settings(DEBUG=True, OTP_REQUEST_PHONE_COOLDOWN_SECONDS=0) def test_newest_otp_still_works_for_login(self): phone = '09121110002' self._request_otp(phone, '111111') second_code = self._request_otp(phone, '222222') response = self.client.post( '/api/auth/login', {'phone': phone, 'otp': second_code}, format='json', ) self.assertEqual(response.status_code, 200) @override_settings(DEBUG=True, OTP_REQUEST_PHONE_COOLDOWN_SECONDS=0) def test_older_otp_is_rejected_after_new_request(self): phone = '09121110003' first_code = self._request_otp(phone, '111111') self._request_otp(phone, '222222') 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 = '09123334444' 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 RequestOTPPhoneRateLimitTests(TestCase): """ 🔒 اثبات محدودیتِ جدیدِ بر اساسِ شماره‌تلفن روی RequestOTPView: درخواست‌های تکراریِ صدورِ OTP برای همون شماره باید بعد از حدِ مجاز بلاک بشن، حتی اگه throttle_scope='otp_request' (IP-محور) رو با تغییرِ IP دور بزنیم. همچنین درخواستِ بلاک‌شده نباید OTPCode جدید بسازه یا پیامک بفرسته. """ def setUp(self): self.client = APIClient() cache.clear() # cooldown بین‌درخواستیِ جدید رو اینجا خاموش می‌کنیم چون این تست‌ها دارن # سقفِ *ساعتی* رو می‌سنجن (با چند درخواستِ پشتِ‌سرِهم بدونِ فاصله)، نه # cooldown رو؛ cooldown تستِ اختصاصیِ خودش رو داره # (RequestOTPPhoneCooldownTests). @override_settings(DEBUG=True, OTP_REQUEST_PHONE_MAX_ATTEMPTS=3, OTP_REQUEST_PHONE_COOLDOWN_SECONDS=0) def test_repeated_requests_for_same_phone_are_eventually_blocked(self): phone = '09127778888' # otp_request (IP-محور) پیش‌فرض 5/hour است؛ برای اینکه فقط # محدودیتِ شماره‌تلفن رو بسنجیم، throttle IP-محور رو موقتاً باز # می‌کنیم تا تداخلی با شمارشِ این تست نداشته باشه. with patch.dict(ScopedRateThrottle.THROTTLE_RATES, {'otp_request': '1000/hour'}): with patch('accounts.views.send_otp_sms'): for _ in range(3): response = self.client.post( '/api/auth/otp/request', {'phone': phone}, format='json', ) self.assertEqual(response.status_code, 200) blocked_response = self.client.post( '/api/auth/otp/request', {'phone': phone}, format='json', ) self.assertEqual(blocked_response.status_code, 429) @override_settings(DEBUG=True, OTP_REQUEST_PHONE_MAX_ATTEMPTS=3, OTP_REQUEST_PHONE_COOLDOWN_SECONDS=0) def test_changing_client_ip_does_not_bypass_the_phone_limit(self): phone = '09127778889' with patch.dict(ScopedRateThrottle.THROTTLE_RATES, {'otp_request': '1000/hour'}): with patch('accounts.views.send_otp_sms'): for i in range(3): # هر درخواست از یک IP متفاوت میاد؛ throttle IP-محور # (که به‌هرحال بالا باز شده) اینجا اهمیتی نداره — نکته # اینه که محدودیتِ شماره‌تلفن مستقل از IP باید اعمال بشه. response = self.client.post( '/api/auth/otp/request', {'phone': phone}, format='json', REMOTE_ADDR=f'10.0.0.{i + 1}', ) self.assertEqual(response.status_code, 200) blocked_response = self.client.post( '/api/auth/otp/request', {'phone': phone}, format='json', REMOTE_ADDR='10.0.0.99', ) self.assertEqual(blocked_response.status_code, 429) @override_settings(DEBUG=True, OTP_REQUEST_PHONE_MAX_ATTEMPTS=1, OTP_REQUEST_PHONE_COOLDOWN_SECONDS=0) def test_throttled_request_creates_no_new_otp_and_sends_no_sms(self): phone = '09127778890' with patch.dict(ScopedRateThrottle.THROTTLE_RATES, {'otp_request': '1000/hour'}): with patch('accounts.views.send_otp_sms') as mock_send: first_response = self.client.post( '/api/auth/otp/request', {'phone': phone}, format='json', ) self.assertEqual(first_response.status_code, 200) self.assertEqual(mock_send.call_count, 1) otp_count_before = OTPCode.objects.filter(phone=phone).count() blocked_response = self.client.post( '/api/auth/otp/request', {'phone': phone}, format='json', ) self.assertEqual(blocked_response.status_code, 429) # نه OTPCode جدیدی ساخته شده، نه پیامکِ جدیدی ارسال شده. self.assertEqual(mock_send.call_count, 1) self.assertEqual( OTPCode.objects.filter(phone=phone).count(), otp_count_before, ) class RequestOTPPhoneRateLimitConcurrencyTests(TransactionTestCase): """ 🔒 اثباتِ ایمنی محدودیتِ شماره‌تلفن زیرِ همزمانی: دو درخواستِ *واقعاً* هم‌زمان (با threading.Barrier هم‌زمان‌شده، نه صرفاً پشتِ‌سرِهم) برای یک شماره، با سقفِ مجازِ ۱، نباید هر دو موفق بشن. الگویِ cache.add که یک عملیاتِ اتمیکِ سطحِ backend cache است (نه یک read-then-write ناامن) دقیقاً همینو تضمین می‌کنه. از TransactionTestCase استفاده می‌شه چون TestCase هر تست رو داخلِ یک تراکنشِ بیرونیِ خودش می‌پیچه که برای درخواست‌هایِ واقعاً هم‌زمان از threadهایِ جدا مناسب نیست. """ def setUp(self): cache.clear() @override_settings(DEBUG=True, OTP_REQUEST_PHONE_MAX_ATTEMPTS=1) def test_two_simultaneous_requests_do_not_both_pass_the_phone_limit(self): import threading phone = '09127778891' barrier = threading.Barrier(2) results = [] results_lock = threading.Lock() def _make_request(): client = APIClient() barrier.wait() # هر دو thread دقیقاً هم‌زمان درخواست بفرستن response = client.post( '/api/auth/otp/request', {'phone': phone}, format='json', ) with results_lock: results.append(response.status_code) with patch.dict(ScopedRateThrottle.THROTTLE_RATES, {'otp_request': '1000/hour'}): with patch('accounts.views.send_otp_sms'): threads = [threading.Thread(target=_make_request) for _ in range(2)] for t in threads: t.start() for t in threads: t.join(timeout=10) self.assertEqual(len(results), 2) # دقیقاً یکی باید موفق بشه (200) و اون یکی باید بلاک بشه (429)؛ # هیچ‌وقت هر دو نباید 200 بگیرن. self.assertEqual(sorted(results), [200, 429]) self.assertEqual(OTPCode.objects.filter(phone=phone).count(), 1) class RequestOTPPhoneCooldownTests(TestCase): """ 🔒 اثباتِ محدودیتِ جدیدِ cooldown بین‌درخواستی برای RequestOTPView: هدفِ این محدودیت جلوگیریِ burst (چند درخواست برای همون شماره در عرضِ چند ثانیه/بدونِ فاصله) است — چیزی که سقفِ ساعتیِ per-phone موجود (OTP_REQUEST_PHONE_MAX_ATTEMPTS) به‌تنهایی جلوش رو نمی‌گیره، چون فقط حجمِ کلی رو محدود می‌کنه نه فاصله‌ی بینِ تلاش‌ها رو. """ def setUp(self): self.client = APIClient() cache.clear() @override_settings(DEBUG=True, OTP_REQUEST_PHONE_COOLDOWN_SECONDS=60) def test_immediate_resend_for_same_phone_is_blocked_by_cooldown(self): phone = '09127779001' with patch('accounts.views.send_otp_sms'): first_response = self.client.post( '/api/auth/otp/request', {'phone': phone}, format='json', ) self.assertEqual(first_response.status_code, 200) second_response = self.client.post( '/api/auth/otp/request', {'phone': phone}, format='json', ) self.assertEqual(second_response.status_code, 429) @override_settings(DEBUG=True, OTP_REQUEST_PHONE_COOLDOWN_SECONDS=60, OTP_REQUEST_PHONE_MAX_ATTEMPTS=5) def test_cooldown_blocks_burst_without_consuming_the_hourly_quota(self): # پنج درخواستِ پشتِ‌سرِهم بدونِ فاصله برای همون شماره: فقط اولی باید # موفق بشه، بقیه باید با cooldown (429) بلاک بشن — و چون # درخواست‌هایِ بلاک‌شده توسطِ cooldown هیچ‌وقت به شمارنده‌ی ساعتی # نمی‌رسن، سهمیه‌ی ساعتی مصرف نمی‌شه. phone = '09127779002' # throttle IP-محورِ otp_request (پیش‌فرض 5/hour) رو موقتاً باز # می‌کنیم تا اینجا فقط cooldown/سهمیه‌ی ساعتیِ per-phone سنجیده بشه، # نه throttle IP-محورِ جدا (که تستِ اختصاصیِ خودش رو داره). with patch.dict(ScopedRateThrottle.THROTTLE_RATES, {'otp_request': '1000/hour'}): with patch('accounts.views.send_otp_sms') as mock_send: statuses = [] for _ in range(5): response = self.client.post( '/api/auth/otp/request', {'phone': phone}, format='json', ) statuses.append(response.status_code) self.assertEqual(statuses, [200, 429, 429, 429, 429]) mock_send.assert_called_once() self.assertEqual(OTPCode.objects.filter(phone=phone).count(), 1) # سهمیه‌ی ساعتی هنوز دست‌نخورده است: بعد از پایانِ cooldown، # همچنان می‌شه تا سقفِ مجاز درخواست داد. # گذشتِ زمان: هم کلیدِ کش و هم cooldownِ دیتابیسی (از expires_at). cache.delete(f'otp_request_phone_cooldown:{phone}') OTPCode.objects.filter(phone=phone).update( expires_at=timezone.now() + timezone.timedelta(seconds=settings.OTP_TTL_SECONDS - 61), ) with patch('accounts.views.send_otp_sms') as mock_send_again: allowed_again = self.client.post( '/api/auth/otp/request', {'phone': phone}, format='json', ) self.assertEqual(allowed_again.status_code, 200) mock_send_again.assert_called_once() @override_settings(DEBUG=True, OTP_REQUEST_PHONE_COOLDOWN_SECONDS=60) def test_cooldown_blocked_request_creates_no_otp_and_sends_no_sms(self): phone = '09127779003' with patch('accounts.views.send_otp_sms') as mock_send: self.client.post('/api/auth/otp/request', {'phone': phone}, format='json') otp_count_before = OTPCode.objects.filter(phone=phone).count() blocked_response = self.client.post( '/api/auth/otp/request', {'phone': phone}, format='json', ) self.assertEqual(blocked_response.status_code, 429) self.assertEqual(mock_send.call_count, 1) self.assertEqual(OTPCode.objects.filter(phone=phone).count(), otp_count_before) @override_settings(DEBUG=True, OTP_REQUEST_PHONE_COOLDOWN_SECONDS=1) def test_cooldown_expires_after_configured_window(self): import time phone = '09127779004' with patch('accounts.views.send_otp_sms'): first_response = self.client.post( '/api/auth/otp/request', {'phone': phone}, format='json', ) self.assertEqual(first_response.status_code, 200) time.sleep(1.2) second_response = self.client.post( '/api/auth/otp/request', {'phone': phone}, format='json', ) self.assertEqual(second_response.status_code, 200) @override_settings(DEBUG=True, OTP_REQUEST_PHONE_COOLDOWN_SECONDS=60) def test_cooldown_is_per_phone_not_global(self): with patch('accounts.views.send_otp_sms'): first_response = self.client.post( '/api/auth/otp/request', {'phone': '09127779005'}, format='json', ) second_response = self.client.post( '/api/auth/otp/request', {'phone': '09127779006'}, format='json', ) self.assertEqual(first_response.status_code, 200) self.assertEqual(second_response.status_code, 200) def test_default_cooldown_is_120_seconds(self): # ⚠️ تصمیمِ محصول/امنیت: کاربر باید هر ۲ دقیقه بتونه OTP دوباره # بگیره؛ این باید دقیقاً همون مقداریه که enforcement سمتِ سرور # اعمال می‌کنه (نه صرفاً یک عددِ hardcode‌شده در فرانت). self.assertEqual(settings.OTP_REQUEST_PHONE_COOLDOWN_SECONDS, 120) @override_settings(DEBUG=True, OTP_REQUEST_PHONE_COOLDOWN_SECONDS=120) def test_cooldown_response_reports_retry_after_header_and_body(self): phone = '09127779008' with patch('accounts.views.send_otp_sms'): first_response = self.client.post( '/api/auth/otp/request', {'phone': phone}, format='json', ) self.assertEqual(first_response.status_code, 200) blocked_response = self.client.post( '/api/auth/otp/request', {'phone': phone}, format='json', ) self.assertEqual(blocked_response.status_code, 429) # هدرِ استانداردِ Retry-After — فرانت باید countdown رو دقیقاً از # همین مقدار شروع کنه، نه از یک عددِ ثابتِ سمتِ کلاینت. self.assertIn('Retry-After', blocked_response.headers) retry_after_header = int(blocked_response.headers['Retry-After']) self.assertGreaterEqual(retry_after_header, 1) self.assertLessEqual(retry_after_header, 120) # همون مقدار در بدنه‌ی پاسخ هم برگردونده می‌شه (retryAfter، هم‌راستا # با استایلِ camelCase بقیه‌ی فیلدهای response در این پروژه). self.assertIn('retryAfter', blocked_response.data) self.assertEqual(blocked_response.data['retryAfter'], retry_after_header) @override_settings(DEBUG=True, OTP_REQUEST_PHONE_COOLDOWN_SECONDS=120) def test_retry_after_counts_down_as_time_passes(self): import time phone = '09127779009' with patch('accounts.views.send_otp_sms'): self.client.post('/api/auth/otp/request', {'phone': phone}, format='json') immediate_retry = self.client.post( '/api/auth/otp/request', {'phone': phone}, format='json', ) self.assertEqual(immediate_retry.status_code, 429) immediate_retry_after = immediate_retry.data['retryAfter'] time.sleep(1.1) later_retry = self.client.post( '/api/auth/otp/request', {'phone': phone}, format='json', ) self.assertEqual(later_retry.status_code, 429) later_retry_after = later_retry.data['retryAfter'] # با گذشتِ زمان، عددِ retryAfter باید کم بشه (نه ثابت بمونه) — این # همون چیزیه که فرانت برای countdown دقیق بهش نیاز داره. self.assertLess(later_retry_after, immediate_retry_after) class RequestOTPPhoneCooldownConcurrencyTests(TransactionTestCase): """ 🔒 اثباتِ ایمنیِ cooldown زیرِ همزمانیِ واقعی (نه صرفاً پشتِ‌سرِهم): دقیقاً همون الگویِ RequestOTPPhoneRateLimitConcurrencyTests، این‌بار برایِ کلیدِ cooldown. از TransactionTestCase استفاده می‌شه چون TestCase هر تست رو داخلِ یک تراکنشِ بیرونیِ خودش می‌پیچه که برایِ درخواست‌هایِ واقعاً هم‌زمان از threadهایِ جدا مناسب نیست. """ def setUp(self): cache.clear() @override_settings(DEBUG=True, OTP_REQUEST_PHONE_COOLDOWN_SECONDS=60) def test_two_simultaneous_requests_for_same_phone_only_one_passes_cooldown(self): import threading phone = '09127779007' barrier = threading.Barrier(2) results = [] results_lock = threading.Lock() def _make_request(): client = APIClient() barrier.wait() response = client.post( '/api/auth/otp/request', {'phone': phone}, format='json', ) with results_lock: results.append(response.status_code) with patch('accounts.views.send_otp_sms'): threads = [threading.Thread(target=_make_request) for _ in range(2)] for t in threads: t.start() for t in threads: t.join(timeout=10) self.assertEqual(len(results), 2) self.assertEqual(sorted(results), [200, 429]) self.assertEqual(OTPCode.objects.filter(phone=phone).count(), 1) @override_settings(DEBUG=True, OTP_REQUEST_PHONE_COOLDOWN_SECONDS=120) def test_cooldown_is_cleared_when_sms_delivery_fails(self): # ⚠️ فیکسِ باگ: قبلاً cooldown *قبل از* فراخوانیِ send_otp_sms ثبت # می‌شد. اگه ارسالِ پیامک با OTPDeliveryError شکست می‌خورد، کاربر # هیچ کدی دریافت نکرده بود ولی همچنان تا ۱۲۰ ثانیه بلاک می‌موند. # این تست تضمین می‌کنه که با شکستِ ارسال، cooldown پاک می‌شه و # کاربر می‌تونه فوراً دوباره تلاش کنه. phone = '09127779010' with patch('accounts.views.send_otp_sms', side_effect=OTPDeliveryError('boom')): first_response = self.client.post( '/api/auth/otp/request', {'phone': phone}, format='json', ) self.assertEqual(first_response.status_code, 503) # اثباتِ اینکه cooldown دیگه فعال نیست: بلافاصله یک تلاشِ دیگه (این‌بار # موفق) باید 429 نگیره. with patch('accounts.views.send_otp_sms') as mock_send: retry_response = self.client.post( '/api/auth/otp/request', {'phone': phone}, format='json', ) self.assertEqual(retry_response.status_code, 200) mock_send.assert_called_once() @override_settings(DEBUG=True, OTP_REQUEST_PHONE_COOLDOWN_SECONDS=120) def test_cooldown_still_active_after_successful_delivery(self): # اثباتِ عدمِ رگرسیون: وقتی ارسال موفق باشه، cooldown همچنان طبقِ # رفتارِ قبلی فعال می‌مونه (فقط شکستِ تحویل باید cooldown رو پاک کنه). phone = '09127779011' with patch('accounts.views.send_otp_sms'): first_response = self.client.post( '/api/auth/otp/request', {'phone': phone}, format='json', ) self.assertEqual(first_response.status_code, 200) blocked_response = self.client.post( '/api/auth/otp/request', {'phone': phone}, format='json', ) self.assertEqual(blocked_response.status_code, 429) self.assertIn('Retry-After', blocked_response.headers) class RefreshTokenThrottleTests(TestCase): """ 🔒 اثبات محدودیتِ جدیدِ throttle_scope='token_refresh' روی RefreshTokenView: درخواست‌های عادی و کافی موفق باشن، درخواست‌های بیش‌ازحد با ۴۲۹ رد بشن، و یک درخواستِ بلاک‌شده هیچ Session جدیدی نسازه/revoke نکنه و هیچ توکنِ جدیدی تولید نکنه. """ def setUp(self): cache.clear() self.client = APIClient() self.user = User.objects.create(is_guest=True, device_id='device-refresh-throttle-test') def _create_session(self, token_suffix): return Session.objects.create( user=self.user, refresh_token=f'test-refresh-token-{token_suffix}', device_id='device-refresh-throttle-test', expires_at=timezone.now() + timezone.timedelta(days=30), revoked=False, ) def test_requests_within_the_limit_still_succeed(self): with patch.dict(ScopedRateThrottle.THROTTLE_RATES, {'token_refresh': '3/hour'}): for i in range(3): session = self._create_session(i) response = self.client.post( '/api/auth/refresh', {'refreshToken': session.refresh_token}, format='json', ) self.assertEqual(response.status_code, 200) def test_excessive_refresh_requests_are_eventually_rejected(self): with patch.dict(ScopedRateThrottle.THROTTLE_RATES, {'token_refresh': '3/hour'}): for i in range(3): session = self._create_session(i) response = self.client.post( '/api/auth/refresh', {'refreshToken': session.refresh_token}, format='json', ) self.assertEqual(response.status_code, 200) extra_session = self._create_session('extra') blocked_response = self.client.post( '/api/auth/refresh', {'refreshToken': extra_session.refresh_token}, format='json', ) self.assertEqual(blocked_response.status_code, 429) def test_throttled_request_does_not_revoke_or_create_session_or_token(self): with patch.dict(ScopedRateThrottle.THROTTLE_RATES, {'token_refresh': '1/hour'}): first_session = self._create_session('first') first_response = self.client.post( '/api/auth/refresh', {'refreshToken': first_session.refresh_token}, format='json', ) self.assertEqual(first_response.status_code, 200) session_count_before = Session.objects.count() blocked_session = self._create_session('blocked') with patch('accounts.views.issue_access_token') as mock_issue_token: blocked_response = self.client.post( '/api/auth/refresh', {'refreshToken': blocked_session.refresh_token}, format='json', ) self.assertEqual(blocked_response.status_code, 429) mock_issue_token.assert_not_called() # هیچ Session جدیدی ساخته نشده (فقط همون session ای که قبل از # درخواستِ بلاک‌شده دستی ساختیم) و session ای که با آن تلاش # کردیم هنوز revoked=False و دست‌نخورده است. self.assertEqual(Session.objects.count(), session_count_before + 1) blocked_session.refresh_from_db() self.assertFalse(blocked_session.revoked) 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): # override_settings(REST_FRAMEWORK=...) به‌تنهایی رفتارِ # ScopedRateThrottle رو تغییر نمی‌ده: DRF مقدارِ THROTTLE_RATES رو # یک‌بار موقعِ import، به‌عنوانِ class attribute، از دیکشنریِ اصلیِ # settings.py می‌گیره؛ override_settings دیکشنریِ settings رو عوض # می‌کنه ولی این class attribute هنوز به نسخه‌ی قدیمی اشاره می‌کنه. # برای سنجیدنِ نرخِ واقعی، مستقیماً patch.dict روی همون دیکشنریِ # زنده‌ی ScopedRateThrottle.THROTTLE_RATES لازمه. 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) class UserProfileViewTests(TestCase): """تست‌های GET /api/user/profile — پروفایل کاربر جاری.""" def setUp(self): self.client = APIClient() def _get_profile(self, token=None): kwargs = {} if token: kwargs['HTTP_AUTHORIZATION'] = f'Bearer {token}' return self.client.get('/api/user/profile', **kwargs) def test_requires_authentication(self): response = self._get_profile() self.assertEqual(response.status_code, 401) def test_returns_profile_for_guest_without_existing_profile_row(self): user = User.objects.create(device_id='device-profile-1', is_guest=True, name='') token = issue_access_token(user.id) response = self._get_profile(token) self.assertEqual(response.status_code, 200) data = response.json() self.assertEqual(data['userId'], str(user.id)) self.assertEqual(data['isGuest'], True) self.assertIsNone(data['phone']) self.assertIsNone(data['gradeId']) self.assertIsNone(data['birthYear']) self.assertIsNone(data['birthDate']) self.assertEqual(data['bio'], '') # get_or_create باید یک رکورد Profile واقعی ساخته باشه. self.assertTrue(Profile.objects.filter(user_id=user.id).exists()) def test_returns_existing_profile_fields_for_registered_user(self): user = User.objects.create( phone='09120000001', is_guest=False, name='علی', avatar='avatar.png', grade_id=2, ) Profile.objects.create(user=user, birth_year=1384, bio='سلام') token = issue_access_token(user.id) response = self._get_profile(token) self.assertEqual(response.status_code, 200) data = response.json() self.assertEqual(data['phone'], '09120000001') self.assertEqual(data['name'], 'علی') self.assertEqual(data['avatar'], 'avatar.png') self.assertEqual(data['isGuest'], False) self.assertEqual(data['gradeId'], 2) self.assertEqual(data['birthYear'], 1384) self.assertEqual(data['bio'], 'سلام') def test_does_not_leak_other_users_profile(self): other = User.objects.create(device_id='device-other', is_guest=True, name='') Profile.objects.create(user=other, bio='محرمانه') user = User.objects.create(device_id='device-profile-2', is_guest=True, name='') token = issue_access_token(user.id) response = self._get_profile(token) self.assertEqual(response.status_code, 200) self.assertEqual(response.json()['userId'], str(user.id)) self.assertNotEqual(response.json()['bio'], 'محرمانه') class SendSmsAdminActionTests(TestCase): """ اکشنِ ادمینِ send_sms_to_selected در UserAdmin — قدمِ اولیه‌ی مینیمالِ فیچرِ «ارسالِ پیامک از ادمین»: بدونِ مدلِ کمپین، زمان‌بندی یا صفِ پس‌زمینه؛ فقط ارسالِ بلادرنگِ یک پیام به کاربرانِ انتخاب‌شده در changelist، با استفاده از accounts.services.send_sms موجود. """ def setUp(self): AuthUser = get_user_model() self.staff_user = AuthUser.objects.create_superuser( username='sms-admin', email='sms-admin@example.com', password='pass12345', ) self.client.force_login(self.staff_user) # last_activity_at=now تا این کاربرهای عمومیِ setUp با فیلترِ # get_inactive_users (که last_activity_at is null رو هم inactive # حساب می‌کنه) قاطی نشن؛ تست‌های target='inactive' کاربرِ # inactive/never-active موردنظرِ خودشون رو صریحاً می‌سازن. now = timezone.now() self.user1 = User.objects.create( name='u1', is_guest=False, phone='0910000001', last_activity_at=now, ) self.user2 = User.objects.create( name='u2', is_guest=False, phone='0910000002', last_activity_at=now, ) self.guest_no_phone = User.objects.create( name='guest', is_guest=True, phone=None, last_activity_at=now, ) self.changelist_url = reverse('admin:accounts_user_changelist') def test_action_registered_on_user_admin(self): model_admin = admin.site._registry[User] self.assertIn('send_sms_to_selected', model_admin.actions) @patch('accounts.admin.send_sms') def test_intermediate_page_shown_without_sending(self, mock_send_sms): """ اولین POST (فقط انتخابِ اکشن، بدونِ 'apply') باید صفحه‌ی میانیِ فرم رو نشون بده، نه ارسالِ فوریِ پیامک. """ response = self.client.post(self.changelist_url, { 'action': 'send_sms_to_selected', '_selected_action': [self.user1.pk, self.user2.pk], 'index': 0, }) self.assertEqual(response.status_code, 200) self.assertTemplateUsed(response, 'accounts/send_sms_form.html') self.assertIn('form', response.context) mock_send_sms.assert_not_called() @patch('accounts.admin.send_sms') def test_apply_sends_sms_to_users_with_phone_and_skips_guest(self, mock_send_sms): response = self.client.post(self.changelist_url, { 'action': 'send_sms_to_selected', '_selected_action': [self.user1.pk, self.user2.pk, self.guest_no_phone.pk], 'apply': 'ارسال پیامک', 'message': 'سلام از طرف ادمین', }) self.assertEqual(response.status_code, 200) self.assertEqual(mock_send_sms.call_count, 2) mock_send_sms.assert_any_call(phone='0910000001', message='سلام از طرف ادمین') mock_send_sms.assert_any_call(phone='0910000002', message='سلام از طرف ادمین') @patch('accounts.admin.send_sms') def test_apply_with_empty_message_does_not_send_and_reshows_form(self, mock_send_sms): response = self.client.post(self.changelist_url, { 'action': 'send_sms_to_selected', '_selected_action': [self.user1.pk], 'apply': 'ارسال پیامک', 'message': '', }) self.assertEqual(response.status_code, 200) self.assertTemplateUsed(response, 'accounts/send_sms_form.html') mock_send_sms.assert_not_called() @patch('accounts.admin.send_sms') def test_apply_continues_after_one_failed_delivery(self, mock_send_sms): """ شکستِ ارسال برای یک کاربر (SMSDeliveryError) نباید کلِ اکشن رو متوقف کنه؛ باقیِ کاربرانِ انتخاب‌شده باید همچنان امتحان بشن. """ mock_send_sms.side_effect = [SMSDeliveryError('boom'), None] response = self.client.post(self.changelist_url, { 'action': 'send_sms_to_selected', '_selected_action': [self.user1.pk, self.user2.pk], 'apply': 'ارسال پیامک', 'message': 'متن', }) self.assertEqual(response.status_code, 200) self.assertEqual(mock_send_sms.call_count, 2) @patch('accounts.admin.send_sms') def test_apply_defaults_to_selected_target_when_field_missing(self, mock_send_sms): """ سازگاری با قبل: اگر فرم بدونِ فیلدِ 'target' پست بشه (مثلِ کلاینتِ قدیمی)، initial='selected' باعثِ می‌شه رفتارِ قبلی (ارسال به queryset انتخاب‌شده) حفظ بشه. """ response = self.client.post(self.changelist_url, { 'action': 'send_sms_to_selected', '_selected_action': [self.user1.pk], 'apply': 'ارسال پیامک', 'message': 'متن', 'target': 'selected', }) self.assertEqual(response.status_code, 200) mock_send_sms.assert_called_once_with(phone='0910000001', message='متن') @patch('accounts.admin.send_sms') def test_apply_with_inactive_target_uses_get_inactive_users(self, mock_send_sms): """ وقتی target='inactive' انتخاب شده، گیرنده‌ها باید از accounts.services.get_inactive_users(since) بیان — نه از queryset انتخاب‌شده در changelist — و cutoff باید از تاریخِ واردشده توسطِ ادمین ساخته بشه (بدونِ عددِ روزِ hardcode‌شده). """ now = timezone.now() active_user = User.objects.create( name='active', is_guest=False, phone='0910000009', last_activity_at=now, ) inactive_user = User.objects.create( name='inactive', is_guest=False, phone='0910000010', last_activity_at=now - timezone.timedelta(days=90), ) never_active_user = User.objects.create( name='never-active', is_guest=False, phone='0910000011', last_activity_at=None, ) cutoff_date = (now - timezone.timedelta(days=30)).date() # انتخابِ changelist عمداً کاربرِ فعال رو هم شامل می‌شه تا مطمئن # بشیم با target='inactive' نادیده گرفته می‌شه. response = self.client.post(self.changelist_url, { 'action': 'send_sms_to_selected', '_selected_action': [active_user.pk], 'apply': 'ارسال پیامک', 'message': 'یادآوری غیرفعال‌ها', 'target': 'inactive', 'inactive_since': cutoff_date.isoformat(), }) self.assertEqual(response.status_code, 200) self.assertEqual(mock_send_sms.call_count, 2) mock_send_sms.assert_any_call(phone='0910000010', message='یادآوری غیرفعال‌ها') mock_send_sms.assert_any_call(phone='0910000011', message='یادآوری غیرفعال‌ها') called_phones = {call.kwargs['phone'] for call in mock_send_sms.call_args_list} self.assertNotIn('0910000009', called_phones) @patch('accounts.admin.send_sms') def test_apply_with_inactive_target_requires_date(self, mock_send_sms): """ اگر target='inactive' باشه ولی تاریخ خالی باشه، فرم نباید valid بشه و هیچ پیامکی ارسال نشه (بدونِ fallback به عددِ روزِ پیش‌فرض). """ response = self.client.post(self.changelist_url, { 'action': 'send_sms_to_selected', '_selected_action': [self.user1.pk], 'apply': 'ارسال پیامک', 'message': 'متن', 'target': 'inactive', 'inactive_since': '', }) self.assertEqual(response.status_code, 200) self.assertTemplateUsed(response, 'accounts/send_sms_form.html') self.assertFalse(response.context['form'].is_valid()) mock_send_sms.assert_not_called() @patch('accounts.admin.send_sms') def test_apply_with_inactive_target_still_skips_users_without_phone(self, mock_send_sms): """ رفتارِ skip کاربرانِ بدونِ شماره‌تلفن باید برای هدف‌گیریِ 'inactive' هم مثلِ 'selected' حفظ بشه. """ User.objects.create( name='inactive-no-phone', is_guest=True, phone=None, last_activity_at=timezone.now() - timezone.timedelta(days=90), ) cutoff_date = (timezone.now() - timezone.timedelta(days=30)).date() response = self.client.post(self.changelist_url, { 'action': 'send_sms_to_selected', '_selected_action': [self.user1.pk], 'apply': 'ارسال پیامک', 'message': 'متن', 'target': 'inactive', 'inactive_since': cutoff_date.isoformat(), }) self.assertEqual(response.status_code, 200) mock_send_sms.assert_not_called() from django.contrib.messages import get_messages message_texts = [m.message for m in get_messages(response.wsgi_request)] self.assertTrue(any('بدونِ شماره‌تلفن' in m for m in message_texts)) @override_settings(ADMIN_SMS_MAX_RECIPIENTS=2) @patch('accounts.admin.send_sms') def test_apply_blocks_send_when_selected_recipients_exceed_cap(self, mock_send_sms): """ محافظتِ سقف باید برای target='selected' هم اعمال بشه: اگر تعدادِ کاربرانِ انتخاب‌شده در changelist از ADMIN_SMS_MAX_RECIPIENTS بیشتر باشه، هیچ پیامکی ارسال نشه. """ response = self.client.post(self.changelist_url, { 'action': 'send_sms_to_selected', '_selected_action': [self.user1.pk, self.user2.pk, self.guest_no_phone.pk], 'apply': 'ارسال پیامک', 'message': 'متن', 'target': 'selected', }) self.assertEqual(response.status_code, 200) self.assertTemplateUsed(response, 'accounts/send_sms_form.html') mock_send_sms.assert_not_called() from django.contrib.messages import get_messages message_texts = [m.message for m in get_messages(response.wsgi_request)] self.assertTrue(any('سقفِ مجاز' in m for m in message_texts)) @override_settings(ADMIN_SMS_MAX_RECIPIENTS=1) @patch('accounts.admin.send_sms') def test_apply_blocks_send_when_inactive_recipients_exceed_cap(self, mock_send_sms): """ محافظتِ سقف برای target='inactive' هم اعمال بشه: اگر get_inactive_users(since) بیش از سقف برگردونه، هیچ پیامکی ارسال نشه — بدون تغییر در خودِ منطقِ get_inactive_users. """ now = timezone.now() User.objects.create( name='inactive-1', is_guest=False, phone='0910000021', last_activity_at=now - timezone.timedelta(days=90), ) User.objects.create( name='inactive-2', is_guest=False, phone='0910000022', last_activity_at=now - timezone.timedelta(days=90), ) cutoff_date = (now - timezone.timedelta(days=30)).date() response = self.client.post(self.changelist_url, { 'action': 'send_sms_to_selected', '_selected_action': [self.user1.pk], 'apply': 'ارسال پیامک', 'message': 'متن', 'target': 'inactive', 'inactive_since': cutoff_date.isoformat(), }) self.assertEqual(response.status_code, 200) self.assertTemplateUsed(response, 'accounts/send_sms_form.html') mock_send_sms.assert_not_called() @override_settings(ADMIN_SMS_MAX_RECIPIENTS=2) @patch('accounts.admin.send_sms') def test_apply_sends_when_recipients_are_within_cap(self, mock_send_sms): """ سقف نباید رفتارِ باتچ‌های نرمال (زیرِ سقف) رو تغییر بده — ارسال باید مثلِ قبل انجام بشه. """ response = self.client.post(self.changelist_url, { 'action': 'send_sms_to_selected', '_selected_action': [self.user1.pk, self.user2.pk], 'apply': 'ارسال پیامک', 'message': 'متن', 'target': 'selected', }) self.assertEqual(response.status_code, 200) self.assertEqual(mock_send_sms.call_count, 2) @override_settings(ADMIN_SMS_MAX_RECIPIENTS=1) @patch('accounts.admin.send_sms') def test_apply_cap_check_does_not_count_phoneless_users_differently(self, mock_send_sms): """ سقف بر اساسِ تعدادِ کلِ گیرنده‌ها (قبل از فیلترِ داشتنِ شماره‌تلفن) محاسبه می‌شه؛ این تست فقط تأیید می‌کنه که با سقفِ ۱، یک queryset دو نفره (even شاملِ کاربرِ بدونِ شماره) هنوز block می‌شه. """ response = self.client.post(self.changelist_url, { 'action': 'send_sms_to_selected', '_selected_action': [self.user1.pk, self.guest_no_phone.pk], 'apply': 'ارسال پیامک', 'message': 'متن', 'target': 'selected', }) self.assertEqual(response.status_code, 200) mock_send_sms.assert_not_called() class SupervisorGroupAccessTests(TestCase): """ تست‌های صفحه‌ی «ناظر» (Supervisor) داخلِ همون Adminِ موجود: - دستورِ مدیریتیِ setup_supervisor_group فقط پرمیشنِ view_* رو ست می‌کنه (بدونِ add/change/delete)، روی User/Order/Transaction/ Subscription. - عضوِ گروهِ Supervisor می‌تونه changelist این چهار مدل رو ببینه، ولی نمی‌تونه add/change/delete کنه — نه با کلیک روی دکمه (چون اصلاً نشون داده نمی‌شه)، نه با درخواستِ مستقیمِ URL. - این محدودیت از طریقِ Permission/Group واقعیِ جنگو تأمینه، نه صرفاً پنهون‌کردنِ لینک. """ def setUp(self): from payment.models import Order from subscription.models import Plan, Subscription call_command('setup_supervisor_group') self.supervisor_group = Group.objects.get(name='Supervisor') AuthUser = get_user_model() self.supervisor = AuthUser.objects.create_user( username='supervisor-1', email='supervisor@example.com', password='pass12345', is_staff=True, ) self.supervisor.groups.add(self.supervisor_group) self.plain_staff = AuthUser.objects.create_user( username='plain-staff-1', email='plain-staff@example.com', password='pass12345', is_staff=True, ) self.target_user = User.objects.create(name='target', is_guest=False, phone='0910000099') self.plan = Plan.objects.create(title='یک‌ماهه', price=10000) self.order = Order.objects.create( user_id=self.target_user.id, amount=10000, idempotency_key='sup-order-1', ) self.subscription = Subscription.objects.create( user_id=self.target_user.id, plan=self.plan, status='active', ) def test_setup_command_grants_only_view_permissions(self): codenames = set( self.supervisor_group.permissions.values_list('content_type__app_label', 'codename') ) self.assertEqual( codenames, { ('accounts', 'view_user'), ('payment', 'view_order'), ('payment', 'view_transaction'), ('subscription', 'view_subscription'), }, ) for _, codename in codenames: self.assertTrue(codename.startswith('view_')) def test_setup_command_is_idempotent_and_strips_extra_permissions(self): """ اگه یک پرمیشنِ change/delete دستی به گروه اضافه شده باشه، اجرای دوباره‌ی دستور باید اون رو حذف کنه (چون گروه باید فقط view_* داشته باشه). """ change_user_perm = Permission.objects.get( content_type__app_label='accounts', codename='change_user', ) self.supervisor_group.permissions.add(change_user_perm) call_command('setup_supervisor_group') self.supervisor_group.refresh_from_db() codenames = set(self.supervisor_group.permissions.values_list('codename', flat=True)) self.assertNotIn('change_user', codenames) self.assertIn('view_user', codenames) def test_supervisor_can_view_user_changelist(self): self.client.force_login(self.supervisor) response = self.client.get(reverse('admin:accounts_user_changelist')) self.assertEqual(response.status_code, 200) def test_supervisor_can_view_order_and_transaction_changelist(self): self.client.force_login(self.supervisor) response = self.client.get(reverse('admin:payment_order_changelist')) self.assertEqual(response.status_code, 200) response = self.client.get(reverse('admin:payment_transaction_changelist')) self.assertEqual(response.status_code, 200) def test_supervisor_can_view_subscription_changelist(self): self.client.force_login(self.supervisor) response = self.client.get(reverse('admin:subscription_subscription_changelist')) self.assertEqual(response.status_code, 200) def test_supervisor_cannot_access_user_add_page(self): self.client.force_login(self.supervisor) response = self.client.get(reverse('admin:accounts_user_add')) self.assertEqual(response.status_code, 403) @patch('accounts.admin.send_sms') def test_supervisor_cannot_execute_send_sms_action_via_direct_post(self, mock_send_sms): """ عضوِ Supervisor فقط view_user داره، نه change_user. اکشنِ send_sms_to_selected باید permissions=['change'] داشته باشه تا جنگو خودش (get_actions/response_action) این اکشن رو برای Supervisor فیلتر کنه — یعنی حتی با POSTِ مستقیم و دستیِ 'apply' (بدونِ دیدنِ دکمه‌ی اکشن در UI)، هیچ پیامکی ارسال نشه. """ self.client.force_login(self.supervisor) response = self.client.post(reverse('admin:accounts_user_changelist'), { 'action': 'send_sms_to_selected', '_selected_action': [self.target_user.pk], 'apply': 'ارسال پیامک', 'message': 'تلاشِ غیرمجاز برای ارسال', }) self.assertEqual(response.status_code, 200) mock_send_sms.assert_not_called() def test_supervisor_can_view_user_change_page_readonly(self): """ طبقِ رفتارِ پیش‌فرضِ ModelAdmin جنگو (از نسخه‌ی ۲٫۱ به بعد)، وقتی کاربر فقط view permission داره (نه change)، GET به change_view باید 200 برگردونه ولی همه‌ی فیلدها read-only باشن — این دقیقاً همون رفتارِ استانداردی‌ست که ما ازش استفاده می‌کنیم و هیچ override سفارشی‌ای اضافه نکردیم. """ self.client.force_login(self.supervisor) response = self.client.get(reverse('admin:accounts_user_change', args=[self.target_user.pk])) self.assertEqual(response.status_code, 200) self.assertNotContains(response, 'name="_save"') self.assertNotContains(response, 'name="_continue"') def test_supervisor_post_to_user_change_does_not_modify_record(self): """ حتی اگه سوپروایزر مستقیماً به URL ویرایش POST بزنه (بدونِ دیدنِ دکمه‌ی ذخیره)، چون change permission نداره باید 403 بگیره و رکورد دست‌نخورده بمونه. """ self.client.force_login(self.supervisor) original_name = self.target_user.name response = self.client.post( reverse('admin:accounts_user_change', args=[self.target_user.pk]), {'name': 'دستکاری‌شده', 'is_guest': 'on'}, ) self.assertEqual(response.status_code, 403) self.target_user.refresh_from_db() self.assertEqual(self.target_user.name, original_name) def test_supervisor_cannot_access_order_delete_page(self): from payment.models import Order self.client.force_login(self.supervisor) response = self.client.get(reverse('admin:payment_order_delete', args=[self.order.pk])) self.assertEqual(response.status_code, 403) self.assertTrue(Order.objects.filter(pk=self.order.pk).exists()) def test_supervisor_cannot_access_subscription_add_page(self): self.client.force_login(self.supervisor) response = self.client.get(reverse('admin:subscription_subscription_add')) self.assertEqual(response.status_code, 403) def test_supervisor_dashboard_shows_only_permitted_links(self): self.client.force_login(self.supervisor) response = self.client.get(reverse('admin:supervisor-dashboard')) self.assertEqual(response.status_code, 200) self.assertContains(response, reverse('admin:accounts_user_changelist')) self.assertContains(response, reverse('admin:payment_order_changelist')) self.assertContains(response, reverse('admin:payment_transaction_changelist')) self.assertContains(response, reverse('admin:subscription_subscription_changelist')) def test_plain_staff_without_group_gets_403_on_dashboard(self): """ یک کاربرِ staff که عضوِ گروهِ Supervisor نیست باید هنگامِ باز کردنِ /admin/supervisor/ با ۴۰۳ (PermissionDenied) روبه‌رو بشه — صرفِ is_staff بودن دیگه برای ورود به این صفحه کافی نیست. """ self.client.force_login(self.plain_staff) response = self.client.get(reverse('admin:supervisor-dashboard')) self.assertEqual(response.status_code, 403) def test_superuser_without_group_gets_403_on_dashboard(self): """ حتی یک superuser هم اگه عضوِ گروهِ Supervisor نباشه، باید ۴۰۳ بگیره — قانونِ دسترسی صرفاً بر اساسِ عضویتِ گروهه، نه سطحِ دسترسیِ کلیِ کاربر. """ AuthUser = get_user_model() superuser = AuthUser.objects.create_superuser( username='super-1', email='super@example.com', password='pass12345', ) self.client.force_login(superuser) response = self.client.get(reverse('admin:supervisor-dashboard')) self.assertEqual(response.status_code, 403) def test_supervisor_can_still_access_dashboard_after_fix(self): """ سنجه‌ی کنترل: کاربرِ staff که واقعاً عضوِ گروهِ Supervisor هست، باید همچنان ۲۰۰ بگیره و لینک‌های مجاز رو ببینه. """ self.client.force_login(self.supervisor) response = self.client.get(reverse('admin:supervisor-dashboard')) self.assertEqual(response.status_code, 200) self.assertContains(response, reverse('admin:accounts_user_changelist')) def test_anonymous_user_redirected_from_supervisor_dashboard(self): response = self.client.get(reverse('admin:supervisor-dashboard')) self.assertEqual(response.status_code, 302) def test_supervisor_group_membership_alone_is_not_enough_without_is_staff(self): """ عضویتِ گروه به‌تنهایی کافی نیست؛ سیستمِ Admin هنوز is_staff رو هم لازم داره — این تضمین می‌کنه محدودیت واقعاً از مسیرِ استانداردِ Admin/Permission جنگو رد می‌شه، نه یک چک سفارشیِ جدا. """ AuthUser = get_user_model() non_staff_supervisor = AuthUser.objects.create_user( username='non-staff-supervisor', email='ns@example.com', password='pass12345', is_staff=False, ) non_staff_supervisor.groups.add(self.supervisor_group) self.client.force_login(non_staff_supervisor) response = self.client.get(reverse('admin:accounts_user_changelist')) self.assertEqual(response.status_code, 302) class AuthUserRowCacheTests(TestCase): """کشِ کوتاهِ ردیفِ کاربر در TadrisinoTokenAuthentication.""" def setUp(self): cache.clear() self.client = APIClient() self.user = User.objects.create(device_id='dev-authcache', is_guest=True, name='n') self.token = issue_access_token(self.user.id) # هر endpointِ authenticated؛ /api/user/hearts سبک است. self.url = '/api/user/hearts' def _hit(self): return self.client.get(self.url, HTTP_AUTHORIZATION=f'Bearer {self.token}') @override_settings(AUTH_USER_CACHE_TTL=0) def test_disabled_by_default_loads_every_request(self): with patch('accounts.authentication._load_user_or_fail', wraps=_orig_loader()) as m: self._hit(); self._hit() self.assertEqual(m.call_count, 2) @override_settings(AUTH_USER_CACHE_TTL=30) def test_enabled_caches_after_first_load(self): with patch('accounts.authentication._load_user_or_fail', wraps=_orig_loader()) as m: self.assertEqual(self._hit().status_code, 200) self.assertEqual(self._hit().status_code, 200) self.assertEqual(m.call_count, 1) @override_settings(AUTH_USER_CACHE_TTL=30) def test_user_save_invalidates_cache(self): self._hit() # cache populated with patch('accounts.authentication._load_user_or_fail', wraps=_orig_loader()) as m: self.user.name = 'changed' self.user.save(update_fields=['name']) # post_save → invalidate self._hit() self.assertEqual(m.call_count, 1) @override_settings(AUTH_USER_CACHE_TTL=30) def test_set_grade_still_works_with_cache_on(self): r = self.client.post( '/api/user/grade', {'gradeId': 3}, format='json', HTTP_AUTHORIZATION=f'Bearer {self.token}') self.assertEqual(r.status_code, 200) self.user.refresh_from_db() self.assertEqual(self.user.grade_id, 3) # درخواستِ بعدی باید گریدِ تازه را ببیند (کش پس از save باطل شده). prof = self.client.get( '/api/user/profile', HTTP_AUTHORIZATION=f'Bearer {self.token}') self.assertEqual(prof.data['gradeId'], 3) def _orig_loader(): """نسخه‌ی اصلیِ لودر برای wraps کردنِ Mock (بدونِ recursion).""" from accounts.models import User as _U from rest_framework.exceptions import AuthenticationFailed def _load(user_id): try: return _U.objects.get(pk=user_id) except _U.DoesNotExist: raise AuthenticationFailed('کاربر پیدا نشد.') return _load class UpdateUserProfileTests(TestCase): """ تست‌های PATCH /api/user/profile. صفحه‌ی «بیا باهم آشنا بشیم» در اپ اسم می‌گیرد و تا پیش از این هیچ مسیری برای نوشتنش نبود — پروفایل فقط خواندنی بود. """ def setUp(self): self.client = APIClient() cache.clear() self.user = User.objects.create(device_id='device-patch-1', is_guest=True, name='') self.token = issue_access_token(self.user.id) def tearDown(self): cache.clear() def _patch(self, body, token=None): kwargs = {'format': 'json'} token = self.token if token is None else token if token: kwargs['HTTP_AUTHORIZATION'] = f'Bearer {token}' return self.client.patch('/api/user/profile', body, **kwargs) def test_requires_authentication(self): self.assertEqual(self._patch({'name': 'سارا'}, token='').status_code, 401) def test_sets_name_and_returns_the_updated_profile(self): response = self._patch({'name': 'سارا'}) self.assertEqual(response.status_code, 200) self.assertEqual(response.data['name'], 'سارا') self.user.refresh_from_db() self.assertEqual(self.user.name, 'سارا') def test_partial_update_leaves_other_fields_alone(self): self._patch({'name': 'علی', 'bio': 'عاشقِ ریاضی'}) response = self._patch({'bio': 'کتاب‌خوان'}) self.assertEqual(response.status_code, 200) self.assertEqual(response.data['name'], 'علی') self.assertEqual(response.data['bio'], 'کتاب‌خوان') def test_creates_the_profile_row_lazily(self): self.assertFalse(Profile.objects.filter(user_id=self.user.id).exists()) response = self._patch({'birthYear': 1391}) self.assertEqual(response.status_code, 200) self.assertEqual(response.data['birthYear'], 1391) self.assertTrue(Profile.objects.filter(user_id=self.user.id).exists()) def test_empty_body_is_rejected(self): self.assertEqual(self._patch({}).status_code, 400) def test_phone_and_guest_flag_cannot_be_changed(self): response = self._patch({'name': 'رضا', 'phone': '09120000000', 'isGuest': False}) self.assertEqual(response.status_code, 200) self.user.refresh_from_db() self.assertIsNone(self.user.phone) self.assertTrue(self.user.is_guest) def test_out_of_range_birth_year_is_rejected(self): self.assertEqual(self._patch({'birthYear': 2026}).status_code, 400) class TestLoginPhoneTests(TestCase): """ 🔒 شماره‌های تستِ تیم (TEST_LOGIN_PHONES): باید ورودِ بدونِ پیامک را برای همان چند شماره ممکن کنند و برای هیچ شماره‌ی دیگری نه. این تست‌ها با DEBUG=False اجرا می‌شوند چون دقیقاً همان حالتی است که روی سرور هست و DEV_MASTER_OTP در آن خاموش است. """ PHONE = '09120000001' CODE = '834512' def setUp(self): self.client = APIClient() cache.clear() @override_settings(DEBUG=False, DEV_MASTER_OTP=None, TEST_LOGIN_PHONES={PHONE: CODE}) def test_configured_phone_logs_in_without_sms(self): with patch('accounts.views.send_otp_sms') as mock_send: request = self.client.post( '/api/auth/otp/request', {'phone': self.PHONE}, format='json', ) self.assertEqual(request.status_code, 200) # مهم‌ترین نکته‌ی این مسیر: هیچ پیامکی نباید فرستاده شود، وگرنه بدونِ # کاوه‌نگار همین‌جا ۵۰۳ می‌گرفتیم و کل هدف از بین می‌رفت. mock_send.assert_not_called() self.assertEqual(request.data['devOtp'], '') login = self.client.post( '/api/auth/login', {'phone': self.PHONE, 'otp': self.CODE}, format='json', ) self.assertEqual(login.status_code, 200) self.assertTrue(login.data['token']) @override_settings(DEBUG=False, DEV_MASTER_OTP=None, TEST_LOGIN_PHONES={PHONE: CODE}) def test_code_is_never_returned_in_the_response(self): response = self.client.post( '/api/auth/otp/request', {'phone': self.PHONE}, format='json', ) self.assertNotIn(self.CODE, str(response.data)) @override_settings(DEBUG=False, DEV_MASTER_OTP=None, TEST_LOGIN_PHONES={PHONE: CODE}) def test_wrong_code_for_a_configured_phone_is_rejected(self): response = self.client.post( '/api/auth/login', {'phone': self.PHONE, 'otp': '000000'}, format='json', ) self.assertEqual(response.status_code, 400) @override_settings(DEBUG=False, DEV_MASTER_OTP=None, TEST_LOGIN_PHONES={PHONE: CODE}) def test_the_code_does_not_work_for_any_other_phone(self): """تفاوتِ اصلی با DEV_MASTER_OTP: کد فقط مالِ همان یک شماره است.""" response = self.client.post( '/api/auth/login', {'phone': '09129999999', 'otp': self.CODE}, format='json', ) self.assertEqual(response.status_code, 400) @override_settings(DEBUG=False, DEV_MASTER_OTP=None, TEST_LOGIN_PHONES={}) def test_empty_configuration_changes_nothing(self): """بدونِ تنظیم، مسیرِ OTP دقیقاً همان رفتارِ قبلی را دارد.""" with patch('accounts.views.send_otp_sms') as mock_send: response = self.client.post( '/api/auth/otp/request', {'phone': '09121112233'}, format='json', ) self.assertEqual(response.status_code, 200) mock_send.assert_called_once() @override_settings(DEBUG=False, DEV_MASTER_OTP=None, TEST_LOGIN_PHONES={PHONE: CODE}, OTP_LOGIN_PHONE_MAX_ATTEMPTS=3) def test_brute_force_on_a_configured_phone_is_still_rate_limited(self): for _ in range(3): self.client.post( '/api/auth/login', {'phone': self.PHONE, 'otp': '111111'}, format='json', ) blocked = self.client.post( '/api/auth/login', {'phone': self.PHONE, 'otp': '111111'}, format='json', ) self.assertEqual(blocked.status_code, 429) class TestLoginPhoneParsingTests(TestCase): """ورودیِ خرابِ TEST_LOGIN_PHONES باید سرویس را بالا نیاورد، نه اینکه بی‌صدا رد شود.""" def _parse(self, raw): from config.settings import _parse_test_login_phones return _parse_test_login_phones(raw) def test_valid_pairs_are_parsed(self): self.assertEqual( self._parse(' 09120000001:834512 , +989120000002:110947 '), {'09120000001': '834512', '09120000002': '110947'}, ) def test_empty_value_yields_no_test_phones(self): self.assertEqual(self._parse(''), {}) def test_malformed_entry_raises(self): from django.core.exceptions import ImproperlyConfigured for raw in ('09120000001', '09120000001:12', 'abc:123456', '09120000001:abcdef', '09120000001:1234567'): with self.subTest(raw=raw), self.assertRaises(ImproperlyConfigured): self._parse(raw)