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

tests/test_websocket_adapter_async.py
214 строк7 KB
User
ci: настроен Black и flake8 для автоматического форматирования кода
04 июл 2026, 08:14
04 июл 2026, 08:144778ed6
"""Асинхронные тесты для WebSocketAdapter."""

import asyncio
import pytest
from unittest.mock import MagicMock, AsyncMock, patch

from liquidcode.kernel import Kernel
from liquidcode.routing import Router, route
from liquidcode.container import Container
from liquidcode.websocket import WebSocketAdapter


class TestWebSocketAdapterAsync:
    """Асинхронные тесты для WebSocketAdapter."""

    @pytest.mark.asyncio
    async def test_handle_connection_basic(self):
        """Базовая обработка WebSocket-соединения."""
        router = Router()
        container = Container()

        class TestController:
            @route("/ws/test", methods=["WEBSOCKET"])
            def handle(self, 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"

        with patch.object(mock_websocket, "__aiter__", return_value=mock_messages()):
            # Call the async method
            task = asyncio.create_task(adapter._handle_connection(mock_websocket))

            # Give it a chance to run
            await asyncio.sleep(0.1)

            # Cancel the task (it would run forever waiting for more messages)
            task.cancel()
            try:
                await task
            except asyncio.CancelledError:
                pass

        # Verify websocket.send was called
        assert mock_websocket.send.called
        mock_websocket.close.assert_called()

    @pytest.mark.asyncio
    async def test_handle_connection_with_multiple_messages(self):
        """Обработка нескольких WebSocket-сообщений."""
        router = Router()
        container = Container()

        call_count = [0]

        class TestController:
            @route("/ws/test", methods=["WEBSOCKET"])
            def handle(self, request):
                call_count[0] += 1
                return {"response": f"message {call_count[0]}"}

        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 with multiple messages
        async def mock_messages():
            yield "message 1"
            yield "message 2"

        with patch.object(mock_websocket, "__aiter__", return_value=mock_messages()):
            task = asyncio.create_task(adapter._handle_connection(mock_websocket))

            await asyncio.sleep(0.2)

            task.cancel()
            try:
                await task
            except asyncio.CancelledError:
                pass

        # Verify both messages were processed
        assert call_count[0] == 2
        assert mock_websocket.send.call_count == 2

    @pytest.mark.asyncio
    async def test_run_method(self):
        """Метод run запускает сервер."""
        router = Router()
        container = Container()
        kernel = Kernel(router, container)
        adapter = WebSocketAdapter(kernel)

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

        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_instance.result = MagicMock(return_value=None)
                mock_future.return_value = mock_future_instance

                # Run the async method
                task = asyncio.create_task(adapter.run(host="localhost", port=8765))

                await asyncio.sleep(0.1)

                task.cancel()
                try:
                    await task
                except asyncio.CancelledError:
                    pass

        # Verify serve was called
        assert mock_serve.__aenter__.called
        mock_serve.__aexit__.assert_called()

    @pytest.mark.asyncio
    async def test_handle_connection_logs_connection(self):
        """handle_connection логирует открытие соединения."""
        import logging
        from websockets.exceptions import ConnectionClosed

        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()

        # Mock async for loop that raises ConnectionClosed
        async def mock_messages():
            yield "test message"
            raise ConnectionClosed()

        with patch.object(mock_websocket, "__aiter__", return_value=mock_messages()):
            task = asyncio.create_task(adapter._handle_connection(mock_websocket))

            await asyncio.sleep(0.1)

            try:
                await task
            except asyncio.CancelledError:
                pass

        # Verify close was called
        mock_websocket.close.assert_called()

    @pytest.mark.asyncio
    async def test_handle_connection_closes_on_error(self):
        """handle_connection закрывает соединение при ошибке."""
        from websockets.exceptions import ConnectionClosed

        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(
            side_effect=ConnectionClosed(rcvd=1000, sent=None)
        )
        mock_websocket.close = AsyncMock()

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

        with patch.object(mock_websocket, "__aiter__", return_value=mock_messages()):
            task = asyncio.create_task(adapter._handle_connection(mock_websocket))

            await asyncio.sleep(0.1)

            try:
                await task
            except asyncio.CancelledError:
                pass

        # Verify close was called even after error
        mock_websocket.close.assert_called()