revoke token test

This commit is contained in:
Sayyid Hamid Mahdavi 2026-04-07 09:47:00 +03:30
parent c1c55c45a6
commit be4df13c78
5 changed files with 124 additions and 14 deletions

View file

@ -1,5 +1,6 @@
from django.contrib import admin from django.contrib import admin
from .models import SMSPolicy from .models import SMSPolicy, Config
class SMSPolicyAdmin(admin.ModelAdmin): class SMSPolicyAdmin(admin.ModelAdmin):
def has_add_permission(self, request): def has_add_permission(self, request):
@ -9,4 +10,8 @@ class SMSPolicyAdmin(admin.ModelAdmin):
return False return False
class ConfigAdmin(admin.ModelAdmin):
list_display = ['key', 'value', 'value_type', 'get_value', 'description', 'comment']
admin.site.register(Config, ConfigAdmin)
admin.site.register(SMSPolicy, SMSPolicyAdmin) admin.site.register(SMSPolicy, SMSPolicyAdmin)

View file

@ -1,3 +1,4 @@
import base64
import json import json
import uuid import uuid
from datetime import timedelta from datetime import timedelta
@ -8,15 +9,18 @@ from django.test import TestCase
from django.urls import reverse from django.urls import reverse
from django.utils import timezone from django.utils import timezone
from oauth2_provider.models import get_access_token_model, get_application_model from oauth2_provider.models import get_access_token_model, get_application_model
from rest_framework.test import APIClient from rest_framework.test import APIClient, APITestCase
from apps.core.models import Config from apps.core.models import Config
from apps.gooyal_oauth2.models import Scope
from apps.users.models import User from apps.users.models import User
AccessToken = get_access_token_model() AccessToken = get_access_token_model()
Application = get_application_model() Application = get_application_model()
def mock_notifications_push_user_success(user_uuid, title, message, priority=5, extras=None): def mock_notifications_push_user_success(user_uuid, title, message, priority=5, extras=None):
import uuid as sys_uuid import uuid as sys_uuid
class Tmp(): class Tmp():
@ -26,32 +30,43 @@ def mock_notifications_push_user_success(user_uuid, title, message, priority=5,
return data return data
class GooyalOAuth2Tests(TestCase): class GooyalOAuth2Tests(APITestCase):
user_uuid = uuid.UUID('b14e8b86-8f4a-44d9-b29d-badceb47005f') user_uuid = uuid.UUID('b14e8b86-8f4a-44d9-b29d-badceb47005f')
access_token_1 = 'au4naVsdKCbKNOhnElPyXcrwSnqqFbm' access_token_1 = 'au4naVsdKCbKNOhnElPyXcrwSnqqFbm'
access_token_2 = 'vu4naVsdKCbKNOhnElPyXcrwSnqqFbm' access_token_2 = 'vu4naVsdKCbKNOhnElPyXcrwSnqqFbm'
application_uuid = uuid.UUID('a14e8b86-8f4a-44d9-b29d-badceb47005f') application_uuid = uuid.UUID('a14e8b86-8f4a-44d9-b29d-badceb47005f')
client_id = '4INGOCMoulE0fNY1SQlTbPtsWqqxGj2DdqjADq6u' client_id = '4INGOCMoulE0fNY1SQlTbPtsWqqxGj2DdqjADq6u'
client_secret = '4INGOCMoulE0fNY1SQlTbPtsWqqxGj2DdqjADq6u'
visitor_uuid = uuid.UUID('b14e8b86-8f4a-44d9-b29d-badceb47005a') visitor_uuid = uuid.UUID('b14e8b86-8f4a-44d9-b29d-badceb47005a')
expire_datetime = timezone.now() + timedelta(seconds=3600) expire_datetime = timezone.now() + timedelta(seconds=3600)
expire_datetime.isoformat() expire_datetime.isoformat()
client = APIClient()
def setUp(self): def setUp(self):
from django.conf import settings
settings.SMS_SEND = False
self.notifications_push_user_success_patcher = patch('apps.users.models.User.notify', self.notifications_push_user_success_patcher = patch('apps.users.models.User.notify',
mock_notifications_push_user_success) mock_notifications_push_user_success)
self.notifications_push_user_success_patcher.start() 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)
user, _ = User.objects.get_or_create(pk=self.user_uuid) scope = Scope.objects.create(name='accounts.status:get', description='accounts.status:get')
self.user = user
self.application = Application.objects.create(
self.application, _created = Application.objects.get_or_create(
client_id=self.client_id, client_id=self.client_id,
client_secret=self.client_secret,
authorization_grant_type='password',
hash_client_secret=False,
uuid=self.application_uuid, uuid=self.application_uuid,
user_id=self.user_uuid, user_id=self.user_uuid,
max_allowed_session=1 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.sys_date_patcher = patch('simata_safte.models.get_sys_date', mock_get_sys_date)
@ -90,6 +105,7 @@ class GooyalOAuth2Tests(TestCase):
**{ **{
"token": self.access_token_2, "token": self.access_token_2,
"user": self.user, "user": self.user,
# "client_id": self.client_id, # "client_id": self.client_id,
# "client_owner": owner, # "client_owner": owner,
"application_id": self.application_uuid, "application_id": self.application_uuid,
@ -100,12 +116,81 @@ class GooyalOAuth2Tests(TestCase):
auth_2 = self._create_authorization_header(access_token_2.token) 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) response = self.client.get(reverse("core:status"), HTTP_AUTHORIZATION=auth_1)
self.assertEqual(response.status_code, 200) self.assertEqual(response.status_code, 200)
def test_1(self): response = self.client.get(reverse("core:status"), HTTP_AUTHORIZATION=auth_2)
self.assertTrue(True) 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',
"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']
print(AccessToken.objects.count())
data = {
"token": access_token,
}
result = self.client.post(reverse("gooyal_oauth2:revoke-token"), data=data)
print(result.content)
print(result.status_code)
print(AccessToken.objects.count())

View file

@ -3,7 +3,7 @@ from django.urls import re_path, path
from oauth2_provider import views from oauth2_provider import views
from rest_framework import routers from rest_framework import routers
from .views.introspect import IntrospectTokenView, IntrospectApplicationView, TokenView from .views.oauth_views import IntrospectTokenView, IntrospectApplicationView, TokenView, RevokeTokenView
from .views import apis as api_views from .views import apis as api_views
from .views import pages from .views import pages
app_name = "gooyal_oauth2" app_name = "gooyal_oauth2"
@ -14,7 +14,7 @@ app_name = "gooyal_oauth2"
base_urlpatterns = [ base_urlpatterns = [
re_path(r"^authorize/$", views.AuthorizationView.as_view(), name="authorize"), re_path(r"^authorize/$", views.AuthorizationView.as_view(), name="authorize"),
re_path(r"^token/$", TokenView.as_view(), name="token"), re_path(r"^token/$", TokenView.as_view(), name="token"),
re_path(r"^revoke_token/$", views.RevokeTokenView.as_view(), name="revoke-token"), re_path(r"^revoke_token/$", RevokeTokenView.as_view(), name="revoke-token"),
re_path(r"^introspect/$", IntrospectTokenView.as_view(), name="introspect"), re_path(r"^introspect/$", IntrospectTokenView.as_view(), name="introspect"),
re_path(r"^introspect_application/$", IntrospectApplicationView.as_view(), name="introspect-application"), re_path(r"^introspect_application/$", IntrospectApplicationView.as_view(), name="introspect-application"),
] ]

View file

@ -261,3 +261,9 @@ class OAuth2Validator(BaseOAuth2Validator): # pylint: disable=w0223
return result return result
else: else:
raise ServiceUnavailable(code='max_allowed_session_reached') raise ServiceUnavailable(code='max_allowed_session_reached')
else:
return result
def revoke_token(self, token, token_type_hint, request, *args, **kwargs):
return super().revoke_token(token, token_type_hint, request, *args, **kwargs)

View file

@ -167,3 +167,17 @@ class TokenView(OAuthLibMixin, View):
response[k] = v response[k] = v
return response return response
@method_decorator(csrf_exempt, name="dispatch")
@method_decorator(login_not_required, name="dispatch")
class RevokeTokenView(OAuthLibMixin, View):
"""
Implements an endpoint to revoke access or refresh tokens
"""
def post(self, request, *args, **kwargs):
url, headers, body, status = self.create_revocation_response(request)
response = HttpResponse(content=body or "", status=status)
for k, v in headers.items():
response[k] = v
return response