tests/unit/test_wsgi.py - develop-0.4 - liquid-g/liquid-code

tests/unit/test_wsgi.py
372 строки10 KB
User
0.4.4 - унифицирована загрузка middleware в application.py; дописаны тесты: - Всего тестов: 341 - Общее покрытие: 64%
05 июл 2026, 14:02
05 июл 2026, 14:0227abc28
"""Unit-тесты для WSGIAdapter."""

import pytest
import json
from unittest.mock import Mock, MagicMock

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


class TestWSGIAdapter:
    """Тесты WSGI адаптера."""

    def test_creation(self):
        """Создание WSGIAdapter."""
        app = Application()
        wsgi = WSGIAdapter(app)
        
        assert wsgi.app is app
        assert wsgi.kernel is app.kernel

    def test_basic_get_request(self):
        """Базовый GET запрос."""
        app = Application()
        
        @app.route("/test")
        def handler():
            return {"message": "hello"}
        
        wsgi = WSGIAdapter(app)
        
        environ = {
            "REQUEST_METHOD": "GET",
            "PATH_INFO": "/test",
            "QUERY_STRING": "",
            "HTTP_HOST": "localhost:8000",
            "SERVER_NAME": "localhost",
            "SERVER_PORT": "8000",
            "wsgi.url_scheme": "http",
        }
        
        start_response = Mock()
        
        result = wsgi(environ, start_response)
        
        assert start_response.called
        status, headers = start_response.call_args[0]
        assert status == "200 OK"
        
        content = json.loads(result[0].decode("utf-8"))
        assert content["message"] == "hello"

    def test_post_request_with_body(self):
        """POST запрос с телом."""
        app = Application()
        
        @app.route("/api/test", methods=["POST"])
        def handler():
            return {"received": "ok"}
        
        wsgi = WSGIAdapter(app)
        
        environ = {
            "REQUEST_METHOD": "POST",
            "PATH_INFO": "/api/test",
            "QUERY_STRING": "",
            "CONTENT_TYPE": "application/json",
            "CONTENT_LENGTH": "13",
            "wsgi.input": MagicMock(read=Mock(return_value=b'{"test": 123}')),
        }
        
        start_response = Mock()
        
        result = wsgi(environ, start_response)
        
        assert start_response.called
        status, headers = start_response.call_args[0]
        assert status == "200 OK"

    def test_query_string(self):
        """Запрос с query string."""
        app = Application()
        
        @app.route("/search")
        def handler():
            return {"query": "searched"}
        
        wsgi = WSGIAdapter(app)
        
        environ = {
            "REQUEST_METHOD": "GET",
            "PATH_INFO": "/search",
            "QUERY_STRING": "q=test",
            "HTTP_HOST": "localhost:8000",
        }
        
        start_response = Mock()
        
        result = wsgi(environ, start_response)
        
        assert start_response.called

    def test_headers_extraction(self):
        """Извлечение заголовков."""
        app = Application()
        
        @app.route("/headers")
        def handler():
            return {"ok": True}
        
        wsgi = WSGIAdapter(app)
        
        environ = {
            "REQUEST_METHOD": "GET",
            "PATH_INFO": "/headers",
            "QUERY_STRING": "",
            "HTTP_AUTHORIZATION": "Bearer token",
            "HTTP_X_CUSTOM_HEADER": "value",
            "CONTENT_TYPE": "application/json",
        }
        
        start_response = Mock()
        
        result = wsgi(environ, start_response)
        
        assert start_response.called

    def test_script_name_prefix(self):
        """SCRIPT_NAME префикс."""
        app = Application()
        
        @app.route("/api/test")
        def handler():
            return {"ok": True}
        
        wsgi = WSGIAdapter(app)
        
        environ = {
            "REQUEST_METHOD": "GET",
            "PATH_INFO": "/app/api/test",
            "SCRIPT_NAME": "/app",
            "QUERY_STRING": "",
        }
        
        start_response = Mock()
        
        result = wsgi(environ, start_response)
        
        assert start_response.called

    def test_empty_path_with_script_name(self):
        """Пустой путь после SCRIPT_NAME."""
        app = Application()
        
        @app.route("/")
        def handler():
            return {"ok": True}
        
        wsgi = WSGIAdapter(app)
        
        environ = {
            "REQUEST_METHOD": "GET",
            "PATH_INFO": "/app/",
            "SCRIPT_NAME": "/app",
            "QUERY_STRING": "",
        }
        
        start_response = Mock()
        
        result = wsgi(environ, start_response)
        
        assert start_response.called

    def test_get_status_text(self):
        """Статус тексты."""
        app = Application()
        wsgi = WSGIAdapter(app)
        
        assert wsgi._get_status_text(200) == "OK"
        assert wsgi._get_status_text(201) == "Created"
        assert wsgi._get_status_text(404) == "Not Found"
        assert wsgi._get_status_text(500) == "Internal Server Error"
        assert wsgi._get_status_text(999) == "Unknown Status"

    def test_json_response_content_type(self):
        """JSON ответ с Content-Type."""
        app = Application()
        
        @app.route("/api/json")
        def handler():
            return {"data": "json"}
        
        wsgi = WSGIAdapter(app)
        
        environ = {
            "REQUEST_METHOD": "GET",
            "PATH_INFO": "/api/json",
            "QUERY_STRING": "",
        }
        
        start_response = Mock()
        
        result = wsgi(environ, start_response)
        
        status, headers = start_response.call_args[0]
        content_type_header = [h for h in headers if h[0].lower() == "content-type"]
        assert len(content_type_header) > 0
        assert "application/json" in content_type_header[0][1]

    def test_bytes_response(self):
        """Байтовый ответ."""
        app = Application()
        
        @app.route("/bytes")
        def handler():
            return b"raw bytes", 200, {"Content-Type": "text/plain"}
        
        wsgi = WSGIAdapter(app)
        
        environ = {
            "REQUEST_METHOD": "GET",
            "PATH_INFO": "/bytes",
            "QUERY_STRING": "",
        }
        
        start_response = Mock()
        
        result = wsgi(environ, start_response)
        
        assert start_response.called
        content = result[0]
        assert content == b"raw bytes"

    def test_exception_handling(self):
        """Обработка исключений."""
        app = Application()
        
        @app.route("/error")
        def handler():
            raise ValueError("Test error")
        
        wsgi = WSGIAdapter(app)
        
        environ = {
            "REQUEST_METHOD": "GET",
            "PATH_INFO": "/error",
            "QUERY_STRING": "",
        }
        
        start_response = Mock()
        
        result = wsgi(environ, start_response)
        
        status, headers = start_response.call_args[0]
        assert status == "500 Internal Server Error"
        
        content = json.loads(result[0].decode("utf-8"))
        assert "error" in content

    def test_missing_method(self):
        """Отсутствующий метод в environ."""
        app = Application()
        
        @app.route("/")
        def handler():
            return {"ok": True}
        
        wsgi = WSGIAdapter(app)
        
        environ = {
            "PATH_INFO": "/",
            "QUERY_STRING": "",
        }
        
        start_response = Mock()
        
        result = wsgi(environ, start_response)
        
        assert start_response.called

    def test_missing_path(self):
        """Отсутствующий путь в environ."""
        app = Application()
        
        @app.route("/")
        def handler():
            return {"ok": True}
        
        wsgi = WSGIAdapter(app)
        
        environ = {
            "REQUEST_METHOD": "GET",
            "QUERY_STRING": "",
        }
        
        start_response = Mock()
        
        result = wsgi(environ, start_response)
        
        assert start_response.called

    def test_content_length_error(self):
        """Ошибка чтения CONTENT_LENGTH."""
        app = Application()
        
        @app.route("/")
        def handler():
            return {"ok": True}
        
        wsgi = WSGIAdapter(app)
        
        environ = {
            "REQUEST_METHOD": "GET",
            "PATH_INFO": "/",
            "QUERY_STRING": "",
            "CONTENT_LENGTH": "invalid",
        }
        
        start_response = Mock()
        
        result = wsgi(environ, start_response)
        
        assert start_response.called

    def test_wsgi_input_error(self):
        """Ошибка чтения wsgi.input."""
        app = Application()
        
        @app.route("/")
        def handler():
            return {"ok": True}
        
        wsgi = WSGIAdapter(app)
        
        environ = {
            "REQUEST_METHOD": "GET",
            "PATH_INFO": "/",
            "QUERY_STRING": "",
            "CONTENT_LENGTH": "100",
            "wsgi.input": MagicMock(read=Mock(side_effect=Exception("Read error"))),
        }
        
        start_response = Mock()
        
        result = wsgi(environ, start_response)
        
        assert start_response.called

    def test_make_wsgi_app_factory(self):
        """Фабрика make_wsgi_app."""
        app = Application()
        
        @app.route("/test")
        def handler():
            return {"ok": True}
        
        wsgi_app = make_wsgi_app(app)
        
        assert isinstance(wsgi_app, WSGIAdapter)
        
        environ = {
            "REQUEST_METHOD": "GET",
            "PATH_INFO": "/test",
            "QUERY_STRING": "",
        }
        
        start_response = Mock()
        
        result = wsgi_app(environ, start_response)
        
        assert start_response.called