154 lines
5.4 KiB
Python
154 lines
5.4 KiB
Python
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
|
|
|
|
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)
|
|
|
|
|
|
|
|
|
|
|