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

This commit is contained in:
2026-07-16 19:53:53 +03:00
parent 63537f24a0
commit 3d1e6b47bd
10 changed files with 674 additions and 62 deletions

View File

@@ -9,6 +9,7 @@
""" """
import json import json
from typing import Any
from django.core.management import call_command from django.core.management import call_command
from django.test import TestCase from django.test import TestCase
@@ -16,8 +17,10 @@ from rest_framework.parsers import JSONParser
from rest_framework.request import Request from rest_framework.request import Request
from rest_framework.test import APIRequestFactory from rest_framework.test import APIRequestFactory
from net.xeaf.rack.enums import HttpMethod
from net.xeaf.rack.management.userstory import AccountTestData from net.xeaf.rack.management.userstory import AccountTestData
from net.xeaf.rack.management.userstory import SessionTestData from net.xeaf.rack.management.userstory import SessionTestData
from net.xeaf.rack.managers import SessionManager
from net.xeaf.rack.models import AccountModel from net.xeaf.rack.models import AccountModel
from net.xeaf.rack.models import SessionModel from net.xeaf.rack.models import SessionModel
@@ -38,12 +41,19 @@ class CoreTestCase(TestCase):
call_command("rack_test_database") call_command("rack_test_database")
self.factory = APIRequestFactory() self.factory = APIRequestFactory()
def create_request(self, method, path, data=None, query_params=None, auth=None): def create_request(self,
method: HttpMethod,
path: str,
headers: dict[str, str] | None = None,
data: dict[str, Any] | None = None,
query_params: dict[str, str] | None = None,
auth: SessionModel | None = None) -> Request:
""" """
Создает DRF Request объект для тестирования Создает объект запроса для тестирования
:param method: HTTP метод (GET, POST, PUT, PATCH, DELETE) :param method: HTTP метод (GET, POST, PUT, PATCH, DELETE)
:param path: URL путь :param path: URL путь
:param headers: Заголовки запроса
:param data: Данные для тела запроса :param data: Данные для тела запроса
:param query_params: Параметры запроса :param query_params: Параметры запроса
:param auth: Объект аутентификации (сессия) :param auth: Объект аутентификации (сессия)
@@ -51,63 +61,120 @@ class CoreTestCase(TestCase):
:return: DRF Request объект :return: DRF Request объект
""" """
method = method.upper() # Уточняем значения параметров
data = data or {} headers = self.__prepare_headers(headers or {})
query_params = query_params or {} query_params = query_params or {}
data = data or {}
# Создаем WSGI request в зависимости от метода # Создаем внутренний объект запроса
if method == 'GET': match method:
wsgi_request = self.factory.get(path, data=query_params) case HttpMethod.GET:
elif method == 'POST': wsgi_request = self.factory.get(path, data=query_params, **headers)
wsgi_request = self.factory.post(path, data=data, format='json') case HttpMethod.POST:
elif method == 'PUT': wsgi_request = self.factory.post(path, data=data, format='json', **headers)
wsgi_request = self.factory.put(path, data=data, format='json') case HttpMethod.PUT:
elif method == 'PATCH': wsgi_request = self.factory.put(path, data=data, format='json', **headers)
wsgi_request = self.factory.patch(path, data=data, format='json') case HttpMethod.PATCH:
elif method == 'DELETE': wsgi_request = self.factory.patch(path, data=data, format='json', **headers)
wsgi_request = self.factory.delete(path, data=data, format='json') case HttpMethod.DELETE:
else: wsgi_request = self.factory.delete(path, **headers)
case _:
wsgi_request = self.factory.generic(
method,
path,
content_type='application/json',
data=json.dumps(data) if data else ''
)
wsgi_request = self.factory.generic( # Преобразуем к объекту запроса DRF
method,
path,
content_type='application/json',
data=json.dumps(data) if data else ''
)
# Преобразуем в DRF Request
drf_request = Request(wsgi_request, parsers=[JSONParser()]) drf_request = Request(wsgi_request, parsers=[JSONParser()])
# Добавляем аутентификацию если передана if method != HttpMethod.GET:
if auth: self.__assign_query_params(drf_request, query_params)
drf_request.auth = auth self.__assign_auth(drf_request, auth)
if hasattr(auth, 'account'):
drf_request.user = auth.account
# Добавляем query_params для не-GET запросов
if query_params and method != 'GET':
# noinspection PyTypeChecker
drf_request.query_params._mutable = True
for key, value in query_params.items():
drf_request.query_params[key] = value
drf_request.query_params._mutable = False
return drf_request return drf_request
def create_authenticated_request(self, method, path, token, data=None, query_params=None): def create_authenticated_request(self,
method: HttpMethod,
path: str,
headers: dict[str, str] | None = None,
data: dict[str, Any] | None = None,
query_params: dict[str, str] | None = None,
token: str | None = None) -> Request:
""" """
Создает аутентифицированный DRF Request Создает объект авторизованного запроса для тестирования
:param method: HTTP метод :param method: HTTP метод (GET, POST, PUT, PATCH, DELETE)
:param path: URL путь :param path: URL путь
:param token: Токен сессии :param headers: Заголовки запроса
:param data: Данные для тела запроса :param data: Данные для тела запроса
:param query_params: Параметры запроса :param query_params: Параметры запроса
:return: DRF Request с аутентификацией :param token: Токен сессии авторизации
:return: DRF Request объект
""" """
session = SessionModel.objects.get(token=token) # Получаем сессию
return self.create_request(method, path, data, query_params, auth=session) if token is None:
session = None
else:
session = SessionManager.find_by_token(token)
# Возвращаем объект запроса
return self.create_request(method=method, path=path, headers=headers, data=data, query_params=query_params, auth=session)
@classmethod
def __prepare_headers(cls, headers: dict[str, str]) -> dict[str, str]:
"""
Подготавливаем объект заголовков для передачи в фабрику
:param headers: Заголовки запроса
:return: Объект заголовков для фабрики
"""
wsgi_headers = {}
for key, value in headers.items():
if key.startswith('HTTP_'):
wsgi_headers[key] = value
else:
norm_key = key.upper().replace('-', '_')
if norm_key in ('CONTENT_TYPE', 'CONTENT_LENGTH'):
wsgi_headers[norm_key.lower()] = value
else:
wsgi_headers[f'HTTP_{norm_key}'] = value
return wsgi_headers
@classmethod
def __assign_query_params(cls, request: Request, query_params: dict[str, Any]) -> None:
"""
Присваивает параметры запроса в DRF Request объект
:param request: DRF Request объект
:param query_params: Параметры запроса
"""
# noinspection PyTypeChecker
request.query_params._mutable = True
for key, value in query_params.items():
request.query_params[key] = value
request.query_params._mutable = False
@classmethod
def __assign_auth(cls, request: Request, auth: SessionModel | None) -> None:
"""
Присваивает аутентификацию в DRF Request объект
:param request: Объект запроса
:param auth: Объект сессии пользователя
"""
if auth:
request.auth = auth
if hasattr(auth, 'account'):
request.user = auth.account
@classmethod @classmethod
def get_i_ivanov_account(cls) -> AccountModel: def get_i_ivanov_account(cls) -> AccountModel:
@@ -129,17 +196,26 @@ class CoreTestCase(TestCase):
return SessionTestData.get_i_ivanov_session() return SessionTestData.get_i_ivanov_session()
def create_i_ivanov_request(self, method, path, data=None, query_params=None): def create_i_ivanov_request(self, method: HttpMethod, path: str, headers: dict[str, str] | None = None, data: dict[str, Any] | None = None, query_params: dict[str, Any] | None = None) -> Request:
""" """
Создает запрос от имени Ивана И. Иванова Создает запрос от имени Ивана И. Иванова
:param method: HTTP метод (GET, POST, PUT, PATCH, DELETE)
:param path: URL путь
:param headers: Заголовки запроса
:param data: Данные для тела запроса
:param query_params: Параметры запроса
:return: Объект запроса с аутентификацией Ивана И. Иванова
""" """
return self.create_authenticated_request( return self.create_authenticated_request(
method, method=method,
path, path=path,
SessionTestData.I_IVANOV_SESSION_TOKEN, headers=headers,
data, data=data,
query_params query_params=query_params,
token=SessionTestData.I_IVANOV_SESSION_TOKEN,
) )
@classmethod @classmethod
@@ -161,15 +237,24 @@ class CoreTestCase(TestCase):
return SessionTestData.get_p_petrov_session() return SessionTestData.get_p_petrov_session()
def create_p_petrov_request(self, method, path, data=None, query_params=None): def create_p_petrov_request(self, method: HttpMethod, path: str, headers: dict[str, str] | None = None, data: dict[str, Any] | None = None, query_params: dict[str, Any] | None = None) -> Request:
""" """
Создает запрос от имени Петра П. Петрова Создает запрос от имени Петра П. Петрова
:param method: HTTP метод (GET, POST, PUT, PATCH, DELETE)
:param path: URL путь
:param headers: Заголовки запроса
:param data: Данные для тела запроса
:param query_params: Параметры запроса
:return: Объект запроса с аутентификацией Петра П. Петрова
""" """
return self.create_authenticated_request( return self.create_authenticated_request(
method, method=method,
path, path=path,
SessionTestData.P_PETROV_SESSION_TOKEN, headers=headers,
data, data=data,
query_params query_params=query_params,
token=SessionTestData.P_PETROV_SESSION_TOKEN
) )

View File

@@ -7,7 +7,6 @@
""" """
Описание класса CoreViewSet Описание класса CoreViewSet
""" """
from typing import Any from typing import Any
from typing import Type from typing import Type
from uuid import UUID from uuid import UUID
@@ -59,7 +58,7 @@ class CoreViewSet(ViewSet):
if default is not None: if default is not None:
return default return default
raise AttributeError(f"{self.__class__.__name__} must define 'serializer' attribute") raise ImproperlyConfigured(f"{self.__class__.__name__} must define 'serializer' attribute")
@classmethod @classmethod
def _process_pk_param(cls, **kwargs: Any) -> UUID | None: def _process_pk_param(cls, **kwargs: Any) -> UUID | None:
@@ -134,6 +133,19 @@ class CoreViewSet(ViewSet):
return self._get_request_method() == HttpMethod.PATCH return self._get_request_method() == HttpMethod.PATCH
def _get_request_id(self) -> UUID:
"""
Возвращает идентификатор запроса
:return: Идентификатор запроса
"""
try:
request_id = self.request.headers.get("X-Request-ID", None)
return UUID(request_id)
except ValueError:
return UUID()
def _get_request_method(self) -> str: def _get_request_method(self) -> str:
""" """
Возвращает имя метода HTTP протокола запроса Возвращает имя метода HTTP протокола запроса

View File

@@ -12,4 +12,5 @@ from .account_manager import AccountManager
from .configuration_manager import ConfigurationManager from .configuration_manager import ConfigurationManager
from .crypto_manager import CryptoManager from .crypto_manager import CryptoManager
from .locale_manager import LocaleManager from .locale_manager import LocaleManager
from .request_manager import RequestManager
from .session_manager import SessionManager from .session_manager import SessionManager

View File

@@ -0,0 +1,120 @@
# DRF Rack
# Библиотека классов расширений для Django REST Framework
#
# Автор: Николай В. Анохин <n.anokhin@xeaf.net>
# Все права защищены. Лицензия: MIT
"""
Описание класса RequestManager
"""
import uuid
from typing import Any
from uuid import UUID
from django.core.exceptions import ImproperlyConfigured
from rest_framework.request import Request
from net.xeaf.rack.models import AccountModel
from net.xeaf.rack.models import SessionModel
from net.xeaf.rack.utils.exceptions import BadRequestException
from net.xeaf.rack.utils.exceptions import UnauthorizedException
class RequestManager:
"""
Реализует методы работы с запросами
"""
REQUEST_ID_META = "X-Request-ID"
""" Идентификатор запроса """
@classmethod
def get_request_id(cls, request: Request) -> UUID:
"""
Возвращает идентификатор запроса
:param request: Объект запроса
:return: Идентификатор запроса
"""
request_id = request.headers.get(cls.REQUEST_ID_META, None)
if request_id is None:
return uuid.uuid4()
try:
return UUID(str(request_id))
except ValueError:
raise BadRequestException()
@classmethod
def get_request_method(cls, request: Request) -> str:
"""
Возвращает имя метода HTTP протокола запроса
:param request: Объект запроса
:return: Имя метода HTTP протокола запроса
"""
method = request.method
if not method:
raise ImproperlyConfigured()
return method.upper()
@classmethod
def get_request_session[T: SessionModel](cls, request: Request) -> T | None:
"""
Возвращает сессию, связанную с текущим запросом
:param request: Объект запроса
:return: Сессия, связанная с текущим запросом
"""
# noinspection PyTypeChecker
session: SessionModel | None = request.auth
return session
@classmethod
def get_request_account[T: AccountModel](cls, request: Request) -> T:
"""
Возвращает аккаунт, связанный с текущим запросом
:param request: Объект запроса
:return: Аккаунт, связанный с текущим запросом
"""
session = cls.get_request_session(request)
if session is not None:
# noinspection PyTypeChecker
account: T | None = session.account
if account is not None:
return account
raise UnauthorizedException()
@classmethod
def get_request_context(cls, request: Request) -> dict[str, Any]:
"""
Возвращает контекст запроса
:param request: Объект запроса
:return: Объект контекста запроса
"""
try:
account = cls.get_request_account(request)
except UnauthorizedException:
account = None
return {
"id": cls.get_request_id(request),
"account": account,
"request": request
}

View File

@@ -11,4 +11,11 @@
from .configuration_manager_tests import ConfigurationManagerTests from .configuration_manager_tests import ConfigurationManagerTests
from .crypto_manager_tests import CryptoManagerTests from .crypto_manager_tests import CryptoManagerTests
from .locale_manager_tests import LocaleManagerTests from .locale_manager_tests import LocaleManagerTests
from .request_manager_get_request_account_tests import RequestManagerGetRequestAccountTests
from .request_manager_get_request_context_tests import RequestManagerGetRequestContextTests
from .request_manager_get_request_id_tests import RequestManagerGetRequestIdTests
from .request_manager_get_request_method_tests import RequestManagerGetRequestMethodTests
from .request_manager_get_request_session_tests import RequestManagerGetRequestSessionTests
from .sesson_manager_tests import SessionManagerTests from .sesson_manager_tests import SessionManagerTests

View File

@@ -0,0 +1,71 @@
# DRF Rack
# Библиотека классов расширений для Django REST Framework
#
# Автор: Николай В. Анохин <n.anokhin@xeaf.net>
# Все права защищены. Лицензия: MIT
"""
Описание класса RequestManagerGetRequestAccountTests
"""
from rest_framework.request import Request
from net.xeaf.rack.core.tests import CoreTestCase
from net.xeaf.rack.enums import HttpMethod
from net.xeaf.rack.management.userstory import AccountTestData
from net.xeaf.rack.managers import RequestManager
from net.xeaf.rack.models import SessionModel
from net.xeaf.rack.utils.exceptions import UnauthorizedException
class RequestManagerGetRequestAccountTests(CoreTestCase):
"""
Тесты для RequestManager.get_request_account()
"""
def _create_test_request(self, session: SessionModel | None) -> Request:
"""
Создает запрос для тестирования
:param session: Модель сессии
:return: Объект запроса
"""
return self.create_request(method=HttpMethod.GET, path="/test", auth=session)
def test_get_request_account_authenticated(self):
"""
Получение аккаунта из аутентифицированного запроса
"""
session = self.get_i_ivanov_session()
request = self._create_test_request(session=session)
result = RequestManager.get_request_account(request)
self.assertIsNotNone(result)
self.assertEqual(str(result.id), str(AccountTestData.I_IVANOV_ID))
self.assertEqual(result.username, AccountTestData.I_IVANOV_USERNAME)
def test_get_request_account_unauthenticated_raises_error(self):
"""
Запрос без аутентификации - ошибка UnauthorizedException
"""
request = self._create_test_request(session=None)
with self.assertRaises(UnauthorizedException):
RequestManager.get_request_account(request)
def test_get_request_account_petrov(self):
"""
Получение аккаунта Петрова
"""
session = self.get_p_petrov_session()
request = self._create_test_request(session=session)
result = RequestManager.get_request_account(request)
self.assertIsNotNone(result)
self.assertEqual(str(result.id), str(AccountTestData.P_PETROV_ID))
self.assertEqual(result.username, AccountTestData.P_PETROV_USERNAME)

View File

@@ -0,0 +1,83 @@
# DRF Rack
# Библиотека классов расширений для Django REST Framework
#
# Автор: Николай В. Анохин <n.anokhin@xeaf.net>
# Все права защищены. Лицензия: MIT
"""
Описание класса RequestManagerGetRequestContextTests
"""
from typing import Any
from uuid import UUID
from rest_framework.request import Request
from net.xeaf.rack.core.tests import CoreTestCase
from net.xeaf.rack.enums import HttpMethod
from net.xeaf.rack.management.userstory import AccountTestData
from net.xeaf.rack.managers import RequestManager
from net.xeaf.rack.models import SessionModel
class RequestManagerGetRequestContextTests(CoreTestCase):
"""
Тесты для RequestManager.get_request_session()
"""
def _create_test_request(self, session: SessionModel | None, headers: dict[str, Any] | None = None) -> Request:
"""
Создает запрос для тестирования
:param session: Модель сессии
:param headers: Заголовки запроса
:return: Объект запроса
"""
return self.create_request(method=HttpMethod.GET, path="/test", headers=headers, auth=session)
def test_get_request_context_with_auth(self):
"""
RM-005-1: Контекст с аутентификацией
"""
session = self.get_i_ivanov_session()
request = self._create_test_request(session=session)
result = RequestManager.get_request_context(request)
self.assertIn('id', result)
self.assertIn('account', result)
self.assertIn('request', result)
self.assertIsInstance(result['id'], UUID)
self.assertEqual(str(result['account'].id), str(AccountTestData.I_IVANOV_ID))
self.assertEqual(result['request'], request)
def test_get_request_context_without_auth(self):
"""
Контекст без аутентификации - account = None
"""
request = self._create_test_request(session=None)
result = RequestManager.get_request_context(request)
self.assertIn('id', result)
self.assertIn('account', result)
self.assertIn('request', result)
self.assertIsInstance(result['id'], UUID)
self.assertIsNone(result['account'])
self.assertEqual(result['request'], request)
def test_get_request_context_with_request_id(self):
"""
Контекст с указанным Request-ID
"""
request_id = "bd428358-6944-4657-8044-a3dc1c4dff8a"
request = self._create_test_request(session=None, headers={RequestManager.REQUEST_ID_META: request_id})
result = RequestManager.get_request_context(request)
self.assertIsInstance(result['id'], UUID)
self.assertEqual(str(result['id']), request_id)

View File

@@ -0,0 +1,66 @@
# DRF Rack
# Библиотека классов расширений для Django REST Framework
#
# Автор: Николай В. Анохин <n.anokhin@xeaf.net>
# Все права защищены. Лицензия: MIT
"""
Описание класса RequestManagerGetRequestIdTests
"""
from uuid import UUID
from net.xeaf.rack.core.tests import CoreTestCase
from net.xeaf.rack.enums import HttpMethod
from net.xeaf.rack.managers import RequestManager
from net.xeaf.rack.utils.exceptions import BadRequestException
class RequestManagerGetRequestIdTests(CoreTestCase):
"""
Тесты для RequestManager.get_request_id()
"""
def _create_test_request(self, request_id: str | None = None):
"""
Создание тестового запроса с указанным Request-ID
"""
headers = {RequestManager.REQUEST_ID_META: request_id} if request_id else {}
return self.create_request(
method=HttpMethod.GET,
path="/test",
headers=headers
)
def test_get_request_id_valid(self):
"""
Получение валидного Request-ID из заголовка
"""
request_id = "bd428358-6944-4657-8044-a3dc1c4dff8a"
request = self._create_test_request(request_id)
result = RequestManager.get_request_id(request)
self.assertIsInstance(result, UUID)
self.assertEqual(str(result), request_id)
def test_get_request_id_missing(self):
"""
Заголовок X-Request-ID отсутствует - новый UUID
"""
request = self._create_test_request()
result = RequestManager.get_request_id(request)
self.assertIsInstance(result, UUID)
def test_get_request_id_invalid(self):
"""
Невалидный X-Request-ID - возникает исключение BadRequestException
"""
request = self._create_test_request('invalid-uuid')
with self.assertRaises(BadRequestException):
RequestManager.get_request_id(request)

View File

@@ -0,0 +1,92 @@
# DRF Rack
# Библиотека классов расширений для Django REST Framework
#
# Автор: Николай В. Анохин <n.anokhin@xeaf.net>
# Все права защищены. Лицензия: MIT
"""
Описание класса RequestManagerGetRequestMethodTests
"""
from django.core.exceptions import ImproperlyConfigured
from rest_framework.request import Request
from net.xeaf.rack.core.tests import CoreTestCase
from net.xeaf.rack.enums import HttpMethod
from net.xeaf.rack.managers import RequestManager
class RequestManagerGetRequestMethodTests(CoreTestCase):
"""
Тесты для RequestManager.get_request_method()
"""
def _create_test_request(self, method: HttpMethod) -> Request:
"""
Создает запрос для тестирования
:param method: Метод запроса
:return: Объект запроса
"""
return self.create_request(method=method, path="/test")
def test_get_request_method_get(self):
"""
GET запрос
"""
request = self._create_test_request(HttpMethod.GET)
result = RequestManager.get_request_method(request)
self.assertEqual(result, HttpMethod.GET)
def test_get_request_method_post(self):
"""
POST запрос
"""
request = self._create_test_request(HttpMethod.POST)
result = RequestManager.get_request_method(request)
self.assertEqual(result, HttpMethod.POST)
def test_get_request_method_put(self):
"""
PUT запрос
"""
request = self._create_test_request(HttpMethod.PUT)
result = RequestManager.get_request_method(request)
self.assertEqual(result, HttpMethod.PUT)
def test_get_request_method_patch(self):
"""
PATCH запрос
"""
request = self._create_test_request(HttpMethod.PATCH)
result = RequestManager.get_request_method(request)
self.assertEqual(result, HttpMethod.PATCH)
def test_get_request_method_delete(self):
"""
DELETE запрос
"""
request = self._create_test_request(HttpMethod.DELETE)
result = RequestManager.get_request_method(request)
self.assertEqual(result, HttpMethod.DELETE)
def test_get_request_method_missing_raises_error(self):
"""
Метод отсутствует - ошибка ImproperlyConfigured
"""
request = self._create_test_request(HttpMethod.DELETE)
request.method = None
with self.assertRaises(ImproperlyConfigured):
RequestManager.get_request_method(request)

View File

@@ -0,0 +1,75 @@
# DRF Rack
# Библиотека классов расширений для Django REST Framework
#
# Автор: Николай В. Анохин <n.anokhin@xeaf.net>
# Все права защищены. Лицензия: MIT
"""
Описание класса RequestManagerGetRequestSessionTests
"""
from rest_framework.request import Request
from net.xeaf.rack.core.tests import CoreTestCase
from net.xeaf.rack.enums import HttpMethod
from net.xeaf.rack.management.userstory import AccountTestData
from net.xeaf.rack.management.userstory import SessionTestData
from net.xeaf.rack.managers import RequestManager
from net.xeaf.rack.models import SessionModel
class RequestManagerGetRequestSessionTests(CoreTestCase):
"""
Тесты для RequestManager.get_request_session()
"""
def _create_test_request(self, session: SessionModel | None) -> Request:
"""
Создает запрос для тестирования
:param session: Модель сессии
:return: Объект запроса
"""
return self.create_request(method=HttpMethod.GET, path="/test", auth=session)
# noinspection DuplicatedCode
def test_get_request_session_authenticated(self):
"""
Получение сессии из аутентифицированного запроса
"""
session = self.get_i_ivanov_session()
request = self._create_test_request(session=session)
result = RequestManager.get_request_session(request)
self.assertIsNotNone(result)
self.assertIsInstance(result, SessionModel)
self.assertEqual(result.token, SessionTestData.I_IVANOV_SESSION_TOKEN)
self.assertEqual(str(result.account.id), str(AccountTestData.I_IVANOV_ID))
def test_get_request_session_unauthenticated(self):
"""
Запрос без аутентификации - возвращает None
"""
request = self._create_test_request(session=None)
result = RequestManager.get_request_session(request)
self.assertIsNone(result)
# noinspection DuplicatedCode
def test_get_request_session_petrov(self):
"""
Получение сессии Петрова
"""
session = self.get_p_petrov_session()
request = self._create_test_request(session=session)
result = RequestManager.get_request_session(request)
self.assertIsNotNone(result)
self.assertIsInstance(result, SessionModel)
self.assertEqual(result.token, SessionTestData.P_PETROV_SESSION_TOKEN)
self.assertEqual(str(result.account.id), str(AccountTestData.P_PETROV_ID))