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 import status 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 AccessToken = get_access_token_model() Application = get_application_model() def mock_notifications_push_user_success(user_uuid, title, message, priority=5, extras=None): import uuid as sys_uuid class Tmp(): uuid = sys_uuid.uuid4() data = Tmp() return data class GooyalOAuth2Tests(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.notifications_push_user_success_patcher = patch('apps.users.models.User.notify', mock_notifications_push_user_success) self.notifications_push_user_success_patcher.start() self.user_phone_number = '+989100000000' self.user = User.objects.create(pk=self.user_uuid, phone_number=self.user_phone_number) scope = Scope.objects.create(name='accounts.status:get', description='accounts.status:get') 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', ) # 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', "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, status.HTTP_409_CONFLICT) 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', "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_loginByOTP_allOK_success(self): print(self.login()) response = self.client.get(reverse("core:status")) print(response.status_code) def test_revoke_token(self): self.user.set_otp() data = { "grant_type": "password", "username": self.user_phone_number, "password": '77501', "scope": 'accounts.status:get', "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'] revoke_data = { "token": access_token, } result = self.client.post(reverse("gooyal_oauth2:revoke-token"), data=revoke_data) self.assertEqual(result.status_code, 200) self.assertEqual(AccessToken.objects.count(), 0) self.user.refresh_from_db() self.user.set_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'] access_token_object = AccessToken.objects.first() revoke_data = { "token": access_token_object.pk, } result = self.client.post(reverse("gooyal_oauth2:revoke-token"), data=revoke_data) self.assertEqual(result.status_code, 200) self.assertEqual(AccessToken.objects.count(), 0)