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

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

import pytest
from unittest.mock import Mock

from liquidcode.contracts import Request, Response
from liquidcode.http import HttpRequest, HttpResponse
from liquidcode.kernel import Kernel
from liquidcode.routing import Router, Route, route
from liquidcode.container import Container
from liquidcode.middleware import LoggingMiddleware
from liquidcode.kernel.kernel import HttpNotFound


class TestKernel:
    """Тесты для Kernel."""

    def test_kernel_init(self):
        """Инициализация Kernel."""
        router = Router()
        container = Container()
        kernel = Kernel(router, container)

        assert kernel.router == router
        assert kernel.container == container
        assert kernel.response_class == HttpResponse

    def test_kernel_with_response_class(self):
        """Kernel с кастомным классом ответа."""
        router = Router()
        container = Container()

        class CustomResponse(Response):
            def __init__(self):
                pass

            def get_content(self):
                return None

            def get_status(self):
                return 200

            def get_headers(self):
                return {}

            def set_header(self, key, value):
                pass

            def set_content(self, content):
                pass

            def set_status(self, status):
                pass

        kernel = Kernel(router, container, response_class=CustomResponse)
        assert kernel.response_class == CustomResponse

    def test_handle_controller(self):
        """Обработка запроса к контроллеру."""
        router = Router()
        container = Container()

        class TestController:
            @route("/test", methods=["GET"])
            def index(self):
                return {"message": "Hello"}

        container.register(TestController)
        router.add_routes_from_controller(TestController)

        kernel = Kernel(router, container)

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

        assert response.get_status() == 200
        assert response.get_content() == {"message": "Hello"}

    def test_handle_not_found(self):
        """Маршрут не найден - 404."""
        router = Router()
        container = Container()
        kernel = Kernel(router, container)

        # Добавляем exception hook для обработки HttpNotFound
        def exception_hook(request, e):
            status = getattr(e, "status_code", 500)
            html = f"<html><body><h1>Ошибка {status}</h1></body></html>"
            return HttpResponse(
                content=html.encode("utf-8"),
                status=status,
                headers={"Content-Type": "text/html"},
            )

        kernel.add_exception(exception_hook)

        request = HttpRequest(path="/nonexistent", method="GET")
        response = kernel.handle(request)

        assert response.get_status() == 404

    def test_add_middleware(self):
        """Добавление middleware в цепочку."""
        router = Router()
        container = Container()
        kernel = Kernel(router, container)

        middleware = Mock(return_value=HttpResponse())
        kernel.add_middleware(middleware)

        @route("/test", methods=["GET"])
        def handler():
            return {"test": True}

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

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

        middleware.assert_called_once()

    def test_before_request_hook(self):
        """Хук before_request."""
        router = Router()
        container = Container()
        kernel = Kernel(router, container)

        hook = Mock(return_value=None)
        kernel.add_before_request(hook)

        @route("/test", methods=["GET"])
        def handler():
            return {"test": True}

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

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

        hook.assert_called_once()

    def test_before_request_hook_returns_response(self):
        """before_request возвращает response - прерывает цепочку."""
        router = Router()
        container = Container()
        kernel = Kernel(router, container)

        hook = Mock(return_value=HttpResponse(content="blocked", status=403))
        kernel.add_before_request(hook)

        @route("/test", methods=["GET"])
        def handler():
            return {"test": True}

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

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

        assert response.get_content() == "blocked"
        assert response.get_status() == 403

    def test_after_request_hook(self):
        """Хук after_request."""
        router = Router()
        container = Container()
        kernel = Kernel(router, container)

        modified_response = HttpResponse(content="modified", status=200)
        hook = Mock(return_value=modified_response)
        kernel.add_after_request(hook)

        @route("/test", methods=["GET"])
        def handler():
            return {"test": True}

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

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

        hook.assert_called_once()
        assert response.get_content() == "modified"

    def test_exception_in_controller_returns_500(self):
        """Исключение в контроллере возвращает 500 ответ (а не вызывает exception-хук)."""
        router = Router()
        container = Container()
        kernel = Kernel(router, container)

        hook = Mock(return_value=HttpResponse(content="error", status=500))
        kernel.add_exception(hook)

        @route("/test", methods=["GET"])
        def handler():
            raise ValueError("Test error")

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

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

        # Exception-хук не вызывается, т.к. исключение перехватывается внутри контроллера
        hook.assert_not_called()
        assert response.get_status() == 500
        assert "Ошибка выполнения контроллера" in str(response.get_content())

    def test_exception_hook_called_for_kernel_errors(self):
        """exception-хук вызывается для ошибок вне контроллера."""
        router = Router()
        container = Container()
        kernel = Kernel(router, container)

        hook = Mock(return_value=HttpResponse(content="custom error", status=500))
        kernel.add_exception(hook)

        # Пробросим исключение на более высоком уровне
        def error_middleware(request, call_next):
            raise RuntimeError("Middleware error")

        kernel.add_middleware(error_middleware)

        @route("/test", methods=["GET"])
        def handler():
            return {"test": True}

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

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

        hook.assert_called_once()
        assert response.get_content() == "custom error"

    def test_terminate_hook(self):
        """Хук terminate."""
        router = Router()
        container = Container()
        kernel = Kernel(router, container)

        hook = Mock()
        kernel.add_terminate(hook)

        @route("/test", methods=["GET"])
        def handler():
            return {"test": True}

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

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

        hook.assert_called_once_with(request, response)

    def test_middleware_modifies_response(self):
        """Middleware модифицирует ответ."""
        router = Router()
        container = Container()
        kernel = Kernel(router, container)

        def custom_middleware(request, call_next):
            response = call_next(request)
            response.set_header("X-Custom", "modified")
            return response

        kernel.add_middleware(custom_middleware)

        @route("/test", methods=["GET"])
        def handler():
            return {"test": True}

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

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

        assert response.get_headers()["X-Custom"] == "modified"

    def test_multiple_middlewares(self):
        """Несколько middleware в цепочке."""
        router = Router()
        container = Container()
        kernel = Kernel(router, container)

        call_order = []

        def middleware1(request, call_next):
            call_order.append("mw1-before")
            response = call_next(request)
            call_order.append("mw1-after")
            return response

        def middleware2(request, call_next):
            call_order.append("mw2-before")
            response = call_next(request)
            call_order.append("mw2-after")
            return response

        kernel.add_middleware(middleware1)
        kernel.add_middleware(middleware2)

        @route("/test", methods=["GET"])
        def handler():
            call_order.append("handler")
            return {"test": True}

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

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

        assert call_order == [
            "mw1-before",
            "mw2-before",
            "handler",
            "mw2-after",
            "mw1-after",
        ]

    def test_profiler_enabled(self):
        """Профайлер включен."""
        router = Router()
        container = Container()
        kernel = Kernel(router, container, profiler_enabled=True)

        @route("/test", methods=["GET"])
        def handler():
            return {"test": True}

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

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

        assert response.get_status() == 200
        assert kernel.profiler_enabled is True

    def test_group_middlewares_with_profiler(self):
        """Групповые middleware с profiler."""
        router = Router()
        container = Container()
        kernel = Kernel(router, container, profiler_enabled=True)

        def group_mw(request, call_next):
            response = call_next(request)
            response.set_header("X-Group-MW", "done")
            return response

        @route("/test", methods=["GET"])
        def handler():
            return {"test": True}

        # Добавляем групповые middleware как 4-й аргумент
        router.add_route("/test", handler, ["GET"], [group_mw])

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

        # Проверяем, что group middleware выполнился с profiler
        assert response.get_status() == 200
        assert response.get_headers()["X-Group-MW"] == "done"

    def test_before_request_hook_blocks_request(self):
        """before_request прерывает выполнение."""
        router = Router()
        container = Container()
        kernel = Kernel(router, container)

        hook = Mock(return_value=HttpResponse(content="blocked", status=401))
        kernel.add_before_request(hook)

        @route("/test", methods=["GET"])
        def handler():
            return {"test": True}

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

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

        assert response.get_status() == 401
        assert response.get_content() == "blocked"
        hook.assert_called_once()

    def test_exception_hook_not_found(self):
        """exception-хук не перехватывает HttpNotFound, когда hook не найден."""
        router = Router()
        container = Container()
        kernel = Kernel(router, container)

        # Добавляем exception hook, который не перехватывает HttpNotFound
        def custom_hook(request, e):
            if not isinstance(e, HttpNotFound):
                return HttpResponse(content="Other error", status=500)
            return None  # Не перехватываем HttpNotFound

        kernel.add_exception(custom_hook)

        # Не добавляем маршрут - будет HttpNotFound
        request = HttpRequest(path="/nonexistent", method="GET")

        # Without exception hook, HttpNotFound should propagate
        # But the test expects it to be caught - so we need to handle it
        # Let's test the scenario where hook returns None (doesn't catch)
        with pytest.raises(HttpNotFound):
            kernel.handle(request)

    def test_controller_method_not_found(self):
        """Метод контроллера не найден."""
        router = Router()
        container = Container()
        kernel = Kernel(router, container)

        @route("/test", methods=["GET"])
        def handler():
            return {"test": True}

        router.add_route(
            "/test",
            (type("FakeController", (), {}), "nonexistent_method"),
            ["GET"],
        )

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

        assert response.get_status() == 500
        assert "Метод nonexistent_method не найден" in str(response.get_content())

    def test_exception_hook_handles_http_not_found(self):
        """exception-хук обрабатывает HttpNotFound (404)."""
        router = Router()
        container = Container()
        kernel = Kernel(router, container)

        # Добавляем exception hook для HttpNotFound
        def http_not_found_hook(request, e):
            assert isinstance(e, HttpNotFound)
            return HttpResponse(content="Custom 404", status=404)

        kernel.add_exception(http_not_found_hook)

        # Не добавляем маршрут - будет 404
        request = HttpRequest(path="/nonexistent", method="GET")
        response = kernel.handle(request)

        assert response.get_status() == 404
        assert response.get_content() == "Custom 404"

    def test_group_middlewares_execute(self):
        """Групповые middleware выполняются."""
        router = Router()
        container = Container()
        kernel = Kernel(router, container)

        call_order = []

        def group_mw1(request, call_next):
            call_order.append("group_mw1_before")
            response = call_next(request)
            call_order.append("group_mw1_after")
            return response

        def group_mw2(request, call_next):
            call_order.append("group_mw2_before")
            response = call_next(request)
            call_order.append("group_mw2_after")
            return response

        @route("/test", methods=["GET"])
        def handler():
            call_order.append("handler")
            return {"test": True}

        # Добавляем групповые middleware как 4-й аргумент
        router.add_route("/test", handler, ["GET"], [group_mw1, group_mw2])

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

        assert call_order == [
            "group_mw1_before",
            "group_mw2_before",
            "handler",
            "group_mw2_after",
            "group_mw1_after",
        ]
        assert response.get_status() == 200

    def test_group_middlewares_modify_response(self):
        """Групповые middleware модифицируют ответ."""
        router = Router()
        container = Container()
        kernel = Kernel(router, container)

        def modify_middleware(request, call_next):
            response = call_next(request)
            response.set_header("X-Group-MW", "modified")
            return response

        @route("/test", methods=["GET"])
        def handler():
            return {"test": True}

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

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

        assert response.get_headers()["X-Group-MW"] == "modified"


class TestKernelProfiler:
    """Тесты профайлинга в Kernel."""

    def test_profiler_with_middleware(self):
        """Профайлер с middleware."""
        router = Router()
        container = Container()
        kernel = Kernel(router, container, profiler_enabled=True)

        def test_middleware(request, call_next):
            return call_next(request)

        kernel.add_middleware(test_middleware)

        @route("/test", methods=["GET"])
        def handler():
            return {"test": True}

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

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

        assert response.get_status() == 200