Feature/test for clear expired (#680)

* document force migration sequence forr application model

* add tests for clear_expired

* add tests for clear_tokens

* add tests for clear_expired

* fix isort

* Update tests/test_models.py

Co-Authored-By: ecilveks <[email protected]>

* CR changes

* fix isort for upstream

* add logging about what is to be deleted

* also log when no refresh tokens are to be deleted

Author: ecilveks

docs/advanced_topics.rst

@@ -48,6 +48,15 @@ Be aware that, when you intend to swap the application model, you should create
 migration defining the swapped application model prior to setting OAUTH2_PROVIDER_APPLICATION_MODEL.
 You'll run into models.E022 in Core system checks if you don't get the order right.
 
+You can force your migration providing the custom model to run in the right order by
+adding::
+
+    run_before = [
+        ('oauth2_provider', '0001_initial'),
+    ]
+
+to the migration class.
+
 That's all, now Django OAuth Toolkit will use your model wherever an Application instance is needed.
 
     **Notice:** `OAUTH2_PROVIDER_APPLICATION_MODEL` is the only setting variable that is not namespaced, this

oauth2_provider/models.py

@@ -1,5 +1,6 @@
 from datetime import timedelta
 from urllib.parse import parse_qsl, urlparse
+import logging
 
 from django.apps import apps
 from django.conf import settings
@@ -14,6 +15,8 @@
 from .settings import oauth2_settings
 from .validators import RedirectURIValidator, WildcardSet
 
+logger = logging.getLogger(__name__)
+
 
 class AbstractApplication(models.Model):
     """
@@ -436,7 +439,30 @@ def clear_expired():
 
     with transaction.atomic():
         if refresh_expire_at:
-            refresh_token_model.objects.filter(revoked__lt=refresh_expire_at).delete()
-            refresh_token_model.objects.filter(access_token__expires__lt=refresh_expire_at).delete()
-        access_token_model.objects.filter(refresh_token__isnull=True, expires__lt=now).delete()
-        grant_model.objects.filter(expires__lt=now).delete()
+            revoked = refresh_token_model.objects.filter(
+                revoked__lt=refresh_expire_at,
+            )
+            expired = refresh_token_model.objects.filter(
+                access_token__expires__lt=refresh_expire_at,
+            )
+
+            logger.info('%s Revoked refresh tokens to be deleted', revoked.count())
+            logger.info('%s Expired refresh tokens to be deleted', expired.count())
+
+            revoked.delete()
+            expired.delete()
+        else:
+            logger.info('refresh_expire_at is %s. No refresh tokens deleted.',
+                         refresh_expire_at)
+
+        access_tokens = access_token_model.objects.filter(
+            refresh_token__isnull=True,
+            expires__lt=now
+        )
+        grants = grant_model.objects.filter(expires__lt=now)
+
+        logger.info('%s Expired access tokens to be deleted', access_tokens.count())
+        logger.info('%s Expired grant tokens to be deleted', grants.count())
+
+        access_tokens.delete()
+        grants.delete()

tests/test_models.py

@@ -1,11 +1,14 @@
+from datetime import datetime as dt
+
+import pytest
 from django.contrib.auth import get_user_model
-from django.core.exceptions import ValidationError
+from django.core.exceptions import ImproperlyConfigured, ValidationError
 from django.test import TestCase
 from django.test.utils import override_settings
 from django.utils import timezone
 
 from oauth2_provider.models import (
-    get_access_token_model, get_application_model,
+    clear_expired, get_access_token_model, get_application_model,
     get_grant_model, get_refresh_token_model
 )
 from oauth2_provider.settings import oauth2_settings
@@ -19,6 +22,7 @@
 
 
 class TestModels(TestCase):
+
     def setUp(self):
         self.user = UserModel.objects.create_user("test_user", "[email protected]", "123456")
 
@@ -118,6 +122,7 @@ def test_scopes_property(self):
     OAUTH2_PROVIDER_GRANT_MODEL="tests.SampleGrant"
 )
 class TestCustomModels(TestCase):
+
     def setUp(self):
         self.user = UserModel.objects.create_user("test_user", "[email protected]", "123456")
 
@@ -260,6 +265,7 @@ def test_expires_can_be_none(self):
 
 
 class TestAccessTokenModel(TestCase):
+
     def setUp(self):
         self.user = UserModel.objects.create_user("test_user", "[email protected]", "123456")
 
@@ -289,3 +295,54 @@ class TestRefreshTokenModel(TestCase):
     def test_str(self):
         refresh_token = RefreshToken(token="test_token")
         self.assertEqual("%s" % refresh_token, refresh_token.token)
+
+
+class TestClearExpired(TestCase):
+
+    def setUp(self):
+        self.user = UserModel.objects.create_user("test_user", "[email protected]", "123456")
+        # Insert two tokens on database.
+        AccessToken.objects.create(
+            id=1,
+            token="555",
+            expires=dt.now(),
+            scope=2,
+            application_id=3,
+            user_id=1,
+            created=dt.now(),
+            updated=dt.now(),
+            source_refresh_token_id="0",
+            )
+        AccessToken.objects.create(
+            id=2,
+            token="666",
+            expires=dt.now(),
+            scope=2,
+            application_id=3,
+            user_id=1,
+            created=dt.now(),
+            updated=dt.now(),
+            source_refresh_token_id="1",
+            )
+
+    def test_clear_expired_tokens(self):
+        oauth2_settings.REFRESH_TOKEN_EXPIRE_SECONDS = 60
+        assert clear_expired() is None
+
+    def test_clear_expired_tokens_incorect_timetype(self):
+        oauth2_settings.REFRESH_TOKEN_EXPIRE_SECONDS = "A"
+        with pytest.raises(ImproperlyConfigured) as excinfo:
+            clear_expired()
+        result = excinfo.value.__class__.__name__
+        assert result == "ImproperlyConfigured"
+
+    def test_clear_expired_tokens_with_tokens(self):
+        self.client.login(username="test_user", password="123456")
+        oauth2_settings.REFRESH_TOKEN_EXPIRE_SECONDS = 0
+        ttokens = AccessToken.objects.count()
+        expiredt = AccessToken.objects.filter(expires__lte=dt.now()).count()
+        assert ttokens == 2
+        assert expiredt == 2
+        clear_expired()
+        expiredt = AccessToken.objects.filter(expires__lte=dt.now()).count()
+        assert expiredt == 0