132 lines
8.2 KiB
Python
132 lines
8.2 KiB
Python
"""Customer OTP journey, repeat submits, and recoverable failure contracts."""
|
|
from datetime import timedelta
|
|
from unittest.mock import Mock, patch
|
|
|
|
from django.test import override_settings
|
|
from django.utils import timezone
|
|
import requests
|
|
|
|
from apps.cuser.models import SmsOpt
|
|
from apps.jwt_auth.test_regressions import RegressionFixtures
|
|
|
|
|
|
class CustomerLoginTests(RegressionFixtures):
|
|
def test_unrecognized_provider_reply_keeps_delivered_code_usable(self):
|
|
for body in ('1', '654321987', '<string>accepted</string>', ''):
|
|
with self.subTest(body=body):
|
|
SmsOpt.objects.all().delete()
|
|
with patch('utils.services.sms.requests.get', return_value=Mock(text=body)):
|
|
sent = self.post('login/mobile/', {'mobile': self.user.mobile})
|
|
self.assertEqual(sent.status_code, 202)
|
|
self.assertEqual(sent.data['result']['delivery_status'], 'unconfirmed')
|
|
code = SmsOpt.objects.get(mobile=self.user.mobile).opt
|
|
verified = self.post('login/mobile/verify/', {'mobile': self.user.mobile, 'opt': code})
|
|
self.assertEqual(verified.status_code, 200)
|
|
self.assertEqual(self.post('token/', {'uuid_opt': verified.data[0]['uuid']}).status_code, 200)
|
|
|
|
def test_lost_provider_reply_keeps_code_and_still_throttles_repeat_sms(self):
|
|
for error in (requests.Timeout, requests.ConnectionError):
|
|
with self.subTest(error=error):
|
|
SmsOpt.objects.all().delete()
|
|
with patch('utils.services.sms.requests.get', side_effect=error) as send:
|
|
sent = self.post('login/mobile/', {'mobile': self.user.mobile})
|
|
self.assertEqual(sent.status_code, 202)
|
|
self.assertEqual(self.post('login/mobile/', {'mobile': self.user.mobile}).status_code, 429)
|
|
send.assert_called_once()
|
|
code = SmsOpt.objects.get().opt
|
|
self.assertEqual(self.post('login/mobile/verify/', {'mobile': self.user.mobile, 'opt': code}).status_code, 200)
|
|
|
|
def test_uncertain_resend_retains_previous_delivered_code(self):
|
|
original = self.otp()
|
|
SmsOpt.objects.filter(pk=original.pk).update(created_at=timezone.now() - timedelta(seconds=61))
|
|
with patch('utils.services.sms.requests.get', return_value=Mock(text='unexpected')):
|
|
self.assertEqual(self.post('login/mobile/', {'mobile': self.user.mobile}).status_code, 202)
|
|
self.assertEqual(SmsOpt.objects.filter(mobile=self.user.mobile).count(), 2)
|
|
self.assertEqual(self.post('login/mobile/verify/', {'mobile': self.user.mobile, 'opt': original.opt}).status_code, 200)
|
|
self.assertFalse(SmsOpt.objects.filter(mobile=self.user.mobile).exists())
|
|
|
|
@override_settings(SMS_SUCCESS_CODES='0,1', SMS_FAILURE_CODES='error,-1')
|
|
def test_configured_response_codes_and_explicit_rejection(self):
|
|
with patch('utils.services.sms.requests.get', return_value=Mock(text='\ufeff1\r\n')):
|
|
sent = self.post('login/mobile/', {'mobile': self.user.mobile})
|
|
self.assertEqual(sent.status_code, 200)
|
|
self.assertEqual(sent.data['result']['delivery_status'], 'accepted')
|
|
original = SmsOpt.objects.get()
|
|
SmsOpt.objects.filter(pk=original.pk).update(created_at=timezone.now() - timedelta(seconds=61))
|
|
with patch('utils.services.sms.requests.get', return_value=Mock(text='-1')):
|
|
self.assertEqual(self.post('login/mobile/', {'mobile': self.user.mobile}).status_code, 503)
|
|
self.assertEqual(SmsOpt.objects.get().pk, original.pk)
|
|
|
|
def test_complete_customer_journey_login_refresh_and_logout(self):
|
|
with patch('apps.jwt_auth.services.DefaultSmsProvider.send_sms', return_value=True) as send:
|
|
sent = self.post('login/mobile/', {'mobile': '۰۹۱۲۳۴۵۶۷۸۹'})
|
|
self.assertEqual(sent.status_code, 200)
|
|
self.assertEqual(sent.data['result']['mobile'], self.user.mobile)
|
|
self.assertEqual(sent.data['result']['retry_after'], 60)
|
|
send.assert_called_once()
|
|
code = SmsOpt.objects.get(mobile=self.user.mobile).opt
|
|
options = self.post('login/mobile/verify/', {'mobile': self.user.mobile, 'opt': code})
|
|
self.assertEqual(options.status_code, 200)
|
|
self.assertEqual(len(options.data), 1)
|
|
self.assertEqual(options.data[0]['cuser_role']['role_type']['role_name_en'], 'workspace_customer')
|
|
tokens = self.post('token/', {'uuid_opt': options.data[0]['uuid']})
|
|
self.assertEqual(tokens.status_code, 200)
|
|
self.client.credentials(HTTP_AUTHORIZATION='Bearer ' + tokens.data['access'])
|
|
self.assertEqual(self.client.get('/cuser/info/').status_code, 200)
|
|
rotated = self.post('token/refresh/', {'refresh': tokens.data['refresh']})
|
|
self.assertEqual(rotated.status_code, 200)
|
|
self.assertNotEqual(rotated.data['refresh'], tokens.data['refresh'])
|
|
self.assertEqual(self.post('logout/', {'refresh': tokens.data['refresh']}).status_code, 204)
|
|
self.client.credentials(HTTP_AUTHORIZATION='Bearer ' + rotated.data['access'])
|
|
self.assertEqual(self.client.get('/cuser/info/').status_code, 401)
|
|
self.assertEqual(self.post('token/refresh/', {'refresh': rotated.data['refresh']}).status_code, 401)
|
|
|
|
def test_repeat_send_does_not_invalidate_code_or_send_another_sms(self):
|
|
with patch('apps.jwt_auth.services.DefaultSmsProvider.send_sms', return_value=True) as send:
|
|
self.assertEqual(self.post('login/mobile/', {'mobile': self.user.mobile}).status_code, 200)
|
|
original = SmsOpt.objects.get(mobile=self.user.mobile)
|
|
for _ in range(3):
|
|
repeated = self.post('login/mobile/', {'mobile': self.user.mobile})
|
|
self.assertEqual(repeated.status_code, 429)
|
|
self.assertIn('قبلاً ارسال', repeated.data['detail'])
|
|
self.assertGreater(int(repeated['Retry-After']), 0)
|
|
self.assertEqual(SmsOpt.objects.get().pk, original.pk)
|
|
self.assertEqual(SmsOpt.objects.get().opt, original.opt)
|
|
send.assert_called_once()
|
|
verified = self.post('login/mobile/verify/', {'mobile': self.user.mobile, 'opt': original.opt})
|
|
self.assertEqual(verified.status_code, 200)
|
|
|
|
def test_resend_after_interval_replaces_code_once(self):
|
|
original = self.otp()
|
|
SmsOpt.objects.filter(pk=original.pk).update(created_at=timezone.now() - timedelta(seconds=61))
|
|
with patch('apps.jwt_auth.services.DefaultSmsProvider.send_sms', return_value=True) as send:
|
|
response = self.post('login/mobile/', {'mobile': self.user.mobile})
|
|
self.assertEqual(response.status_code, 200)
|
|
self.assertFalse(SmsOpt.objects.filter(pk=original.pk).exists())
|
|
self.assertEqual(SmsOpt.objects.filter(mobile=self.user.mobile).count(), 1)
|
|
send.assert_called_once()
|
|
|
|
def test_expired_code_does_not_prevent_requesting_a_new_one(self):
|
|
self.otp(expired=True)
|
|
with patch('apps.jwt_auth.services.DefaultSmsProvider.send_sms', return_value=True):
|
|
self.assertEqual(self.post('login/mobile/', {'mobile': self.user.mobile}).status_code, 200)
|
|
|
|
def test_panel_disabled_between_send_and_verify_has_clear_error(self):
|
|
code = self.otp()
|
|
self.role.customer_panel = False
|
|
self.role.save(update_fields=['customer_panel'])
|
|
response = self.post('login/mobile/verify/', {'mobile': self.user.mobile, 'opt': code.opt})
|
|
self.assertEqual(response.status_code, 403)
|
|
self.assertIn('پنل مشتریان', str(response.data))
|
|
|
|
def test_double_verify_and_double_token_exchange_cannot_create_multiple_sessions(self):
|
|
from apps.jwt_auth.models import JwtToken
|
|
code = self.otp()
|
|
first = self.post('login/mobile/verify/', {'mobile': self.user.mobile, 'opt': code.opt})
|
|
self.assertEqual(first.status_code, 200)
|
|
self.assertEqual(self.post('login/mobile/verify/', {'mobile': self.user.mobile, 'opt': code.opt}).status_code, 400)
|
|
option = first.data[0]['uuid']
|
|
self.assertEqual(self.post('token/', {'uuid_opt': option}).status_code, 200)
|
|
self.assertEqual(self.post('token/', {'uuid_opt': option}).status_code, 401)
|
|
self.assertEqual(JwtToken.objects.count(), 1)
|