patx/micropie
Bug fixes: Multipart file binding now awaits queue notifications instead of busy-spinning. Added configurable per-chunk body_timeout, session_timeout, and session_cookie_secure options to App. In-memory sessions now honor each timeout passed to save(). HTTP and WebSocket cookies share the configurable Secure policy. Core diagnostics and exceptions now use the micropie logger.
Commit a686ec8 · patx · 2026-07-14T21:32:05-04:00
Comments
No comments yet.
Diff
diff --git a/docs/apidocs/howto/sessions.rst b/docs/apidocs/howto/sessions.rst
index b29f6de..ed0553b 100644
--- a/docs/apidocs/howto/sessions.rst
+++ b/docs/apidocs/howto/sessions.rst
@@ -68,10 +68,19 @@ Assign your back‑end when constructing your application:
Expiring sessions
-----------------
-The global constant :data:`micropie.SESSION_TIMEOUT` controls how long
-a session remains valid (in seconds). The default is eight hours.
-Back‑ends may choose to enforce this timeout in whatever manner makes
-sense for their storage layer.
+Pass ``session_timeout`` to :class:`~micropie.App` to control how long
+sessions remain valid (in seconds). Its default is the eight-hour
+:data:`micropie.SESSION_TIMEOUT` constant. MicroPie passes the configured
+value to :meth:`~micropie.SessionBackend.save` for both HTTP and WebSocket
+sessions:
+
+.. code-block:: python
+
+ app = MyApp(session_timeout=30 * 60) # 30 minutes
+
+The in-memory back-end applies the supplied value independently to each
+session as a sliding inactivity timeout. Custom back-ends should likewise
+honour the value passed to ``save()`` according to their storage model.
Security considerations
-----------------------
@@ -79,6 +88,13 @@ Security considerations
Sessions are sent to the client as cookies and should be treated as
untrusted input. Avoid storing sensitive information directly in the
session. If you implement your own back‑end, ensure that session
-identifiers are random, unique and that the cookie includes the
-``HttpOnly``, ``Secure`` and ``SameSite=Lax`` directives. The
-in‑memory back‑end provided by MicroPie already issues these flags.
\ No newline at end of file
+identifiers are random and unique. MicroPie generates cookies with
+``HttpOnly``, ``Secure`` and ``SameSite=Lax`` by default.
+
+Browsers do not send ``Secure`` cookies over plain HTTP. For local HTTP
+development, disable that attribute explicitly while retaining it in
+production:
+
+.. code-block:: python
+
+ app = MyApp(session_cookie_secure=False)
diff --git a/docs/apidocs/reference/app.rst b/docs/apidocs/reference/app.rst
index 0fb883c..4153556 100644
--- a/docs/apidocs/reference/app.rst
+++ b/docs/apidocs/reference/app.rst
@@ -13,12 +13,23 @@ Uvicorn.
Constructor
-----------
-.. class:: App(session_backend=None)
+.. class:: App(session_backend=None, *, body_timeout=5.0, session_timeout=SESSION_TIMEOUT, session_cookie_secure=True)
Create a new application. If *session_backend* is provided it must
be an instance of :class:`~micropie.SessionBackend`. When omitted,
MicroPie uses an in‑memory back‑end.
+ *body_timeout* is the maximum number of seconds MicroPie waits for
+ each non-multipart request-body chunk. The timeout restarts after
+ every chunk, so it limits idle time rather than total upload time.
+ Set it to ``None`` to disable the framework deadline.
+
+ *session_timeout* is passed to the session back-end whenever session
+ data is saved. *session_cookie_secure* controls whether generated
+ session cookies include the ``Secure`` attribute. It defaults to
+ ``True``; applications served over plain HTTP during local development
+ can set it to ``False`` explicitly.
+
Attributes
----------
@@ -38,6 +49,21 @@ Attributes
The active :class:`~micropie.SessionBackend` instance used to load
and save session dictionaries for both HTTP and WebSocket flows.
+.. attribute:: body_timeout
+
+ The configured per-chunk idle timeout for non-multipart request
+ bodies, or ``None`` when timeout enforcement is disabled.
+
+.. attribute:: session_timeout
+
+ The expiration timeout passed to the session back-end when HTTP or
+ WebSocket session data is saved.
+
+.. attribute:: session_cookie_secure
+
+ Whether HTTP and WebSocket session cookies include the ``Secure``
+ attribute.
+
.. attribute:: startup_handlers
A list of asynchronous callables that run during the ASGI
@@ -81,3 +107,11 @@ Additionally, MicroPie defines several private helper methods such as
``_parse_cookies`` and ``_send_response``. These are considered
internal and not part of the public API. They may change without
notice.
+
+Framework diagnostics
+---------------------
+
+MicroPie writes framework diagnostics and exception tracebacks to the
+standard ``micropie`` Python logger. Configure this logger through the
+:mod:`logging` package to integrate it with your deployment's handlers
+and formatters.
diff --git a/docs/apidocs/reference/session.rst b/docs/apidocs/reference/session.rst
index c3c3aa4..8d2f5c3 100644
--- a/docs/apidocs/reference/session.rst
+++ b/docs/apidocs/reference/session.rst
@@ -42,12 +42,15 @@ InMemorySessionBackend
.. method:: load(session_id)
Return the session dictionary if it exists and has not expired
- according to :data:`micropie.SESSION_TIMEOUT`, otherwise return
- an empty dictionary.
+ according to the timeout supplied when it was saved, otherwise
+ return an empty dictionary. Successful loads refresh the
+ inactivity timer.
.. method:: save(session_id, data, timeout)
- Store *data* under *session_id* and update the last access time.
+ Store *data* under *session_id*, retain *timeout* for that session,
+ and update the last access time. Empty data or a non-positive
+ timeout deletes the session.
SESSION_TIMEOUT
---------------
@@ -55,9 +58,10 @@ SESSION_TIMEOUT
.. data:: SESSION_TIMEOUT
The default session expiration time in seconds (eight hours). You
- can override this constant in your own code by assigning a new value
- to ``micropie.SESSION_TIMEOUT`` or by using a custom back-end that
- persists with a different timeout policy.
+ can select a different value per application with
+ ``App(session_timeout=...)``. The constant supplies the default; it
+ is not consulted as mutable runtime policy by existing applications
+ or saved in-memory sessions.
See also the :doc:`../howto/sessions` guide for examples of using and
implementing session back‑ends.
diff --git a/docs/apidocs/reference/websocket.rst b/docs/apidocs/reference/websocket.rst
index afee227..c1e8c4b 100644
--- a/docs/apidocs/reference/websocket.rst
+++ b/docs/apidocs/reference/websocket.rst
@@ -10,11 +10,13 @@ request and passes it as the first argument to your WebSocket handler.
Constructor
-----------
-.. class:: WebSocket(receive, send)
+.. class:: WebSocket(receive, send, *, session_cookie_secure=True)
Create a new WebSocket wrapper around the ASGI ``receive`` and
``send`` callables. You do not instantiate this class yourself;
- MicroPie does so internally.
+ MicroPie does so internally and supplies the owning application's
+ session-cookie policy. Direct construction keeps secure cookies by
+ default.
Methods
-------
diff --git a/micropie.py b/micropie.py
index 813ba3f..59a97d8 100644
--- a/micropie.py
+++ b/micropie.py
@@ -13,9 +13,9 @@ __license__ = "BSD3"
import asyncio
import contextvars
import inspect
+import logging
import re
import time
-import traceback
import uuid
from abc import ABC, abstractmethod
from typing import Any, Awaitable, Callable, Dict, List, NamedTuple, Optional, Tuple
@@ -41,6 +41,9 @@ except ImportError:
MULTIPART_INSTALLED = False
+logger = logging.getLogger(__name__)
+
+
_PARAM_EMPTY = inspect.Parameter.empty
_VAR_POSITIONAL = inspect.Parameter.VAR_POSITIONAL
_POSITIONAL_PARAM_KINDS = (
@@ -57,6 +60,18 @@ _JSON_HEADERS_BYTES = [_JSON_HEADER_BYTES]
_ROUTE_ATTR_MISSING = object()
+def _format_session_cookie(session_id: str, *, secure: bool) -> str:
+ attributes = [
+ f"session_id={session_id}",
+ "Path=/",
+ "SameSite=Lax",
+ "HttpOnly",
+ ]
+ if secure:
+ attributes.append("Secure")
+ return "; ".join(attributes) + ";"
+
+
class _HandlerParam(NamedTuple):
name: str
kind: Any
@@ -103,31 +118,37 @@ class InMemorySessionBackend(SessionBackend):
def __init__(self):
self.sessions: Dict[str, Dict[str, Any]] = {}
self.last_access: Dict[str, float] = {}
+ self.timeouts: Dict[str, int] = {}
self._next_cleanup: float = 0.0
self._cleanup_interval: float = 60.0
def _cleanup(self, now: Optional[float] = None, *, force: bool = False):
- """Remove expired sessions based on SESSION_TIMEOUT."""
+ """Remove sessions whose configured inactivity timeout has elapsed."""
if now is None:
now = time.time()
if not force and now < self._next_cleanup:
return
self._next_cleanup = now + self._cleanup_interval
expired = [
- sid for sid, ts in self.last_access.items() if now - ts >= SESSION_TIMEOUT
+ sid
+ for sid, ts in self.last_access.items()
+ if now - ts >= self.timeouts.get(sid, SESSION_TIMEOUT)
]
for sid in expired:
self.sessions.pop(sid, None)
self.last_access.pop(sid, None)
+ self.timeouts.pop(sid, None)
async def load(self, session_id: str) -> Dict[str, Any]:
now = time.time()
self._cleanup(now)
last_access = self.last_access.get(session_id)
if last_access is not None:
- if now - last_access >= SESSION_TIMEOUT:
+ timeout = self.timeouts.get(session_id, SESSION_TIMEOUT)
+ if now - last_access >= timeout:
self.sessions.pop(session_id, None)
self.last_access.pop(session_id, None)
+ self.timeouts.pop(session_id, None)
return {}
self.last_access[session_id] = now
return self.sessions.get(session_id, {})
@@ -135,13 +156,15 @@ class InMemorySessionBackend(SessionBackend):
async def save(self, session_id: str, data: Dict[str, Any], timeout: int) -> None:
self._cleanup()
- if not data:
+ if not data or timeout <= 0:
# treat empty as delete
self.sessions.pop(session_id, None)
self.last_access.pop(session_id, None)
+ self.timeouts.pop(session_id, None)
else:
self.sessions[session_id] = data
self.last_access[session_id] = time.time()
+ self.timeouts[session_id] = timeout
# -----------------------------
@@ -231,6 +254,8 @@ class WebSocket:
self,
receive: Callable[[], Awaitable[Dict[str, Any]]],
send: Callable[[Dict[str, Any]], Awaitable[None]],
+ *,
+ session_cookie_secure: bool = True,
) -> None:
"""
Initialize a WebSocket instance.
@@ -238,11 +263,13 @@ class WebSocket:
Args:
receive: The ASGI receive callable.
send: The ASGI send callable.
+ session_cookie_secure: Whether session cookies include the Secure flag.
"""
self.receive = receive
self.send = send
self.accepted = False
self.session_id: Optional[str] = None
+ self.session_cookie_secure = session_cookie_secure
async def accept(
self, subprotocol: Optional[str] = None, session_id: Optional[str] = None
@@ -265,7 +292,9 @@ class WebSocket:
headers.append(
(
"Set-Cookie",
- f"session_id={session_id}; Path=/; SameSite=Lax; HttpOnly; Secure;",
+ _format_session_cookie(
+ session_id, secure=self.session_cookie_secure
+ ),
)
)
self.session_id = session_id
@@ -453,7 +482,18 @@ class App:
and startup/shutdown handlers via 'startup_handlers' and 'shutdown_handlers'.
"""
- def __init__(self, session_backend: Optional[SessionBackend] = None) -> None:
+ def __init__(
+ self,
+ session_backend: Optional[SessionBackend] = None,
+ *,
+ body_timeout: Optional[float] = 5.0,
+ session_timeout: int = SESSION_TIMEOUT,
+ session_cookie_secure: bool = True,
+ ) -> None:
+ if body_timeout is not None and body_timeout <= 0:
+ raise ValueError("body_timeout must be positive or None")
+ if session_timeout <= 0:
+ raise ValueError("session_timeout must be positive")
if JINJA_INSTALLED:
self.env = Environment(
loader=FileSystemLoader("templates"),
@@ -465,6 +505,9 @@ class App:
self.session_backend: SessionBackend = (
session_backend or InMemorySessionBackend()
)
+ self.body_timeout = body_timeout
+ self.session_timeout = session_timeout
+ self.session_cookie_secure = session_cookie_secure
self.middlewares: List[HttpMiddleware] = []
self.ws_middlewares: List[WebSocketMiddleware] = []
self.startup_handlers: List[Callable[[], Awaitable[None]]] = []
@@ -628,14 +671,20 @@ class App:
parse_task: Optional[asyncio.Task] = (
None # background multipart task if started
)
+ multipart_file_notifications: Optional[asyncio.Queue] = None
async def _cancel_parse_task():
- if parse_task is not None and not parse_task.done():
+ if parse_task is None:
+ return
+ if not parse_task.done():
parse_task.cancel()
- try:
- await parse_task
- except asyncio.CancelledError:
- pass
+ try:
+ await parse_task
+ except asyncio.CancelledError:
+ pass
+ except Exception:
+ # The parser runner logs the original exception.
+ pass
async def _early_exit(
code: int, body: Any, headers: Optional[List[Tuple[str, str]]] = None
@@ -651,14 +700,18 @@ class App:
"""
if name in request.files:
return request.files[name]
- if parse_task is None:
+ if parse_task is None or multipart_file_notifications is None:
return None
- while True:
- if name in request.files:
- return request.files[name]
+ while name not in request.files:
if parse_task.done():
- return None
- await asyncio.sleep(0)
+ try:
+ await parse_task
+ except Exception:
+ # The parser runner logs the original exception.
+ pass
+ break
+ await multipart_file_notifications.get()
+ return request.files.get(name)
try:
# Parse query/cookies/session
@@ -676,32 +729,50 @@ class App:
):
if "multipart/form-data" in content_type:
if not MULTIPART_INSTALLED:
- print("For multipart form data support install 'multipart'.")
+ logger.error(
+ "Multipart form data requires the 'multipart' package"
+ )
await _early_exit(500, "500 Internal Server Error")
return
boundary_match = re.search(r"boundary=([^;]+)", content_type)
if not boundary_match:
await _early_exit(400, "400 Bad Request: Missing boundary")
return
- # Start parsing in the background; do NOT await here so handlers/middleware can run concurrently.
- parse_task = asyncio.create_task(
- self._parse_multipart_into_request(
- receive,
- boundary_match.group(1).encode("utf-8"),
- request,
- file_queue_maxsize=2048,
- )
- )
+ multipart_file_notifications = asyncio.Queue()
+
+ async def _run_multipart_parser() -> None:
+ try:
+ await self._parse_multipart_into_request(
+ receive,
+ boundary_match.group(1).encode("utf-8"),
+ request,
+ file_notifications=multipart_file_notifications,
+ file_queue_maxsize=2048,
+ )
+ except asyncio.CancelledError:
+ raise
+ except Exception:
+ logger.exception("Failed to parse multipart request body")
+ raise
+ finally:
+ multipart_file_notifications.put_nowait(None)
+
+ # Parse in the background so handlers can drain file queues
+ # while later upload chunks are still arriving.
+ parse_task = asyncio.create_task(_run_multipart_parser())
else:
body_chunks: List[bytes] = []
try:
- async with asyncio.timeout(5): # Timeout after 5 seconds
- while True:
+ while True:
+ if self.body_timeout is None:
msg = await receive()
- if chunk := msg.get("body", b""):
- body_chunks.append(chunk)
- if not msg.get("more_body"):
- break
+ else:
+ async with asyncio.timeout(self.body_timeout):
+ msg = await receive()
+ if chunk := msg.get("body", b""):
+ body_chunks.append(chunk)
+ if not msg.get("more_body"):
+ break
except asyncio.TimeoutError:
await _early_exit(
408, "408 Request Timeout: Failed to receive body"
@@ -856,7 +927,9 @@ class App:
else handler(*func_args)
)
except Exception:
- traceback.print_exc()
+ logger.exception(
+ "Unhandled exception in HTTP handler for %s", scope.get("path", "")
+ )
await _early_exit(500, "500 Internal Server Error")
return
@@ -866,7 +939,8 @@ class App:
await parse_task
request.body_parsed = True
except Exception:
- traceback.print_exc()
+ # The parser runner logs the original exception.
+ pass
# Normalize response
if isinstance(result, tuple):
@@ -899,11 +973,13 @@ class App:
extra_headers.append(
(
"Set-Cookie",
- f"session_id={session_id}; Path=/; SameSite=Lax; HttpOnly; Secure;",
+ _format_session_cookie(
+ session_id, secure=self.session_cookie_secure
+ ),
)
)
await self.session_backend.save(
- session_id, request.session, SESSION_TIMEOUT
+ session_id, request.session, self.session_timeout
)
elif session_id:
# Empty session and existing cookie -> treat as logout/delete
@@ -1044,7 +1120,11 @@ class App:
path_param_count = len(path_params)
query_params = request.query_params
session = request.session
- ws = WebSocket(receive, send)
+ ws = WebSocket(
+ receive,
+ send,
+ session_cookie_secure=self.session_cookie_secure,
+ )
func_args.append(ws) # First non-self parameter is WebSocket object
for param in handler_info.params:
if param.name == "ws": # Skip the WebSocket parameter
@@ -1081,7 +1161,10 @@ class App:
except ConnectionClosed:
pass # Normal closure, no need to send another close message
except Exception as e:
- traceback.print_exc()
+ logger.exception(
+ "Unhandled exception in WebSocket handler for %s",
+ scope.get("path", ""),
+ )
await self._send_websocket_close(send, 1011, f"Handler error: {str(e)}")
return
@@ -1092,7 +1175,7 @@ class App:
# Save / clear session after middlewares
if request.session:
await self.session_backend.save(
- ws.session_id, request.session, SESSION_TIMEOUT
+ ws.session_id, request.session, self.session_timeout
)
elif had_session_id:
# Treat empty session as logout/delete
@@ -1126,6 +1209,7 @@ class App:
boundary: bytes,
request: "Request",
*,
+ file_notifications: Optional[asyncio.Queue] = None,
file_queue_maxsize: int = 2048,
) -> None:
"""
@@ -1178,6 +1262,8 @@ class App:
or "application/octet-stream",
"content": current_queue,
}
+ if file_notifications is not None:
+ file_notifications.put_nowait(current_field_name)
else:
# Text field
request.body_params.setdefault(current_field_name, [])
@@ -1224,7 +1310,9 @@ class App:
has_content_type = False
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}")
+ logger.warning(
+ "Rejected response header containing a newline: %r: %r", k, v
+ )
continue
if k.lower() == "content-type":
has_content_type = True
@@ -1333,7 +1421,7 @@ class App:
The rendered template as a string.
"""
if not JINJA_INSTALLED:
- print("To use the `_render_template` method install 'jinja2'.")
+ logger.error("Template rendering requires the 'jinja2' package")
return "500 Internal Server Error: Jinja2 not installed."
assert self.env is not None
template = await asyncio.to_thread(self.env.get_template, name)
diff --git a/tests.py b/tests.py
index 0edd055..f7485f4 100644
--- a/tests.py
+++ b/tests.py
@@ -178,6 +178,36 @@ class TestSession(MicroPieTestCase):
expired_data = await backend.load(session_id)
self.assertEqual(expired_data, {}, "Expired session should return empty dict")
+ async def test_in_memory_backend_honors_each_saved_timeout(self):
+ """Each in-memory session should use the timeout supplied to save()."""
+ backend = InMemorySessionBackend()
+
+ with patch("micropie.time.time", return_value=100.0):
+ await backend.save("short", {"user": "alice"}, 10)
+ await backend.save("long", {"user": "bob"}, 100)
+
+ backend._cleanup(now=111.0, force=True)
+ self.assertNotIn("short", backend.sessions)
+ self.assertNotIn("short", backend.timeouts)
+ self.assertIn("long", backend.sessions)
+
+ with patch("micropie.time.time", return_value=150.0):
+ self.assertEqual(await backend.load("long"), {"user": "bob"})
+ with patch("micropie.time.time", return_value=250.0):
+ self.assertEqual(await backend.load("long"), {})
+
+ self.assertNotIn("long", backend.sessions)
+ self.assertNotIn("long", backend.timeouts)
+
+ async def test_nonpositive_timeout_deletes_in_memory_session(self):
+ """A non-positive timeout should delete even non-empty session data."""
+ backend = InMemorySessionBackend()
+ await backend.save("abc", {"user": "alice"}, 10)
+ await backend.save("abc", {"user": "alice"}, 0)
+
+ self.assertEqual(await backend.load("abc"), {})
+ self.assertNotIn("abc", backend.timeouts)
+
async def test_cookie_parsing(self):
"""Test parsing of cookie header."""
cookie_header = "session_id=abc123; theme=dark; user=john"
@@ -226,8 +256,58 @@ class TestSession(MicroPieTestCase):
),
"Set-Cookie header with session_id not found",
)
+ self.assertTrue(
+ any(
+ h[0] == b"Set-Cookie" and b"; Secure;" in h[1]
+ for h in set_cookie_call["headers"]
+ ),
+ "Session cookies should be Secure by default",
+ )
self.assertEqual(set_cookie_call["status"], 200, "Status should be 200")
+ async def test_app_session_configuration_controls_save_and_cookie(self):
+ """App session settings should reach the backend and cookie formatter."""
+ backend = CountingSessionBackend()
+ self.app = App(
+ session_backend=backend,
+ session_timeout=123,
+ session_cookie_secure=False,
+ )
+
+ async def set_session(self):
+ self.request.session["user"] = "alice"
+ return "OK"
+
+ setattr(self.app, "set_session", set_session.__get__(self.app, App))
+ 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)
+
+ self.assertEqual(backend.save_calls[-1][2], 123)
+ start = next(
+ call[0][0]
+ for call in send.call_args_list
+ if call[0][0]["type"] == "http.response.start"
+ )
+ cookie = next(
+ value for name, value in start["headers"] if name == b"Set-Cookie"
+ )
+ self.assertIn(b"HttpOnly", cookie)
+ self.assertNotIn(b"Secure", cookie)
+
+ async def test_app_rejects_invalid_timeout_configuration(self):
+ """Enabled body and session timeouts must be positive."""
+ with self.assertRaisesRegex(ValueError, "body_timeout"):
+ App(body_timeout=0)
+ with self.assertRaisesRegex(ValueError, "session_timeout"):
+ App(session_timeout=0)
+
+ self.assertIsNone(App(body_timeout=None).body_timeout)
+
class TestFastPaths(MicroPieTestCase):
"""Tests for optimized dispatch/session paths."""
@@ -390,6 +470,286 @@ class TestFastPaths(MicroPieTestCase):
send.assert_any_call({"type": "websocket.send", "text": "lobby:alice"})
+class TestRequestBodyTimeouts(MicroPieTestCase):
+ """Tests for configurable non-multipart body receive timeouts."""
+
+ async def test_timeout_resets_for_each_body_chunk(self):
+ """A slow body succeeds while each individual chunk arrives in time."""
+ self.app = App(body_timeout=0.2)
+
+ async def submit(self, value):
+ return value
+
+ setattr(self.app, "submit", submit.__get__(self.app, App))
+ chunks = [
+ {"type": "http.request", "body": b"val", "more_body": True},
+ {"type": "http.request", "body": b"ue=", "more_body": True},
+ {"type": "http.request", "body": b"slow", "more_body": False},
+ ]
+
+ async def receive():
+ await asyncio.sleep(0.08)
+ return chunks.pop(0)
+
+ scope = self.create_mock_scope(
+ path="/submit",
+ method="POST",
+ headers=[(b"content-type", b"application/x-www-form-urlencoded")],
+ )
+ send = AsyncMock()
+
+ await self.app(scope, receive, send)
+
+ send.assert_any_call(
+ {"type": "http.response.body", "body": b"slow", "more_body": False}
+ )
+
+ async def test_stalled_body_returns_408(self):
+ """A receive call exceeding the idle timeout should return 408."""
+ self.app = App(body_timeout=0.01)
+
+ async def receive():
+ await asyncio.sleep(1)
+ return {"type": "http.request", "body": b"value=late", "more_body": False}
+
+ scope = self.create_mock_scope(
+ path="/index",
+ method="POST",
+ headers=[(b"content-type", b"application/x-www-form-urlencoded")],
+ )
+ send = AsyncMock()
+
+ await self.app(scope, receive, send)
+
+ send.assert_any_call(
+ {
+ "type": "http.response.start",
+ "status": 408,
+ "headers": [(b"Content-Type", b"text/html; charset=utf-8")],
+ }
+ )
+
+ async def test_body_timeout_can_be_disabled(self):
+ """None should allow a receive call to wait without a framework deadline."""
+ self.app = App(body_timeout=None)
+
+ async def submit(self, value):
+ return value
+
+ setattr(self.app, "submit", submit.__get__(self.app, App))
+
+ async def receive():
+ await asyncio.sleep(0.02)
+ return {
+ "type": "http.request",
+ "body": b"value=received",
+ "more_body": False,
+ }
+
+ scope = self.create_mock_scope(
+ path="/submit",
+ method="POST",
+ headers=[(b"content-type", b"application/x-www-form-urlencoded")],
+ )
+ send = AsyncMock()
+
+ await self.app(scope, receive, send)
+
+ send.assert_any_call(
+ {
+ "type": "http.response.body",
+ "body": b"received",
+ "more_body": False,
+ }
+ )
+
+
+class TestMultipartCoordination(MicroPieTestCase):
+ """Tests for notification-driven multipart file binding."""
+
+ def multipart_scope(self):
+ return self.create_mock_scope(
+ path="/upload",
+ method="POST",
+ headers=[
+ (b"content-type", b"multipart/form-data; boundary=test-boundary")
+ ],
+ )
+
+ async def test_file_notification_starts_handler_before_parse_completion(self):
+ """Handlers can drain a file queue while the parser is still active."""
+ parser_may_finish = asyncio.Event()
+ handler_received_file = asyncio.Event()
+
+ async def fake_parse(
+ receive,
+ boundary,
+ request,
+ *,
+ file_notifications=None,
+ file_queue_maxsize=2048,
+ ):
+ content = asyncio.Queue(maxsize=file_queue_maxsize)
+ request.files["upload"] = {
+ "filename": "example.txt",
+ "content_type": "text/plain",
+ "content": content,
+ }
+ file_notifications.put_nowait("upload")
+ await content.put(b"streamed")
+ await content.put(None)
+ await parser_may_finish.wait()
+
+ self.app._parse_multipart_into_request = fake_parse
+
+ async def upload(self, upload):
+ chunk = await upload["content"].get()
+ handler_received_file.set()
+ parser_may_finish.set()
+ return chunk
+
+ setattr(self.app, "upload", upload.__get__(self.app, App))
+ send = AsyncMock()
+
+ with patch("micropie.MULTIPART_INSTALLED", True):
+ await asyncio.wait_for(
+ self.app(self.multipart_scope(), AsyncMock(), send), timeout=1
+ )
+
+ self.assertTrue(handler_received_file.is_set())
+ send.assert_any_call(
+ {
+ "type": "http.response.body",
+ "body": b"streamed",
+ "more_body": False,
+ }
+ )
+
+ async def test_parser_completion_wakes_missing_file_waiter(self):
+ """A missing file should resolve to a 400 without polling or hanging."""
+
+ async def fake_parse(
+ receive,
+ boundary,
+ request,
+ *,
+ file_notifications=None,
+ file_queue_maxsize=2048,
+ ):
+ return None
+
+ self.app._parse_multipart_into_request = fake_parse
+
+ async def upload(self, upload):
+ return "unreachable"
+
+ setattr(self.app, "upload", upload.__get__(self.app, App))
+ send = AsyncMock()
+
+ with patch("micropie.MULTIPART_INSTALLED", True):
+ await asyncio.wait_for(
+ self.app(self.multipart_scope(), AsyncMock(), send), timeout=1
+ )
+
+ send.assert_any_call(
+ {
+ "type": "http.response.start",
+ "status": 400,
+ "headers": [(b"Content-Type", b"text/html; charset=utf-8")],
+ }
+ )
+
+ async def test_parser_failure_is_logged_and_retrieved(self):
+ """Parser failures should wake waiters and be consumed by the request task."""
+
+ async def fake_parse(
+ receive,
+ boundary,
+ request,
+ *,
+ file_notifications=None,
+ file_queue_maxsize=2048,
+ ):
+ raise RuntimeError("broken multipart body")
+
+ self.app._parse_multipart_into_request = fake_parse
+
+ async def upload(self, upload):
+ return "unreachable"
+
+ setattr(self.app, "upload", upload.__get__(self.app, App))
+ send = AsyncMock()
+
+ with patch("micropie.MULTIPART_INSTALLED", True), self.assertLogs(
+ "micropie", level="ERROR"
+ ) as logs:
+ await asyncio.wait_for(
+ self.app(self.multipart_scope(), AsyncMock(), send), timeout=1
+ )
+
+ self.assertTrue(
+ any(
+ "Failed to parse multipart request body" in line
+ for line in logs.output
+ )
+ )
+ send.assert_any_call(
+ {
+ "type": "http.response.start",
+ "status": 400,
+ "headers": [(b"Content-Type", b"text/html; charset=utf-8")],
+ }
+ )
+
+ async def test_early_response_cancels_and_retrieves_parser(self):
+ """Middleware short-circuiting should cancel the background parser cleanly."""
+ parser_started = asyncio.Event()
+ parser_cancelled = asyncio.Event()
+
+ async def fake_parse(
+ receive,
+ boundary,
+ request,
+ *,
+ file_notifications=None,
+ file_queue_maxsize=2048,
+ ):
+ parser_started.set()
+ try:
+ await asyncio.Future()
+ except asyncio.CancelledError:
+ parser_cancelled.set()
+ raise
+
+ class RejectUpload(HttpMiddleware):
+ async def before_request(self, request):
+ await parser_started.wait()
+ return {"status_code": 413, "body": "too large"}
+
+ async def after_request(
+ self, request, status_code, response_body, extra_headers
+ ):
+ return None
+
+ self.app._parse_multipart_into_request = fake_parse
+ self.app.middlewares.append(RejectUpload())
+ send = AsyncMock()
+
+ with patch("micropie.MULTIPART_INSTALLED", True):
+ await asyncio.wait_for(
+ self.app(self.multipart_scope(), AsyncMock(), send), timeout=1
+ )
+
+ self.assertTrue(parser_cancelled.is_set())
+ send.assert_any_call(
+ {
+ "type": "http.response.start",
+ "status": 413,
+ "headers": [(b"Content-Type", b"text/html; charset=utf-8")],
+ }
+ )
+
+
class TestRouting(MicroPieTestCase):
"""Tests for HTTP and WebSocket routing."""
@@ -656,6 +1016,74 @@ class TestWebSocket(MicroPieTestCase):
}
)
+ async def test_websocket_cookie_respects_secure_configuration(self):
+ """WebSocket handshake cookies should use the App cookie policy."""
+ self.app = App(session_cookie_secure=False)
+
+ async def ws_cookie(self, ws):
+ await ws.accept(session_id="abc")
+ await ws.close()
+
+ setattr(self.app, "ws_cookie", ws_cookie.__get__(self.app, App))
+ scope = self.create_mock_scope(path="/cookie", scope_type="websocket")
+ receive = AsyncMock(return_value={"type": "websocket.connect"})
+ send = AsyncMock()
+
+ await self.app(scope, receive, send)
+
+ accept = next(
+ call[0][0]
+ for call in send.call_args_list
+ if call[0][0]["type"] == "websocket.accept"
+ )
+ cookie = next(
+ value for name, value in accept["headers"] if name == b"Set-Cookie"
+ )
+ self.assertIn(b"HttpOnly", cookie)
+ self.assertNotIn(b"Secure", cookie)
+
+ async def test_websocket_session_uses_app_timeout(self):
+ """WebSocket session saves should receive the configured timeout."""
+ backend = CountingSessionBackend()
+ self.app = App(session_backend=backend, session_timeout=321)
+
+ async def ws_session(self, ws):
+ self.request.session["user"] = "alice"
+ await ws.accept()
+ await ws.close()
+
+ setattr(self.app, "ws_session", ws_session.__get__(self.app, App))
+ scope = self.create_mock_scope(path="/session", scope_type="websocket")
+ receive = AsyncMock(return_value={"type": "websocket.connect"})
+ send = AsyncMock()
+
+ await self.app(scope, receive, send)
+
+ self.assertEqual(backend.save_calls[-1][2], 321)
+
+ async def test_websocket_handler_exception_is_logged(self):
+ """WebSocket handler failures should include traceback logging."""
+
+ async def ws_broken(self, ws):
+ raise RuntimeError("websocket boom")
+
+ setattr(self.app, "ws_broken", ws_broken.__get__(self.app, App))
+ scope = self.create_mock_scope(path="/broken", scope_type="websocket")
+ receive = AsyncMock(return_value={"type": "websocket.connect"})
+ send = AsyncMock()
+
+ with self.assertLogs("micropie", level="ERROR") as logs:
+ await self.app(scope, receive, send)
+
+ self.assertTrue(any(record.exc_info for record in logs.records))
+ send.assert_any_call(
+ {
+ "type": "websocket.close",
+ "code": 1011,
+ "reason": "Handler error: websocket boom",
+ }
+ )
+
class TestMiddleware(MicroPieTestCase):
"""Tests for HTTP and WebSocket middleware."""
@@ -789,6 +1217,31 @@ class TestResponseHandling(MicroPieTestCase):
}
)
+ async def test_http_handler_exception_is_logged(self):
+ """HTTP handler failures should include traceback logging."""
+
+ async def index(self):
+ raise RuntimeError("http boom")
+
+ 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()
+
+ with self.assertLogs("micropie", level="ERROR") as logs:
+ await self.app(scope, receive, send)
+
+ self.assertTrue(any(record.exc_info for record in logs.records))
+ send.assert_any_call(
+ {
+ "type": "http.response.start",
+ "status": 500,
+ "headers": [(b"Content-Type", b"text/html; charset=utf-8")],
+ }
+ )
+
async def test_header_injection(self):
"""Test protection against header injection."""
@@ -803,7 +1256,12 @@ class TestResponseHandling(MicroPieTestCase):
)
send = AsyncMock()
- await self.app(scope, receive, send)
+ with self.assertLogs("micropie", level="WARNING") as logs:
+ await self.app(scope, receive, send)
+
+ self.assertTrue(
+ any("Rejected response header" in line for line in logs.output)
+ )
start_call = None
for call in send.call_args_list:
@@ -840,7 +1298,9 @@ class TestOptionalDependencies(MicroPieTestCase):
async def test_no_multipart_installed(self):
"""Test behavior when multipart is not installed."""
- with patch("micropie.MULTIPART_INSTALLED", False):
+ with patch("micropie.MULTIPART_INSTALLED", False), self.assertLogs(
+ "micropie", level="ERROR"
+ ) as logs:
scope = self.create_mock_scope(
path="/index",
method="POST",
@@ -855,6 +1315,10 @@ class TestOptionalDependencies(MicroPieTestCase):
await self.app(scope, receive, send)
+ self.assertTrue(
+ any("requires the 'multipart' package" in line for line in logs.output)
+ )
+
send.assert_any_call(
{
"type": "http.response.start",
@@ -872,7 +1336,9 @@ class TestOptionalDependencies(MicroPieTestCase):
async def test_no_jinja_installed(self):
"""Test behavior when Jinja2 is not installed."""
- with patch("micropie.JINJA_INSTALLED", False):
+ with patch("micropie.JINJA_INSTALLED", False), self.assertLogs(
+ "micropie", level="ERROR"
+ ) as logs:
async def index(self):
return await self._render_template("test.html")
@@ -887,6 +1353,10 @@ class TestOptionalDependencies(MicroPieTestCase):
await self.app(scope, receive, send)
+ self.assertTrue(
+ any("requires the 'jinja2' package" in line for line in logs.output)
+ )
+
send.assert_any_call(
{
"type": "http.response.start",