Enhance clear_expired tests. (#1088)

Author: n2ygk

oauth2_provider/management/commands/cleartokens.py

@@ -3,7 +3,7 @@
 from ...modelsimportclear_expired
 
 
-classCommand(BaseCommand):
+classCommand(BaseCommand):# pragma: no cover
 help="Can be run as a cronjob or directly to clean out expired tokens"
 
 defhandle(self, *args, **options):

tests/test_models.py

@@ -1,3 +1,5 @@
+fromdatetimeimporttimedelta
+
 importpytest
 fromdjango.contrib.authimportget_user_model
 fromdjango.core.exceptionsimportImproperlyConfigured, ValidationError
@@ -294,31 +296,66 @@ def test_str(self):
 classTestClearExpired(BaseTestModels):
 defsetUp(self):
 super().setUp()
-# Insert two tokens on database.
+# Insert many tokens, both expired and not, and grants.
+self.num_tokens=100
+now=timezone.now()
+earlier=now-timedelta(seconds=100)
+later=now+timedelta(seconds=100)
 app=Application.objects.create(
 name="test_app",
 redirect_uris="http://localhost http://example.com http://example.org",
 user=self.user,
 client_type=Application.CLIENT_CONFIDENTIAL,
 authorization_grant_type=Application.GRANT_AUTHORIZATION_CODE,
  )
-AccessToken.objects.create(
-token="555",
-expires=timezone.now(),
-scope=2,
-application=app,
-user=self.user,
-created=timezone.now(),
-updated=timezone.now(),
+# make 200 access tokens, half current and half expired.
+expired_access_tokens=AccessToken.objects.bulk_create(
+AccessToken(token="expired AccessToken{}".format(i), expires=earlier)
+foriinrange(self.num_tokens)
  )
-AccessToken.objects.create(
-token="666",
-expires=timezone.now(),
-scope=2,
-application=app,
-user=self.user,
-created=timezone.now(),
-updated=timezone.now(),
+current_access_tokens=AccessToken.objects.bulk_create(
+AccessToken(token=f"current AccessToken {i}", expires=later) foriinrange(self.num_tokens)
+ )
+# Give the first half of the access tokens a refresh token,
+# alternating between current and expired ones.
+RefreshToken.objects.bulk_create(
+RefreshToken(
+token=f"expired AT's refresh token {i}",
+application=app,
+access_token=expired_access_tokens[i].pk,
+user=self.user,
+ )
+foriinrange(0, len(expired_access_tokens) //2, 2)
+ )
+RefreshToken.objects.bulk_create(
+RefreshToken(
+token=f"current AT's refresh token {i}",
+application=app,
+access_token=current_access_tokens[i].pk,
+user=self.user,
+ )
+foriinrange(1, len(current_access_tokens) //2, 2)
+ )
+# Make some grants, half of which are expired.
+Grant.objects.bulk_create(
+Grant(
+user=self.user,
+code=f"old grant code {i}",
+application=app,
+expires=earlier,
+redirect_uri="https://localhost/redirect",
+ )
+foriinrange(self.num_tokens)
+ )
+Grant.objects.bulk_create(
+Grant(
+user=self.user,
+code=f"new grant code {i}",
+application=app,
+expires=later,
+redirect_uri="https://localhost/redirect",
+ )
+foriinrange(self.num_tokens)
  )
 
 deftest_clear_expired_tokens(self):
@@ -333,15 +370,21 @@ def test_clear_expired_tokens_incorect_timetype(self):
 assertresult=="ImproperlyConfigured"
 
 deftest_clear_expired_tokens_with_tokens(self):
-self.client.login(username="test_user", password="123456")
-self.oauth2_settings.REFRESH_TOKEN_EXPIRE_SECONDS=0
-ttokens=AccessToken.objects.count()
-expiredt=AccessToken.objects.filter(expires__lte=timezone.now()).count()
-assertttokens==2
-assertexpiredt==2
+self.oauth2_settings.CLEAR_EXPIRED_TOKENS_BATCH_SIZE=10
+self.oauth2_settings.CLEAR_EXPIRED_TOKENS_BATCH_INTERVAL=0.0
+at_count=AccessToken.objects.count()
+assertat_count==2*self.num_tokens, f"{2*self.num_tokens} access tokens should exist."
+rt_count=RefreshToken.objects.count()
+assertrt_count==self.num_tokens//2, f"{self.num_tokens//2} refresh tokens should exist."
+gt_count=Grant.objects.count()
+assertgt_count==self.num_tokens*2, f"{self.num_tokens*2} grants should exist."
 clear_expired()
-expiredt=AccessToken.objects.filter(expires__lte=timezone.now()).count()
-assertexpiredt==0
+at_count=AccessToken.objects.count()
+assertat_count==self.num_tokens, "Half the access tokens should not have been deleted."
+rt_count=RefreshToken.objects.count()
+assertrt_count==self.num_tokens//2, "Half of the refresh tokens should have been deleted."
+gt_count=Grant.objects.count()
+assertgt_count==self.num_tokens, "Half the grants should have been deleted."
 
 
 @pytest.mark.django_db