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

tests/test_wsgi.py
279 строк8 KB
User
ci: настроен Black и flake8 для автоматического форматирования кода
04 июл 2026, 08:14
04 июл 2026, 08:144778ed6
"""Тесты для WSGI-адаптера."""

import pytest
from unittest.mock import Mock, patch

from liquidcode.application import Application
from liquidcode.wsgi import WSGIAdapter
from liquidcode.http import HttpRequest, HttpResponse


class TestWSGIAdapter:
    """Тесты WSGIAdapter."""

    def setup_method(self):
        """Создание приложения."""
        self.app = Application()

    def test_init(self):
        """Инициализация адаптера."""
        adapter = WSGIAdapter(self.app)
        assert adapter.app == self.app
        assert adapter.kernel == self.app.kernel

    def test_call_get_request(self):
        """Обработка GET-запроса."""
        adapter = WSGIAdapter(self.app)

        environ = {
            "REQUEST_METHOD": "GET",
            "PATH_INFO": "/test",
            "QUERY_STRING": "",
            "HTTP_HOST": "localhost:8000",
        }
        start_response = Mock()

        result = adapter(environ, start_response)

        start_response.assert_called_once()
        assert len(result) == 1

    def test_call_post_request(self):
        """Обработка POST-запроса."""
        adapter = WSGIAdapter(self.app)

        environ = {
            "REQUEST_METHOD": "POST",
            "PATH_INFO": "/test",
            "QUERY_STRING": "",
            "CONTENT_TYPE": "application/json",
            "CONTENT_LENGTH": "18",
            "wsgi.input": Mock(read=lambda n: b'{"key": "value"}'),
        }
        start_response = Mock()

        result = adapter(environ, start_response)

        start_response.assert_called_once()
        assert len(result) == 1

    def test_call_with_query_params(self):
        """Обработка запроса с query-параметрами."""
        adapter = WSGIAdapter(self.app)

        environ = {
            "REQUEST_METHOD": "GET",
            "PATH_INFO": "/search",
            "QUERY_STRING": "q=test&page=1",
            "HTTP_HOST": "localhost:8000",
        }
        start_response = Mock()

        result = adapter(environ, start_response)

        start_response.assert_called_once()

    def test_call_with_headers(self):
        """Обработка запроса с заголовками."""
        adapter = WSGIAdapter(self.app)

        environ = {
            "REQUEST_METHOD": "GET",
            "PATH_INFO": "/test",
            "QUERY_STRING": "",
            "HTTP_AUTHORIZATION": "Bearer token",
            "HTTP_X_CUSTOM": "value",
        }
        start_response = Mock()

        result = adapter(environ, start_response)

        start_response.assert_called_once()

    def test_call_with_path(self):
        """Обработка запроса по маршруту."""

        @self.app.route("/users/{id}", methods=["GET"])
        def get_user(request, id: int):
            return {"user_id": id}

        adapter = WSGIAdapter(self.app)

        environ = {
            "REQUEST_METHOD": "GET",
            "PATH_INFO": "/users/42",
            "QUERY_STRING": "",
        }
        start_response = Mock()

        result = adapter(environ, start_response)

        start_response.assert_called_once()
        response_body = result[0].decode("utf-8")
        assert "42" in response_body


class TestWSGIExtractHeaders:
    """Тесты _extract_headers."""

    def setup_method(self):
        """Настройка приложения."""
        from liquidcode.application import Application

        self.app = Application()

    def test_basic_headers(self):
        """Базовые заголовки."""
        adapter = WSGIAdapter(self.app)

        environ = {
            "HTTP_HOST": "localhost:8000",
            "HTTP_CONTENT_TYPE": "application/json",
            "HTTP_AUTHORIZATION": "Bearer token",
        }

        headers = adapter._extract_headers(environ)

        assert "Host" in headers
        assert "Content-Type" in headers
        assert "Authorization" in headers

    def test_content_type_header(self):
        """Заголовок CONTENT_TYPE."""
        adapter = WSGIAdapter(self.app)

        environ = {
            "CONTENT_TYPE": "text/html",
            "CONTENT_LENGTH": "100",
        }

        headers = adapter._extract_headers(environ)

        assert "Content-Type" in headers
        assert "Content-Length" in headers


class TestWSGIExtractBody:
    """Тесты _extract_body."""

    def setup_method(self):
        """Настройка приложения."""
        from liquidcode.application import Application

        self.app = Application()

    def test_empty_body(self):
        """Пустое тело запроса."""
        adapter = WSGIAdapter(self.app)

        environ = {
            "CONTENT_LENGTH": "0",
        }

        body = adapter._extract_body(environ)
        assert body == b""

    def test_with_body(self):
        """Запрос с телом."""
        adapter = WSGIAdapter(self.app)

        mock_input = Mock()
        mock_input.read.return_value = b'{"test": "data"}'

        environ = {
            "CONTENT_LENGTH": "18",
            "wsgi.input": mock_input,
        }

        body = adapter._extract_body(environ)
        assert body == b'{"test": "data"}'

    def test_invalid_content_length(self):
        """Невалидная длина контента."""
        adapter = WSGIAdapter(self.app)

        environ = {
            "CONTENT_LENGTH": "invalid",
        }

        body = adapter._extract_body(environ)
        assert body == b""


class TestWSGIStatusText:
    """Тесты _get_status_text."""

    def setup_method(self):
        """Настройка приложения."""
        from liquidcode.application import Application

        self.app = Application()
        self.adapter = WSGIAdapter(self.app)

    def test_status_200(self):
        """Статус 200."""
        assert self.adapter._get_status_text(200) == "OK"

    def test_status_201(self):
        """Статус 201."""
        assert self.adapter._get_status_text(201) == "Created"

    def test_status_404(self):
        """Статус 404."""
        assert self.adapter._get_status_text(404) == "Not Found"

    def test_status_500(self):
        """Статус 500."""
        assert self.adapter._get_status_text(500) == "Internal Server Error"

    def test_unknown_status(self):
        """Неизвестный статус."""
        assert self.adapter._get_status_text(999) == "Unknown Status"


class TestWSGIErrorHandling:
    """Тесты обработки ошибок."""

    def setup_method(self):
        """Настройка приложения."""
        from liquidcode.application import Application

        self.app = Application()

    def test_exception_handling(self):
        """Обработка исключения в контроллере."""

        @self.app.route("/error")
        def error_handler(request):
            raise ValueError("Test error")

        adapter = WSGIAdapter(self.app)

        environ = {
            "REQUEST_METHOD": "GET",
            "PATH_INFO": "/error",
            "QUERY_STRING": "",
        }
        start_response = Mock()

        result = adapter(environ, start_response)

        start_response.assert_called_once()
        response_body = result[0].decode("utf-8")
        assert "error" in response_body.lower()

    def test_kernel_error_handling(self):
        """Обработка ошибки ядра."""
        adapter = WSGIAdapter(self.app)

        environ = {
            "REQUEST_METHOD": "GET",
            "PATH_INFO": "/nonexistent",
            "QUERY_STRING": "",
        }
        start_response = Mock()

        result = adapter(environ, start_response)

        start_response.assert_called_once()
        assert len(result) == 1