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

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

import pytest
from unittest.mock import Mock, patch

from liquidcode.base_server import (
    LiquidHTTPServer,
    RequestHandler,
    create_server,
)
from liquidcode.http import HttpRequest, HttpResponse
from liquidcode.kernel import Kernel
from liquidcode.routing import Router
from liquidcode.container import Container


class TestLiquidHTTPServer:
    """Тесты LiquidHTTPServer."""

    def test_init(self):
        """Инициализация сервера."""
        kernel = Mock()
        server = LiquidHTTPServer(("localhost", 8000), Mock, kernel)

        assert server.kernel == kernel


class TestCreateServer:
    """Тесты create_server."""

    def test_create_server(self):
        """Создание сервера."""
        kernel = Mock(spec=Kernel)
        server = create_server("localhost", 8000, kernel)

        assert isinstance(server, LiquidHTTPServer)
        assert server.kernel == kernel

    def test_server_address(self):
        """Адрес сервера."""
        kernel = Mock(spec=Kernel)
        server = create_server("127.0.0.1", 9000, kernel)

        assert server.server_address == ("127.0.0.1", 9000)


class TestServerIntegration:
    """Интеграционные тесты сервера."""

    def test_server_with_kernel(self):
        """Сервер с ядром."""
        router = Router()
        container = Container()
        kernel = Kernel(router, container)

        server = create_server("localhost", 0, kernel)

        assert server.kernel == kernel


class TestRequestHandlerMethods:
    """Тесты методов RequestHandler."""

    def test_log_message(self):
        """Логирование сообщения."""
        handler = Mock(spec=RequestHandler)
        handler.address_string.return_value = "127.0.0.1"

        # Патчим logger напрямую в модуле base_server
        with patch("liquidcode.base_server.logger") as mock_logger:
            RequestHandler.log_message(handler, "Test %s", "message")
            mock_logger.info.assert_called_once_with("127.0.0.1 - Test message")

    def test_do_get(self):
        """GET-запрос вызывает _handle_request."""
        handler = Mock(spec=RequestHandler)
        handler._handle_request = Mock()

        RequestHandler.do_GET(handler)

        handler._handle_request.assert_called_once_with("GET")

    def test_do_post(self):
        """POST-запрос вызывает _handle_request."""
        handler = Mock(spec=RequestHandler)
        handler._handle_request = Mock()

        RequestHandler.do_POST(handler)

        handler._handle_request.assert_called_once_with("POST")

    def test_do_put(self):
        """PUT-запрос вызывает _handle_request."""
        handler = Mock(spec=RequestHandler)
        handler._handle_request = Mock()

        RequestHandler.do_PUT(handler)

        handler._handle_request.assert_called_once_with("PUT")

    def test_do_patch(self):
        """PATCH-запрос вызывает _handle_request."""
        handler = Mock(spec=RequestHandler)
        handler._handle_request = Mock()

        RequestHandler.do_PATCH(handler)

        handler._handle_request.assert_called_once_with("PATCH")

    def test_do_delete(self):
        """DELETE-запрос вызывает _handle_request."""
        handler = Mock(spec=RequestHandler)
        handler._handle_request = Mock()

        RequestHandler.do_DELETE(handler)

        handler._handle_request.assert_called_once_with("DELETE")


class TestServerWithMockKernel:
    """Тесты с mock ядром."""

    def test_handle_request_with_response(self):
        """Обработка запроса с ответом."""
        kernel = Mock(spec=Kernel)
        response = HttpResponse(content={"message": "OK"}, status=200)
        kernel.handle.return_value = response

        server = LiquidHTTPServer(("localhost", 8000), RequestHandler, kernel)

        assert server.kernel == kernel

    def test_handle_request_exception(self):
        """Обработка исключения в _handle_request."""
        kernel = Mock(spec=Kernel)
        kernel.handle.side_effect = Exception("Test error")

        server = LiquidHTTPServer(("localhost", 8000), RequestHandler, kernel)
        handler = Mock()
        handler.headers = {"Content-Length": "0"}
        handler.rfile = Mock(read=Mock(return_value=b""))
        handler.server = server
        handler.path = "/test"
        handler._send_json = Mock()

        RequestHandler._handle_request(handler, "GET")

        # Проверяем, что _send_json вызван с 500 статусом
        handler._send_json.assert_called_once()
        call_args = handler._send_json.call_args
        assert call_args[0][1] == 500

    def test_handle_request_with_connection_error(self):
        """Обработка ConnectionAbortedError при отправке."""
        kernel = Mock(spec=Kernel)
        response = Mock()
        response.get_content.return_value = {"message": "OK"}
        response.get_status.return_value = 200
        response.get_headers.return_value = {}
        kernel.handle.return_value = response

        server = LiquidHTTPServer(("localhost", 8000), RequestHandler, kernel)
        handler = Mock()
        handler.headers = {"Content-Length": "0"}
        handler.rfile = Mock(read=Mock(return_value=b""))
        handler.server = server
        handler.path = "/test"
        handler.send_response = Mock()
        handler.send_header = Mock()
        handler.end_headers = Mock()
        handler.wfile = Mock(
            write=Mock(side_effect=ConnectionAbortedError("Connection closed"))
        )

        # Не должно выбросить исключение
        RequestHandler._handle_request(handler, "GET")

        # Проверяем, что logger.debug был вызван
        handler.server.kernel.terminate.assert_called_once()

    def test_handle_request_with_bytes_content(self):
        """Обработка запроса с бинарным контентом (статический файл)."""
        kernel = Mock(spec=Kernel)
        # Бинарный контент (например, изображение)
        binary_content = b"\x89PNG\r\n\x1a\n"
        response = Mock()
        response.get_content.return_value = binary_content
        response.get_status.return_value = 200
        response.get_headers.return_value = {"Content-Type": "image/png"}
        kernel.handle.return_value = response

        server = LiquidHTTPServer(("localhost", 8000), RequestHandler, kernel)
        handler = Mock()
        handler.headers = {"Content-Length": "0"}
        handler.rfile = Mock(read=Mock(return_value=b""))
        handler.server = server
        handler.path = "/test.png"
        handler.send_response = Mock()
        handler.send_header = Mock()
        handler.end_headers = Mock()
        handler.wfile = Mock(write=Mock(return_value=None))

        RequestHandler._handle_request(handler, "GET")

        # Проверяем, что send_response, send_header, end_headers и write были вызваны
        handler.send_response.assert_called_once_with(200)
        handler.end_headers.assert_called_once()
        handler.wfile.write.assert_called_once_with(binary_content)
        handler.server.kernel.terminate.assert_called_once()

    def test_handle_request_with_bytes_content_connection_error(self):
        """Обработка ConnectionAbortedError при отправке бинарного контента."""
        kernel = Mock(spec=Kernel)
        binary_content = b"\x89PNG\r\n\x1a\n"
        response = Mock()
        response.get_content.return_value = binary_content
        response.get_status.return_value = 200
        response.get_headers.return_value = {"Content-Type": "image/png"}
        kernel.handle.return_value = response

        server = LiquidHTTPServer(("localhost", 8000), RequestHandler, kernel)
        handler = Mock()
        handler.headers = {"Content-Length": "0"}
        handler.rfile = Mock(read=Mock(return_value=b""))
        handler.server = server
        handler.path = "/test.png"
        handler.send_response = Mock()
        handler.send_header = Mock()
        handler.end_headers = Mock()
        handler.wfile = Mock(
            write=Mock(side_effect=ConnectionAbortedError("Connection closed"))
        )

        # Не должно выбросить исключение
        RequestHandler._handle_request(handler, "GET")

        # Проверяем, что terminate был вызван
        handler.server.kernel.terminate.assert_called_once()

    def test_send_json_with_non_serializable_data(self):
        """Обработка несериализуемых данных в _send_json."""
        server = LiquidHTTPServer(("localhost", 8000), RequestHandler, Mock())
        handler = Mock()
        handler.server = server
        handler.send_response = Mock()
        handler.send_header = Mock()
        handler.end_headers = Mock()
        handler.wfile = Mock(write=Mock(return_value=None))

        # Пытаемся сериализовать объект, который не поддерживается JSON
        RequestHandler._send_json(handler, {"data": set([1, 2, 3])}, 500, {})

        # Проверяем, что был вызван json.dumps с ошибкой и возвращен стандартный ответ
        handler.send_response.assert_called_once_with(500)
        handler.end_headers.assert_called_once()

    def test_send_json_with_connection_error(self):
        """Обработка ConnectionAbortedError при отправке JSON."""
        server = LiquidHTTPServer(("localhost", 8000), RequestHandler, Mock())
        handler = Mock()
        handler.server = server
        handler.send_response = Mock()
        handler.send_header = Mock(
            side_effect=ConnectionAbortedError("Connection closed")
        )
        handler.end_headers = Mock()
        handler.wfile = Mock(write=Mock(return_value=None))

        # Не должно выбросить исключение
        RequestHandler._send_json(handler, {"message": "OK"}, 200, {})

        # Проверяем, что send_header был вызван и logger.debug был вызван


class TestHTTPServerManager:
    """Тесты HTTPServerManager."""

    def test_init(self):
        """Инициализация."""
        kernel = Mock(spec=Kernel)
        from liquidcode.server.http import HTTPServerManager

        manager = HTTPServerManager(kernel, host="0.0.0.0", port=8080)
        assert manager.kernel == kernel
        assert manager.host == "0.0.0.0"
        assert manager.port == 8080
        assert manager._debug is False

    def test_set_debug(self):
        """Установка режима отладки."""
        kernel = Mock(spec=Kernel)
        from liquidcode.server.http import HTTPServerManager

        manager = HTTPServerManager(kernel)
        manager.set_debug(True)
        assert manager._debug is True
        manager.set_debug(False)
        assert manager._debug is False

    def test_run_with_defaults(self):
        """Запуск с параметрами по умолчанию."""
        kernel = Mock(spec=Kernel)
        from liquidcode.server.http import HTTPServerManager, logger

        manager = HTTPServerManager(kernel)

        with patch("liquidcode.server.http.create_server") as mock_create:
            # Создаем mock сервер, который поддерживает serve_forever и shutdown
            mock_server = Mock()
            mock_server.serve_forever = Mock(side_effect=KeyboardInterrupt)
            mock_create.return_value = mock_server

            with patch.object(logger, "info") as mock_logger_info:
                # Запуск с исключением KeyboardInterrupt
                try:
                    manager.run()
                except KeyboardInterrupt:
                    pass

                mock_create.assert_called_once_with("localhost", 8000, kernel)
                # Проверяем, что logger.info был вызван
                assert mock_logger_info.called
                # Проверяем, что хотя бы один вызов содержит ожидаемую строку
                call_args_list = [
                    call[0][0] for call in mock_logger_info.call_args_list
                ]
                assert any("Сервер запущен" in args for args in call_args_list)

    def test_run_with_custom_params(self):
        """Запуск с кастомными параметрами."""
        kernel = Mock(spec=Kernel)
        from liquidcode.server.http import HTTPServerManager

        manager = HTTPServerManager(kernel)

        with patch("liquidcode.server.http.create_server") as mock_create:
            mock_server = Mock()
            mock_create.return_value = mock_server

            manager.run(host="0.0.0.0", port=9000)

            mock_create.assert_called_once_with("0.0.0.0", 9000, kernel)