tests/test_middleware.py - master - liquid-g/liquid-code

tests/test_middleware.py
429 строк14 KB
User
ci: настроен Black и flake8 для автоматического форматирования кода
04 июл 2026, 08:14
04 июл 2026, 08:144778ed6
"""Тесты для Middleware."""

import pytest
from unittest.mock import Mock, patch

from liquidcode.contracts import Request, Response
from liquidcode.http import HttpRequest, HttpResponse
from liquidcode.middleware import (
    Middleware,
    LoggingMiddleware,
    CORSMiddleware,
    SessionMiddleware,
    SessionStore,
    BodyParserMiddleware,
    ErrorHandlingMiddleware,
)
from liquidcode.template import TemplateManager


class TestMiddlewareBase:
    """Тесты для базового класса Middleware."""

    def test_middleware_call(self):
        """Базовый middleware вызывает call_next."""

        class TestMiddleware(Middleware):
            pass

        call_next = Mock(return_value=HttpResponse())
        request = Mock(spec=Request)

        middleware = TestMiddleware()
        response = middleware(request, call_next)

        call_next.assert_called_once_with(request)
        assert isinstance(response, HttpResponse)


class TestLoggingMiddleware:
    """Тесты для LoggingMiddleware."""

    def test_logging_middleware(self):
        """LoggingMiddleware логирует запрос."""
        request = HttpRequest(
            path="/users",
            method="GET",
        )

        call_next = Mock(return_value=HttpResponse(status=200))

        middleware = LoggingMiddleware()
        response = middleware(request, call_next)

        call_next.assert_called_once_with(request)
        assert isinstance(response, HttpResponse)


class TestCORSMiddleware:
    """Тесты для CORSMiddleware."""

    def test_cors_preflight(self):
        """OPTIONS-запрос возвращает preflight-ответ."""
        request = HttpRequest(
            path="/api",
            method="OPTIONS",
        )

        middleware = CORSMiddleware()
        response = middleware(request, lambda r: HttpResponse())

        assert response.get_status() == 200
        headers = response.get_headers()
        assert "Access-Control-Allow-Origin" in headers
        assert "Access-Control-Allow-Methods" in headers

    def test_cors_adds_headers(self):
        """Добавление CORS-заголовков к обычному ответу."""
        request = HttpRequest(
            path="/api",
            method="GET",
        )

        call_next = Mock(return_value=HttpResponse(content={"data": "test"}))

        middleware = CORSMiddleware()
        response = middleware(request, call_next)

        headers = response.get_headers()
        assert "Access-Control-Allow-Origin" in headers
        assert headers["Access-Control-Allow-Origin"] == "*"

    def test_cors_custom_origin(self):
        """Кастомный allow_origin."""
        request = HttpRequest(
            path="/api",
            method="OPTIONS",
        )

        middleware = CORSMiddleware(allow_origin="https://example.com")
        response = middleware(request, lambda r: HttpResponse())

        assert (
            response.get_headers()["Access-Control-Allow-Origin"]
            == "https://example.com"
        )


class TestSessionMiddleware:
    """Тесты для SessionMiddleware."""

    def test_create_new_session(self):
        """Создание новой сессии при отсутствии куки."""
        request = HttpRequest(
            path="/",
            method="GET",
        )

        store = SessionStore()
        middleware = SessionMiddleware(store=store)

        response = middleware(request, lambda r: HttpResponse())

        # Проверяем, что сессия создана
        assert "Set-Cookie" in response.get_headers()
        session_id = request.get_attr("session_id")
        assert session_id is not None
        assert store.exists(session_id)

    def test_existing_session(self):
        """Восстановление существующей сессии."""
        store = SessionStore()
        session_id = "test_session_123"
        store.set(session_id, {"user": "john"})

        # Создаем валидную подпись
        import hashlib
        import hmac

        secret_key = b"test_key"
        signature = hmac.new(
            secret_key, session_id.encode("utf-8"), hashlib.sha256
        ).hexdigest()
        cookie_value = f"{session_id}.{signature}"

        request = HttpRequest(
            path="/",
            method="GET",
            headers={"Cookie": f"session_id={cookie_value}"},
        )

        middleware = SessionMiddleware(store=store, secret_key="test_key")
        response = middleware(request, lambda r: HttpResponse())

        assert request.get_attr("session") == {"user": "john"}

    def test_session_data_in_request(self):
        """Данные сессии доступны в request."""
        request = HttpRequest(
            path="/",
            method="GET",
        )

        store = SessionStore()
        middleware = SessionMiddleware(store=store)
        middleware(request, lambda r: HttpResponse())

        session = request.get_attr("session")
        assert isinstance(session, dict)


class TestBodyParserMiddleware:
    """Тесты для BodyParserMiddleware."""

    def test_parse_json(self):
        """Парсинг JSON-тела."""
        body = b'{"name": "test", "value": 123}'
        request = HttpRequest(
            path="/api",
            method="POST",
            headers={"Content-Type": "application/json"},
            body=body,
        )

        middleware = BodyParserMiddleware()
        middleware(request, lambda r: HttpResponse())

        parsed = request.get_attr("parsed_body")
        # JSON парсит числа как int, а не str
        assert parsed == {"name": "test", "value": 123}

    def test_parse_form_urlencoded(self):
        """Парсинг form-urlencoded."""
        body = b"name=test&value=123"
        request = HttpRequest(
            path="/api",
            method="POST",
            headers={"Content-Type": "application/x-www-form-urlencoded"},
            body=body,
        )

        middleware = BodyParserMiddleware()
        middleware(request, lambda r: HttpResponse())

        parsed = request.get_attr("parsed_body")
        assert parsed == {"name": "test", "value": "123"}

    def test_invalid_json(self):
        """Некорректный JSON."""
        body = b"invalid json"
        request = HttpRequest(
            path="/api",
            method="POST",
            headers={"Content-Type": "application/json"},
            body=body,
        )

        middleware = BodyParserMiddleware()
        middleware(request, lambda r: HttpResponse())

        parsed = request.get_attr("parsed_body")
        assert parsed is None

    def test_empty_body(self):
        """Пустое тело."""
        request = HttpRequest(
            path="/api",
            method="POST",
        )

        middleware = BodyParserMiddleware()
        middleware(request, lambda r: HttpResponse())

        parsed = request.get_attr("parsed_body")
        assert parsed is None


class TestErrorHandlingMiddleware:
    """Тесты для ErrorHandlingMiddleware."""

    def test_no_exception(self):
        """Без исключений - пропускает запрос."""
        request = HttpRequest(
            path="/users",
            method="GET",
        )

        call_next = Mock(
            return_value=HttpResponse(content={"data": "test"}, status=200)
        )
        middleware = ErrorHandlingMiddleware()
        response = middleware(request, call_next)

        call_next.assert_called_once_with(request)
        assert response.get_status() == 200
        assert response.get_content() == {"data": "test"}

    def test_exception_default_status(self):
        """Исключение без status_code - возвращает 500."""
        request = HttpRequest(
            path="/test",
            method="GET",
        )

        exception = ValueError("Test error")
        call_next = Mock(side_effect=exception)
        middleware = ErrorHandlingMiddleware()
        response = middleware(request, call_next)

        assert response.get_status() == 500

    def test_exception_with_status_code(self):
        """Исключение с кастомным status_code."""

        class CustomError(Exception):
            status_code = 404

        request = HttpRequest(
            path="/not-found",
            method="GET",
        )

        exception = CustomError("Not found")
        call_next = Mock(side_effect=exception)
        middleware = ErrorHandlingMiddleware()
        response = middleware(request, call_next)

        assert response.get_status() == 404

    def test_json_response_when_accept_json(self):
        """Возвращает JSON, если Accept: application/json."""
        request = HttpRequest(
            path="/api",
            method="GET",
            headers={"Accept": "application/json"},
        )

        exception = ValueError("API error")
        call_next = Mock(side_effect=exception)
        middleware = ErrorHandlingMiddleware()
        response = middleware(request, call_next)

        assert response.get_status() == 500
        assert response.get_headers()["Content-Type"] == "application/json"
        content = response.get_content()
        assert content["error"] == "API error"
        assert content["status"] == 500
        assert content["path"] == "/api"

    def test_json_response_when_prefer_json(self):
        """Возвращает JSON, если prefer_json=True."""
        request = HttpRequest(
            path="/api",
            method="GET",
            headers={"Accept": "text/html"},
        )

        exception = ValueError("API error")
        call_next = Mock(side_effect=exception)
        middleware = ErrorHandlingMiddleware(prefer_json=True)
        response = middleware(request, call_next)

        assert response.get_headers()["Content-Type"] == "application/json"
        content = response.get_content()
        assert content["error"] == "API error"

    def test_html_response_with_template_manager(self):
        """Возвращает HTML через template_manager из container."""
        from unittest.mock import MagicMock

        request = HttpRequest(
            path="/error",
            method="GET",
        )

        # Mock container с template_manager
        container = MagicMock()
        template_manager = MagicMock()
        template_manager.render.return_value = "<html>Error 500</html>"
        container.get.return_value = template_manager

        exception = ValueError("Test error")
        call_next = Mock(side_effect=exception)
        middleware = ErrorHandlingMiddleware(container=container)
        response = middleware(request, call_next)

        assert response.get_status() == 500
        assert response.get_headers()["Content-Type"] == "text/html"
        content = response.get_content()
        if isinstance(content, bytes):
            content = content.decode("utf-8")
        assert "<html>Error 500</html>" in content
        container.get.assert_called_once_with(TemplateManager)
        template_manager.render.assert_called_once_with(
            "errors/500.html.j2",
            status=500,
            error="Test error",
            path="/error",
            method="GET",
        )

    def test_html_response_with_template_manager_not_found(self):
        """Если шаблон не найден - встроенная страница."""
        from unittest.mock import MagicMock

        request = HttpRequest(
            path="/error",
            method="GET",
        )

        # Mock container с template_manager, который выбрасывает исключение
        container = MagicMock()
        template_manager = MagicMock()
        template_manager.render.side_effect = Exception("Template not found")
        container.get.return_value = template_manager

        exception = ValueError("Test error")
        call_next = Mock(side_effect=exception)
        middleware = ErrorHandlingMiddleware(container=container)
        response = middleware(request, call_next)

        assert response.get_status() == 500
        assert response.get_headers()["Content-Type"] == "text/html"
        content = response.get_content()
        if isinstance(content, bytes):
            content = content.decode("utf-8")
        assert "Ошибка 500" in content

    def test_html_response_without_template_manager(self):
        """Встроенная HTML-страница без template_manager."""
        request = HttpRequest(
            path="/error",
            method="POST",
        )

        exception = ValueError("Test error message")
        call_next = Mock(side_effect=exception)
        middleware = ErrorHandlingMiddleware()
        response = middleware(request, call_next)

        assert response.get_status() == 500
        assert response.get_headers()["Content-Type"] == "text/html"
        content = response.get_content()
        if isinstance(content, bytes):
            content = content.decode("utf-8")
        assert "Ошибка 500" in content
        assert "Test error message" in content
        assert "POST" in content
        assert "/error" in content

    def test_logs_exception(self):
        """Логирует исключение."""
        request = HttpRequest(
            path="/test",
            method="GET",
        )

        exception = ValueError("Test error")
        call_next = Mock(side_effect=exception)
        middleware = ErrorHandlingMiddleware()

        # Проверяем logger.error (используется по умолчанию)
        with patch("logging.Logger.error") as mock_log:
            middleware(request, call_next)
            mock_log.assert_called_once()
            # Проверяем, что вызов содержит ожидаемый текст
            call_args = mock_log.call_args[0][0]
            assert "Ошибка" in call_args
            assert "GET" in call_args
            assert "/test" in call_args