Добавлен класс ExpiredTokenAuthentication

net/xeaf/rack/tests/utils/auth/__init__.py

@@ -0,0 +1,12 @@
+# DRF Rack
+# Библиотека классов расширений для Django REST Framework
+#
+# Автор: Николай В. Анохин <n.anokhin@xeaf.net>
+# Все права защищены. Лицензия: MIT
+
+"""
+Пакет описания тестов классов авторизации
+"""
+
+from .expiring_token_authentication_tests import ExpiringTokenAuthenticationTests
+

net/xeaf/rack/tests/utils/auth/expiring_token_authentication_tests.py

@@ -0,0 +1,173 @@
+# DRF Rack
+# Библиотека классов расширений для Django REST Framework
+#
+# Автор: Николай В. Анохин <n.anokhin@xeaf.net>
+# Все права защищены. Лицензия: MIT
+
+"""
+Описание класса ExpiringTokenAuthenticationTests
+"""
+
+import datetime
+
+from django.utils import timezone
+from django.utils import translation
+from rest_framework.parsers import JSONParser
+from rest_framework.request import Request
+from rest_framework.test import APIRequestFactory
+
+from net.xeaf.rack.core import CoreTestCase
+from net.xeaf.rack.management.userstory import AccountTestData
+from net.xeaf.rack.management.userstory import SessionTestData
+from net.xeaf.rack.models import AccountModel
+from net.xeaf.rack.models import SessionModel
+from net.xeaf.rack.utils.auth import ExpiringTokenAuthentication
+
+
+class ExpiringTokenAuthenticationTests(CoreTestCase):
+    """
+    Тесты для ExpiringTokenAuthentication
+    """
+
+    def setUp(self) -> None:
+        """
+        Подготовка к тестированию
+        """
+
+        super().setUp()
+        self.auth = ExpiringTokenAuthentication()
+        self.factory = APIRequestFactory()
+        self.old_language = translation.get_language()
+
+    def tearDown(self) -> None:
+        """
+        Восстанавливаем язык глобального контекста после тестов
+        """
+
+        super().tearDown()
+        translation.activate(self.old_language)
+
+    @classmethod
+    def _wrap_to_drf_request(cls, wsgi_request) -> Request:
+        """
+        Вспомогательный метод для честного оборачивания Django WSGIRequest в DRF Request
+        """
+
+        return Request(wsgi_request, parsers=[JSONParser()])
+
+    def test_authenticate_adjusts_language_via_query_param(self) -> None:
+        """
+        Должен переключить глобальную локаль Django, используя механизмы LocaleManager
+        """
+
+        # 1. Создаем нативный запрос Django
+        wsgi_req = self.factory.get("/", {"x-ui-language": "en"})
+        # 2. Оборачиваем его в честный DRF Request, у которого ЕСТЬ query_params
+        drf_request = self._wrap_to_drf_request(wsgi_req)
+
+        self.auth.authenticate(drf_request)
+
+        # Проверяем, что LocaleManager отработал "вживую" и активировал перевод
+        self.assertEqual(translation.get_language(), "en")
+        self.assertEqual(getattr(drf_request, "LANGUAGE_CODE", None), "EN")
+
+    def test_authenticate_extracts_token_from_query_params_and_logs_in(self) -> None:
+        """
+        Должен успешно авторизовать Ивана Иванова, если его токен передан в GET параметре
+        """
+
+        wsgi_req = self.factory.get("/", {ExpiringTokenAuthentication.AUTH_TOKEN_PARAM: SessionTestData.I_IVANOV_SESSION_TOKEN})
+        drf_request = self._wrap_to_drf_request(wsgi_req)
+
+        result = self.auth.authenticate(drf_request)
+
+        # Линтер-безопасная проверка контейнера
+        self.assertIsInstance(result, tuple)
+        assert result is not None
+
+        account, session = result
+        self.assertEqual(account.id, AccountTestData.I_IVANOV_ID)
+        self.assertEqual(session.token, SessionTestData.I_IVANOV_SESSION_TOKEN)
+
+    def test_authenticate_fallback_to_super_headers_if_no_query_param(self) -> None:
+        """
+        Должен отработать через стандартный заголовок 'Authorization: Token ...' родительского класса
+        """
+
+        wsgi_req = self.factory.get("/", HTTP_AUTHORIZATION=f"Token {SessionTestData.P_PETROV_SESSION_TOKEN}")
+        drf_request = self._wrap_to_drf_request(wsgi_req)
+
+        result = self.auth.authenticate(drf_request)
+
+        self.assertIsInstance(result, tuple)
+        assert result is not None
+
+        account, session = result
+        self.assertEqual(account.id, AccountTestData.P_PETROV_ID)
+
+    def test_authenticate_credentials_returns_none_tuple_if_token_not_found(self) -> None:
+        """
+        Должен мягко вернуть (None, None), если токена вообще нет в базе сессий
+        """
+
+        result = self.auth.authenticate_credentials("completely_fake_token")
+
+        self.assertEqual(result, (None, None))
+
+    def test_authenticate_credentials_returns_none_tuple_if_session_is_expired(self) -> None:
+        """
+        Должен вернуть (None, None), если сессия Петра Петрова просрочена
+        """
+
+        session = SessionModel.objects.get(token=SessionTestData.P_PETROV_SESSION_TOKEN)
+        session.expires_at = timezone.now() - datetime.timedelta(seconds=1)
+        session.save()
+
+        result = self.auth.authenticate_credentials(SessionTestData.P_PETROV_SESSION_TOKEN)
+
+        self.assertEqual(result, (None, None))
+
+    def test_authenticate_credentials_returns_none_tuple_if_account_is_not_active(self) -> None:
+        """
+        Должен вернуть (None, None), если AccountManager заблокировал вход (is_active = False)
+        """
+
+        account = AccountModel.objects.get(id=AccountTestData.I_IVANOV_ID)
+        account.is_active = False
+        account.save()
+
+        result = self.auth.authenticate_credentials(SessionTestData.I_IVANOV_SESSION_TOKEN)
+
+        self.assertEqual(result, (None, None))
+
+    def test_authenticate_credentials_returns_none_tuple_if_account_is_deleted(self) -> None:
+        """
+        Должен вернуть (None, None), если пользователь помечен как удаленный
+        """
+
+        account = AccountModel.objects.get(id=AccountTestData.I_IVANOV_ID)
+        account.soft_delete(save=True)
+
+        result = self.auth.authenticate_credentials(SessionTestData.I_IVANOV_SESSION_TOKEN)
+
+        self.assertEqual(result, (None, None))
+
+    def test_authenticate_credentials_renews_session_timestamp_on_success(self) -> None:
+        """
+        Должен обновить время жизни сессии в БД и вернуть валидные объекты при успешном входе
+        """
+
+        session_before = SessionModel.objects.get(token=SessionTestData.I_IVANOV_SESSION_TOKEN)
+
+        result = self.auth.authenticate_credentials(SessionTestData.I_IVANOV_SESSION_TOKEN)
+
+        self.assertIsInstance(result, tuple)
+        assert result is not None
+        account, session = result
+
+        self.assertIsInstance(account, AccountModel)
+        self.assertEqual(account.id, AccountTestData.I_IVANOV_ID)
+
+        self.assertIsInstance(session, SessionModel)
+        session.refresh_from_db()
+        self.assertGreater(session.expires_at, session_before.expires_at)

net/xeaf/rack/utils/auth/__init__.py

@@ -7,3 +7,5 @@
 """
 Классов авторизации
 """
+
+from .expiring_token_authentication import ExpiringTokenAuthentication

net/xeaf/rack/utils/auth/expiring_token_authentication.py

@@ -13,6 +13,7 @@ from typing import Any
 from rest_framework.authentication import TokenAuthentication
 from rest_framework.request import Request
 
+from net.xeaf.rack.managers import AccountManager
 from net.xeaf.rack.managers import LocaleManager
 from net.xeaf.rack.managers import SessionManager
 from net.xeaf.rack.models import AccountModel
@@ -54,8 +55,21 @@ class ExpiringTokenAuthentication(TokenAuthentication):
             return None, None
 
         account = session.account
-        if not account.is_active or account.is_deleted():
+        if not self._check_account(account):
             return None, None
 
         SessionManager.renew_session(session)
         return session.account, session
+
+    @classmethod
+    def _check_account(cls, account: AccountModel) -> bool:
+        """
+        Проверка учетной записи
+
+        :param account: Учетная запись
+        """
+
+        # Можно прошедшим проверку и не удаленным
+        reject = AccountManager.check_account_auth(account)
+        result = reject is None and not account.is_deleted()
+        return result