test: add tests for token family functionality across views, serializers, and models

- Added test_token_family.py with:
  - TestTokenFamilyModels
  - TestTokenFamilyInRefreshTokens
  - TestTokenFamilyInAccessTokens
  - TestFlushExpiredTokenFamiliesCommand
- Added tests for token family behavior in serializers and views
- Updated conftest.py, urls.py, and test_integration.py to support token family tests

Author: juanbailon

tests/conftest.py

@@ -36,11 +36,13 @@ def pytest_configure():
             "rest_framework",
             "rest_framework_simplejwt",
             "rest_framework_simplejwt.token_blacklist",
+            "rest_framework_simplejwt.token_family",
             "tests",
         ),
         PASSWORD_HASHERS=("django.contrib.auth.hashers.MD5PasswordHasher",),
         SIMPLE_JWT={
             "BLACKLIST_AFTER_ROTATION": True,
+            "TOKEN_FAMILY_ENABLED": True,
         },
     )
 

tests/test_integration.py

@@ -1,11 +1,14 @@
 from datetime import timedelta
+from importlib import reload
+from freezegun import freeze_time
 
 from django.contrib.auth import get_user_model
 from django.urls import reverse
 from rest_framework.status import HTTP_200_OK, HTTP_401_UNAUTHORIZED
 
 from rest_framework_simplejwt.settings import api_settings
-from rest_framework_simplejwt.tokens import AccessToken
+from rest_framework_simplejwt.tokens import AccessToken, RefreshToken
+from rest_framework_simplejwt.utils import aware_utcnow
 
 from .utils import APIViewTestCase, override_api_settings
 
@@ -127,3 +130,124 @@ def test_user_can_get_access_and_refresh_tokens_and_use_them(self):
 
         self.assertEqual(res.status_code, HTTP_200_OK)
         self.assertEqual(res.data["foo"], "bar")
+
+    @override_api_settings(
+        AUTH_TOKEN_CLASSES=("rest_framework_simplejwt.tokens.AccessToken",),
+        TOKEN_FAMILY_CHECK_ON_ACCESS=True,
+    )
+    def test_access_token_performs_family_blacklist_check_when_enabled(self):
+        res = self.client.post(
+            reverse("token_obtain_pair"),
+            data={
+                User.USERNAME_FIELD: self.username,
+                "password": self.password,
+            },
+        )
+
+        access = res.data["access"]
+        refresh = res.data["refresh"]
+
+        self.authenticate_with_token(api_settings.AUTH_HEADER_TYPES[0], access)
+
+        res = self.view_get()
+
+        self.assertEqual(res.status_code, HTTP_200_OK)
+        self.assertEqual(res.data["foo"], "bar")
+
+        RefreshToken(str(refresh)).blacklist_family()
+
+        self.authenticate_with_token(api_settings.AUTH_HEADER_TYPES[0], access)
+
+        res = self.view_get()
+
+        self.assertEqual(res.status_code, HTTP_401_UNAUTHORIZED)
+        self.assertEqual("token_not_valid", res.data["code"])
+
+        error_msg = res.data.get("messages")[0].get("message")
+        self.assertIn("family", error_msg)
+        self.assertIn("blacklisted", error_msg)
+
+    @override_api_settings(
+        AUTH_TOKEN_CLASSES=("rest_framework_simplejwt.tokens.AccessToken",),
+        TOKEN_FAMILY_CHECK_ON_ACCESS=True,
+        # We use a smaller family lifetime than the default access token lifetime
+        # so we dont have to reload the tokens and serializers modules
+        TOKEN_FAMILY_LIFETIME=timedelta(minutes=2),
+    )
+    def test_access_token_performs_family_expiration_check_when_enabled(self):
+
+        with freeze_time(aware_utcnow() - timedelta(minutes=2)):
+            res = self.client.post(
+                reverse("token_obtain_pair"),
+                data={
+                    User.USERNAME_FIELD: self.username,
+                    "password": self.password,
+                },
+            )
+
+        access = res.data["access"]
+
+        self.authenticate_with_token(api_settings.AUTH_HEADER_TYPES[0], access)
+
+        res = self.view_get()
+
+        self.assertEqual(res.status_code, HTTP_401_UNAUTHORIZED)
+        self.assertEqual("token_not_valid", res.data["code"])
+
+        error_msg = res.data.get("messages")[0].get("message")
+        self.assertIn("family", error_msg)
+        self.assertIn("expired", error_msg)
+
+    @override_api_settings(
+        AUTH_TOKEN_CLASSES=("rest_framework_simplejwt.tokens.AccessToken",),
+        TOKEN_FAMILY_CHECK_ON_ACCESS=False, 
+        # We use a smaller family lifetime than the default access token lifetime
+        # so we dont have to reload the tokens and serializers modules
+        TOKEN_FAMILY_LIFETIME=timedelta(minutes=2),
+    )
+    def test_access_token_does_not_performs_family_checks_when_disabled(self):
+        res = self.client.post(
+            reverse("token_obtain_pair"),
+            data={
+                User.USERNAME_FIELD: self.username,
+                "password": self.password,
+            },
+        )
+
+        access = res.data["access"]
+        refresh = res.data["refresh"]
+
+        self.authenticate_with_token(api_settings.AUTH_HEADER_TYPES[0], access)
+        res = self.view_get()
+
+        self.assertEqual(res.status_code, HTTP_200_OK)
+        self.assertEqual(res.data["foo"], "bar")
+
+        # blacklisting the token family
+        RefreshToken(str(refresh)).blacklist_family()
+
+        self.authenticate_with_token(api_settings.AUTH_HEADER_TYPES[0], access)
+        res = self.view_get()
+
+        # response must be 200_OK  since the family check for the access token is disabled
+        self.assertEqual(res.status_code, HTTP_200_OK)
+        self.assertEqual(res.data["foo"], "bar")
+
+        # testing for family expiration now
+        with freeze_time(aware_utcnow() - timedelta(minutes=2)):
+            res = self.client.post(
+                reverse("token_obtain_pair"),
+                data={
+                    User.USERNAME_FIELD: self.username,
+                    "password": self.password,
+                },
+            )
+
+        access = res.data["access"]
+
+        self.authenticate_with_token(api_settings.AUTH_HEADER_TYPES[0], access)
+        res = self.view_get()
+    
+        # response must be 200_OK  since the family check for the access token is disabled
+        self.assertEqual(res.status_code, HTTP_200_OK)
+        self.assertEqual(res.data["foo"], "bar")