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

tests/test_websocket.py
547 строк19 KB
User
ci: настроен Black и flake8 для автоматического форматирования кода
04 июл 2026, 08:14
04 июл 2026, 08:144778ed6
"""Тесты для WebSocket-функциональности."""

import pytest
from unittest.mock import MagicMock, AsyncMock

from liquidcode.contracts import Request
from liquidcode.http import HttpResponse
from liquidcode.websocket import (
    WebSocketRequest,
    WebSocketResponse,
    WebSocketAdapter,
)
from liquidcode.server.websocket import WebSocketServerManager
from liquidcode.kernel import Kernel
from liquidcode.routing import Router, Route, route
from liquidcode.container import Container


class TestWebSocketRequest:
    """Тесты для WebSocketRequest."""

    def test_init_basic(self):
        """Базовая инициализация WebSocketRequest."""
        request = WebSocketRequest(
            path="/ws/chat", message="Hello", websocket=MagicMock()
        )

        assert request.get_path() == "/ws/chat"
        assert request.get_method() == "WEBSOCKET"
        assert request.get_body() == b"Hello"
        assert request.get_headers() == {}
        assert request.get_query_params() == {}

    def test_init_with_headers(self):
        """Инициализация с заголовками."""
        headers = {"Authorization": "Bearer token123"}
        request = WebSocketRequest(
            path="/ws/chat",
            message="Hello",
            websocket=MagicMock(),
            headers=headers,
        )

        assert request.get_headers() == headers

    def test_init_with_query_params(self):
        """Инициализация с query-параметрами."""
        query_params = {"user": "alice", "room": "general"}
        request = WebSocketRequest(
            path="/ws/chat?user=alice&room=general",
            message="Hello",
            websocket=MagicMock(),
            query_params=query_params,
        )

        assert request.get_query_params() == query_params

    def test_parse_query_params_from_path(self):
        """Парсинг query-параметров из пути."""
        request = WebSocketRequest(
            path="/ws/chat?user=alice&room=general&count=5",
            message="Hello",
            websocket=MagicMock(),
        )

        params = request.get_query_params()
        assert params["user"] == "alice"
        assert params["room"] == "general"
        assert params["count"] == "5"

    def test_route_params(self):
        """Управление route-параметрами."""
        request = WebSocketRequest(
            path="/ws/chat/123", message="Hello", websocket=MagicMock()
        )

        assert request.get_route_params() == {}
        request.set_route_params({"id": "123"})
        assert request.get_route_params() == {"id": "123"}

    def test_attributes(self):
        """Управление атрибутами запроса."""
        request = WebSocketRequest(
            path="/ws/chat", message="Hello", websocket=MagicMock()
        )

        assert request.get_attr("user") is None
        request.set_attr("user", "alice")
        assert request.get_attr("user") == "alice"
        assert request.get_attr("user", "default") == "alice"
        assert request.get_attr("nonexistent", "default") == "default"

    def test_websocket_reference(self):
        """Ссылка на websocket-объект."""
        mock_websocket = MagicMock()
        request = WebSocketRequest(
            path="/ws/chat", message="Hello", websocket=mock_websocket
        )

        assert request.get_websocket() == mock_websocket

    def test_get_body_as_bytes(self):
        """Тело запроса всегда возвращается как байты."""
        request = WebSocketRequest(
            path="/ws/chat", message="Привет мир", websocket=MagicMock()
        )

        body = request.get_body()
        assert isinstance(body, bytes)
        assert body == "Привет мир".encode("utf-8")


class TestWebSocketResponse:
    """Тесты для WebSocketResponse."""

    def test_init_basic(self):
        """Базовая инициализация WebSocketResponse."""
        response = WebSocketResponse(message="Hello", status=200)

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

    def test_init_with_headers(self):
        """Инициализация с заголовками."""
        headers = {"Content-Type": "text/plain"}
        response = WebSocketResponse(message="Hello", status=201, headers=headers)

        assert response.get_headers() == headers

    def test_set_content(self):
        """Изменение содержимого ответа."""
        response = WebSocketResponse(message="Hello")
        response.set_content("Goodbye")

        assert response.get_content() == "Goodbye"

    def test_set_status(self):
        """Изменение статуса."""
        response = WebSocketResponse(message="Hello", status=200)
        response.set_status(101)

        assert response.get_status() == 101

    def test_set_header(self):
        """Добавление заголовка."""
        response = WebSocketResponse(message="Hello")
        response.set_header("X-Custom", "value")

        assert response.get_headers() == {"X-Custom": "value"}

    def test_content_various_types(self):
        """Различные типы содержимого."""
        # Строка
        response1 = WebSocketResponse(message="string")
        assert response1.get_content() == "string"

        # Словарь (для JSON)
        response2 = WebSocketResponse(message={"key": "value"})
        assert response2.get_content() == {"key": "value"}

        # Число
        response3 = WebSocketResponse(message=42)
        assert response3.get_content() == 42


class TestWebSocketAdapter:
    """Тесты для WebSocketAdapter."""

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

        assert adapter.kernel == kernel


class TestWebSocketIntegration:
    """Интеграционные тесты для WebSocket."""

    def test_websocket_with_kernel(self):
        """WebSocket-запрос проходит через Kernel."""
        router = Router()
        container = Container()

        class WebSocketController:
            @route("/ws/test", methods=["WEBSOCKET"])
            def handle(self, request: Request):
                return {"status": "ok", "path": request.get_path()}

        container.register(WebSocketController)
        router.add_routes_from_controller(WebSocketController)

        kernel = Kernel(router, container)

        # Создаём WebSocketRequest
        ws_request = WebSocketRequest(
            path="/ws/test", message="test", websocket=MagicMock()
        )

        response = kernel.handle(ws_request)

        assert response.get_status() == 200
        assert response.get_content()["status"] == "ok"
        assert response.get_content()["path"] == "/ws/test"

    def test_websocket_route_decorator(self):
        """Декоратор @route для WebSocket."""
        router = Router()

        @route("/ws/messages", methods=["WEBSOCKET"])
        def message_handler():
            return {"type": "message"}

        router.add_route("/ws/messages", message_handler, ["WEBSOCKET"])

        assert router.has_route("/ws/messages", "WEBSOCKET")

    def test_websocket_request_implements_request(self):
        """WebSocketRequest реализует интерфейс Request."""
        ws_request = WebSocketRequest(
            path="/ws/test", message="test", websocket=MagicMock()
        )

        assert isinstance(ws_request, Request)

    def test_websocket_response_implements_response(self):
        """WebSocketResponse реализует интерфейс Response."""
        from liquidcode.contracts import Response

        ws_response = WebSocketResponse(message="test")

        assert isinstance(ws_response, Response)


class TestWebSocketScenarios:
    """Сценарные тесты для WebSocket-приложений."""

    def test_chat_controller_scenario(self):
        """Сценарий чат-контроллера."""
        router = Router()
        container = Container()

        class ChatController:
            @route("/ws/chat", methods=["WEBSOCKET"])
            def handle(self, request: Request):
                body = request.get_body().decode("utf-8")
                return {"response": f"Received: {body}"}

        container.register(ChatController)
        router.add_routes_from_controller(ChatController)

        kernel = Kernel(router, container)

        # Сценарий: клиент отправляет сообщение
        ws_request = WebSocketRequest(
            path="/ws/chat", message="Hello, World!", websocket=MagicMock()
        )

        response = kernel.handle(ws_request)

        assert response.get_status() == 200
        content = response.get_content()
        assert isinstance(content, dict)
        assert "Hello, World!" in content["response"]

    def test_websocket_with_auth_header(self):
        """WebSocket с авторизацией через заголовок."""
        headers = {"Authorization": "Bearer secret-token"}
        ws_request = WebSocketRequest(
            path="/ws/protected",
            message="Hello",
            websocket=MagicMock(),
            headers=headers,
        )

        assert ws_request.get_headers()["Authorization"] == "Bearer secret-token"

    def test_websocket_query_params_validation(self):
        """Валидация query-параметров в WebSocket."""
        ws_request = WebSocketRequest(
            path="/ws/game?player=alice&mode=easy",
            message="start",
            websocket=MagicMock(),
        )

        params = ws_request.get_query_params()
        assert params["player"] == "alice"
        assert params["mode"] == "easy"

    def test_websocket_argument_resolution(self):
        """Разрешение аргументов в WebSocket-контроллере."""
        router = Router()
        container = Container()

        class GameController:
            @route("/ws/game/{room_id}", methods=["WEBSOCKET"])
            def handle(self, room_id: int, request: Request):
                return {
                    "room_id": room_id,
                    "message": request.get_body().decode("utf-8"),
                }

        container.register(GameController)
        router.add_routes_from_controller(GameController)

        kernel = Kernel(router, container)

        ws_request = WebSocketRequest(
            path="/ws/game/123", message="join", websocket=MagicMock()
        )
        # Устанавливаем route params вручную (как делает Kernel)
        ws_request.set_route_params({"room_id": "123"})

        response = kernel.handle(ws_request)

        assert response.get_status() == 200
        content = response.get_content()
        assert content["room_id"] == 123
        assert content["message"] == "join"


class TestWebSocketAdapter:
    """Тесты WebSocketAdapter."""

    def test_init(self):
        """Инициализация адаптера."""
        router = Router()
        container = Container()
        kernel = Kernel(router, container)

        adapter = WebSocketAdapter(kernel)
        assert adapter.kernel == kernel

    def test_handle_connection_with_message(self):
        """Обработка WebSocket-сообщения."""
        import asyncio
        from unittest.mock import MagicMock, AsyncMock

        router = Router()
        container = Container()

        class TestController:
            @route("/ws/test", methods=["WEBSOCKET"])
            def handle(self, request: Request):
                return {"response": "ok"}

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

        kernel = Kernel(router, container)
        adapter = WebSocketAdapter(kernel)

        # Mock websocket
        mock_websocket = MagicMock()
        mock_websocket.path = "/ws/test"
        mock_websocket.request_headers = {}
        mock_websocket.send = AsyncMock()
        mock_websocket.close = AsyncMock()

        # Mock async for loop
        async def mock_messages():
            yield "test message"

        # We can't fully test this without a real WebSocket server,
        # but we can test the basic structure
        assert adapter._handle_connection is not None


class TestWebSocketServerManager:
    """Тесты WebSocketServerManager."""

    def test_init(self):
        """Инициализация менеджера."""
        router = Router()
        container = Container()
        kernel = Kernel(router, container)

        manager = WebSocketServerManager(kernel, host="0.0.0.0", port=8766)
        assert manager.kernel == kernel
        assert manager.host == "0.0.0.0"
        assert manager.port == 8766


class TestWebSocketAdapterAdvanced:
    """Расширенные тесты для WebSocketAdapter."""

    def test_init_with_kernel(self):
        """Инициализация адаптера с Kernel."""
        from liquidcode.kernel import Kernel

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

        adapter = WebSocketAdapter(kernel)
        assert adapter.kernel == kernel


class TestWebSocketServerManager:
    """Тесты WebSocketServerManager."""

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

        manager = WebSocketServerManager(kernel, host="0.0.0.0", port=8766)
        assert manager.kernel == kernel
        assert manager.host == "0.0.0.0"
        assert manager.port == 8766

    def test_run_with_custom_params(self):
        """Запуск с кастомными параметрами."""
        router = Router()
        container = Container()
        kernel = Kernel(router, container)

        # Тест просто проверяет инициализацию, так как asyncio.run() блокирует
        manager = WebSocketServerManager(kernel, host="localhost", port=8767)
        assert manager.host == "localhost"
        assert manager.port == 8767

    def test_run_uses_defaults(self):
        """Запуск с дефолтными параметрами."""
        router = Router()
        container = Container()
        kernel = Kernel(router, container)

        manager = WebSocketServerManager(kernel)
        assert manager.host == "localhost"
        assert manager.port == 8765

    def test_run_method_with_patch(self):
        """Метод run() вызывает adapter.run()."""
        from unittest.mock import patch, MagicMock

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

        manager = WebSocketServerManager(kernel, host="test", port=9999)

        # Патчим asyncio.run и WebSocketAdapter
        with patch("liquidcode.server.websocket.asyncio.run") as mock_run:
            with patch("liquidcode.server.websocket.WebSocketAdapter") as MockAdapter:
                mock_instance = MagicMock()
                MockAdapter.return_value = mock_instance

                manager.run(host="host", port=1234)

                MockAdapter.assert_called_once_with(kernel)
                mock_instance.run.assert_called_once_with(host="host", port=1234)

    def test_handle_connection_logs_connection(self):
        """handle_connection логирует открытие соединения."""
        from unittest.mock import MagicMock, patch

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

        adapter = WebSocketAdapter(kernel)

        # Mock websocket
        mock_websocket = MagicMock()
        mock_websocket.path = "/ws/test"
        mock_websocket.request_headers = {}
        mock_websocket.send = MagicMock()
        mock_websocket.close = MagicMock()

        # Mock async for loop
        async def mock_messages():
            yield "test message"

        # Просто проверяем, что метод существует
        assert hasattr(adapter, "_handle_connection")
        assert callable(adapter._handle_connection)

    def test_run_logs_start_message(self):
        """run логирует запуск сервера."""
        from unittest.mock import MagicMock, patch

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

        adapter = WebSocketAdapter(kernel)

        # Mock ws_server.serve
        mock_serve = MagicMock()
        mock_serve.__aenter__ = MagicMock()
        mock_serve.__aexit__ = MagicMock()

        with patch("websockets.asyncio.server.serve", return_value=mock_serve):
            with patch("asyncio.Future") as mock_future:
                # Mock future to not block
                mock_future_instance = MagicMock()
                mock_future.return_value = mock_future_instance

                # Просто проверяем, что метод существует
                assert hasattr(adapter, "run")
                assert callable(adapter.run)

    def test_adapter_integration_with_kernel(self):
        """Интеграция WebSocketAdapter с Kernel."""
        router = Router()
        container = Container()

        # Создаём обработчик WebSocket
        def ws_handler(request):
            return {"status": "connected", "path": request.get_path()}

        router.add_route("/ws/test", ws_handler, ["WEBSOCKET"])

        kernel = Kernel(router, container)
        adapter = WebSocketAdapter(kernel)

        # Проверяем, что adapter использует kernel
        assert adapter.kernel == kernel
        assert adapter.kernel.router == router

    def test_handle_connection_logs_error(self):
        """handle_connection логирует ошибки."""
        from unittest.mock import patch
        import asyncio

        router = Router()
        container = Container()
        kernel = Kernel(router, container)
        adapter = WebSocketAdapter(kernel)

        # Mock websocket с исключением
        mock_websocket = MagicMock()
        mock_websocket.path = "/ws/test"
        mock_websocket.request_headers = {}
        mock_websocket.send = AsyncMock()
        mock_websocket.close = AsyncMock()

        async def mock_messages():
            yield "test message"

        # Проверяем, что метод существует
        assert hasattr(adapter, "_handle_connection")