tests/unit/kernel/test_kernel_extra.py - develop-0.4 - liquid-g/liquid-code

tests/unit/kernel/test_kernel_extra.py
374 строки12 KB
User
0.4.20 - refactor: централизация обработки исключений в ErrorHandlingMiddleware (T-003)
07 июл 2026, 22:54
07 июл 2026, 22:5426d8fe2
"""Дополнительные unit-тесты для обновленного kernel.py."""

import pytest
from unittest.mock import Mock, patch

from liquidcode.routing import Router
from liquidcode.container import Container
from liquidcode.kernel import Kernel
from liquidcode.http import HttpResponse
from liquidcode.contracts import Request


class MockRequest(Request):
    """Мок для Request."""
    def __init__(self, path="/test", method="GET", headers=None, body=b"", query_params=None):
        self._path = path
        self._method = method
        self._headers = headers or {}
        self._body = body
        self._query_params = query_params or {}
        self._route_params = {}
        self._attributes = {}

    def get_path(self):
        return self._path

    def get_method(self):
        return self._method

    def get_headers(self):
        return self._headers

    def get_body(self):
        return self._body

    def get_query_params(self):
        return self._query_params

    def get_route_params(self):
        return self._route_params

    def set_route_params(self, params):
        self._route_params = params

    def get_attr(self, key):
        return self._attributes.get(key)

    def set_attr(self, key, value):
        self._attributes[key] = value


class TestKernelOptionsRequest:
    """Тесты для обработки OPTIONS запросов (CORS preflight)."""

    def test_options_request_with_routes(self):
        """OPTIONS запрос с существующими маршрутами."""
        router = Router()
        router.add_route("/users", lambda: None, ["GET", "POST"])
        router.add_route("/posts", lambda: None, ["GET"])

        container = Container()
        kernel = Kernel(router, container)

        request = MockRequest(path="/users", method="OPTIONS")
        response = kernel.handle(request)

        assert response.get_status() == 204
        assert "Allow" in response.get_headers()
        assert "GET" in response.get_headers()["Allow"]
        assert "POST" in response.get_headers()["Allow"]

    def test_options_request_no_routes(self):
        """OPTIONS запрос без маршрутов."""
        router = Router()
        router.add_route("/posts", lambda: None, ["GET"])

        container = Container()
        kernel = Kernel(router, container)

        request = MockRequest(path="/users", method="OPTIONS")
        response = kernel.handle(request)

        assert response.get_status() == 404
        assert "Allow" not in response.get_headers()

    def test_options_request_all_methods(self):
        """OPTIONS запрос с разными методами."""
        router = Router()
        router.add_route("/users", lambda: None, ["GET"])
        router.add_route("/users", lambda: None, ["POST"])
        router.add_route("/users", lambda: None, ["PUT", "DELETE"])

        container = Container()
        kernel = Kernel(router, container)

        request = MockRequest(path="/users", method="OPTIONS")
        response = kernel.handle(request)

        assert response.get_status() == 204
        allow_header = response.get_headers()["Allow"]
        assert "GET" in allow_header
        assert "POST" in allow_header
        assert "PUT" in allow_header
        assert "DELETE" in allow_header


class TestKernelEnableProfiler:
    """Тесты для включения профайлера."""

    def test_enable_profiler(self):
        """Включение профайлера."""
        router = Router()
        container = Container()
        kernel = Kernel(router, container, profiler_enabled=False)

        assert kernel.profiler_enabled is False

        kernel.enable_profiler()
        assert kernel.profiler_enabled is True


class TestKernelProcessRequest:
    """Тесты для _process_request."""

    def test_process_request_with_controller_tuple(self):
        """Обработка с контроллером в виде кортежа."""
        router = Router()

        class MyController:
            def index(self):
                return {"test": True}

        router.add_route("/test", (MyController, "index"), ["GET"])

        container = Container()
        container.register(MyController, MyController)

        kernel = Kernel(router, container)

        request = MockRequest(path="/test", method="GET")
        response = kernel.handle(request)

        assert response.get_status() == 200

    def test_process_request_controller_method_not_found(self):
        """Метод контроллера не найден."""
        from liquidcode.middleware import ErrorHandlingMiddleware
        
        router = Router()
        router.add_route("/test", (object, "nonexistent"), ["GET"])

        container = Container()
        kernel = Kernel(router, container)
        kernel.add_middleware(ErrorHandlingMiddleware(container=container, kernel=kernel))

        request = MockRequest(path="/test", method="GET")
        response = kernel.handle(request)

        assert response.get_status() == 500

    def test_process_request_controller_creation_error(self):
        """Ошибка создания контроллера."""
        from liquidcode.middleware import ErrorHandlingMiddleware
        
        router = Router()

        class BadController:
            def __init__(self):
                raise Exception("Init failed")

        router.add_route("/test", (BadController, "index"), ["GET"])

        container = Container()
        container.register(BadController, BadController)

        kernel = Kernel(router, container)
        kernel.add_middleware(ErrorHandlingMiddleware(container=container, kernel=kernel))

        request = MockRequest(path="/test", method="GET")
        response = kernel.handle(request)

        assert response.get_status() == 500


class TestKernelResponseConversion:
    """Тесты для преобразования ответов."""

    def test_response_with_tuple_2(self):
        """Ответ в виде кортежа (data, status)."""
        router = Router()

        def controller():
            return {"test": True}, 201

        router.add_route("/test", controller, ["GET"])

        container = Container()
        kernel = Kernel(router, container)

        request = MockRequest(path="/test", method="GET")
        response = kernel.handle(request)

        assert response.get_status() == 201

    def test_response_with_tuple_3(self):
        """Ответ в виде кортежа (data, status, headers)."""
        router = Router()

        def controller():
            return {"test": True}, 201, {"X-Custom": "value"}

        router.add_route("/test", controller, ["GET"])

        container = Container()
        kernel = Kernel(router, container)

        request = MockRequest(path="/test", method="GET")
        response = kernel.handle(request)

        assert response.get_status() == 201
        assert response.get_headers().get("X-Custom") == "value"

    def test_response_with_tuple_invalid(self):
        """Невалидный кортеж ответа."""
        router = Router()

        def controller():
            return ("only", "two", "items", "extra")

        router.add_route("/test", controller, ["GET"])

        container = Container()
        kernel = Kernel(router, container)

        request = MockRequest(path="/test", method="GET")
        response = kernel.handle(request)

        assert response.get_status() == 500


class TestKernelControllerMethodInvocation:
    """Тесты для вызова метода контроллера."""

    def test_controller_method_with_args(self):
        """Метод с аргументами из параметров маршрута."""
        router = Router()

        class MyController:
            def get_user(self, id: int):
                return {"id": id}

        router.add_route("/users/{id}", (MyController, "get_user"), ["GET"])

        container = Container()
        container.register(MyController, MyController)

        kernel = Kernel(router, container)

        request = MockRequest(path="/users/123", method="GET")
        response = kernel.handle(request)

        assert response.get_status() == 200


class TestKernelHooks:
    """Тесты для хуков."""

    def test_before_request_hook_can_intercept(self):
        """Хук before_request может перехватить запрос."""
        router = Router()
        router.add_route("/test", lambda: {"test": True}, ["GET"])

        container = Container()
        kernel = Kernel(router, container)

        # Добавляем хук, который возвращает свой ответ
        def before_hook(request):
            if request.get_path() == "/test":
                return HttpResponse(content={"intercepted": True}, status=403)

        kernel.add_before_request(before_hook)

        request = MockRequest(path="/test", method="GET")
        response = kernel.handle(request)

        assert response.get_status() == 403
        data = response.get_content()
        assert data.get("intercepted") is True

    def test_after_request_hook_modifies_response(self):
        """Хук after_request может изменить ответ."""
        router = Router()
        router.add_route("/test", lambda: {"test": True}, ["GET"])

        container = Container()
        kernel = Kernel(router, container)

        # Добавляем хук, который добавляет заголовок
        def after_hook(request, response):
            response.set_header("X-After-Hook", "applied")
            return response

        kernel.add_after_request(after_hook)

        request = MockRequest(path="/test", method="GET")
        response = kernel.handle(request)

        assert response.get_headers().get("X-After-Hook") == "applied"

    def test_exception_hook_can_handle(self):
        """Хук exception может обработать исключение."""
        from liquidcode.middleware import ErrorHandlingMiddleware
        
        router = Router()

        def controller():
            raise ValueError("test error")

        router.add_route("/test", controller, ["GET"])

        container = Container()
        kernel = Kernel(router, container)
        kernel.add_middleware(ErrorHandlingMiddleware(container=container, kernel=kernel))

        # Добавляем хук, который возвращает ответ при ошибке
        def exception_hook(request, exception):
            return HttpResponse(content={"error": str(exception)}, status=500)

        kernel.add_exception(exception_hook)

        request = MockRequest(path="/test", method="GET")
        response = kernel.handle(request)

        assert response.get_status() == 500
        data = response.get_content()
        assert "test error" in data.get("error")


class TestKernelMiddlewareChain:
    """Тесты для цепочки middleware."""

    def test_middleware_execution_order(self):
        """Порядок выполнения middleware."""
        router = Router()
        router.add_route("/test", lambda: {"test": True}, ["GET"])

        container = Container()
        kernel = Kernel(router, container)

        execution_order = []

        def mw1(request, call_next):
            execution_order.append("mw1_before")
            response = call_next(request)
            execution_order.append("mw1_after")
            return response

        def mw2(request, call_next):
            execution_order.append("mw2_before")
            response = call_next(request)
            execution_order.append("mw2_after")
            return response

        kernel.add_middleware(mw1)
        kernel.add_middleware(mw2)

        request = MockRequest(path="/test", method="GET")
        response = kernel.handle(request)

        # Проверяем порядок: сначала mw1, потом mw2, потом контроллер
        assert "mw1_before" in execution_order
        assert "mw2_before" in execution_order
        assert "mw2_after" in execution_order
        assert "mw1_after" in execution_order