""" نرمال‌سازیِ شماره و سقف‌های دیتابیسیِ پیامکِ OTP (۲۰۲۶-۰۹-۲۵، همان روزی که کلیدِ کاوه‌نگار وصل شد). security-before-production.md بندِ ۱۷. کش در هر تست پاک می‌شود تا رفتارِ «workerِ دیگر» (کشِ خالی، دیتابیسِ مشترک) شبیه‌سازی شود. هم‌زمانیِ واقعیِ دو پروسه‌ی gunicorn تست نشده است. """ from unittest.mock import patch import requests 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 accounts.models import OTPCode, User from accounts.services import send_sms def _request(phone): return APIClient().post('/api/auth/otp/request', {'phone': phone}, format='json') @override_settings(DEBUG=True, TEST_LOGIN_PHONES={}) class PhoneNormalizationTests(TestCase): def setUp(self): cache.clear() def test_other_spellings_of_the_same_number_share_one_cooldown(self): # همان سنجه‌ای که security-before-production.md برای بستنِ بندِ ۱۷ نوشته. with patch('accounts.views.send_otp_sms') as send: self.assertEqual(_request('09121112233').status_code, 200) for spelling in ('+989121112233', '989121112233', '۰۹۱۲۱۱۱۲۲۳۳', '0912 111 2233'): self.assertEqual(_request(spelling).status_code, 429, spelling) send.assert_called_once() self.assertEqual(set(OTPCode.objects.values_list('phone', flat=True)), {'09121112233'}) def test_login_with_another_spelling_reaches_the_same_account(self): User.objects.create(phone='09121112244', is_guest=False) with patch('accounts.views.secrets.randbelow', return_value=246810), \ patch('accounts.views.send_otp_sms'): _request('+989121112244') body = APIClient().post('/api/auth/login', {'phone': '۰۹۱۲۱۱۱۲۲۴۴', 'otp': '246810'}, format='json').json() self.assertEqual(User.objects.filter(phone__endswith='9121112244').count(), 1) self.assertEqual(body['userId'], str(User.objects.get(phone='09121112244').id)) def test_non_mobile_numbers_are_rejected_before_any_sms(self): with patch('accounts.views.send_otp_sms') as send: for bad in ('+14155552671', '02188888888', '0912123', 'abc'): self.assertEqual(_request(bad).status_code, 400, bad) send.assert_not_called() self.assertFalse(OTPCode.objects.exists()) @override_settings(TEST_LOGIN_PHONES={'09120000101': '4321'}) def test_demo_phone_matches_in_any_spelling(self): response = APIClient().post('/api/auth/login', {'phone': '+989120000101', 'otp': '4321'}, format='json') self.assertEqual(response.status_code, 200) @override_settings(DEBUG=True, TEST_LOGIN_PHONES={}, OTP_REQUEST_PHONE_COOLDOWN_SECONDS=120) class DatabaseBackedLimitTests(TestCase): """کشِ LocMem مالِ هر worker است؛ این سقف‌ها باید با کشِ خالی هم بمانند.""" def setUp(self): cache.clear() def test_cooldown_holds_on_a_worker_with_an_empty_cache(self): with patch('accounts.views.send_otp_sms') as send: self.assertEqual(_request('09121112255').status_code, 200) cache.clear() # درخواستِ بعدی به workerِ دیگری می‌رسد response = _request('09121112255') self.assertEqual(response.status_code, 429) self.assertTrue(1 <= response.json()['retryAfter'] <= 120) send.assert_called_once() @override_settings(OTP_REQUEST_PHONE_COOLDOWN_SECONDS=0, OTP_REQUEST_PHONE_DAILY_MAX=3) def test_daily_cap_per_phone(self): with patch('accounts.views.send_otp_sms') as send: statuses = [] for _ in range(5): cache.clear() # سقفِ ساعتیِ کش را کنار می‌گذارد؛ فقط سقفِ روزانه بماند statuses.append(_request('09121112266').status_code) self.assertEqual(statuses, [200, 200, 200, 429, 429]) self.assertEqual(send.call_count, 3) @override_settings(OTP_REQUEST_PHONE_COOLDOWN_SECONDS=0) def test_codes_older_than_a_day_do_not_count(self): OTPCode.objects.bulk_create([ OTPCode(phone='09121112277', code='000000', expires_at=timezone.now() - timezone.timedelta(days=2)) for _ in range(20) ]) with patch('accounts.views.send_otp_sms'): self.assertEqual(_request('09121112277').status_code, 200) @override_settings(OTP_REQUEST_PHONE_COOLDOWN_SECONDS=0, OTP_SMS_GLOBAL_DAILY_MAX=2) def test_global_daily_cap_stops_sms_pumping(self): with patch('accounts.views.send_otp_sms') as send: self.assertEqual(_request('09121110001').status_code, 200) self.assertEqual(_request('09121110002').status_code, 200) with self.assertLogs('accounts.views', level='ERROR'): response = _request('09121110003') self.assertEqual(response.status_code, 503) self.assertEqual(send.call_count, 2) self.assertFalse(OTPCode.objects.filter(phone='09121110003').exists()) @override_settings(DEBUG=False, KAVENEGAR_API_KEY='k') def test_failed_delivery_frees_the_cooldown_and_quota(self): with patch('accounts.services.requests.post', side_effect=requests.ConnectionError('down')): self.assertEqual(_request('09121112288').status_code, 503) cache.clear() ok = type('R', (), {'status_code': 200, 'json': lambda self: { 'return': {'status': 200}, 'entries': [{'messageid': 1, 'status': 1}]}})() with patch('accounts.services.requests.post', return_value=ok): self.assertEqual(_request('09121112288').status_code, 200) class ProviderLoggingTests(TestCase): @override_settings(KAVENEGAR_API_KEY='k') def test_accepted_message_logs_its_id_not_its_text(self): ok = type('R', (), {'status_code': 200, 'json': lambda self: { 'return': {'status': 200}, 'entries': [{'messageid': 8792343, 'status': 1}]}})() with patch('accounts.services.requests.post', return_value=ok), \ self.assertLogs('accounts.services', level='INFO') as logs: send_sms(phone='09121112299', message='کد ورود شما: 123456') joined = '\n'.join(logs.output) self.assertIn('messageid=8792343', joined) self.assertNotIn('123456', joined) def _kavenegar_reply(http=200, status=200): return type('R', (), {'status_code': http, 'json': lambda self: { 'return': {'status': status}, 'entries': [{'messageid': 1, 'status': 1}] if status == 200 else None}})() @override_settings(DEBUG=False, KAVENEGAR_API_KEY='k', TEST_LOGIN_PHONES={}) class VerifyLookupTests(TestCase): """خطِ حسابِ فعلی متنِ آزاد را لغو می‌کند (وضعیتِ ۱۳)؛ کدِ ورود باید با الگو برود.""" def setUp(self): cache.clear() @override_settings(KAVENEGAR_OTP_TEMPLATE='tadrisino-otp') def test_otp_goes_through_the_template_with_only_the_code(self): with patch('accounts.views.secrets.randbelow', return_value=4321), \ patch('accounts.services.requests.post', return_value=_kavenegar_reply()) as post: self.assertEqual(_request('09121113301').status_code, 200) url, data = post.call_args.args[0], post.call_args.kwargs['data'] self.assertTrue(url.endswith('/verify/lookup.json')) self.assertEqual(data, {'receptor': '09121113301', 'template': 'tadrisino-otp', 'token': '004321'}) @override_settings(KAVENEGAR_OTP_TEMPLATE='') def test_without_a_template_the_plain_send_is_kept(self): with patch('accounts.services.requests.post', return_value=_kavenegar_reply()) as post: self.assertEqual(_request('09121113302').status_code, 200) self.assertTrue(post.call_args.args[0].endswith('/sms/send.json')) self.assertIn('message', post.call_args.kwargs['data']) @override_settings(KAVENEGAR_OTP_TEMPLATE='misspelled') def test_unknown_template_is_an_explicit_503_and_frees_the_cooldown(self): # کاوه‌نگار برای الگوی ناموجود/تأییدنشده HTTP ۴۲۴ می‌دهد (۲۰۲۶-۰۹-۲۵ زنده دیده شد). with patch('accounts.services.requests.post', return_value=_kavenegar_reply(424, 424)): self.assertEqual(_request('09121113303').status_code, 503) self.assertFalse(OTPCode.objects.filter(phone='09121113303').exists()) @override_settings(DEBUG=False, KAVENEGAR_API_KEY='k', KAVENEGAR_OTP_TEMPLATE='SMSVerify', TEST_LOGIN_PHONES={}) class SlowProviderTests(TestCase): """۲۰۲۶-۰۹-۲۷: کاوه‌نگار ۵٫۴ ثانیه طول کشید، پیامک رسید، ولی ما ۵۰۳ دادیم و کد را پاک کردیم.""" def setUp(self): cache.clear() def test_late_reply_keeps_the_code_and_moves_the_user_on(self): with patch('accounts.views.secrets.randbelow', return_value=135790), \ patch('accounts.services.requests.post', side_effect=requests.ReadTimeout('slow')): self.assertEqual(_request('09121114401').status_code, 200) self.assertTrue(OTPCode.objects.filter(phone='09121114401', code='135790', is_used=False).exists()) login = APIClient().post('/api/auth/login', {'phone': '09121114401', 'otp': '135790'}, format='json') self.assertEqual(login.status_code, 200) # cooldown سرِ جایش است، پس دکمه‌ی دوباره پیامکِ دوم نمی‌فرستد with patch('accounts.services.requests.post') as post: self.assertEqual(_request('09121114401').status_code, 429) post.assert_not_called() def test_no_connection_is_still_an_explicit_503(self): with patch('accounts.services.requests.post', side_effect=requests.ConnectTimeout('down')): self.assertEqual(_request('09121114402').status_code, 503) self.assertFalse(OTPCode.objects.filter(phone='09121114402').exists()) def test_timeouts_leave_room_for_a_slow_iranian_network(self): with patch('accounts.services.requests.post', return_value=_kavenegar_reply()) as post: _request('09121114403') self.assertEqual(post.call_args.kwargs['timeout'], (10, 30))