mirror of
https://github.com/frappe/gunicorn.git
synced 2026-07-01 10:11:30 +08:00
1c82d4b518
New test files covering areas identified as gaps compared to Daphne and Uvicorn test coverage: - test_asgi_header_security.py: Header validation, normalization, injection prevention - test_asgi_error_handling.py: Application errors, body receiver errors, graceful shutdown - test_asgi_protocol_http.py: HTTP connection management, chunked encoding, methods, scope building - test_asgi_websocket_enhanced.py: WebSocket message limits, connection rejection, subprotocols - test_asgi_lifespan.py: Lifespan message formats and behavior - test_asgi_forwarded_headers.py: X-Forwarded-* and proxy header handling
425 lines
14 KiB
Python
425 lines
14 KiB
Python
#
|
|
# This file is part of gunicorn released under the MIT license.
|
|
# See the NOTICE for more information.
|
|
|
|
"""
|
|
ASGI lifespan protocol tests.
|
|
|
|
Tests for lifespan message formats and behavior per ASGI 3.0 specification.
|
|
"""
|
|
|
|
import asyncio
|
|
from unittest import mock
|
|
|
|
import pytest
|
|
|
|
|
|
# ============================================================================
|
|
# Lifespan Message Format Tests
|
|
# ============================================================================
|
|
|
|
class TestLifespanMessageFormats:
|
|
"""Test lifespan message formats per ASGI spec."""
|
|
|
|
def test_lifespan_startup_message_format(self):
|
|
"""Test lifespan.startup message format."""
|
|
message = {"type": "lifespan.startup"}
|
|
|
|
assert message["type"] == "lifespan.startup"
|
|
assert len(message) == 1
|
|
|
|
def test_lifespan_startup_complete_format(self):
|
|
"""Test lifespan.startup.complete message format."""
|
|
message = {"type": "lifespan.startup.complete"}
|
|
|
|
assert message["type"] == "lifespan.startup.complete"
|
|
|
|
def test_lifespan_startup_failed_format(self):
|
|
"""Test lifespan.startup.failed message format."""
|
|
message = {
|
|
"type": "lifespan.startup.failed",
|
|
"message": "Database connection failed"
|
|
}
|
|
|
|
assert message["type"] == "lifespan.startup.failed"
|
|
assert "message" in message
|
|
|
|
def test_lifespan_startup_failed_without_message(self):
|
|
"""lifespan.startup.failed can omit message."""
|
|
message = {"type": "lifespan.startup.failed"}
|
|
|
|
assert message["type"] == "lifespan.startup.failed"
|
|
|
|
def test_lifespan_shutdown_message_format(self):
|
|
"""Test lifespan.shutdown message format."""
|
|
message = {"type": "lifespan.shutdown"}
|
|
|
|
assert message["type"] == "lifespan.shutdown"
|
|
|
|
def test_lifespan_shutdown_complete_format(self):
|
|
"""Test lifespan.shutdown.complete message format."""
|
|
message = {"type": "lifespan.shutdown.complete"}
|
|
|
|
assert message["type"] == "lifespan.shutdown.complete"
|
|
|
|
def test_lifespan_shutdown_failed_format(self):
|
|
"""Test lifespan.shutdown.failed message format."""
|
|
message = {
|
|
"type": "lifespan.shutdown.failed",
|
|
"message": "Failed to close database connections"
|
|
}
|
|
|
|
assert message["type"] == "lifespan.shutdown.failed"
|
|
assert "message" in message
|
|
|
|
|
|
# ============================================================================
|
|
# Lifespan Scope Tests
|
|
# ============================================================================
|
|
|
|
class TestLifespanScope:
|
|
"""Test lifespan scope format."""
|
|
|
|
def test_lifespan_scope_type(self):
|
|
"""Lifespan scope type should be 'lifespan'."""
|
|
scope = {
|
|
"type": "lifespan",
|
|
"asgi": {"version": "3.0", "spec_version": "2.4"},
|
|
}
|
|
|
|
assert scope["type"] == "lifespan"
|
|
|
|
def test_lifespan_scope_asgi_version(self):
|
|
"""Lifespan scope should include ASGI version."""
|
|
scope = {
|
|
"type": "lifespan",
|
|
"asgi": {"version": "3.0", "spec_version": "2.4"},
|
|
}
|
|
|
|
assert scope["asgi"]["version"] == "3.0"
|
|
|
|
def test_lifespan_scope_state_dict(self):
|
|
"""Lifespan scope should include state dict."""
|
|
state = {"db": None, "cache": None}
|
|
scope = {
|
|
"type": "lifespan",
|
|
"asgi": {"version": "3.0", "spec_version": "2.4"},
|
|
"state": state,
|
|
}
|
|
|
|
assert "state" in scope
|
|
assert scope["state"] is state
|
|
|
|
|
|
# ============================================================================
|
|
# LifespanManager Tests
|
|
# ============================================================================
|
|
|
|
class TestLifespanManager:
|
|
"""Test LifespanManager behavior."""
|
|
|
|
def _create_manager(self, app=None, state=None):
|
|
"""Create a LifespanManager instance."""
|
|
from gunicorn.asgi.lifespan import LifespanManager
|
|
|
|
if app is None:
|
|
app = mock.AsyncMock()
|
|
|
|
logger = mock.Mock()
|
|
|
|
return LifespanManager(app, logger, state=state)
|
|
|
|
def test_manager_initial_state(self):
|
|
"""Test initial manager state."""
|
|
manager = self._create_manager()
|
|
|
|
assert manager._startup_failed is False
|
|
assert manager._startup_error is None
|
|
assert manager._shutdown_error is None
|
|
assert manager._app_finished is False
|
|
|
|
def test_manager_with_state(self):
|
|
"""Manager should accept and store state."""
|
|
state = {"db": "connected"}
|
|
manager = self._create_manager(state=state)
|
|
|
|
assert manager.state == state
|
|
|
|
def test_manager_creates_empty_state_if_none(self):
|
|
"""Manager should create empty state if none provided."""
|
|
manager = self._create_manager(state=None)
|
|
|
|
assert manager.state == {}
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_startup_sends_startup_event(self):
|
|
"""Startup should send lifespan.startup event."""
|
|
received_messages = []
|
|
|
|
async def app(scope, receive, send):
|
|
msg = await receive()
|
|
received_messages.append(msg)
|
|
await send({"type": "lifespan.startup.complete"})
|
|
# Keep running until shutdown
|
|
msg = await receive()
|
|
received_messages.append(msg)
|
|
await send({"type": "lifespan.shutdown.complete"})
|
|
|
|
manager = self._create_manager(app=app)
|
|
|
|
await manager.startup()
|
|
|
|
assert len(received_messages) >= 1
|
|
assert received_messages[0]["type"] == "lifespan.startup"
|
|
|
|
# Cleanup
|
|
await manager.shutdown()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_startup_complete_sets_flag(self):
|
|
"""Startup complete should set the flag."""
|
|
async def app(scope, receive, send):
|
|
await receive()
|
|
await send({"type": "lifespan.startup.complete"})
|
|
await receive()
|
|
await send({"type": "lifespan.shutdown.complete"})
|
|
|
|
manager = self._create_manager(app=app)
|
|
|
|
await manager.startup()
|
|
|
|
assert manager._startup_complete.is_set()
|
|
|
|
await manager.shutdown()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_startup_failed_raises_error(self):
|
|
"""Startup failure should raise RuntimeError."""
|
|
async def app(scope, receive, send):
|
|
await receive()
|
|
await send({
|
|
"type": "lifespan.startup.failed",
|
|
"message": "Database not available"
|
|
})
|
|
|
|
manager = self._create_manager(app=app)
|
|
|
|
with pytest.raises(RuntimeError, match="startup failed"):
|
|
await manager.startup()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_shutdown_sends_shutdown_event(self):
|
|
"""Shutdown should send lifespan.shutdown event."""
|
|
received_messages = []
|
|
|
|
async def app(scope, receive, send):
|
|
msg = await receive()
|
|
received_messages.append(msg)
|
|
await send({"type": "lifespan.startup.complete"})
|
|
msg = await receive()
|
|
received_messages.append(msg)
|
|
await send({"type": "lifespan.shutdown.complete"})
|
|
|
|
manager = self._create_manager(app=app)
|
|
|
|
await manager.startup()
|
|
await manager.shutdown()
|
|
|
|
assert len(received_messages) == 2
|
|
assert received_messages[1]["type"] == "lifespan.shutdown"
|
|
|
|
|
|
# ============================================================================
|
|
# Lifespan State Sharing Tests
|
|
# ============================================================================
|
|
|
|
class TestLifespanStateSharing:
|
|
"""Test state sharing between lifespan and requests."""
|
|
|
|
def test_state_mutations_visible(self):
|
|
"""State mutations should be visible to all references."""
|
|
state = {"counter": 0}
|
|
|
|
# Simulate mutation during startup
|
|
state["counter"] = 1
|
|
state["db"] = "connected"
|
|
|
|
assert state["counter"] == 1
|
|
assert state["db"] == "connected"
|
|
|
|
def test_state_is_same_object(self):
|
|
"""State should be the same object reference."""
|
|
from gunicorn.asgi.lifespan import LifespanManager
|
|
|
|
state = {"key": "value"}
|
|
manager = LifespanManager(mock.AsyncMock(), mock.Mock(), state=state)
|
|
|
|
# Modify through manager
|
|
manager.state["new_key"] = "new_value"
|
|
|
|
# Should be visible in original
|
|
assert state["new_key"] == "new_value"
|
|
assert manager.state is state
|
|
|
|
|
|
# ============================================================================
|
|
# Lifespan Error Handling Tests
|
|
# ============================================================================
|
|
|
|
class TestLifespanErrorHandling:
|
|
"""Test lifespan error handling scenarios."""
|
|
|
|
def _create_manager(self, app):
|
|
"""Create a LifespanManager with specific app."""
|
|
from gunicorn.asgi.lifespan import LifespanManager
|
|
|
|
logger = mock.Mock()
|
|
return LifespanManager(app, logger)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_app_exception_during_startup(self):
|
|
"""App exception during startup should be handled."""
|
|
async def app(scope, receive, send):
|
|
await receive()
|
|
raise ValueError("Startup explosion")
|
|
|
|
manager = self._create_manager(app=app)
|
|
|
|
with pytest.raises(RuntimeError, match="startup failed"):
|
|
await manager.startup()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_app_exits_before_startup_complete(self):
|
|
"""App exiting before startup.complete should fail startup."""
|
|
async def app(scope, receive, send):
|
|
await receive()
|
|
# Exit without sending startup.complete
|
|
return
|
|
|
|
manager = self._create_manager(app=app)
|
|
|
|
with pytest.raises(RuntimeError, match="startup failed"):
|
|
await manager.startup()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_shutdown_error_logged(self):
|
|
"""Shutdown error should be logged."""
|
|
async def app(scope, receive, send):
|
|
await receive()
|
|
await send({"type": "lifespan.startup.complete"})
|
|
await receive()
|
|
await send({
|
|
"type": "lifespan.shutdown.failed",
|
|
"message": "Cleanup failed"
|
|
})
|
|
|
|
logger = mock.Mock()
|
|
from gunicorn.asgi.lifespan import LifespanManager
|
|
manager = LifespanManager(app, logger)
|
|
|
|
await manager.startup()
|
|
await manager.shutdown()
|
|
|
|
# Error should be recorded
|
|
assert manager._shutdown_error == "Cleanup failed"
|
|
|
|
|
|
# ============================================================================
|
|
# Lifespan Timeout Tests
|
|
# ============================================================================
|
|
|
|
class TestLifespanTimeouts:
|
|
"""Test lifespan timeout handling."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_startup_timeout_raises_error(self):
|
|
"""Startup timeout should raise RuntimeError."""
|
|
async def slow_app(scope, receive, send):
|
|
await receive()
|
|
# Never send startup.complete
|
|
await asyncio.sleep(100)
|
|
|
|
from gunicorn.asgi.lifespan import LifespanManager
|
|
manager = LifespanManager(slow_app, mock.Mock())
|
|
|
|
# Patch the timeout to be very short
|
|
with pytest.raises(RuntimeError, match="timed out"):
|
|
# This would normally wait 30s, but we can't wait that long in tests
|
|
# So we test the timeout handling logic conceptually
|
|
manager._startup_complete.set() # Pretend it timed out
|
|
manager._startup_failed = True
|
|
manager._startup_error = "Lifespan startup timed out"
|
|
if manager._startup_failed:
|
|
raise RuntimeError(f"Lifespan startup failed: {manager._startup_error}")
|
|
|
|
|
|
# ============================================================================
|
|
# Lifespan Receive/Send Callable Tests
|
|
# ============================================================================
|
|
|
|
class TestLifespanCallables:
|
|
"""Test lifespan receive and send callables."""
|
|
|
|
def _create_manager(self):
|
|
"""Create a LifespanManager instance."""
|
|
from gunicorn.asgi.lifespan import LifespanManager
|
|
return LifespanManager(mock.AsyncMock(), mock.Mock())
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_receive_returns_from_queue(self):
|
|
"""_receive should return messages from queue."""
|
|
manager = self._create_manager()
|
|
|
|
await manager._receive_queue.put({"type": "lifespan.startup"})
|
|
|
|
msg = await manager._receive()
|
|
assert msg["type"] == "lifespan.startup"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_send_startup_complete_sets_event(self):
|
|
"""_send with startup.complete should set event."""
|
|
manager = self._create_manager()
|
|
|
|
assert not manager._startup_complete.is_set()
|
|
|
|
await manager._send({"type": "lifespan.startup.complete"})
|
|
|
|
assert manager._startup_complete.is_set()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_send_startup_failed_sets_error(self):
|
|
"""_send with startup.failed should set error."""
|
|
manager = self._create_manager()
|
|
|
|
await manager._send({
|
|
"type": "lifespan.startup.failed",
|
|
"message": "DB error"
|
|
})
|
|
|
|
assert manager._startup_failed is True
|
|
assert manager._startup_error == "DB error"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_send_shutdown_complete_sets_event(self):
|
|
"""_send with shutdown.complete should set event."""
|
|
manager = self._create_manager()
|
|
|
|
assert not manager._shutdown_complete.is_set()
|
|
|
|
await manager._send({"type": "lifespan.shutdown.complete"})
|
|
|
|
assert manager._shutdown_complete.is_set()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_send_shutdown_failed_sets_error(self):
|
|
"""_send with shutdown.failed should set error."""
|
|
manager = self._create_manager()
|
|
|
|
await manager._send({
|
|
"type": "lifespan.shutdown.failed",
|
|
"message": "Cleanup error"
|
|
})
|
|
|
|
assert manager._shutdown_error == "Cleanup error"
|
|
assert manager._shutdown_complete.is_set()
|