Zareian
0580aec7b0
Kavenegar is being switched on, so every OTP request now costs credit: - phone numbers are normalised to 09XXXXXXXXX everywhere (login, OTP, family invite, TEST_LOGIN_PHONES keys) and existing rows are migrated; non-Iranian and landline numbers are rejected before any SMS - cooldown and a daily per-phone cap are counted in the database, so the three gunicorn workers no longer triple them - a global daily OTP SMS cap (default 300) stops SMS pumping - per-IP rates widened for CGNAT / school Wi-Fi (env-overridable) - provider message id is logged; SMS text warns not to share the code Second audit round: an offer no longer shortens a longer paid plan, Zarinpal error bodies no longer 500, island XP reaches the parent report, a non-numeric questionId no longer 500s after the answer is generated. 596 tests green. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
171 lines
7.9 KiB
Python
171 lines
7.9 KiB
Python
"""
|
||
حساب، ورود و چند موردِ دیگر از بازبینیِ ۲۰۲۶-۰۹-۲۵.
|
||
"""
|
||
import time
|
||
from unittest.mock import patch
|
||
|
||
from django.core.cache import cache
|
||
from django.test import TestCase, override_settings
|
||
from rest_framework.test import APIClient
|
||
|
||
from accounts.authentication import issue_access_token
|
||
from accounts.models import OTPCode, User
|
||
from content.models import Choice, Grade, Lesson, Node, Question, Unit
|
||
from progress.models import UserHearts, UserStats
|
||
|
||
|
||
def _client(user):
|
||
client = APIClient()
|
||
client.credentials(HTTP_AUTHORIZATION=f'Bearer {issue_access_token(user.id)}')
|
||
return client
|
||
|
||
|
||
class LogoutTests(TestCase):
|
||
def setUp(self):
|
||
cache.clear()
|
||
|
||
def test_logout_kills_the_current_token_and_its_refresh(self):
|
||
guest = APIClient().post('/api/auth/session', {'type': 'guest', 'deviceId': 'dev-out'},
|
||
format='json').json()
|
||
client = APIClient()
|
||
client.credentials(HTTP_AUTHORIZATION=f'Bearer {guest["token"]}')
|
||
self.assertEqual(client.get('/api/user/profile').status_code, 200)
|
||
|
||
time.sleep(1.1) # زمانِ توکن ثانیهای است؛ توکن باید پیش از خروج صادر شده باشد
|
||
response = client.post('/api/auth/logout', {'refreshToken': guest['refreshToken']}, format='json')
|
||
self.assertEqual(response.status_code, 200)
|
||
|
||
self.assertEqual(client.get('/api/user/profile').status_code, 401)
|
||
refreshed = APIClient().post('/api/auth/refresh', {'refreshToken': guest['refreshToken']}, format='json')
|
||
self.assertEqual(refreshed.status_code, 401)
|
||
|
||
# ورودِ دوباره کار میکند
|
||
again = APIClient().post('/api/auth/session', {'type': 'guest', 'deviceId': 'dev-out'},
|
||
format='json').json()
|
||
fresh = APIClient()
|
||
fresh.credentials(HTTP_AUTHORIZATION=f'Bearer {again["token"]}')
|
||
self.assertEqual(fresh.get('/api/user/profile').status_code, 200)
|
||
|
||
|
||
@override_settings(DEBUG=True, OTP_REQUEST_PHONE_COOLDOWN_SECONDS=0)
|
||
class OtpTests(TestCase):
|
||
def setUp(self):
|
||
cache.clear()
|
||
|
||
def _request(self, phone, code='123456'):
|
||
with patch('accounts.views.secrets.randbelow', return_value=int(code)), \
|
||
patch('accounts.views.send_otp_sms'):
|
||
APIClient().post('/api/auth/otp/request', {'phone': phone}, format='json')
|
||
|
||
def test_a_code_dies_after_five_wrong_guesses(self):
|
||
phone = '09121234001'
|
||
self._request(phone)
|
||
for guess in ('000001', '000002', '000003', '000004', '000005'):
|
||
APIClient().post('/api/auth/login', {'phone': phone, 'otp': guess}, format='json')
|
||
self.assertTrue(OTPCode.objects.get(phone=phone).is_used)
|
||
right = APIClient().post('/api/auth/login', {'phone': phone, 'otp': '123456'}, format='json')
|
||
self.assertEqual(right.status_code, 400)
|
||
|
||
|
||
@override_settings(DEBUG=True, OTP_REQUEST_PHONE_COOLDOWN_SECONDS=0,
|
||
TEST_LOGIN_PHONES={'09120000101': '4321'})
|
||
class GuestSignupKeepsProgressTests(TestCase):
|
||
def setUp(self):
|
||
cache.clear()
|
||
guest = APIClient().post('/api/auth/session', {'type': 'guest', 'deviceId': 'dev-up'},
|
||
format='json').json()
|
||
self.guest_id = int(guest['userId'])
|
||
UserStats.objects.create(user_id=self.guest_id, xp=120)
|
||
self.client = APIClient()
|
||
self.client.credentials(HTTP_AUTHORIZATION=f'Bearer {guest["token"]}')
|
||
|
||
def _login(self, phone, otp):
|
||
return self.client.post('/api/auth/login', {'phone': phone, 'otp': otp}, format='json')
|
||
|
||
def test_the_guest_becomes_the_registered_user(self):
|
||
with patch('accounts.views.secrets.randbelow', return_value=654321), \
|
||
patch('accounts.views.send_otp_sms'):
|
||
self.client.post('/api/auth/otp/request', {'phone': '09121234002'}, format='json')
|
||
body = self._login('09121234002', '654321').json()
|
||
|
||
self.assertEqual(int(body['userId']), self.guest_id)
|
||
user = User.objects.get(pk=self.guest_id)
|
||
self.assertEqual((user.phone, user.is_guest), ('09121234002', False))
|
||
self.assertEqual(UserStats.objects.get(user_id=self.guest_id).xp, 120)
|
||
|
||
def test_an_existing_account_is_not_merged(self):
|
||
existing = User.objects.create(phone='09121234003', is_guest=False)
|
||
with patch('accounts.views.secrets.randbelow', return_value=111222), \
|
||
patch('accounts.views.send_otp_sms'):
|
||
self.client.post('/api/auth/otp/request', {'phone': '09121234003'}, format='json')
|
||
body = self._login('09121234003', '111222').json()
|
||
self.assertEqual(int(body['userId']), existing.id)
|
||
self.assertTrue(User.objects.get(pk=self.guest_id).is_guest)
|
||
|
||
def test_demo_test_phones_never_absorb_a_guest(self):
|
||
body = self._login('09120000101', '4321').json()
|
||
self.assertNotEqual(int(body['userId']), self.guest_id)
|
||
self.assertTrue(User.objects.get(pk=self.guest_id).is_guest)
|
||
|
||
|
||
class DisplayNameTests(TestCase):
|
||
def test_contact_details_are_stripped_from_the_name(self):
|
||
user = User.objects.create(phone='09121234004', is_guest=False)
|
||
body = _client(user).patch('/api/user/profile', {'name': 'سارا 0912 345 6789 @sara'},
|
||
format='json').json()
|
||
self.assertEqual(body['name'], 'سارا')
|
||
|
||
|
||
class HeartsTests(TestCase):
|
||
def setUp(self):
|
||
cache.clear()
|
||
grade = Grade.objects.create(title='پایه نهم', icon='', color='')
|
||
node = Node.objects.create(unit=Unit.objects.create(grade=grade, title='ف', order=1), title='ن', order=1)
|
||
self.question = Question.objects.create(lesson=Lesson.objects.create(node=node, title='د'),
|
||
type='multiple_choice', prompt='؟')
|
||
self.right = Choice.objects.create(question=self.question, value='۱', is_correct=True)
|
||
self.user = User.objects.create(phone='09121234005', is_guest=False)
|
||
UserHearts.objects.create(user_id=self.user.id, current_hearts=0)
|
||
|
||
def _answer(self):
|
||
return _client(self.user).post(f'/api/questions/{self.question.id}/answer',
|
||
{'selectedOptionId': self.right.id}, format='json')
|
||
|
||
def test_off_by_default_so_the_current_app_keeps_working(self):
|
||
self.assertEqual(self._answer().status_code, 200)
|
||
|
||
@override_settings(HEARTS_ENFORCED=True)
|
||
def test_when_enforced_no_hearts_means_no_answer(self):
|
||
response = self._answer()
|
||
self.assertEqual(response.status_code, 403)
|
||
self.assertEqual(response.json()['code'], 'no_hearts')
|
||
|
||
|
||
class FamilyTwoChildrenTests(TestCase):
|
||
def test_a_parent_of_two_can_see_each_child(self):
|
||
parent = User.objects.create(phone='09121234010', is_guest=False)
|
||
kids = []
|
||
for phone in ('09121234011', '09121234012'):
|
||
kid = User.objects.create(phone=phone, is_guest=False)
|
||
code = _client(parent).post('/api/family/link/invite', {'childPhone': phone},
|
||
format='json').json()['inviteCode']
|
||
_client(kid).post('/api/family/link/accept', {'inviteCode': code}, format='json')
|
||
kids.append(kid)
|
||
client = _client(parent)
|
||
self.assertEqual(client.get('/api/family/settings').status_code, 200)
|
||
self.assertEqual(client.get('/api/family/weekly-report').status_code, 200)
|
||
self.assertEqual(client.get(f'/api/family/settings?childUserId={kids[0].id}').status_code, 200)
|
||
self.assertEqual(client.get('/api/family/settings?childUserId=999999').status_code, 404)
|
||
|
||
|
||
class AdminLoginThrottleTests(TestCase):
|
||
def setUp(self):
|
||
cache.clear()
|
||
|
||
@override_settings(ADMIN_LOGIN_MAX_ATTEMPTS=3)
|
||
def test_password_guessing_is_capped(self):
|
||
codes = [self.client.post('/admin/login/', {'username': 'a', 'password': 'b'}).status_code
|
||
for _ in range(4)]
|
||
self.assertEqual(codes[-1], 429)
|
||
self.assertNotEqual(codes[0], 429)
|