import base64 import json import uuid from datetime import timedelta from unicodedata import category from unittest.mock import patch from django.test import TestCase from django.urls import reverse from django.utils import timezone from oauth2_provider.models import get_access_token_model, get_application_model from rest_framework.test import APIClient, APITestCase from apps.core.models import Config from apps.gooyal_oauth2.models import Scope from apps.users.models import User, InviteRequest AccessToken = get_access_token_model() Application = get_application_model() class UserTests(APITestCase): user_uuid = uuid.UUID('b14e8b86-8f4a-44d9-b29d-badceb47005f') access_token_1 = 'au4naVsdKCbKNOhnElPyXcrwSnqqFbm' access_token_2 = 'vu4naVsdKCbKNOhnElPyXcrwSnqqFbm' application_uuid = uuid.UUID('a14e8b86-8f4a-44d9-b29d-badceb47005f') client_id = '4INGOCMoulE0fNY1SQlTbPtsWqqxGj2DdqjADq6u' client_secret = '4INGOCMoulE0fNY1SQlTbPtsWqqxGj2DdqjADq6u' visitor_uuid = uuid.UUID('b14e8b86-8f4a-44d9-b29d-badceb47005a') expire_datetime = timezone.now() + timedelta(seconds=3600) expire_datetime.isoformat() client = APIClient() def setUp(self): from django.conf import settings settings.SMS_SEND = False self.user_phone_number = '+989100000000' self.user = User.objects.create(pk=self.user_uuid, phone_number=self.user_phone_number) Scope.objects.create(name='accounts.status:get', description='accounts.status:get') Scope.objects.create(name='accounts.account:retrieve', description='accounts.account:retrieve') self.application = Application.objects.create( client_id=self.client_id, client_secret=self.client_secret, authorization_grant_type='password', hash_client_secret=False, uuid=self.application_uuid, user_id=self.user_uuid, max_allowed_session=1, allowed_scope='accounts.status:get accounts.account:retrieve', ) # self.sys_date_patcher = patch('simata_safte.models.get_sys_date', mock_get_sys_date) # self.set_id_patcher = patch('simata_safte.models.set_id', mock_set_id) # self.sign_pdf_patcher = patch('simata_safte.models.sign_pdf', mock_sign_pdf) # self.check_pdf_signature_patcher = patch('simata_safte.models.check_pdf_signature', mock_check_pdf_signature) def tearDown(self): super().tearDown() def _create_authorization_header(self, token): return "Bearer {0}".format(token) def test_authentication_allow(self): access_token_1 = AccessToken.objects.create( **{ "token": self.access_token_1, "user": self.user, # "client_id": self.client_id, # "client_owner": owner, "application_id": self.application_uuid, "scope": 'accounts.status:get accounts.account:retrieve', "expires": self.expire_datetime.isoformat(), }, ) auth_1 = self._create_authorization_header(access_token_1.token) response = self.client.get(reverse("core:status"), HTTP_AUTHORIZATION=auth_1) self.assertContains(response, 'status') access_token_2 = AccessToken.objects.create( **{ "token": self.access_token_2, "user": self.user, # "client_id": self.client_id, # "client_owner": owner, "application_id": self.application_uuid, "scope": 'accounts.status:get', "expires": self.expire_datetime.isoformat(), }, ) auth_2 = self._create_authorization_header(access_token_2.token) response = self.client.get(reverse("core:status"), HTTP_AUTHORIZATION=auth_1) self.assertEqual(response.status_code, 200) response = self.client.get(reverse("core:status"), HTTP_AUTHORIZATION=auth_2) self.assertEqual(response.status_code, 503) self.application.max_allowed_session = 0 self.application.save() self.application.refresh_from_db() response = self.client.get(reverse("core:status"), HTTP_AUTHORIZATION=auth_2) self.assertEqual(response.status_code, 200) def basic_auth_string(self, username, password): """ساخت Basic Auth string""" import base64 user_pass = f"{username}:{password}" basic_credentials = base64.b64encode(user_pass.encode('utf-8')).decode('utf-8') return basic_credentials def login(self): self.user.set_otp() data = { "grant_type": "password", "username": self.user_phone_number, "password": '77501', "scope": 'accounts.status:get accounts.account:retrieve', "auth_fields": 'phone_number:otp' } self.client.credentials( HTTP_AUTHORIZATION='Basic ' + self.basic_auth_string(self.client_id, self.client_secret) ) result = self.client.post(reverse("gooyal_oauth2:token"), data=data) access_token = result.json()['access_token'] self.client.credentials(HTTP_AUTHORIZATION='Bearer ' + access_token) self.assertEqual(result.status_code, 200) return result def test_getSessions_allOK_success(self): self.login() response = self.client.get(reverse("users:user_sessions_api")) self.assertEqual(response.json()['results'][0]['is_current'] , True) class InviteRequestTests(APITestCase): user_uuid = uuid.UUID('c14e8b86-8f4a-44d9-b29d-badceb47005f') inviter_uuid = uuid.UUID('d14e8b86-8f4a-44d9-b29d-badceb47005f') application_uuid = uuid.UUID('e14e8b86-8f4a-44d9-b29d-badceb47005f') client_id = 'inviteClientId1234567890123456789012' client_secret = 'inviteClientSecret123456789012345678' mobile_number = '+989120000000' client = APIClient() def setUp(self): self.user = User.objects.create(pk=self.user_uuid, phone_number='+989100000001') self.inviter = User.objects.create(pk=self.inviter_uuid, phone_number='+989100000002') Scope.objects.create(name='accounts.invite:submit', description='accounts.invite:submit') self.application = Application.objects.create( name='invite-test-app', client_id=self.client_id, client_secret=self.client_secret, client_type=Application.CLIENT_CONFIDENTIAL, authorization_grant_type=Application.GRANT_CLIENT_CREDENTIALS, hash_client_secret=False, uuid=self.application_uuid, user_id=self.user_uuid, allowed_scope='accounts.invite:submit', ) def basic_auth_header(self): credentials = f'{self.client_id}:{self.client_secret}'.encode('utf-8') return 'Basic ' + base64.b64encode(credentials).decode('utf-8') def test_create_invite_request_success(self): response = self.client.post( reverse('users:invite_request_api'), data={ 'invite_code': str(self.inviter_uuid), 'mobile_number': self.mobile_number, }, format='json', HTTP_AUTHORIZATION=self.basic_auth_header(), ) self.assertEqual(response.status_code, 201) self.assertEqual(InviteRequest.objects.count(), 1) invite_request = InviteRequest.objects.get() self.assertEqual(invite_request.inviter_id, self.inviter_uuid) self.assertEqual(invite_request.mobile_number, self.mobile_number) self.assertEqual(invite_request.application_id, self.application_uuid) def test_create_invite_request_invalid_invite_code(self): response = self.client.post( reverse('users:invite_request_api'), data={ 'invite_code': str(uuid.uuid4()), 'mobile_number': self.mobile_number, }, format='json', HTTP_AUTHORIZATION=self.basic_auth_header(), ) self.assertEqual(response.status_code, 400) self.assertEqual(InviteRequest.objects.count(), 0) def test_create_invite_request_invalid_client_credentials(self): response = self.client.post( reverse('users:invite_request_api'), data={ 'invite_code': str(self.inviter_uuid), 'mobile_number': self.mobile_number, }, format='json', HTTP_AUTHORIZATION='Basic ' + base64.b64encode(b'wrong:credentials').decode('utf-8'), ) self.assertEqual(response.status_code, 401) self.assertEqual(InviteRequest.objects.count(), 0) def test_create_invite_request_missing_scope(self): self.application.allowed_scope = '' self.application.save(update_fields=['allowed_scope']) response = self.client.post( reverse('users:invite_request_api'), data={ 'invite_code': str(self.inviter_uuid), 'mobile_number': self.mobile_number, }, format='json', HTTP_AUTHORIZATION=self.basic_auth_header(), ) self.assertEqual(response.status_code, 403) self.assertEqual(InviteRequest.objects.count(), 0)