list session api

apps/gooyal_oauth2/tests.py

@@ -180,17 +180,32 @@ class GooyalOAuth2Tests(APITestCase):
         result = self.client.post(reverse("gooyal_oauth2:token"), data=data)
         access_token = result.json()['access_token']
 
-        print(AccessToken.objects.count())
+        revoke_data = {
-
-        data = {
             "token": access_token,
         }
 
-        result = self.client.post(reverse("gooyal_oauth2:revoke-token"), data=data)
+        result = self.client.post(reverse("gooyal_oauth2:revoke-token"), data=revoke_data)
-        print(result.content)
+        self.assertEqual(result.status_code, 200)
-        print(result.status_code)
+        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)
+
+
+
 
-        print(AccessToken.objects.count())
 
 
 

apps/gooyal_oauth2/urls.py

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

apps/gooyal_oauth2/validators.py

@@ -266,4 +266,24 @@ class OAuth2Validator(BaseOAuth2Validator):  # pylint: disable=w0223
                 return result
 
     def revoke_token(self, token, token_type_hint, request, *args, **kwargs):
-        return super().revoke_token(token, token_type_hint, request, *args, **kwargs)
+        """
+                Revoke an access or refresh token.
+
+                :param token: The token string.
+                :param token_type_hint: access_token or refresh_token.
+                :param request: The HTTP Request (oauthlib.common.Request)
+                """
+        if token_type_hint not in ["access_token", "refresh_token"]:
+            token_type_hint = None
+
+        token_types = {
+            "access_token": AccessToken,
+            "refresh_token": RefreshToken,
+        }
+
+        token_type = token_types.get(token_type_hint, AccessToken)
+
+        try:
+            token_type.objects.get(pk=token).revoke()
+        except:
+            token_type.objects.get(token=token).revoke()

apps/gooyal_oauth2/views/oauth_views.py

@@ -169,7 +169,7 @@ class TokenView(OAuthLibMixin, View):
 
 @method_decorator(csrf_exempt, name="dispatch")
 @method_decorator(login_not_required, name="dispatch")
-class RevokeTokenView(OAuthLibMixin, View):
+class GooyalRevokeTokenView(OAuthLibMixin, View):
     """
     Implements an endpoint to revoke access or refresh tokens
     """

apps/users/models.py

@@ -195,6 +195,8 @@ class User(AbstractUser):
 
             if result:
                 self.otp = None
+                self.otp_expire = None
+
                 if not self.date_joined:
                     self.date_joined = timezone.now()
             else:
@@ -202,7 +204,9 @@ class User(AbstractUser):
             self.save()
         else:
             result = False
-
+            self.otp = None
+            self.otp_expire = None
+            self.save()
 
         return result
 

apps/users/serializers.py

@@ -138,7 +138,21 @@ class ChangePasswordSerializer(serializers.Serializer):
 
 AccessToken = get_access_token_model()
 class SessionSerializer(serializers.ModelSerializer):
+    is_current = serializers.SerializerMethodField()
+
+    def get_is_current(self, obj):
+        request = self.context.get('request')
+        token = request and request.auth and request.auth.token
+        if token == obj.token:
+            result = True
+
+        else:
+            result = False
+
+        return result
+
+
     class Meta:
         model = AccessToken
-        fields = ('uuid', 'token', 'created', "detail")
+        fields = ('uuid', 'created', "detail", "is_current")
-        read_only_fields = ['uuid', 'token', 'created', "detail"]
+        read_only_fields = ['uuid', 'created', "detail", "is_current"   ]

apps/users/tests.py

@@ -1,3 +1,154 @@
-from django.test import TestCase
+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)
+
+
+
+
 
-# Create your tests here.