patx/micropie
Fix multipart error handling in micropie.py
Commit 55034e9 · patx · 2025-07-04T17:09:20-04:00
Fix multipart error handling in micropie.py Moved MULTIPART_INSTALLED check to _asgi_app_http to send a single 500 response with "500 Internal Server Error" and stop processing, fixing the double-response issue in test_no_multipart_installed.
Comments
No comments yet.
Diff
diff --git a/examples/explicit_routing/README.md b/examples/explicit_routing/README.md
new file mode 100644
index 0000000..3b63093
--- /dev/null
+++ b/examples/explicit_routing/README.md
@@ -0,0 +1,77 @@
+## Explicit Routing with MicroPie
+
+MicroPie now supports explicit routing through the `ExplicitApp` class, `route` and `ws_route` decorators, and router middleware. This allows you to define routes declaratively with type-safe parameters and debug them easily.
+
+### Key Features
+- **Decorators**: Use `@route` for HTTP and `@ws_route` for WebSocket routes.
+- **Type-Safe Parameters**: Support for `int`, `str`, `float`, and `uuid` in path parameters (e.g., `/users/{user}/{age:int}`).
+- **Debugging**: Use `app.list_routes()` to inspect registered routes.
+- **Validation**: HTTP methods and path formats are validated at decoration time.
+- **Subprotocols**: WebSocket routes support optional subprotocols.
+
+### Example Usage
+
+#### HTTP Routing
+```python
+from micropie_routing import ExplicitApp, route
+
+class MyApp(ExplicitApp):
+ @route("/greet/{name}", method=["GET", "POST"])
+ async def greet(self, name: str = "Guest"):
+ return f"Hello, {name}!"
+
+ @route("/user/{id:int}", method="GET")
+ async def get_user(self, id: int):
+ return {"user_id": id}
+
+app = MyApp()
+print(app.list_routes()) # Debug routes
+```
+
+**Access:**
+- `GET /greet/Alice` → `"Hello, Alice!"`
+- `GET /user/123` → `{"user_id": 123}`
+
+#### WebSocket Routing
+```python
+from micropie_routing import ExplicitApp, ws_route
+from micropie import WebSocket
+
+class MyApp(ExplicitApp):
+ @ws_route("/ws/chat/{room}")
+ async def ws_chat(self, ws: WebSocket, room: str):
+ await ws.accept()
+ while True:
+ msg = await ws.receive_text()
+ await ws.send_text(f"Room {room}: {msg}")
+
+app = MyApp()
+```
+
+**Connect:**
+- `ws://127.0.0.1:8000/ws/chat/lobby` → Echoes messages with room prefix.
+
+### Debugging Routes
+Use `app.list_routes()` to inspect registered routes:
+```python
+print(app.list_routes())
+```
+Output:
+```json
+{
+ "http": [
+ {"path": "/greet/{name}", "methods_or_subprotocol": ["GET", "POST"], "handler": "greet"},
+ {"path": "/user/{id:int}", "methods_or_subprotocol": ["GET"], "handler": "get_user"}
+ ],
+ "websocket": [
+ {"path": "/ws/chat/{room}", "methods_or_subprotocol": null, "handler": "ws_chat"}
+ ]
+}
+```
+
+### Notes
+- **Path Syntax**: Use `{name}` for strings or `{name:type}` for `int`, `float`, or `uuid`.
+- **Validation**: Invalid HTTP methods or paths raise `InvalidMethodError` or `InvalidPathError` at decoration time.
+- **Performance**: Regex patterns are pre-compiled for faster matching.
+
+Check the [examples/explicit_routing](https://github.com/patx/micropie/tree/main/examples/explicit_routing) folder for more advanced usage.
diff --git a/examples/explicit_routing/http.py b/examples/explicit_routing/app.py
similarity index 100%
rename from examples/explicit_routing/http.py
rename to examples/explicit_routing/app.py
diff --git a/examples/explicit_routing/micropie_routing.py b/examples/explicit_routing/micropie_routing.py
index 947ddcf..5b4fcb3 100644
--- a/examples/explicit_routing/micropie_routing.py
+++ b/examples/explicit_routing/micropie_routing.py
@@ -1,51 +1,105 @@
import re
from typing import Dict, List, Optional, Tuple, Any, Callable, Type, Union
+import uuid
from micropie import App, HttpMiddleware, WebSocketMiddleware, Request, WebSocketRequest
+# Specific exceptions for better error handling
class RouteError(Exception):
- """Custom exception for route-related errors."""
+ """Base exception for routing errors."""
pass
+class InvalidPathError(RouteError):
+ """Raised for invalid route path formats."""
+ pass
+
+class UnsupportedTypeError(RouteError):
+ """Raised for unsupported parameter types."""
+ pass
+
+class InvalidMethodError(RouteError):
+ """Raised for invalid HTTP methods."""
+ pass
+
+class BaseRouter:
+ """Base class for HTTP and WebSocket routers to share logic."""
+ def __init__(self):
+ # Map route paths to (methods/subprotocol, compiled regex, handler, param_types)
+ self.routes: Dict[str, Tuple[Any, re.Pattern, Callable, List[Type]]] = {}
+ self._param_types = {
+ "int": (int, r"(\d+)"),
+ "str": (str, r"([^/]+)"),
+ "float": (float, r"([-+]?\d*\.?\d+)"),
+ "uuid": (uuid.UUID, r"([0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12})")
+ }
+
+ def _process_param(self, match: re.Match, param_types: List[Type]) -> str:
+ """Process a route parameter and store its type."""
+ param_name = match.group(1)
+ param_type_str = match.group(2) or "str" # Default to str if no type specified
+ if param_type_str not in self._param_types:
+ raise UnsupportedTypeError(f"Unsupported parameter type: {param_type_str}")
+ param_type, regex = self._param_types[param_type_str]
+ param_types.append(param_type)
+ return regex
+
+ def add_route(self, path: str, handler: Callable, methods_or_subprotocol: Any) -> None:
+ """
+ Register a route with its handler and methods/subprotocol.
+
+ Args:
+ path: The route pattern (e.g., "/users/{user}/{record:int}")
+ handler: The handler function
+ methods_or_subprotocol: List of HTTP methods or WebSocket subprotocol
+ """
+ if not path.startswith("/"):
+ raise InvalidPathError(f"Route path must start with '/': {path}")
+ param_types = []
+ # Support both {name} and {name:type} syntax
+ pattern = re.sub(r"{([^:}]*)?(?::([^}]+))?}", lambda m: self._process_param(m, param_types), path)
+ try:
+ compiled_pattern = re.compile(f"^{pattern}$")
+ except re.error as e:
+ raise InvalidPathError(f"Invalid route pattern: {path} ({str(e)})")
+ self.routes[path] = (methods_or_subprotocol, compiled_pattern, handler, param_types)
+
+ def list_routes(self) -> List[Dict[str, Any]]:
+ """
+ Return a list of registered routes for debugging.
+
+ Returns:
+ List of dictionaries containing route details.
+ """
+ return [
+ {"path": path, "methods_or_subprotocol": details[0], "handler": details[2].__name__}
+ for path, details in self.routes.items()
+ ]
+
class ExplicitRouter(HttpMiddleware):
+ """Middleware for explicit HTTP routing with type-safe parameters."""
def __init__(self):
- # Map route paths to (methods, regex pattern, handler_name, param_types)
- self.routes: Dict[str, Tuple[List[str], str, str, List[Type]]] = {}
-
+ super().__init__()
+ self.router = BaseRouter()
+
def add_route(self, path: str, handler: Callable, methods: List[str]) -> None:
"""
- Register an explicit route with its handler and HTTP methods.
+ Register an explicit HTTP route.
Args:
- path: The route pattern (e.g., "/api/users/{user:str}/records/{record:int}")
+ path: The route pattern (e.g., "/users/{user}/{record:int}")
handler: The handler function
methods: List of HTTP methods (e.g., ["GET", "POST"])
"""
if not methods:
- raise RouteError("At least one HTTP method must be specified")
- # Normalize methods to uppercase
+ raise InvalidMethodError("At least one HTTP method must be specified")
+ valid_methods = {"GET", "POST", "PUT", "PATCH", "DELETE", "HEAD", "OPTIONS"}
methods = [m.upper() for m in methods]
- # Parse parameter types from path
- param_types = []
- pattern = re.sub(r"{([^:]+):([^}]+)}", lambda m: self._process_param(m, param_types), path)
- pattern = f"^{pattern}$"
- # Store handler name instead of handler function
- self.routes[path] = (methods, pattern, handler.__name__, param_types)
-
- def _process_param(self, match: re.Match, param_types: List[Type]) -> str:
- """Process a route parameter and store its type."""
- param_name, param_type = match.group(1), match.group(2)
- if param_type == "int":
- param_types.append(int)
- return r"(\d+)"
- elif param_type == "str":
- param_types.append(str)
- return r"([^/]+)"
- else:
- raise RouteError(f"Unsupported parameter type: {param_type}")
-
+ if not all(m in valid_methods for m in methods):
+ raise InvalidMethodError(f"Invalid HTTP methods: {', '.join(set(methods) - valid_methods)}")
+ self.router.add_route(path, handler, methods)
+
async def before_request(self, request: Request) -> Optional[Dict]:
"""
- Match the request path and set path parameters for MicroPie routing.
+ Match the request path and set path parameters.
Args:
request: The MicroPie Request object
@@ -54,25 +108,22 @@ class ExplicitRouter(HttpMiddleware):
Dictionary with response details to short-circuit, or None to continue.
"""
path = request.scope["path"]
-
- for route_path, (methods, pattern, handler_name, param_types) in self.routes.items():
+ for route_path, (methods, pattern, handler, param_types) in self.router.routes.items():
if request.method not in methods:
continue
- match = re.match(pattern, path)
+ match = pattern.match(path)
if match:
try:
- # Convert parameters to their specified types
params = [
param_type(param) for param, param_type in zip(match.groups(), param_types)
]
request.path_params = params
- request._route_handler = handler_name # Set handler name as string
+ request._route_handler = handler.__name__
return None
except ValueError as e:
return {"status_code": 400, "body": f"Invalid parameter format: {str(e)}"}
-
return None
-
+
async def after_request(
self,
request: Request,
@@ -83,40 +134,25 @@ class ExplicitRouter(HttpMiddleware):
return None
class WebSocketExplicitRouter(WebSocketMiddleware):
+ """Middleware for explicit WebSocket routing with type-safe parameters."""
def __init__(self):
- # Map WebSocket route paths to (regex pattern, handler_name, param_types)
- self.routes: Dict[str, Tuple[str, str, List[Type]]] = {}
-
- def add_route(self, path: str, handler: Callable) -> None:
+ super().__init__()
+ self.router = BaseRouter()
+
+ def add_route(self, path: str, handler: Callable, subprotocol: Optional[str] = None) -> None:
"""
- Register an explicit WebSocket route with its handler.
+ Register an explicit WebSocket route.
Args:
- path: The route pattern (e.g., "/ws/users/{user:str}/chat")
+ path: The route pattern (e.g., "/ws/users/{user}/chat")
handler: The handler function
+ subprotocol: Optional WebSocket subprotocol
"""
- # Parse parameter types from path
- param_types = []
- pattern = re.sub(r"{([^:]+):([^}]+)}", lambda m: self._process_param(m, param_types), path)
- pattern = f"^{pattern}$"
- # Store handler name instead of handler function
- self.routes[path] = (pattern, handler.__name__, param_types)
-
- def _process_param(self, match: re.Match, param_types: List[Type]) -> str:
- """Process a route parameter and store its type."""
- param_name, param_type = match.group(1), match.group(2)
- if param_type == "int":
- param_types.append(int)
- return r"(\d+)"
- elif param_type == "str":
- param_types.append(str)
- return r"([^/]+)"
- else:
- raise RouteError(f"Unsupported parameter type: {param_type}")
-
+ self.router.add_route(path, handler, subprotocol)
+
async def before_websocket(self, request: WebSocketRequest) -> Optional[Dict]:
"""
- Match the WebSocket path and set path parameters for routing.
+ Match the WebSocket path and set path parameters.
Args:
request: The WebSocketRequest object
@@ -125,59 +161,95 @@ class WebSocketExplicitRouter(WebSocketMiddleware):
Dictionary with close details to reject, or None to continue.
"""
path = request.scope["path"]
-
- for route_path, (pattern, handler_name, param_types) in self.routes.items():
- match = re.match(pattern, path)
+ for route_path, (subprotocol, pattern, handler, param_types) in self.router.routes.items():
+ match = pattern.match(path)
if match:
try:
- # Convert parameters to their specified types
params = [
param_type(param) for param, param_type in zip(match.groups(), param_types)
]
request.path_params = params
- request._ws_route_handler = handler_name # Set handler name as string
+ request._ws_route_handler = handler.__name__
+ request._ws_subprotocol = subprotocol
return None
except ValueError as e:
return {"code": 1008, "reason": f"Invalid parameter format: {str(e)}"}
-
return None
-
+
async def after_websocket(self, request: WebSocketRequest) -> None:
- """Post-processing after WebSocket handler execution."""
- pass
+ """Log WebSocket session closure for debugging."""
+ print(f"WebSocket session closed for path: {request.scope['path']}")
def route(path: str, method: Union[str, List[str]] = "GET"):
- """Decorator to register a route for an HTTP handler method."""
+ """
+ Decorator to register an HTTP route with validation.
+
+ Args:
+ path: The route path (e.g., "/users/{user}")
+ method: HTTP method(s) as a string or list
+
+ Raises:
+ InvalidMethodError: If invalid HTTP methods are provided
+ InvalidPathError: If the path format is invalid
+ """
def decorator(handler: Callable) -> Callable:
- # Normalize method to a list
methods = [method] if isinstance(method, str) else method
+ valid_methods = {"GET", "POST", "PUT", "PATCH", "DELETE", "HEAD", "OPTIONS"}
+ methods = [m.upper() for m in methods]
+ if not all(m in valid_methods for m in methods):
+ raise InvalidMethodError(f"Invalid HTTP methods: {', '.join(set(methods) - valid_methods)}")
+ if not path.startswith("/"):
+ raise InvalidPathError(f"Route path must start with '/': {path}")
handler._route = (path, methods)
return handler
return decorator
-def ws_route(path: str):
- """Decorator to register a route for a WebSocket handler method."""
+def ws_route(path: str, subprotocol: Optional[str] = None):
+ """
+ Decorator to register a WebSocket route with optional subprotocol.
+
+ Args:
+ path: The WebSocket route path (e.g., "/ws/chat/{room}")
+ subprotocol: Optional WebSocket subprotocol
+
+ Raises:
+ InvalidPathError: If the path format is invalid
+ """
def decorator(handler: Callable) -> Callable:
- handler._ws_route = path
+ if not path.startswith("/"):
+ raise InvalidPathError(f"Route path must start with '/': {path}")
+ handler._ws_route = (path, subprotocol)
return handler
return decorator
class ExplicitApp(App):
- """A subclass of MicroPie.App that automatically registers HTTP and WebSocket routes."""
- def __init__(self):
- super().__init__()
+ """A MicroPie App subclass with explicit routing support."""
+ def __init__(self, session_backend=None):
+ super().__init__(session_backend=session_backend)
self.router = ExplicitRouter()
self.ws_router = WebSocketExplicitRouter()
self.middlewares.append(self.router)
self.ws_middlewares.append(self.ws_router)
self._register_routes()
-
+
def _register_routes(self):
- """Automatically register HTTP and WebSocket routes from decorated methods."""
+ """Register HTTP and WebSocket routes from decorated methods."""
for name, method in self.__class__.__dict__.items():
if hasattr(method, "_route"):
path, methods = method._route
self.router.add_route(path, getattr(self, name), methods)
if hasattr(method, "_ws_route"):
- path = method._ws_route
- self.ws_router.add_route(path, getattr(self, name))
+ path, subprotocol = method._ws_route
+ self.ws_router.add_route(path, getattr(self, name), subprotocol)
+
+ def list_routes(self) -> Dict[str, List[Dict[str, Any]]]:
+ """
+ Return all registered HTTP and WebSocket routes for debugging.
+
+ Returns:
+ Dictionary with 'http' and 'websocket' keys containing route details.
+ """
+ return {
+ "http": self.router.router.list_routes(),
+ "websocket": self.ws_router.router.list_routes()
+ }
diff --git a/examples/middleware/basic.py b/examples/middleware/basic.py
new file mode 100644
index 0000000..713bb1d
--- /dev/null
+++ b/examples/middleware/basic.py
@@ -0,0 +1,16 @@
+from micropie import App, HttpMiddleware
+
+class MiddlewareExample(HttpMiddleware):
+ async def before_request(self, request):
+ print("Hook before request")
+
+ async def after_request(self, request, status_code, response_body, extra_headers):
+ print("Hook after request")
+
+class Root(App):
+ async def index(self):
+ print("Hello, World!")
+ return "Hello, World!"
+
+app = Root()
+app.middlewares.append(MiddlewareExample())
diff --git a/examples/socketio/webtrc/__pycache__/app.cpython-312.pyc b/examples/socketio/webtrc/__pycache__/app.cpython-312.pyc
new file mode 100644
index 0000000..90a7439
Binary files /dev/null and b/examples/socketio/webtrc/__pycache__/app.cpython-312.pyc differ
diff --git a/micropie.py b/micropie.py
index d96f824..5350aa0 100644
--- a/micropie.py
+++ b/micropie.py
@@ -421,6 +421,10 @@ class App:
await self._send_response(send, 400, "400 Bad Request: Bad JSON")
return
elif "multipart/form-data" in content_type:
+ if not MULTIPART_INSTALLED:
+ print("For multipart form data support install 'multipart'.")
+ await self._send_response(send, 500, "500 Internal Server Error")
+ return
if boundary := re.search(r"boundary=([^;]+)", content_type):
reader = asyncio.StreamReader()
reader.feed_data(body_data)
@@ -674,11 +678,6 @@ class App:
Returns:
tuple[dict, dict]: A tuple containing form_data and files.
"""
- if not MULTIPART_INSTALLED:
- print("For multipart form data support install 'multipart'.")
- await self._send_response(None, 500, "500 Internal Server Error")
- return None
-
with PushMultipartParser(boundary) as parser:
form_data: dict = {}
files: dict = {}
@@ -874,7 +873,7 @@ class App:
"""
if not JINJA_INSTALLED:
print("To use the `_render_template` method install 'jinja2'.")
- return "500 Internal Server Error"
+ return "500 Internal Server Error: Jinja2 not installed."
assert self.env is not None
template = await asyncio.to_thread(self.env.get_template, name)
return await template.render_async(**kwargs)
diff --git a/tests.py b/tests.py
index e781fd6..1c26030 100644
--- a/tests.py
+++ b/tests.py
@@ -1,23 +1,34 @@
import asyncio
import unittest
import uuid
+from unittest.mock import AsyncMock, patch
from urllib.parse import parse_qs
-from unittest.mock import AsyncMock
-from micropie import App, InMemorySessionBackend, Request, SESSION_TIMEOUT
+from micropie import App, InMemorySessionBackend, Request, WebSocketRequest, SESSION_TIMEOUT, ConnectionClosed, HttpMiddleware
-class TestMicroPie(unittest.TestCase):
- def setUp(self):
- """Set up the test environment with a new event loop."""
+class MicroPieTestCase(unittest.IsolatedAsyncioTestCase):
+ """Base test case for MicroPie tests with common setup."""
+
+ async def asyncSetUp(self):
+ """Initialize the App instance for each test."""
self.app = App(session_backend=InMemorySessionBackend())
- self.loop = asyncio.new_event_loop()
- asyncio.set_event_loop(self.loop)
- def tearDown(self):
- """Clean up the event loop after each test."""
- self.loop.close()
+ def create_mock_scope(self, path="/index", method="GET", headers=None, query_string=b"", scope_type="http"):
+ """Create a mock ASGI scope for testing."""
+ if headers is None:
+ headers = []
+ return {
+ "type": scope_type,
+ "method": method,
+ "path": path,
+ "headers": headers,
+ "query_string": query_string
+ }
+
+class TestRequest(MicroPieTestCase):
+ """Tests for the Request and WebSocketRequest classes."""
- def test_request_initialization(self):
- """Test Request object initialization."""
+ async def test_request_initialization(self):
+ """Verify that the Request object initializes correctly with scope data."""
scope = {
"type": "http",
"method": "GET",
@@ -26,78 +37,96 @@ class TestMicroPie(unittest.TestCase):
"query_string": b"param1=value1"
}
request = Request(scope)
- # Fix: Explicitly parse query_params in Request initialization
request.query_params = parse_qs(scope.get("query_string", b"").decode("utf-8", "ignore"))
- self.assertEqual(request.method, "GET")
- self.assertEqual(request.headers["host"], "example.com")
- self.assertEqual(request.query_params, {"param1": ["value1"]})
- self.assertEqual(request.session, {})
+ self.assertEqual(request.method, "GET", "Request method should be GET")
+ self.assertEqual(request.headers["host"], "example.com", "Host header should be set")
+ self.assertEqual(request.query_params, {"param1": ["value1"]}, "Query params should be parsed")
+ self.assertEqual(request.session, {}, "Session should be empty initially")
- def test_in_memory_session_backend(self):
+ async def test_websocket_request_initialization(self):
+ """Verify that WebSocketRequest initializes correctly."""
+ scope = {
+ "type": "websocket",
+ "path": "/ws_test",
+ "headers": [(b"host", b"example.com")],
+ "query_string": b"param1=value1"
+ }
+ request = WebSocketRequest(scope)
+ request.query_params = parse_qs(scope.get("query_string", b"").decode("utf-8", "ignore"))
+ self.assertEqual(request.scope["path"], "/ws_test", "WebSocketRequest path should be set")
+ self.assertEqual(request.query_params, {"param1": ["value1"]}, "Query params should be parsed")
+
+class TestSession(MicroPieTestCase):
+ """Tests for session management and cookie parsing."""
+
+ async def test_in_memory_session_backend(self):
"""Test InMemorySessionBackend load and save operations."""
backend = InMemorySessionBackend()
session_id = str(uuid.uuid4())
session_data = {"user_id": "123", "name": "Test User"}
- # Test saving and loading session data
- self.loop.run_until_complete(backend.save(session_id, session_data, SESSION_TIMEOUT))
- loaded_data = self.loop.run_until_complete(backend.load(session_id))
- self.assertEqual(loaded_data, session_data)
+ await backend.save(session_id, session_data, SESSION_TIMEOUT)
+ loaded_data = await backend.load(session_id)
+ self.assertEqual(loaded_data, session_data, "Loaded session data should match saved data")
- # Test session timeout (simulating expired session)
- backend.last_access[session_id] = 0 # Set to far past
- expired_data = self.loop.run_until_complete(backend.load(session_id))
- self.assertEqual(expired_data, {})
+ backend.last_access[session_id] = 0 # Simulate expired session
+ expired_data = await backend.load(session_id)
+ self.assertEqual(expired_data, {}, "Expired session should return empty dict")
- def test_cookie_parsing(self):
- """Test cookie parsing in App."""
+ async def test_cookie_parsing(self):
+ """Test parsing of cookie header."""
cookie_header = "session_id=abc123; theme=dark; user=john"
cookies = self.app._parse_cookies(cookie_header)
self.assertEqual(cookies, {
"session_id": "abc123",
"theme": "dark",
"user": "john"
- })
+ }, "Cookies should be parsed correctly")
+ self.assertEqual(self.app._parse_cookies(""), {}, "Empty cookie header should return empty dict")
- # Test empty cookie header
- self.assertEqual(self.app._parse_cookies(""), {})
+ async def test_session_management(self):
+ """Test session handling in request processing."""
+ async def set_session(self):
+ self.request.session["user"] = "test_user"
+ return 200, "Session set"
- def test_redirect(self):
- """Test redirect response generation."""
- location = "/new-page"
- status_code, body, headers = self.app._redirect(location)
- self.assertEqual(status_code, 302)
- self.assertEqual(body, "")
- self.assertIn(("Location", location), headers)
+ setattr(self.app, "set_session", set_session.__get__(self.app, App))
- # Test with extra headers
- extra_headers = [("X-Custom", "Value")]
- status_code, body, headers = self.app._redirect(location, extra_headers)
- self.assertIn(("X-Custom", "Value"), headers)
+ scope = self.create_mock_scope(path="/set_session")
+ receive = AsyncMock(return_value={"type": "http.request", "body": b"", "more_body": False})
+ send = AsyncMock()
+
+ await self.app(scope, receive, send)
- async def async_test_app_handler(self):
- """Test App handling a simple request."""
- # Define a simple handler in the app
+ set_cookie_call = None
+ for call in send.call_args_list:
+ args = call[0][0]
+ if args["type"] == "http.response.start" and any(h[0] == b"Set-Cookie" for h in args["headers"]):
+ set_cookie_call = args
+ break
+ self.assertIsNotNone(set_cookie_call, "Set-Cookie header not found")
+ self.assertTrue(
+ any(h[0] == b"Set-Cookie" and b"session_id=" in h[1] for h in set_cookie_call["headers"]),
+ "Set-Cookie header with session_id not found"
+ )
+ self.assertEqual(set_cookie_call["status"], 200, "Status should be 200")
+
+class TestRouting(MicroPieTestCase):
+ """Tests for HTTP and WebSocket routing."""
+
+ async def test_app_handler(self):
+ """Test handling of a simple HTTP request with query parameter."""
async def index(self, name="World"):
return 200, f"Hello, {name}!"
setattr(self.app, "index", index.__get__(self.app, App))
- # Mock ASGI scope, receive, and send
- scope = {
- "type": "http",
- "method": "GET",
- "path": "/index",
- "headers": [],
- "query_string": b"name=Test"
- }
+ scope = self.create_mock_scope(path="/index", query_string=b"name=Test")
receive = AsyncMock(return_value={"type": "http.request", "body": b"", "more_body": False})
send = AsyncMock()
- # Run the app
await self.app(scope, receive, send)
- # Verify response
send.assert_any_call({
"type": "http.response.start",
"status": 200,
@@ -109,67 +138,121 @@ class TestMicroPie(unittest.TestCase):
"more_body": False
})
- def test_app_handler(self):
- """Run async test for app handler."""
- self.loop.run_until_complete(self.async_test_app_handler())
+ async def test_404_response(self):
+ """Test 404 response for non-existent route."""
+ scope = self.create_mock_scope(path="/nonexistent")
+ receive = AsyncMock(return_value={"type": "http.request", "body": b"", "more_body": False})
+ send = AsyncMock()
- async def async_test_session_management(self):
- """Test session management in request handling."""
- # Define a handler that uses session
- async def set_session(self):
- self.request.session["user"] = "test_user"
- return 200, "Session set"
+ await self.app(scope, receive, send)
- setattr(self.app, "set_session", set_session.__get__(self.app, App))
+ send.assert_any_call({
+ "type": "http.response.start",
+ "status": 404,
+ "headers": [(b"Content-Type", b"text/html; charset=utf-8")]
+ })
+ send.assert_any_call({
+ "type": "http.response.body",
+ "body": b"404 Not Found",
+ "more_body": False
+ })
- # Mock ASGI scope, receive, and send
- session_id = str(uuid.uuid4())
- scope = {
- "type": "http",
- "method": "GET",
- "path": "/set_session",
- "headers": [], # Remove existing session_id to force Set-Cookie
- "query_string": b""
- }
+ async def test_missing_parameter(self):
+ """Test handler with missing required parameter."""
+ async def index(self, required_param):
+ return "Should not reach here"
+
+ setattr(self.app, "index", index.__get__(self.app, App))
+
+ scope = self.create_mock_scope(path="/index")
receive = AsyncMock(return_value={"type": "http.request", "body": b"", "more_body": False})
send = AsyncMock()
- # Run the app
await self.app(scope, receive, send)
- # Verify session was saved
- # Since no session_id was provided, a new one should have been generated
- session_data = await self.app.session_backend.load(session_id)
- self.assertEqual(session_data, {}) # Session not saved under this ID
+ send.assert_any_call({
+ "type": "http.response.start",
+ "status": 400,
+ "headers": [(b"Content-Type", b"text/html; charset=utf-8")]
+ })
+ send.assert_any_call({
+ "type": "http.response.body",
+ "body": b"400 Bad Request: Missing required parameter 'required_param'",
+ "more_body": False
+ })
- # Verify Set-Cookie header was sent with a new session_id
- calls = send.call_args_list
- set_cookie_call = None
- for call in calls:
- args = call[0][0]
- if args["type"] == "http.response.start" and any(h[0] == b"Set-Cookie" for h in args["headers"]):
- set_cookie_call = args
- break
- self.assertIsNotNone(set_cookie_call, "Set-Cookie header not found")
- self.assertEqual(set_cookie_call["status"], 200)
- self.assertTrue(
- any(h[0] == b"Set-Cookie" and b"session_id=" in h[1] for h in set_cookie_call["headers"]),
- "Set-Cookie header with session_id not found"
- )
+class TestWebSocket(MicroPieTestCase):
+ """Tests for WebSocket handling."""
- def test_session_management(self):
- """Run async test for session management."""
- self.loop.run_until_complete(self.async_test_session_management())
+ async def test_websocket_handler(self):
+ """Test WebSocket connection and message handling."""
+ async def ws_echo(self, ws):
+ await ws.accept()
+ msg = await ws.receive_text()
+ await ws.send_text(f"Echo: {msg}")
+ await ws.close(1000, "Done")
- async def async_test_404_response(self):
- """Test 404 response for unknown route."""
- scope = {
- "type": "http",
- "method": "GET",
- "path": "/nonexistent",
- "headers": [],
- "query_string": b""
- }
+ setattr(self.app, "ws_echo", ws_echo.__get__(self.app, App))
+
+ scope = self.create_mock_scope(path="/echo", scope_type="websocket")
+ receive = AsyncMock(side_effect=[
+ {"type": "websocket.connect"},
+ {"type": "websocket.receive", "text": "Hello"},
+ {"type": "websocket.disconnect", "code": 1000}
+ ])
+ send = AsyncMock()
+
+ await self.app(scope, receive, send)
+
+ send.assert_any_call({
+ "type": "websocket.accept",
+ "subprotocol": None,
+ "headers": []
+ })
+ send.assert_any_call({
+ "type": "websocket.send",
+ "text": "Echo: Hello"
+ })
+ send.assert_any_call({
+ "type": "websocket.close",
+ "code": 1000,
+ "reason": "Done"
+ })
+
+ async def test_websocket_missing_handler(self):
+ """Test WebSocket 1008 response for non-existent route."""
+ scope = self.create_mock_scope(path="/nonexistent", scope_type="websocket")
+ receive = AsyncMock(return_value={"type": "websocket.connect"})
+ send = AsyncMock()
+
+ await self.app(scope, receive, send)
+
+ send.assert_any_call({
+ "type": "websocket.close",
+ "code": 1008,
+ "reason": "No matching WebSocket route"
+ })
+
+class TestMiddleware(MicroPieTestCase):
+ """Tests for HTTP and WebSocket middleware."""
+
+ async def test_http_middleware(self):
+ """Test HTTP middleware before and after request."""
+ class TestMiddleware(HttpMiddleware):
+ async def before_request(self, request):
+ request.custom_data = "set_by_middleware"
+ return None
+ async def after_request(self, request, status_code, response_body, extra_headers):
+ return {"status_code": 201, "body": f"{response_body} + middleware", "headers": extra_headers}
+
+ self.app.middlewares.append(TestMiddleware())
+
+ async def index(self):
+ return f"Data: {self.request.custom_data}"
+
+ setattr(self.app, "index", index.__get__(self.app, App))
+
+ scope = self.create_mock_scope(path="/index")
receive = AsyncMock(return_value={"type": "http.request", "body": b"", "more_body": False})
send = AsyncMock()
@@ -177,18 +260,167 @@ class TestMicroPie(unittest.TestCase):
send.assert_any_call({
"type": "http.response.start",
- "status": 404,
+ "status": 201,
"headers": [(b"Content-Type", b"text/html; charset=utf-8")]
})
send.assert_any_call({
"type": "http.response.body",
- "body": b"404 Not Found",
+ "body": b"Data: set_by_middleware + middleware",
"more_body": False
})
- def test_404_response(self):
- """Run async test for 404 response."""
- self.loop.run_until_complete(self.async_test_404_response())
+class TestResponseHandling(MicroPieTestCase):
+ """Tests for response handling and edge cases."""
+
+ async def test_json_handling(self):
+ """Test JSON request and response handling."""
+ async def json_handler(self):
+ return self.request.get_json
+
+ setattr(self.app, "json_handler", json_handler.__get__(self.app, App))
+
+ scope = self.create_mock_scope(
+ path="/json_handler",
+ method="POST",
+ headers=[(b"content-type", b"application/json")]
+ )
+ receive = AsyncMock(return_value={"type": "http.request", "body": b'{"key": "value"}', "more_body": False})
+ send = AsyncMock()
+
+ with patch("micropie.json") as mock_json:
+ mock_json.loads.return_value = {"key": "value"}
+ mock_json.dumps.return_value = b'{"key": "value"}'
+
+ await self.app(scope, receive, send)
+
+ mock_json.loads.assert_called_once()
+ mock_json.dumps.assert_called_once()
+ send.assert_any_call({
+ "type": "http.response.start",
+ "status": 200,
+ "headers": [(b"Content-Type", b"application/json")]
+ })
+ send.assert_any_call({
+ "type": "http.response.body",
+ "body": b'{"key": "value"}',
+ "more_body": False
+ })
+
+ async def test_invalid_json(self):
+ """Test handling of invalid JSON in POST request."""
+ scope = self.create_mock_scope(
+ path="/index",
+ method="POST",
+ headers=[(b"content-type", b"application/json")]
+ )
+ receive = AsyncMock(return_value={"type": "http.request", "body": b"{invalid}", "more_body": False})
+ send = AsyncMock()
+
+ await self.app(scope, receive, send)
+
+ send.assert_any_call({
+ "type": "http.response.start",
+ "status": 400,
+ "headers": [(b"Content-Type", b"text/html; charset=utf-8")]
+ })
+ send.assert_any_call({
+ "type": "http.response.body",
+ "body": b"400 Bad Request: Bad JSON",
+ "more_body": False
+ })
+
+ async def test_header_injection(self):
+ """Test protection against header injection."""
+ async def index(self):
+ return 200, "Test", [("Bad-Header", "value\r\nInject: malicious")]
+
+ setattr(self.app, "index", index.__get__(self.app, App))
+
+ scope = self.create_mock_scope(path="/index")
+ receive = AsyncMock(return_value={"type": "http.request", "body": b"", "more_body": False})
+ send = AsyncMock()
+
+ await self.app(scope, receive, send)
+
+ start_call = None
+ for call in send.call_args_list:
+ args = call[0][0]
+ if args["type"] == "http.response.start":
+ start_call = args
+ break
+ self.assertIsNotNone(start_call, "Response start call not found")
+ self.assertEqual(start_call["status"], 200, "Status should be 200")
+ self.assertEqual(
+ start_call["headers"],
+ [(b"Content-Type", b"text/html; charset=utf-8")],
+ "Malicious header should be filtered out"
+ )
+ send.assert_any_call({
+ "type": "http.response.body",
+ "body": b"Test",
+ "more_body": False
+ })
+
+ async def test_redirect(self):
+ """Test redirect response generation."""
+ location = "/new-page"
+ extra_headers = [("X-Custom", "Value")]
+ status_code, body, headers = self.app._redirect(location, extra_headers)
+ self.assertEqual(status_code, 302, "Redirect should return 302 status")
+ self.assertEqual(body, "", "Redirect body should be empty")
+ self.assertIn(("Location", location), headers, "Location header should be set")
+ self.assertIn(("X-Custom", "Value"), headers, "Extra headers should be included")
+
+class TestOptionalDependencies(MicroPieTestCase):
+ """Tests for behavior with missing optional dependencies."""
+
+ async def test_no_multipart_installed(self):
+ """Test behavior when multipart is not installed."""
+ with patch("micropie.MULTIPART_INSTALLED", False):
+ scope = self.create_mock_scope(
+ path="/index",
+ method="POST",
+ headers=[(b"content-type", b"multipart/form-data; boundary=----boundary")]
+ )
+ receive = AsyncMock(return_value={"type": "http.request", "body": b"", "more_body": False})
+ send = AsyncMock()
+
+ await self.app(scope, receive, send)
+
+ send.assert_any_call({
+ "type": "http.response.start",
+ "status": 500,
+ "headers": [(b"Content-Type", b"text/html; charset=utf-8")]
+ })
+ send.assert_any_call({
+ "type": "http.response.body",
+ "body": b"500 Internal Server Error",
+ "more_body": False
+ })
+
+ async def test_no_jinja_installed(self):
+ """Test behavior when Jinja2 is not installed."""
+ with patch("micropie.JINJA_INSTALLED", False):
+ async def index(self):
+ return await self._render_template("test.html")
+ setattr(self.app, "index", index.__get__(self.app, App))
+
+ scope = self.create_mock_scope(path="/index")
+ receive = AsyncMock(return_value={"type": "http.request", "body": b"", "more_body": False})
+ send = AsyncMock()
+
+ await self.app(scope, receive, send)
+
+ send.assert_any_call({
+ "type": "http.response.start",
+ "status": 200,
+ "headers": [(b"Content-Type", b"text/html; charset=utf-8")]
+ })
+ send.assert_any_call({
+ "type": "http.response.body",
+ "body": b"500 Internal Server Error: Jinja2 not installed.",
+ "more_body": False
+ })
if __name__ == "__main__":
unittest.main()