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 from apps.core.models import Config 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(TestCase): 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' visitor_uuid = uuid.UUID('b14e8b86-8f4a-44d9-b29d-badceb47005a') expire_datetime = timezone.now() + timedelta(seconds=3600) expire_datetime.isoformat() def setUp(self): self.notifications_push_user_success_patcher = patch('apps.users.models.User.notify', mock_notifications_push_user_success) self.notifications_push_user_success_patcher.start() user, _ = User.objects.get_or_create(pk=self.user_uuid) self.user = user self.application, _created = Application.objects.get_or_create( client_id=self.client_id, uuid=self.application_uuid, user_id=self.user_uuid, max_allowed_session=1 ) # 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_2) self.assertEqual(response.status_code, 503) response = self.client.get(reverse("core:status"), HTTP_AUTHORIZATION=auth_1) self.assertEqual(response.status_code, 200) def test_1(self): self.assertTrue(True)