patx/micropie
update benchmarks and remove development version with websockets, continuing development locally
Commit 343b2a8 · patx · 2025-06-03T20:30:27-04:00
Comments
No comments yet.
Diff
diff --git a/README.md b/README.md
index ccc33e3..109c7c2 100644
--- a/README.md
+++ b/README.md
@@ -9,7 +9,7 @@
- 🔒 **Sessions:** Simple, plugable, session management using cookies.
- 🎨 **Templates:** Jinja2, if installed, for rendering dynamic HTML pages.
- ⚙️ **Middleware:** Support for custom request middleware enabling functions like rate limiting, authentication, logging, and more.
-- ✨ **ASGI-Powered:** Built w/ asynchronous support for modern web servers like Uvicorn and Daphne, enabling high concurrency.
+- ✨ **ASGI-Powered:** Built w/ asynchronous support for modern web servers like Uvicorn, Hypercorn, and Daphne, enabling high concurrency.
- 🛠️ **Lightweight Design:** Only optional dependencies for flexibility and faster development/deployment.
- ⚡ **Blazing Fast:** Check out how MicroPie compares to other popular ASGI frameworks below!
@@ -235,17 +235,17 @@ MicroPie allows you to take full advantage of these benefits while maintaining s
## Benchmark Results
-Below is a performance comparison of various ASGI frameworks using their "Hello World" examples from each framework's website. Ran with `uvicorn` with 4 workers and `wrk -t12 -c1000 -d30s http://127.0.0.1:8000/`:
-
-| Framework | Requests/sec | Transfer/sec | Avg Latency | Stdev Latency | Max Latency | Socket Errors (timeouts) |
-|-----------------|--------------|--------------|-------------|---------------|-------------|--------------------------|
-| **Muffin** | 6508.80 | 0.90MB | 132.62ms | 69.71ms | 2.00s | 533 |
-| **Starlette** | 6340.40 | 0.86MB | 130.72ms | 75.55ms | 2.00s | 621 |
-| **BlackSheep** | 5928.99 | 0.98MB | 142.48ms | 73.61ms | 1.99s | 526 |
-| **MicroPie** | 5447.04 | 0.85MB | 157.04ms | 71.55ms | 2.00s | 470 |
-| **Litestar** | 5088.38 | 730.46KB | 151.59ms | 81.75ms | 2.00s | 662 |
-| **Sanic** | 4236.29 | 682.61KB | 196.80ms | 80.56ms | 2.00s | 452 |
-| **FastAPI** | 2352.53 | 326.23KB | 396.95ms | 112.41ms | 2.00s | 516 |
+The table below summarizes the performance of various ASGI frameworks based on a 15-second `wrk` test with 4 threads and 64 connections, measuring a simple "hello world" JSON response. [Learn More](https://gist.github.com/patx/26ad4babd662105007a6e728f182e1db).
+
+| Framework | Total Requests | Req/Sec | Transfer/Sec (MB/s) | Avg Latency (ms) | Stdev Latency (ms) | Max Latency (ms) |
+|-------------|----------------|-----------|---------------------|------------------|--------------------|------------------|
+| Muffin | 889,891 | 58,931.31 | 7.98 | 1.08 | 0.52 | 29.21 |
+| Blacksheep | 831,432 | 55,060.05 | 7.98 | 1.15 | 0.39 | 15.11 |
+| MicroPie | 791,721 | 52,685.82 | 8.09 | 1.35 | 1.09 | 21.59 |
+| Starlette | 779,092 | 51,930.45 | 7.03 | 1.22 | 0.39 | 17.42 |
+| Litestar | 610,059 | 40,401.18 | 5.47 | 1.57 | 0.63 | 33.66 |
+| Sanic | 536,203 | 35,508.58 | 5.45 | 1.84 | 0.97 | 37.83 |
+| FastAPI | 281,493 | 18,756.73 | 2.54 | 3.52 | 1.82 | 56.73 |
## **Suggestions or Feedback?**
We welcome suggestions, bug reports, and pull requests!
diff --git a/docs/index.html b/docs/index.html
index 9612da3..5e71d83 100644
--- a/docs/index.html
+++ b/docs/index.html
@@ -117,7 +117,7 @@
<span class="c2">class</span> MyApp(<span class="c9">App</span>):
<span class="c2">async def</span> index(<span class="c9">self</span>):
- return <span class="c9">'Hello World!'</span>
+ return <span class="c9">"Hello World!"</span>
app = MyApp() <small><em># Run with `uvicorn app:app`</em></small>
</code></pre>
diff --git a/pyproject.toml b/pyproject.toml
index 2e66675..6d2644d 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -4,7 +4,7 @@ build-backend = "flit_core.buildapi"
[project]
name = "MicroPie"
-version = "0.9.9.8"
+version = "0.10"
description = "An ultra micro ASGI web framework"
keywords = ["micropie", "asgi", "microframework", "http"]
readme = "README.md"
@@ -18,7 +18,8 @@ classifiers = [
]
[project.optional-dependencies]
-all = ["jinja2", "multipart", "aiofiles", "orjson"]
+standard = ["jinja2", "multipart", "aiofiles"]
+all = ["jinja2", "multipart", "aiofiles", "orjson", "uvicorn"]
[project.urls]
Homepage = "https://patx.github.io/micropie"
diff --git a/unstable/MicroPie.py b/unstable/MicroPie.py
deleted file mode 100644
index 9780353..0000000
--- a/unstable/MicroPie.py
+++ /dev/null
@@ -1,660 +0,0 @@
-"""
-MicroPie: A simple Python ultra-micro web framework with ASGI
-support. https://patx.github.io/micropie
-
-Copyright 2025 Harrison Erd
-
-Redistribution and use in source and binary forms, with or without
-modification, are permitted provided that the following conditions are met:
-
-1. Redistributions of source code must retain the above copyright notice,
- this list of conditions and the following disclaimer.
-
-2. Redistributions in binary form must reproduce the above copyright notice,
- this list of conditions and the following disclaimer in the documentation
- and/or other materials provided with the distribution.
-
-3. Neither the name of the copyright holder nor the names of its
- contributors may be used to endorse or promote products derived from this
- software without specific prior written permission.
-
-THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS
-IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO,
-THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR
-PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR
-CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL,
-EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO,
-PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS;
-OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY,
-WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR
-OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE,
-EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
-"""
-
-import asyncio
-import contextvars
-import inspect
-import json
-import os
-import re
-import time
-import uuid
-from abc import ABC, abstractmethod
-from typing import Any, Awaitable, Callable, Dict, List, Optional, Tuple
-from urllib.parse import parse_qs
-
-try:
- from jinja2 import Environment, FileSystemLoader, select_autoescape
- JINJA_INSTALLED = True
-except ImportError:
- JINJA_INSTALLED = False
-
-try:
- import aiofiles, aiofiles.os
- from multipart import PushMultipartParser, MultipartSegment
- MULTIPART_INSTALLED = True
-except ImportError:
- MULTIPART_INSTALLED = False
-
-
-# -----------------------------
-# Session Backend Abstraction
-# -----------------------------
-SESSION_TIMEOUT: int = 8 * 3600 # Default 8 hours
-
-class SessionBackend(ABC):
- @abstractmethod
- async def load(self, session_id: str) -> Dict[str, Any]:
- """
- Load session data given a session ID.
-
- Args:
- session_id: str
- """
- pass
-
- @abstractmethod
- async def save(self, session_id: str, data: Dict[str, Any], timeout: int) -> None:
- """
- Save session data.
-
- Args:
- session_id: str
- data: Dict
- timeout: int (in seconds)
- """
- pass
-
-class InMemorySessionBackend(SessionBackend):
- def __init__(self):
- self.sessions: Dict[str, Dict[str, Any]] = {}
- self.last_access: Dict[str, float] = {}
-
- async def load(self, session_id: str) -> Dict[str, Any]:
- now = time.time()
- if session_id in self.sessions and (now - self.last_access.get(session_id, now)) < SESSION_TIMEOUT:
- self.last_access[session_id] = now
- return self.sessions[session_id]
- return {}
-
- async def save(self, session_id: str, data: Dict[str, Any], timeout: int) -> None:
- self.sessions[session_id] = data
- self.last_access[session_id] = time.time()
-
-
-# -----------------------------
-# Request Object
-# -----------------------------
-current_request: contextvars.ContextVar[Any] = contextvars.ContextVar("current_request")
-
-class Request:
- """Represents an HTTP request in the MicroPie framework."""
- def __init__(self, scope: Dict[str, Any]) -> None:
- """
- Initialize a new Request instance.
-
- Args:
- scope: The ASGI scope dictionary for the request.
- """
- self.scope: Dict[str, Any] = scope
- self.method: str = scope["method"]
- self.path_params: List[str] = []
- self.query_params: Dict[str, List[str]] = {}
- self.body_params: Dict[str, List[str]] = {}
- self.get_json: Any = {}
- self.session: Dict[str, Any] = {}
- self.files: Dict[str, Any] = {}
- self.headers: Dict[str, str] = {
- k.decode("utf-8", errors="replace").lower(): v.decode("utf-8", errors="replace")
- for k, v in scope.get("headers", [])
- }
-
-
-# -----------------------------
-# WebSocket Object
-# -----------------------------
-class WebSocket:
- """Represents a WebSocket connection in the MicroPie framework."""
- def __init__(self, scope: Dict[str, Any], receive: Callable[[], Awaitable[Dict[str, Any]]], send: Callable[[Dict[str, Any]], Awaitable[None]]) -> None:
- self.scope = scope
- self.receive = receive
- self.send = send
- self.query_params = parse_qs(scope.get("query_string", b"").decode("utf-8", "ignore"))
- self.headers = {
- k.decode("utf-8", errors="replace").lower(): v.decode("utf-8", errors="replace")
- for k, v in scope.get("headers", [])
- }
- self._closed = False
-
- async def accept(self) -> None:
- if not self._closed:
- await self.send({
- "type": "websocket.accept"
- })
-
- async def send_text(self, data: str) -> None:
- if not self._closed:
- await self.send({
- "type": "websocket.send",
- "text": data
- })
-
- async def send_json(self, data: Any) -> None:
- if not self._closed:
- await self.send({
- "type": "websocket.send",
- "text": json.dumps(data)
- })
-
- async def receive_text(self) -> str:
- if self._closed:
- raise ConnectionError("WebSocket already closed")
- message = await self.receive()
- print(f"Raw WebSocket message: {message}") # Debug raw message
- if message["type"] == "websocket.disconnect":
- self._closed = True
- raise ConnectionError("WebSocket disconnected")
- return message.get("text", "")
-
- async def receive_json(self) -> Any:
- text = await self.receive_text()
- if not text.strip():
- print("Received empty WebSocket message")
- return None
- try:
- return json.loads(text)
- except json.JSONDecodeError as e:
- print(f"Invalid JSON received: {text}, error: {e}")
- raise
-
- async def close(self, code: int = 1000) -> None:
- if not self._closed:
- await self.send({
- "type": "websocket.close",
- "code": code
- })
- self._closed = True
-
-# -----------------------------
-# Middleware Abstraction
-# -----------------------------
-class HttpMiddleware(ABC):
- """
- Pluggable middleware class that allows hooking into the request lifecycle.
- """
-
- @abstractmethod
- async def before_request(self, request: Request) -> None:
- """
- Called before the request is processed.
- """
- pass
-
- @abstractmethod
- async def after_request(
- self,
- request: Request,
- status_code: int,
- response_body: Any,
- extra_headers: List[Tuple[str, str]]
- ) -> None:
- """
- Called after the request is processed, but before the final response
- is sent to the client. You may alter the status_code, response_body,
- or extra_headers if needed.
- """
- pass
-
-
-# -----------------------------
-# Application Base
-# -----------------------------
-class App:
- """
- ASGI application for handling HTTP and WebSocket requests in MicroPie.
- It supports pluggable session backends via the 'session_backend' attribute
- and pluggable middlewares via the 'middlewares' list.
- """
-
- def __init__(self, session_backend: Optional[SessionBackend] = None) -> None:
- if JINJA_INSTALLED:
- self.env = Environment(
- loader=FileSystemLoader("templates"),
- autoescape=select_autoescape(["html", "xml"]),
- enable_async=True
- )
- else:
- self.env = None
- self.session_backend: SessionBackend = session_backend or InMemorySessionBackend()
- self.middlewares: List[HttpMiddleware] = []
-
- @property
- def request(self) -> Request:
- """
- Retrieve the current request from the context variable.
-
- Returns: The current Request instance.
- """
- return current_request.get()
-
- async def __call__(
- self,
- scope: Dict[str, Any],
- receive: Callable[[], Awaitable[Dict[str, Any]]],
- send: Callable[[Dict[str, Any]], Awaitable[None]]
- ) -> None:
- """
- ASGI callable interface for the server.
-
- Args:
- scope: The ASGI scope dictionary.
- receive: The callable to receive ASGI events.
- send: The callable to send ASGI events.
- """
- if scope["type"] == "http":
- await self._asgi_app_http(scope, receive, send)
- elif scope["type"] == "websocket":
- await self._handle_websocket(scope, receive, send)
- else:
- pass # Handle lifespan and other scope types in the future.
-
- async def _asgi_app_http(
- self,
- scope: Dict[str, Any],
- receive: Callable[[], Awaitable[Dict[str, Any]]],
- send: Callable[[Dict[str, Any]], Awaitable[None]]
- ) -> None:
- """
- ASGI application entry point for handling HTTP requests.
-
- Args:
- scope: The ASGI scope dictionary.
- receive: The callable to receive ASGI events.
- send: The callable to send ASGI events.
- """
- request: Request = Request(scope)
- token = current_request.set(request)
- status_code: int = 200
- response_body: Any = ""
- extra_headers: List[Tuple[str, str]] = []
- try:
- # Middleware: before request
- for mw in self.middlewares:
- if result := await mw.before_request(request):
- status_code, response_body, extra_headers = (
- result["status_code"],
- result["body"],
- result.get("headers", []),
- )
- await self._send_response(send, status_code, response_body, extra_headers)
- return
-
- # Parse path and find handler
- path: str = scope["path"].lstrip("/")
- parts: List[str] = path.split("/") if path else []
- func_name: str = parts[0] if parts else "index"
- if func_name.startswith("_"):
- await self._send_response(send, 404, "404 Not Found")
- return
-
- request.path_params = parts[1:] if len(parts) > 1 else []
- handler = getattr(self, func_name, None) or getattr(self, "index", None)
- if not handler:
- await self._send_response(send, 404, "404 Not Found")
- return
-
- # Parse request details
- request.query_params = parse_qs(scope.get("query_string", b"").decode("utf-8", "ignore"))
- cookies = self._parse_cookies(request.headers.get("cookie", ""))
- request.session = await self.session_backend.load(cookies.get("session_id", "")) or {}
-
- # Parse body parameters.
- if request.method in ("POST", "PUT", "PATCH"):
- body_data = bytearray()
- while True:
- msg: Dict[str, Any] = await receive()
- body_data += msg.get("body", b"")
- if not msg.get("more_body"):
- break
- content_type = request.headers.get("content-type", "")
- if "application/json" in content_type:
- try:
- request.get_json = json.loads(body_data.decode("utf-8"))
- if isinstance(request.get_json, dict):
- request.body_params = {k: [str(v)] for k, v in request.get_json.items()}
- except Exception as e:
- print(f"Request error: {e}")
- await self._send_response(send, 400, "400 Bad Request: Bad JSON")
- return
- elif "multipart/form-data" in content_type:
- if boundary := re.search(r"boundary=([^;]+)", content_type):
- reader = asyncio.StreamReader()
- reader.feed_data(body_data)
- reader.feed_eof()
- request.body_params, request.files = await self._parse_multipart(reader, boundary.group(1).encode("utf-8"))
- else:
- await self._send_response(send, 400, "400 Bad Request: Missing boundary")
- return
- else:
- request.body_params = parse_qs(body_data.decode("utf-8", "ignore"))
-
- # Build function arguments from path, query, body, files, and session values.
- sig = inspect.signature(handler)
- func_args: List[Any] = []
- for param in sig.parameters.values():
- param_value = None
- if request.path_params:
- param_value = request.path_params.pop(0)
- elif param.name in request.query_params:
- param_value = request.query_params[param.name][0]
- elif param.name in request.body_params:
- param_value = request.body_params[param.name][0] if request.body_params[param.name] else ""
- elif param.name in request.files:
- param_value = request.files[param.name]
- elif param.name in request.session:
- param_value = request.session[param.name]
- elif param.default is not param.empty:
- param_value = param.default
- else:
- status_code = 400
- response_body = f"400 Bad Request: Missing required parameter '{param.name}'"
- await self._send_response(send, status_code, response_body)
- return
- func_args.append(param_value)
-
- if handler == getattr(self, "index", None) and not func_args and path:
- await self._send_response(send, 404, "404 Not Found")
- return
-
- # Execute handler
- try:
- result = await handler(*func_args) if inspect.iscoroutinefunction(handler) else handler(*func_args)
- except Exception as e:
- print(f"Request error: {e}")
- await self._send_response(send, 500, "500 Internal Server Error")
- return
-
- # Normalize response
- if isinstance(result, tuple):
- status_code, response_body = result[0], result[1]
- extra_headers = result[2] if len(result) > 2 else []
- else:
- response_body = result
- if isinstance(response_body, (dict, list)):
- response_body = json.dumps(response_body)
- extra_headers.append(("Content-Type", "application/json"))
-
- # Save session
- if request.session:
- session_id = cookies.get("session_id") or str(uuid.uuid4())
- await self.session_backend.save(session_id, request.session, SESSION_TIMEOUT)
- if not cookies.get("session_id"):
- extra_headers.append(("Set-Cookie", f"session_id={session_id}; Path=/; SameSite=Lax; HttpOnly; Secure;"))
-
- # Middleware: after request
- for mw in self.middlewares:
- if result := await mw.after_request(request, status_code, response_body, extra_headers):
- status_code, response_body, extra_headers = (
- result.get("status_code", status_code),
- result.get("body", response_body),
- result.get("headers", extra_headers)
- )
-
- await self._send_response(send, status_code, response_body, extra_headers)
-
- finally:
- current_request.reset(token)
-
- async def _handle_websocket(
- self,
- scope: Dict[str, Any],
- receive: Callable[[], Awaitable[Dict[str, Any]]],
- send: Callable[[Dict[str, Any]], Awaitable[None]]
- ) -> None:
- """
- Handle WebSocket connections.
-
- Args:
- scope: The ASGI scope dictionary for the WebSocket connection.
- receive: The callable to receive ASGI events.
- send: The callable to send ASGI events.
- """
- websocket = WebSocket(scope, receive, send)
- path: str = scope["path"].lstrip("/")
- parts: List[str] = path.split("/") if path else []
- func_name: str = parts[0] if parts else "ws_index"
- if func_name.startswith("_"):
- await websocket.close(1008) # Policy violation
- return
-
- handler_name = f"ws_{func_name}"
- handler = getattr(self, handler_name, None) or getattr(self, "ws_index", None)
- if not handler:
- await websocket.close(1008) # Policy violation
- return
-
- closed = False
- try:
- await handler(websocket, parts[1:] if len(parts) > 1 else [])
- except Exception as e:
- print(f"WebSocket error: {e}")
- await websocket.close(1011) # Internal error
- closed = True
- finally:
- if not closed:
- await websocket.close()
-
- def _parse_cookies(self, cookie_header: str) -> Dict[str, str]:
- """
- Parse the Cookie header and return a dictionary of cookie names and values.
-
- Args:
- cookie_header: The raw Cookie header string.
-
- Returns:
- A dictionary mapping cookie names to their corresponding values.
- """
- cookies: Dict[str, str] = {}
- if not cookie_header:
- return cookies
- for cookie in cookie_header.split(";"):
- if "=" in cookie:
- k, v = cookie.strip().split("=", 1)
- cookies[k] = v
- return cookies
-
- async def _parse_multipart(self, reader: asyncio.StreamReader, boundary: bytes):
- """
- Asynchronously parses a multipart form-data request.
-
- This method processes incoming multipart form-data, handling
- both text fields and file uploads. It reads data from the provided
- asyncio stream reader and extracts form values and files,
- saving uploaded files to a designated directory.
-
- Args:
- reader (asyncio.StreamReader): The stream reader from which
- to read the multipart data.
- boundary (bytes): The boundary string used to separate form
- fields in the multipart request.
-
- Returns:
- tuple[dict, dict]: A tuple containing form_data & files.
- """
- if not MULTIPART_INSTALLED:
- print("For multipart form data support install 'multipart' and 'aiofiles'.")
- await self._send_response(send, 500, "500 Internal Server Error")
- return
-
- with PushMultipartParser(boundary) as parser:
- form_data: dict = {}
- files: dict = {}
- current_field_name: Optional[str] = None
- current_filename: Optional[str] = None
- current_content_type: Optional[str] = None
- current_file: Optional[aiofiles.threadpool.binary.AsyncBufferedIOBase] = None
- form_value: str = ""
- upload_directory: str = "uploads"
- await aiofiles.os.makedirs(upload_directory, exist_ok=True)
- while not parser.closed:
- chunk: bytes = await reader.read(65536)
- if not chunk:
- break
- for result in parser.parse(chunk):
- if isinstance(result, MultipartSegment):
- current_field_name = result.name
- current_filename = result.filename
- current_content_type = None
- form_value = ""
- for header, value in result.headerlist:
- if header.lower() == "content-type":
- current_content_type = value
-
- if current_filename:
- safe_filename: str = f"{uuid.uuid4()}_{current_filename}"
- safe_filename = re.sub(r"[^a-zA-Z0-9_.-]", "_", safe_filename)
- file_path: str = os.path.join(upload_directory, safe_filename)
- current_file = await aiofiles.open(file_path, "wb")
- else:
- # Initialize form_data with an empty list for text fields
- if current_field_name not in form_data:
- form_data[current_field_name] = []
- elif result:
- if current_file:
- await current_file.write(result)
- else:
- form_value += result.decode("utf-8", "ignore")
- else:
- if current_file:
- await current_file.close()
- current_file = None
- files[current_field_name] = {
- "filename": current_filename,
- "content_type": current_content_type or "application/octet-stream",
- "saved_path": os.path.join(upload_directory, safe_filename),
- }
- else:
- # Append form_value to form_data, even if empty
- if current_field_name:
- form_data[current_field_name].append(form_value or "")
- form_value = ""
- # Ensure any remaining form_value is appended
- if current_field_name and not current_filename:
- form_data[current_field_name].append(form_value or "")
- return form_data, files
-
- async def _send_response(
- self,
- send: Callable[[Dict[str, Any]], Awaitable[None]],
- status_code: int,
- body: Any,
- extra_headers: Optional[List[Tuple[str, str]]] = None
- ) -> None:
- """
- Send an HTTP response using the ASGI send callable.
-
- Args:
- send: The ASGI send callable.
- status_code: The HTTP status code for the response.
- body: The response body, which may be a string, bytes, or
- generator.
- extra_headers: Optional list of extra header tuples.
- """
- if extra_headers is None:
- extra_headers = []
- sanitized_headers: List[Tuple[str, str]] = []
- for k, v in extra_headers:
- if "\n" in k or "\r" in k or "\n" in v or "\r" in v:
- print(f"Header injection attempt detected: {k}: {v}")
- continue
- sanitized_headers.append((k, v))
- if not any(h[0].lower() == "content-type" for h in sanitized_headers):
- sanitized_headers.append(("Content-Type", "text/html; charset=utf-8"))
- await send({
- "type": "http.response.start",
- "status": status_code,
- "headers": [(k.encode("latin-1"), v.encode("latin-1")) for k, v in sanitized_headers],
- })
- if hasattr(body, "__aiter__"):
- async for chunk in body:
- if isinstance(chunk, str):
- chunk = chunk.encode("utf-8")
- await send({
- "type": "http.response.body",
- "body": chunk,
- "more_body": True
- })
- await send({"type": "http.response.body", "body": b"", "more_body": False})
- return
- if hasattr(body, "__iter__") and not isinstance(body, (bytes, str)):
- for chunk in body:
- if isinstance(chunk, str):
- chunk = chunk.encode("utf-8")
- await send({
- "type": "http.response.body",
- "body": chunk,
- "more_body": True
- })
- await send({"type": "http.response.body", "body": b"", "more_body": False})
- return
- response_body = (body if isinstance(body, bytes)
- else str(body).encode("utf-8"))
- await send({
- "type": "http.response.body",
- "body": response_body,
- "more_body": False
- })
-
- def _redirect(self, location: str, extra_headers: list = None) -> Tuple[int, str]:
- """
- Generate an HTTP redirect response.
-
- Args:
- location: The URL to redirect to.
- extra_headers: Optional list of tuples (header_name, header_value) to include in the response.
-
- Returns:
- A tuple containing the HTTP status code, the HTML body, and headers list.
- """
- headers = [("Location", location)]
- if extra_headers:
- headers.extend(extra_headers)
- return 302, "", headers
-
- async def _render_template(self, name: str, **kwargs: Any) -> str:
- """
- Render a template asynchronously using Jinja2.
-
- Args:
- name: The name of the template file.
- **kwargs: Additional keyword arguments for the template.
-
- Returns:
- The rendered template as a string.
- """
- if not JINJA_INSTALLED:
- print("To use the `_render_template` method install 'jinja2'.")
- return 500, "500 Internal Server Error"
- 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/unstable/tests.py b/unstable/tests.py
deleted file mode 100644
index 7080ed3..0000000
--- a/unstable/tests.py
+++ /dev/null
@@ -1,990 +0,0 @@
-import asyncio
-import os
-import shutil
-import time
-import uuid
-import pytest
-from MicroPie import App, Request, WebSocket, HttpMiddleware, InMemorySessionBackend
-from urllib.parse import parse_qs
-
-# Mock MULTIPART_INSTALLED and JINJA_INSTALLED for testing optional dependencies
-MULTIPART_INSTALLED = True
-JINJA_INSTALLED = True
-
-# Import optional dependencies safely
-try:
- import aiofiles
- from multipart import PushMultipartParser, MultipartSegment
-except ImportError:
- pass
-
-try:
- from jinja2 import Environment
-except ImportError:
- pass
-
-# Setup fixture for uploads directory
[email protected](autouse=True)
-def setup_uploads():
- upload_dir = "uploads"
- if os.path.exists(upload_dir):
- shutil.rmtree(upload_dir)
- yield
- if os.path.exists(upload_dir):
- shutil.rmtree(upload_dir)
-
-# Setup fixture for templates directory
[email protected]
-def setup_templates():
- os.makedirs("templates", exist_ok=True)
- yield
- if os.path.exists("templates"):
- shutil.rmtree("templates")
-
-# Test 1: Basic HTTP GET Request
[email protected]
-async def test_basic_get_request():
- class TestApp(App):
- async def index(self):
- return "Hello, World!"
-
- app = TestApp()
- scope = {
- "type": "http",
- "method": "GET",
- "path": "/",
- "headers": [],
- "query_string": b"",
- }
- sent_messages = []
-
- async def mock_send(message):
- sent_messages.append(message)
-
- async def mock_receive():
- return {"type": "http.request", "body": b"", "more_body": False}
-
- await app(scope, mock_receive, mock_send)
- assert len(sent_messages) == 2
- assert sent_messages[0]["type"] == "http.response.start"
- assert sent_messages[0]["status"] == 200
- assert sent_messages[1]["type"] == "http.response.body"
- assert sent_messages[1]["body"] == b"Hello, World!"
-
-# Test 2: HTTP GET with Path Parameters
[email protected]
-async def test_get_with_path_params():
- class TestApp(App):
- async def user(self, user_id):
- return f"User {user_id}"
-
- app = TestApp()
- scope = {
- "type": "http",
- "method": "GET",
- "path": "/user/123",
- "headers": [],
- "query_string": b"",
- }
- sent_messages = []
-
- async def mock_send(message):
- sent_messages.append(message)
-
- async def mock_receive():
- return {"type": "http.request", "body": b"", "more_body": False}
-
- await app(scope, mock_receive, mock_send)
- assert len(sent_messages) == 2
- assert sent_messages[0]["status"] == 200
- assert sent_messages[1]["body"] == b"User 123"
-
-# Test 3: HTTP GET with Query Parameters
[email protected]
-async def test_get_with_query_params():
- class TestApp(App):
- async def search(self, query):
- return f"Search for {query}"
-
- app = TestApp()
- scope = {
- "type": "http",
- "method": "GET",
- "path": "/search",
- "headers": [],
- "query_string": b"query=python",
- }
- sent_messages = []
-
- async def mock_send(message):
- sent_messages.append(message)
-
- async def mock_receive():
- return {"type": "http.request", "body": b"", "more_body": False}
-
- await app(scope, mock_receive, mock_send)
- assert len(sent_messages) == 2
- assert sent_messages[0]["status"] == 200
- assert sent_messages[1]["body"] == b"Search for python"
-
-# Test 4: HTTP POST with Form Data
[email protected]
-async def test_post_with_form_data():
- class TestApp(App):
- async def login(self, username, password):
- return "Login successful" if username == "admin" and password == "secret" else ("Invalid credentials", 401)
-
- app = TestApp()
- scope = {
- "type": "http",
- "method": "POST",
- "path": "/login",
- "headers": [(b"content-type", b"application/x-www-form-urlencoded")],
- "query_string": b"",
- }
- sent_messages = []
-
- async def mock_send(message):
- sent_messages.append(message)
-
- async def mock_receive():
- return {"type": "http.request", "body": b"username=admin&password=secret", "more_body": False}
-
- await app(scope, mock_receive, mock_send)
- assert len(sent_messages) == 2
- assert sent_messages[0]["status"] == 200
- assert sent_messages[1]["body"] == b"Login successful"
-
-# Test 5: HTTP POST with JSON Data
[email protected]
-async def test_post_with_json_data():
- class TestApp(App):
- async def create_user(self, name):
- return f"User {name} created"
-
- app = TestApp()
- scope = {
- "type": "http",
- "method": "POST",
- "path": "/create_user",
- "headers": [(b"content-type", b"application/json")],
- "query_string": b"",
- }
- sent_messages = []
-
- async def mock_send(message):
- sent_messages.append(message)
-
- async def mock_receive():
- return {"type": "http.request", "body": b'{"name": "Alice"}', "more_body": False}
-
- await app(scope, mock_receive, mock_send)
- assert len(sent_messages) == 2
- assert sent_messages[0]["status"] == 200
- assert sent_messages[1]["body"] == b"User Alice created"
-
-# Test 6: HTTP POST with Multipart File Upload
[email protected]
-async def test_post_with_multipart_file_upload():
- class TestApp(App):
- async def upload(self, file):
- return f"File {file['filename']} uploaded to {file['saved_path']}"
-
- app = TestApp()
- scope = {
- "type": "http",
- "method": "POST",
- "path": "/upload",
- "headers": [(b"content-type", b"multipart/form-data; boundary=boundary")],
- "query_string": b"",
- }
- sent_messages = []
-
- async def mock_send(message):
- sent_messages.append(message)
-
- body = (
- b"--boundary\r\n"
- b'Content-Disposition: form-data; name="file"; filename="test.txt"\r\n'
- b"Content-Type: text/plain\r\n"
- b"\r\n"
- b"Hello, World!\r\n"
- b"--boundary--\r\n"
- )
-
- async def mock_receive():
- return {"type": "http.request", "body": body, "more_body": False}
-
- await app(scope, mock_receive, mock_send)
- assert len(sent_messages) == 2
- assert sent_messages[0]["status"] == 200
- assert b"File test.txt uploaded to" in sent_messages[1]["body"]
- files = os.listdir("uploads")
- assert len(files) == 1
- with open(os.path.join("uploads", files[0]), "rb") as f:
- assert f.read() == b"Hello, World!"
-
-# Test 7: Session Management
[email protected]
-async def test_session_management():
- class TestApp(App):
- async def login(self, username):
- request = self.request
- request.session["username"] = username
- return "Logged in"
-
- async def profile(self):
- request = self.request
- return f"Welcome, {request.session.get('username', 'Guest')}"
-
- app = TestApp()
- scope_login = {
- "type": "http",
- "method": "POST",
- "path": "/login",
- "headers": [],
- "query_string": b"username=alice",
- }
- sent_messages_login = []
-
- async def mock_send_login(message):
- sent_messages_login.append(message)
-
- async def mock_receive_login():
- return {"type": "http.request", "body": b"", "more_body": False}
-
- await app(scope_login, mock_receive_login, mock_send_login)
- assert sent_messages_login[0]["status"] == 200
-
- session_id = [h[1].decode().split(";")[0].split("=")[1] for h in sent_messages_login[0]["headers"] if h[0] == b"Set-Cookie"][0]
- scope_profile = {
- "type": "http",
- "method": "GET",
- "path": "/profile",
- "headers": [(b"cookie", f"session_id={session_id}".encode())],
- "query_string": b"",
- }
- sent_messages_profile = []
-
- async def mock_send_profile(message):
- sent_messages_profile.append(message)
-
- async def mock_receive_profile():
- return {"type": "http.request", "body": b"", "more_body": False}
-
- await app(scope_profile, mock_receive_profile, mock_send_profile)
- assert sent_messages_profile[0]["status"] == 200
- assert sent_messages_profile[1]["body"] == b"Welcome, alice"
-
-# Test 8: WebSocket Connection
[email protected]
-async def test_websocket_connection():
- class TestApp(App):
- async def ws_echo(self, websocket, path_params):
- await websocket.accept()
- message = await websocket.receive_text()
- await websocket.send_text(f"Echo: {message}")
- await websocket.close()
-
- app = TestApp()
- scope = {
- "type": "websocket",
- "path": "/echo",
- "headers": [],
- "query_string": b"",
- }
- sent_messages = []
- received_messages = [
- {"type": "websocket.connect"},
- {"type": "websocket.receive", "text": "Hello"},
- {"type": "websocket.disconnect", "code": 1000},
- ]
-
- async def mock_send(message):
- sent_messages.append(message)
-
- async def mock_receive():
- return received_messages.pop(0)
-
- await app(scope, mock_receive, mock_send)
- assert len(sent_messages) == 3
- assert sent_messages[0]["type"] == "websocket.accept"
- assert sent_messages[1]["type"] == "websocket.send"
- assert sent_messages[1]["text"] == "Echo: Hello"
- assert sent_messages[2]["type"] == "websocket.close"
- assert sent_messages[2]["code"] == 1000
-
-# Test 9: HTTP Middleware
[email protected]
-async def test_http_middleware():
- class CustomHeaderMiddleware(HttpMiddleware):
- async def before_request(self, request):
- pass
-
- async def after_request(self, request, status_code, response_body, extra_headers):
- extra_headers.append(("X-Custom-Header", "Test"))
- return {"headers": extra_headers}
-
- class TestApp(App):
- async def index(self):
- return "Hello"
-
- app = TestApp()
- app.middlewares.append(CustomHeaderMiddleware())
- scope = {
- "type": "http",
- "method": "GET",
- "path": "/",
- "headers": [],
- "query_string": b"",
- }
- sent_messages = []
-
- async def mock_send(message):
- sent_messages.append(message)
-
- async def mock_receive():
- return {"type": "http.request", "body": b"", "more_body": False}
-
- await app(scope, mock_receive, mock_send)
- assert len(sent_messages) == 2
- assert any(h[0] == b"X-Custom-Header" and h[1] == b"Test" for h in sent_messages[0]["headers"])
-
-# Test 10: Template Rendering
[email protected]
-async def test_template_rendering(setup_templates):
- with open("templates/hello.html", "w") as f:
- f.write("Hello, {{ name }}!")
-
- class TestApp(App):
- async def index(self):
- return await self._render_template("hello.html", name="World")
-
- app = TestApp()
- scope = {
- "type": "http",
- "method": "GET",
- "path": "/",
- "headers": [],
- "query_string": b"",
- }
- sent_messages = []
-
- async def mock_send(message):
- sent_messages.append(message)
-
- async def mock_receive():
- return {"type": "http.request", "body": b"", "more_body": False}
-
- await app(scope, mock_receive, mock_send)
- assert sent_messages[0]["status"] == 200
- assert sent_messages[1]["body"] == b"Hello, World!"
-
-# Test 11: 404 Not Found
[email protected]
-async def test_404_not_found():
- class TestApp(App):
- pass
-
- app = TestApp()
- scope = {
- "type": "http",
- "method": "GET",
- "path": "/nonexistent",
- "headers": [],
- "query_string": b"",
- }
- sent_messages = []
-
- async def mock_send(message):
- sent_messages.append(message)
-
- async def mock_receive():
- return {"type": "http.request", "body": b"", "more_body": False}
-
- await app(scope, mock_receive, mock_send)
- assert sent_messages[0]["status"] == 404
- assert sent_messages[1]["body"] == b"404 Not Found"
-
-# Test 12: 400 Bad Request (Missing Parameter)
[email protected]
-async def test_400_missing_parameter():
- class TestApp(App):
- async def index(self, required_param):
- return "Should not reach here"
-
- app = TestApp()
- scope = {
- "type": "http",
- "method": "GET",
- "path": "/",
- "headers": [],
- "query_string": b"",
- }
- sent_messages = []
-
- async def mock_send(message):
- sent_messages.append(message)
-
- async def mock_receive():
- return {"type": "http.request", "body": b"", "more_body": False}
-
- await app(scope, mock_receive, mock_send)
- assert sent_messages[0]["status"] == 400
- assert b"Missing required parameter" in sent_messages[1]["body"]
-
-# Test 13: 500 Internal Server Error
[email protected]
-async def test_500_internal_server_error():
- class TestApp(App):
- async def index(self):
- raise Exception("Test error")
-
- app = TestApp()
- scope = {
- "type": "http",
- "method": "GET",
- "path": "/",
- "headers": [],
- "query_string": b"",
- }
- sent_messages = []
-
- async def mock_send(message):
- sent_messages.append(message)
-
- async def mock_receive():
- return {"type": "http.request", "body": b"", "more_body": False}
-
- await app(scope, mock_receive, mock_send)
- assert sent_messages[0]["status"] == 500
- assert sent_messages[1]["body"] == b"500 Internal Server Error"
-
-# Test 14: WebSocket Error Handling
[email protected]
-async def test_websocket_error_handling():
- class TestApp(App):
- async def ws_index(self, websocket, path_params):
- raise Exception("Test error")
-
- app = TestApp()
- scope = {
- "type": "websocket",
- "path": "/",
- "headers": [],
- "query_string": b"",
- }
- sent_messages = []
-
- async def mock_send(message):
- sent_messages.append(message)
-
- async def mock_receive():
- return {"type": "websocket.connect"}
-
- await app(scope, mock_receive, mock_send)
- assert len(sent_messages) == 1
- assert sent_messages[0]["type"] == "websocket.close"
- assert sent_messages[0]["code"] == 1011
-
-# Test 15: Parse Cookies
-def test_parse_cookies():
- app = App()
- cookies = app._parse_cookies("session_id=abc123; user=alice")
- assert cookies == {"session_id": "abc123", "user": "alice"}
-
-# Test 16: Redirect
[email protected]
-async def test_redirect():
- class TestApp(App):
- async def index(self):
- return self._redirect("/new_location")
-
- app = TestApp()
- scope = {
- "type": "http",
- "method": "GET",
- "path": "/",
- "headers": [],
- "query_string": b"",
- }
- sent_messages = []
-
- async def mock_send(message):
- sent_messages.append(message)
-
- async def mock_receive():
- return {"type": "http.request", "body": b"", "more_body": False}
-
- await app(scope, mock_receive, mock_send)
- assert sent_messages[0]["status"] == 302
- assert any(h[0] == b"Location" and h[1] == b"/new_location" for h in sent_messages[0]["headers"])
-
-# Test 17: In-Memory Session Backend
[email protected]
-async def test_in_memory_session_backend():
- backend = InMemorySessionBackend()
- session_id = "test_session"
- data = {"key": "value"}
- await backend.save(session_id, data, 3600)
- loaded_data = await backend.load(session_id)
- assert loaded_data == data
- # Simulate session timeout
- backend.last_access[session_id] = time.time() - 8 * 3600 - 1
- assert await backend.load(session_id) == {}
-
-# Test 18: Synchronous Handler
[email protected]
-async def test_synchronous_handler():
- class TestApp(App):
- def index(self):
- return "Sync Hello"
-
- app = TestApp()
- scope = {
- "type": "http",
- "method": "GET",
- "path": "/",
- "headers": [],
- "query_string": b"",
- }
- sent_messages = []
-
- async def mock_send(message):
- sent_messages.append(message)
-
- async def mock_receive():
- return {"type": "http.request", "body": b"", "more_body": False}
-
- await app(scope, mock_receive, mock_send)
- assert sent_messages[0]["status"] == 200
- assert sent_messages[1]["body"] == b"Sync Hello"
-
-# Test 19: Asynchronous Streaming Response
[email protected]
-async def test_async_streaming_response():
- class TestApp(App):
- async def stream(self):
- async def generate():
- yield "Chunk 1"
- await asyncio.sleep(0.1)
- yield "Chunk 2"
- return generate()
-
- app = TestApp()
- scope = {
- "type": "http",
- "method": "GET",
- "path": "/stream",
- "headers": [],
- "query_string": b"",
- }
- sent_messages = []
-
- async def mock_send(message):
- sent_messages.append(message)
-
- async def mock_receive():
- return {"type": "http.request", "body": b"", "more_body": False}
-
- await app(scope, mock_receive, mock_send)
- assert len(sent_messages) == 4
- assert sent_messages[1]["body"] == b"Chunk 1"
- assert sent_messages[2]["body"] == b"Chunk 2"
- assert sent_messages[3]["body"] == b""
-
-# Test 20: Synchronous Generator Response
[email protected]
-async def test_sync_generator_response():
- class TestApp(App):
- def stream(self):
- def generate():
- yield "Chunk 1"
- yield "Chunk 2"
- return generate()
-
- app = TestApp()
- scope = {
- "type": "http",
- "method": "GET",
- "path": "/stream",
- "headers": [],
- "query_string": b"",
- }
- sent_messages = []
-
- async def mock_send(message):
- sent_messages.append(message)
-
- async def mock_receive():
- return {"type": "http.request", "body": b"", "more_body": False}
-
- await app(scope, mock_receive, mock_send)
- assert len(sent_messages) == 4
- assert sent_messages[1]["body"] == b"Chunk 1"
- assert sent_messages[2]["body"] == b"Chunk 2"
-
-# Test 21: JSON Response
[email protected]
-async def test_json_response():
- class TestApp(App):
- async def data(self):
- return {"key": "value"}
-
- app = TestApp()
- scope = {
- "type": "http",
- "method": "GET",
- "path": "/data",
- "headers": [],
- "query_string": b"",
- }
- sent_messages = []
-
- async def mock_send(message):
- sent_messages.append(message)
-
- async def mock_receive():
- return {"type": "http.request", "body": b"", "more_body": False}
-
- await app(scope, mock_receive, mock_send)
- assert any(h[0] == b"Content-Type" and h[1] == b"application/json" for h in sent_messages[0]["headers"])
- assert sent_messages[1]["body"] == b'{"key": "value"}'
-
-# Test 22: Protected Path (Starting with '_')
[email protected]
-async def test_protected_path():
- class TestApp(App):
- async def _hidden(self):
- return "Should not reach here"
-
- app = TestApp()
- scope = {
- "type": "http",
- "method": "GET",
- "path": "/_hidden",
- "headers": [],
- "query_string": b"",
- }
- sent_messages = []
-
- async def mock_send(message):
- sent_messages.append(message)
-
- async def mock_receive():
- return {"type": "http.request", "body": b"", "more_body": False}
-
- await app(scope, mock_receive, mock_send)
- assert sent_messages[0]["status"] == 404
-
-# Test 23: WebSocket Protected Path
[email protected]
-async def test_websocket_protected_path():
- class TestApp(App):
- async def _ws_hidden(self, websocket, path_params):
- pass
-
- app = TestApp()
- scope = {
- "type": "websocket",
- "path": "/_hidden",
- "headers": [],
- "query_string": b"",
- }
- sent_messages = []
-
- async def mock_send(message):
- sent_messages.append(message)
-
- async def mock_receive():
- return {"type": "websocket.connect"}
-
- await app(scope, mock_receive, mock_send)
- assert sent_messages[0]["type"] == "websocket.close"
- assert sent_messages[0]["code"] == 1008
-
-# Test 24: Invalid JSON
[email protected]
-async def test_invalid_json():
- class TestApp(App):
- async def index(self):
- pass
-
- app = TestApp()
- scope = {
- "type": "http",
- "method": "POST",
- "path": "/index",
- "headers": [(b"content-type", b"application/json")],
- "query_string": b"",
- }
- sent_messages = []
-
- async def mock_send(message):
- sent_messages.append(message)
-
- async def mock_receive():
- return {"type": "http.request", "body": b"{invalid}", "more_body": False}
-
- await app(scope, mock_receive, mock_send)
- assert sent_messages[0]["status"] == 400
- assert sent_messages[1]["body"] == b"400 Bad Request: Bad JSON"
-
-# Test 25: Header Injection Prevention
[email protected]
-async def test_header_injection_prevention():
- class TestApp(App):
- async def index(self):
- return "Hello", 200, [("X-Test", "Value\nInjection")]
-
- app = TestApp()
- scope = {
- "type": "http",
- "method": "GET",
- "path": "/",
- "headers": [],
- "query_string": b"",
- }
- sent_messages = []
-
- async def mock_send(message):
- sent_messages.append(message)
-
- async def mock_receive():
- return {"type": "http.request", "body": b"", "more_body": False}
-
- await app(scope, mock_receive, mock_send)
- assert not any(b"\n" in h[1] for h in sent_messages[0]["headers"])
-
-# Test 26: Missing Jinja2 Dependency
[email protected]
-async def test_missing_jinja2(monkeypatch):
- monkeypatch.setattr("MicroPie.JINJA_INSTALLED", False)
- class TestApp(App):
- async def index(self):
- return await self._render_template("hello.html", name="World")
-
- app = TestApp()
- scope = {
- "type": "http",
- "method": "GET",
- "path": "/",
- "headers": [],
- "query_string": b"",
- }
- sent_messages = []
-
- async def mock_send(message):
- sent_messages.append(message)
-
- async def mock_receive():
- return {"type": "http.request", "body": b"", "more_body": False}
-
- await app(scope, mock_receive, mock_send)
- assert sent_messages[0]["status"] == 500
- assert sent_messages[1]["body"] == b"500 Internal Server Error"
-
-# Test 27: Missing Multipart Dependency
[email protected]
-async def test_missing_multipart(monkeypatch):
- monkeypatch.setattr("MicroPie.MULTIPART_INSTALLED", False)
- class TestApp(App):
- async def upload(self, file):
- return "Should not reach here"
-
- app = TestApp()
- scope = {
- "type": "http",
- "method": "POST",
- "path": "/upload",
- "headers": [(b"content-type", b"multipart/form-data; boundary=boundary")],
- "query_string": b"",
- }
- sent_messages = []
-
- async def mock_send(message):
- sent_messages.append(message)
-
- async def mock_receive():
- return {"type": "http.request", "body": b"", "more_body": False}
-
- await app(scope, mock_receive, mock_send)
- assert sent_messages[0]["status"] == 500
- assert sent_messages[1]["body"] == b"500 Internal Server Error"
-
-# Test 28: WebSocket Send JSON
[email protected]
-async def test_websocket_send_json():
- class TestApp(App):
- async def ws_json(self, websocket, path_params):
- await websocket.accept()
- await websocket.send_json({"message": "Hello"})
- await websocket.close()
-
- app = TestApp()
- scope = {
- "type": "websocket",
- "path": "/json",
- "headers": [],
- "query_string": b"",
- }
- sent_messages = []
-
- async def mock_send(message):
- sent_messages.append(message)
-
- async def mock_receive():
- return {"type": "websocket.connect"}
-
- await app(scope, mock_receive, mock_send)
- assert len(sent_messages) == 3
- assert sent_messages[0]["type"] == "websocket.accept"
- assert sent_messages[1]["type"] == "websocket.send"
- assert sent_messages[1]["text"] == '{"message": "Hello"}'
- assert sent_messages[2]["type"] == "websocket.close"
-
-# Test 29: WebSocket Receive JSON
[email protected]
-async def test_websocket_receive_json():
- class TestApp(App):
- async def ws_json(self, websocket, path_params):
- await websocket.accept()
- data = await websocket.receive_json()
- await websocket.send_text(f"Received: {data['message']}")
- await websocket.close()
-
- app = TestApp()
- scope = {
- "type": "websocket",
- "path": "/json",
- "headers": [],
- "query_string": b"",
- }
- sent_messages = []
- received_messages = [
- {"type": "websocket.connect"},
- {"type": "websocket.receive", "text": '{"message": "Hello"}'},
- {"type": "websocket.disconnect", "code": 1000},
- ]
-
- async def mock_send(message):
- sent_messages.append(message)
-
- async def mock_receive():
- return received_messages.pop(0)
-
- await app(scope, mock_receive, mock_send)
- assert len(sent_messages) == 3
- assert sent_messages[0]["type"] == "websocket.accept"
- assert sent_messages[1]["type"] == "websocket.send"
- assert sent_messages[1]["text"] == "Received: Hello"
- assert sent_messages[2]["type"] == "websocket.close"
-
-# Test 30: WebSocket Disconnect
[email protected]
-async def test_websocket_disconnect():
- class TestApp(App):
- async def ws_disconnect(self, websocket, path_params):
- await websocket.accept()
- await websocket.receive_text() # Should raise ConnectionError
- await websocket.close()
-
- app = TestApp()
- scope = {
- "type": "websocket",
- "path": "/disconnect",
- "headers": [],
- "query_string": b"",
- }
- sent_messages = []
- received_messages = [
- {"type": "websocket.connect"},
- {"type": "websocket.disconnect", "code": 1000},
- ]
-
- async def mock_send(message):
- sent_messages.append(message)
-
- async def mock_receive():
- return received_messages.pop(0)
-
- await app(scope, mock_receive, mock_send)
- assert len(sent_messages) == 2
- assert sent_messages[0]["type"] == "websocket.accept"
- assert sent_messages[1]["type"] == "websocket.close"
-
-# Test 31: Middleware Before Request Early Exit
[email protected]
-async def test_middleware_before_request_early_exit():
- class EarlyExitMiddleware(HttpMiddleware):
- async def before_request(self, request):
- return {"status_code": 403, "body": "Forbidden"}
-
- async def after_request(self, request, status_code, response_body, extra_headers):
- pass
-
- class TestApp(App):
- async def index(self):
- return "Should not reach here"
-
- app = TestApp()
- app.middlewares.append(EarlyExitMiddleware())
- scope = {
- "type": "http",
- "method": "GET",
- "path": "/",
- "headers": [],
- "query_string": b"",
- }
- sent_messages = []
-
- async def mock_send(message):
- sent_messages.append(message)
-
- async def mock_receive():
- return {"type": "http.request", "body": b"", "more_body": False}
-
- await app(scope, mock_receive, mock_send)
- assert sent_messages[0]["status"] == 403
- assert sent_messages[1]["body"] == b"Forbidden"
-
-# Test 32: Empty Cookie Header
-def test_empty_cookie_header():
- app = App()
- cookies = app._parse_cookies("")
- assert cookies == {}
-
-# Test 33: Multipart Form Data Without File
[email protected]
-async def test_multipart_form_data_without_file():
- class TestApp(App):
- async def form(self, field):
- return f"Field: {field[0]}"
-
- app = TestApp()
- scope = {
- "type": "http",
- "method": "POST",
- "path": "/form",
- "headers": [(b"content-type", b"multipart/form-data; boundary=boundary")],
- "query_string": b"",
- }
- sent_messages = []
-
- async def mock_send(message):
- sent_messages.append(message)
-
- body = (
- b"--boundary\r\n"
- b'Content-Disposition: form-data; name="field"\r\n'
- b"\r\n"
- b"test_value\r\n"
- b"--boundary--\r\n"
- )
-
- async def mock_receive():
- return {"type": "http.request", "body": body, "more_body": False}
-
- await app(scope, mock_receive, mock_send)
- assert sent_messages[0]["status"] == 200
- assert sent_messages[1]["body"] == b"Field: test_value"