patx/wred
Isolate example app state and clean up Python style
Commit f879b5b · patx · 2026-10-18T12:25:04-04:00
Isolate example app state and clean up Python style Give each example app its own rate limiter, separate ASGI startup from the factory, and support injected databases. Apply Pocoo-style formatting to examples and tests, configure Ruff, and correct the README's UTF-8 decoding description.
Comments
No comments yet.
Diff
diff --git a/README.md b/README.md
index ccafe98..da9292e 100644
--- a/README.md
+++ b/README.md
@@ -115,7 +115,7 @@ async def search(request):
```
Query parameters preserve blank values; repeated keys keep the last value.
-Invalid percent-encoded UTF-8 is replaced with `�`.
+Invalid UTF-8 in raw or percent-encoded query bytes raises `UnicodeDecodeError`.
JSON can be an object, array, or scalar. Validate inputs and convert types yourself.
Authentication middleware can populate the scope for handlers to check.
diff --git a/example_app/app.py b/example_app/app.py
index 62d3498..2b4a558 100644
--- a/example_app/app.py
+++ b/example_app/app.py
@@ -1,11 +1,14 @@
-"""Run the notebook demo with MongoDB listening on localhost:27017:
+# -*- coding: utf-8 -*-
+"""
+ example_app.app
+ ~~~~~~~~~~~~~~~
- uv sync --extra example
- uv run --extra example weft example_app.app:app
+ Build the notebook demo explicitly with ``create_app``. Importing this
+ module does not create an application or a database client. See
+ ``example_app.asgi`` for the server entry point and startup instructions.
-Open http://127.0.0.1:8000. MongoDB preserves accounts, sessions, and notes.
-Set MONGODB_URI and MONGODB_DATABASE to choose another server/database.
-Rate limits are process-local. Use HTTPS outside localhost.
+ :copyright: (c) 2026 by Harrison Erd.
+ :license: BSD-3-Clause, see LICENSE for more details.
"""
import os
@@ -13,17 +16,19 @@ import os
from wred import App
from . import auth, notes, pages
+from .rate_limit import RateLimiter
from .storage import Database
class NotebookApp:
"""Own the example's database lifecycle while keeping Wred HTTP-only."""
- def __init__(self, mongodb_uri, database_name):
- self.database = Database(mongodb_uri, database_name)
+ def __init__(self, database):
+ self.database = database
self.http = App()
- pages.register_routes(self.http)
- auth.register_routes(self.http, self.database)
+ rate_limit = RateLimiter()
+ pages.register_routes(self.http, rate_limit)
+ auth.register_routes(self.http, self.database, rate_limit)
notes.register_routes(self.http, self.database)
async def __call__(self, scope, receive, send):
@@ -36,7 +41,10 @@ class NotebookApp:
await self.database.initialize()
except Exception as error:
await self.database.close()
- await send({'type': 'lifespan.startup.failed', 'message': str(error)})
+ await send({
+ 'type': 'lifespan.startup.failed',
+ 'message': str(error),
+ })
return
await send({'type': 'lifespan.startup.complete'})
elif message['type'] == 'lifespan.shutdown':
@@ -45,12 +53,24 @@ class NotebookApp:
return
-def create_app(mongodb_uri=None, database_name=None):
- """Build an app with one native async MongoDB client per worker."""
- return NotebookApp(
- mongodb_uri if mongodb_uri is not None else os.environ.get('MONGODB_URI', 'mongodb://127.0.0.1:27017'),
- database_name if database_name is not None else os.environ.get('MONGODB_DATABASE', 'wred_notebook'),
- )
+def create_app(mongodb_uri=None, database_name=None, *, database=None):
+ """Build an app with independent rate limits and an owned database.
+ Without ``database``, create a MongoDB client using the explicit URI and
+ name, falling back to ``MONGODB_URI`` and ``MONGODB_DATABASE``. Defaults
+ are ``mongodb://127.0.0.1:27017`` and ``wred_notebook``.
-app = create_app()
+ An injected ``database`` takes precedence over URI and name settings.
+ The app calls its async ``initialize`` and ``close`` methods during
+ lifespan startup and shutdown, including closing it on startup failure.
+ """
+ if database is None:
+ database = Database(
+ mongodb_uri
+ if mongodb_uri is not None
+ else os.environ.get('MONGODB_URI', 'mongodb://127.0.0.1:27017'),
+ database_name
+ if database_name is not None
+ else os.environ.get('MONGODB_DATABASE', 'wred_notebook'),
+ )
+ return NotebookApp(database)
diff --git a/example_app/asgi.py b/example_app/asgi.py
new file mode 100644
index 0000000..405da61
--- /dev/null
+++ b/example_app/asgi.py
@@ -0,0 +1,24 @@
+# -*- coding: utf-8 -*-
+"""
+ example_app.asgi
+ ~~~~~~~~~~~~~~~~
+
+ Run the notebook demo with MongoDB listening on localhost:27017::
+
+ python -m pip install -e . pymongo
+ weft example_app.asgi:app
+
+ Open http://127.0.0.1:8000. MongoDB preserves accounts, sessions,
+ and notes.
+ Set MONGODB_URI and MONGODB_DATABASE to choose another server/database.
+ Rate limits are isolated per app and process-local. Use HTTPS outside
+ localhost.
+
+ :copyright: (c) 2026 by Harrison Erd.
+ :license: BSD-3-Clause, see LICENSE for more details.
+"""
+
+from .app import create_app
+
+
+app = create_app()
diff --git a/example_app/auth.py b/example_app/auth.py
index 6e18398..22be010 100644
--- a/example_app/auth.py
+++ b/example_app/auth.py
@@ -1,4 +1,13 @@
-"""Signup, login, refresh cookies, and bearer authentication for the demo."""
+# -*- coding: utf-8 -*-
+"""
+ example_app.auth
+ ~~~~~~~~~~~~~~~~
+
+ Signup, login, refresh cookies, and bearer authentication for the demo.
+
+ :copyright: (c) 2026 by Harrison Erd.
+ :license: BSD-3-Clause, see LICENSE for more details.
+"""
from asyncio import to_thread
from functools import wraps
@@ -9,34 +18,38 @@ from time import time
import orjson
-from .rate_limit import rate_limit
-
-token_lifetime = 3600
-refresh_lifetime = 7 * 24 * 3600
-refresh_cookie_name = 'notebook_refresh'
+TOKEN_LIFETIME = 3600
+REFRESH_LIFETIME = 7 * 24 * 3600
+REFRESH_COOKIE_NAME = 'notebook_refresh'
def same_origin_request(request):
- # HTML forms cannot supply this custom header; cross-origin fetch requires a preflight.
+ # HTML forms cannot supply this custom header; cross-origin fetch requires
+ # a preflight.
if request.headers.get('x-notebook-request') != '1':
return False
if request.headers.get('sec-fetch-site', 'same-origin') != 'same-origin':
return False
- # Browsers send Origin on POST; API clients may omit it but still need the custom header.
+ # Browsers send Origin on POST; API clients may omit it but still need the
+ # custom header.
origin = request.headers.get('origin')
- expected = f'{request.scope.get("scheme", "http")}://{request.headers.get("host", "")}'
+ expected = (
+ f'{request.scope.get("scheme", "http")}://'
+ f'{request.headers.get("host", "")}'
+ )
return origin is None or origin == expected
def refresh_cookie(request, value, max_age):
cookie = SimpleCookie()
- cookie[refresh_cookie_name] = value
- morsel = cookie[refresh_cookie_name]
+ cookie[REFRESH_COOKIE_NAME] = value
+ morsel = cookie[REFRESH_COOKIE_NAME]
morsel['path'] = '/api'
morsel['httponly'] = True
morsel['samesite'] = 'Strict'
morsel['max-age'] = max_age
- # HTTPS deployments get Secure; plain HTTP remains usable for the localhost demo.
+ # HTTPS deployments get Secure; plain HTTP remains usable for the localhost
+ # demo.
if request.scope.get('scheme') == 'https':
morsel['secure'] = True
return 'Set-Cookie', morsel.OutputString()
@@ -48,7 +61,7 @@ def read_refresh_cookie(request):
cookie.load(request.headers.get('cookie', ''))
except CookieError:
return ''
- morsel = cookie.get(refresh_cookie_name)
+ morsel = cookie.get(REFRESH_COOKIE_NAME)
return morsel.value if morsel else ''
@@ -56,7 +69,7 @@ def token_response(token, username):
return {
'access_token': token,
'token_type': 'bearer',
- 'expires_in': token_lifetime,
+ 'expires_in': TOKEN_LIFETIME,
'user': {'id': username},
}
@@ -73,12 +86,16 @@ async def read_credentials(request):
raise ValueError('Username and password must be strings')
username = username.strip().casefold()
if not 1 <= len(username) <= 64 or not 15 <= len(password) <= 128:
- raise ValueError('Use a username of 1–64 and a password of 15–128 characters')
+ raise ValueError(
+ 'Use a username of 1–64 and a password of 15–128 characters'
+ )
return username, password
async def hash_password(password, salt):
- return await to_thread(pbkdf2_hmac, 'sha256', password.encode('utf-8'), salt, 600_000)
+ return await to_thread(
+ pbkdf2_hmac, 'sha256', password.encode('utf-8'), salt, 600_000
+ )
def protected(database):
@@ -105,15 +122,25 @@ def protected(database):
body, content_type, status = response[:3]
headers = list(response[3]) if len(response) == 4 else []
else:
- body, content_type, status, headers = response, 'application/json', 200, []
- return body, content_type, status, headers + [('Cache-Control', 'no-store')]
+ body, content_type, status, headers = (
+ response,
+ 'application/json',
+ 200,
+ [],
+ )
+ return (
+ body,
+ content_type,
+ status,
+ headers + [('Cache-Control', 'no-store')],
+ )
return wrapper
return decorator
-def register_routes(app, database):
+def register_routes(app, database, rate_limit):
"""Register authentication endpoints on the supplied Wred app."""
@app.route('/api/signup', methods=['POST'])
@@ -127,7 +154,11 @@ def register_routes(app, database):
password_hash = await hash_password(password, salt)
# MongoDB's unique _id also handles signups in different workers.
if not await database.create_user(username, salt, password_hash):
- return {'error': 'Username already exists'}, 'application/json', 409
+ return (
+ {'error': 'Username already exists'},
+ 'application/json',
+ 409,
+ )
return {'id': username}, 'application/json', 201
@app.route('/api/login', methods=['POST'])
@@ -135,7 +166,12 @@ def register_routes(app, database):
async def login(request):
if not same_origin_request(request):
return (
- {'error': 'A same-origin request with X-Notebook-Request: 1 is required'},
+ {
+ 'error': (
+ 'A same-origin request with '
+ 'X-Notebook-Request: 1 is required'
+ )
+ },
'application/json',
403,
)
@@ -143,11 +179,14 @@ def register_routes(app, database):
username, password = await read_credentials(request)
except ValueError as error:
return {'error': str(error)}, 'application/json', 400
- # Use a dummy record for unknown users so failed logins still perform password hashing.
+ # Use a dummy record for unknown users so failed logins still perform
+ # password hashing.
record = await database.get_user(username)
salt = record['salt'] if record else bytes(16)
password_hash = await hash_password(password, salt)
- if record is None or not compare_digest(password_hash, record['password_hash']):
+ if record is None or not compare_digest(
+ password_hash, record['password_hash']
+ ):
return {'error': 'Invalid credentials'}, 'application/json', 401
login_id = token_urlsafe(16)
refresh = token_urlsafe(32)
@@ -155,14 +194,21 @@ def register_routes(app, database):
now = time()
# Persist digests and Unix timestamps so sessions survive restarts.
await database.create_login(
- username, login_id, sha256(refresh.encode()).hexdigest(), now + refresh_lifetime,
- sha256(access.encode()).hexdigest(), now + token_lifetime,
+ username,
+ login_id,
+ sha256(refresh.encode()).hexdigest(),
+ now + REFRESH_LIFETIME,
+ sha256(access.encode()).hexdigest(),
+ now + TOKEN_LIFETIME,
)
return (
token_response(access, username),
'application/json',
200,
- [('Cache-Control', 'no-store'), refresh_cookie(request, refresh, refresh_lifetime)],
+ [
+ ('Cache-Control', 'no-store'),
+ refresh_cookie(request, refresh, REFRESH_LIFETIME),
+ ],
)
@app.route('/api/refresh', methods=['POST'])
@@ -170,7 +216,12 @@ def register_routes(app, database):
async def refresh(request):
if not same_origin_request(request):
return (
- {'error': 'A same-origin request with X-Notebook-Request: 1 is required'},
+ {
+ 'error': (
+ 'A same-origin request with '
+ 'X-Notebook-Request: 1 is required'
+ )
+ },
'application/json',
403,
)
@@ -178,16 +229,23 @@ def register_routes(app, database):
access = token_urlsafe(32)
now = time()
username = await database.refresh_login(
- digest, sha256(access.encode()).hexdigest(), now + token_lifetime, now,
+ digest,
+ sha256(access.encode()).hexdigest(),
+ now + TOKEN_LIFETIME,
+ now,
)
if username is None:
return (
{'error': 'Unauthorized'},
'application/json',
401,
- [('Cache-Control', 'no-store'), refresh_cookie(request, '', 0)],
+ [
+ ('Cache-Control', 'no-store'),
+ refresh_cookie(request, '', 0),
+ ],
)
- # Reuse the cookie while issuing a fresh access token; concurrent tabs work.
+ # Reuse the cookie while issuing a fresh access token; concurrent tabs
+ # work.
return (
token_response(access, username),
'application/json',
@@ -204,4 +262,9 @@ def register_routes(app, database):
@protected(database)
async def logout(request):
await database.revoke_login(request.login_id)
- return {'logged_out': True}, 'application/json', 200, [refresh_cookie(request, '', 0)]
+ return (
+ {'logged_out': True},
+ 'application/json',
+ 200,
+ [refresh_cookie(request, '', 0)],
+ )
diff --git a/example_app/notes.py b/example_app/notes.py
index 1b9790f..840786b 100644
--- a/example_app/notes.py
+++ b/example_app/notes.py
@@ -1,4 +1,13 @@
-"""Private note CRUD, validation, filtering, pagination, and JSON export."""
+# -*- coding: utf-8 -*-
+"""
+ example_app.notes
+ ~~~~~~~~~~~~~~~~~
+
+ Private note CRUD, validation, filtering, pagination, and JSON export.
+
+ :copyright: (c) 2026 by Harrison Erd.
+ :license: BSD-3-Clause, see LICENSE for more details.
+"""
from datetime import UTC, datetime
from secrets import token_urlsafe
@@ -13,15 +22,22 @@ async def read_note(request, partial=False):
data = await request.json()
except orjson.JSONDecodeError:
raise ValueError('Invalid JSON body') from None
- if not isinstance(data, dict) or not data or data.keys() - {'title', 'body', 'pinned'}:
+ if (
+ not isinstance(data, dict)
+ or not data
+ or data.keys() - {'title', 'body', 'pinned'}
+ ):
raise ValueError('Send an object containing title, body, or pinned')
if not partial and 'title' not in data:
raise ValueError('A title is required')
if 'title' in data and (
- not isinstance(data['title'], str) or not 1 <= len(data['title'].strip()) <= 100
+ not isinstance(data['title'], str)
+ or not 1 <= len(data['title'].strip()) <= 100
):
raise ValueError('Title must contain 1–100 characters')
- if 'body' in data and (not isinstance(data['body'], str) or len(data['body']) > 10_000):
+ if 'body' in data and (
+ not isinstance(data['body'], str) or len(data['body']) > 10_000
+ ):
raise ValueError('Body must be a string of at most 10,000 characters')
if 'pinned' in data and not isinstance(data['pinned'], bool):
raise ValueError('Pinned must be a boolean')
@@ -35,7 +51,11 @@ async def owned_note(database, note_id, user):
if note is None:
return {'error': 'Note not found'}, 'application/json', 404
if note['owner'] != user['id']:
- return {'error': 'This note belongs to another user'}, 'application/json', 403
+ return (
+ {'error': 'This note belongs to another user'},
+ 'application/json',
+ 403,
+ )
return note
@@ -56,14 +76,32 @@ def register_routes(app, database):
limit = int(query.get('limit') or '5')
offset = int(query.get('offset') or '0')
except ValueError:
- return {'error': 'Limit and offset must be integers'}, 'application/json', 400
+ return (
+ {'error': 'Limit and offset must be integers'},
+ 'application/json',
+ 400,
+ )
if not 1 <= limit <= 50 or offset < 0:
- return {'error': 'Limit must be 1–50; offset must be zero or greater'}, 'application/json', 400
+ return (
+ {
+ 'error': (
+ 'Limit must be 1–50; offset must be zero or greater'
+ )
+ },
+ 'application/json',
+ 400,
+ )
search = query.get('q', '').strip().casefold()
pinned = query.get('pinned', '')
if pinned not in ('', 'true', 'false'):
- return {'error': 'Pinned filter must be true or false'}, 'application/json', 400
- return await database.list_notes(request.user['id'], search, pinned, limit, offset)
+ return (
+ {'error': 'Pinned filter must be true or false'},
+ 'application/json',
+ 400,
+ )
+ return await database.list_notes(
+ request.user['id'], search, pinned, limit, offset
+ )
@app.route('/api/notes', methods=['POST'])
@protected(database)
@@ -82,7 +120,12 @@ def register_routes(app, database):
'updated_at': now,
}
await database.create_note(note)
- return note, 'application/json', 201, [('Location', f'/api/notes/{note_id}')]
+ return (
+ note,
+ 'application/json',
+ 201,
+ [('Location', f'/api/notes/{note_id}')],
+ )
@app.route('/api/notes/{note_id}', methods=['GET'])
@protected(database)
@@ -99,7 +142,9 @@ def register_routes(app, database):
data = await read_note(request, partial=request.method == 'PATCH')
except ValueError as error:
return {'error': str(error)}, 'application/json', 400
- data['updated_at'] = datetime.now(UTC).isoformat(timespec='microseconds')
+ data['updated_at'] = datetime.now(UTC).isoformat(
+ timespec='microseconds'
+ )
updated = await database.update_note(note_id, request.user['id'], data)
if updated is None:
return {'error': 'Note not found'}, 'application/json', 404
@@ -115,7 +160,8 @@ def register_routes(app, database):
return {'error': 'Note not found'}, 'application/json', 404
return b'', 'application/json', 204
- # This fixed path wins over /api/notes/{note_id}, even though it is registered afterward.
+ # This fixed path wins over /api/notes/{note_id}, even though it is
+ # registered afterward.
@app.route('/api/notes/export', methods=['GET'])
@protected(database)
async def export_notes(request):
diff --git a/example_app/pages.py b/example_app/pages.py
index 128b826..4d32a01 100644
--- a/example_app/pages.py
+++ b/example_app/pages.py
@@ -1,23 +1,34 @@
-"""Public page/assets, redirect, health check, and a rate-limit demonstration."""
+# -*- coding: utf-8 -*-
+"""
+ example_app.pages
+ ~~~~~~~~~~~~~~~~~
-from pathlib import Path
+ Public page/assets, redirect, health check, and a rate-limit demonstration.
+
+ :copyright: (c) 2026 by Harrison Erd.
+ :license: BSD-3-Clause, see LICENSE for more details.
+"""
-from .rate_limit import rate_limit
+from pathlib import Path
-static_root = Path(__file__).parent / 'static'
+STATIC_ROOT = Path(__file__).parent / 'static'
-def register_routes(app):
+def register_routes(app, rate_limit):
"""Register the browser UI and public demonstration endpoints."""
@app.route('/')
async def home(request):
- body, content_type, status = await app.static('index.html', root=static_root)
+ body, content_type, status = await app.static(
+ 'index.html', root=STATIC_ROOT
+ )
return body, content_type, status, [('Cache-Control', 'no-cache')]
@app.route('/static/{filename}')
async def assets(request, filename):
- body, content_type, status = await app.static(filename, root=static_root)
+ body, content_type, status = await app.static(
+ filename, root=STATIC_ROOT
+ )
return body, content_type, status, [('Cache-Control', 'no-cache')]
@app.route('/ui')
diff --git a/example_app/rate_limit.py b/example_app/rate_limit.py
index 22ecccb..ea25aa3 100644
--- a/example_app/rate_limit.py
+++ b/example_app/rate_limit.py
@@ -1,40 +1,62 @@
-"""Application-owned sliding-window rate limiting, independent of Wred."""
+# -*- coding: utf-8 -*-
+"""
+ example_app.rate_limit
+ ~~~~~~~~~~~~~~~~~~~~~~
+
+ Application-owned sliding-window rate limiting, independent of Wred.
+
+ :copyright: (c) 2026 by Harrison Erd.
+ :license: BSD-3-Clause, see LICENSE for more details.
+"""
from collections import deque
from functools import wraps
from math import ceil
from time import monotonic
-rate_buckets = {} # (handler, client IP, window) -> accepted request timestamps.
-
-
-def rate_limit(requests, seconds):
- def decorator(handler):
- @wraps(handler)
- async def wrapper(request, *args, **kwargs):
- now = monotonic()
- # Drop inactive clients so their buckets do not remain in memory forever.
- for key, timestamps in list(rate_buckets.items()):
- if timestamps[-1] <= now - key[2]:
- del rate_buckets[key]
- # Use ASGI's client address, not an untrusted X-Forwarded-For header.
- client = (request.scope.get('client') or ('unknown', 0))[0]
- key = (handler.__name__, client, seconds)
- timestamps = rate_buckets.setdefault(key, deque())
- while timestamps and timestamps[0] <= now - seconds:
- timestamps.popleft()
- if len(timestamps) >= requests:
- retry_after = max(1, ceil(timestamps[0] + seconds - now))
- return (
- {'error': 'Too many requests', 'retry_after': retry_after},
- 'application/json',
- 429,
- [('Retry-After', str(retry_after)), ('Cache-Control', 'no-store')],
- )
- # Reserve the slot before any await, so concurrent requests cannot bypass it.
- timestamps.append(now)
- return await handler(request, *args, **kwargs)
-
- return wrapper
-
- return decorator
+
+class RateLimiter:
+ """Create rate-limit decorators with state owned by one application."""
+
+ def __init__(self):
+ self._buckets = {}
+
+ def __call__(self, requests, seconds):
+ def decorator(handler):
+ @wraps(handler)
+ async def wrapper(request, *args, **kwargs):
+ now = monotonic()
+ # Drop inactive clients so their buckets do not remain in
+ # memory forever.
+ for key, timestamps in list(self._buckets.items()):
+ if timestamps[-1] <= now - key[2]:
+ del self._buckets[key]
+ # Use ASGI's client address, not an untrusted X-Forwarded-For
+ # header.
+ client = (request.scope.get('client') or ('unknown', 0))[0]
+ key = (handler, client, seconds)
+ timestamps = self._buckets.setdefault(key, deque())
+ while timestamps and timestamps[0] <= now - seconds:
+ timestamps.popleft()
+ if len(timestamps) >= requests:
+ retry_after = max(1, ceil(timestamps[0] + seconds - now))
+ return (
+ {
+ 'error': 'Too many requests',
+ 'retry_after': retry_after,
+ },
+ 'application/json',
+ 429,
+ [
+ ('Retry-After', str(retry_after)),
+ ('Cache-Control', 'no-store'),
+ ],
+ )
+ # Reserve the slot before any await so concurrent requests
+ # cannot bypass it.
+ timestamps.append(now)
+ return await handler(request, *args, **kwargs)
+
+ return wrapper
+
+ return decorator
diff --git a/example_app/storage.py b/example_app/storage.py
index e61dfb4..e2c1adb 100644
--- a/example_app/storage.py
+++ b/example_app/storage.py
@@ -1,4 +1,13 @@
-"""Application-owned MongoDB storage using PyMongo's native async client."""
+# -*- coding: utf-8 -*-
+"""
+ example_app.storage
+ ~~~~~~~~~~~~~~~~~~~
+
+ Application-owned MongoDB storage using PyMongo's native async client.
+
+ :copyright: (c) 2026 by Harrison Erd.
+ :license: BSD-3-Clause, see LICENSE for more details.
+"""
from asyncio import Lock
from datetime import UTC, datetime
@@ -28,11 +37,22 @@ class Database:
async with self._init_lock:
if not self._initialized:
await self.client.admin.command('ping')
- await self.db.sessions.create_index('refresh_digest', unique=True)
- await self.db.sessions.create_index('access_tokens.digest', unique=True)
- await self.db.sessions.create_index('expires_at', expireAfterSeconds=0)
+ await self.db.sessions.create_index(
+ 'refresh_digest', unique=True
+ )
+ await self.db.sessions.create_index(
+ 'access_tokens.digest', unique=True
+ )
+ await self.db.sessions.create_index(
+ 'expires_at', expireAfterSeconds=0
+ )
await self.db.notes.create_index(
- [('owner', 1), ('pinned', -1), ('updated_at', -1), ('id', -1)],
+ [
+ ('owner', 1),
+ ('pinned', -1),
+ ('updated_at', -1),
+ ('id', -1),
+ ],
)
self._initialized = True
@@ -43,7 +63,11 @@ class Database:
await self.initialize()
try:
await self.db.users.insert_one(
- {'_id': username, 'salt': salt, 'password_hash': password_hash},
+ {
+ '_id': username,
+ 'salt': salt,
+ 'password_hash': password_hash,
+ },
)
except DuplicateKeyError:
return False
@@ -53,8 +77,8 @@ class Database:
await self.initialize()
return await self.db.users.find_one({'_id': username})
- async def create_login(self, username, login_id, refresh_digest, refresh_expires,
- access_digest, access_expires):
+ async def create_login(self, username, login_id, refresh_digest,
+ refresh_expires, access_digest, access_expires):
await self.initialize()
# Keep a login and its access credentials in one document. Refresh and
# logout are atomic without requiring replica-set transactions.
@@ -63,20 +87,57 @@ class Database:
'user_id': username,
'refresh_digest': refresh_digest,
'expires_at': timestamp(refresh_expires),
- 'access_tokens': [{'digest': access_digest, 'expires_at': timestamp(access_expires)}],
+ 'access_tokens': [
+ {
+ 'digest': access_digest,
+ 'expires_at': timestamp(access_expires),
+ }
+ ],
})
- async def refresh_login(self, refresh_digest, access_digest, access_expires, now):
+ async def refresh_login(self, refresh_digest, access_digest,
+ access_expires, now):
await self.initialize()
current_time = timestamp(now)
session = await self.db.sessions.find_one_and_update(
- {'refresh_digest': refresh_digest, 'expires_at': {'$gt': current_time}},
- [{'$set': {'access_tokens': {'$concatArrays': [
- # Prune old access tokens while retaining tokens used by other tabs.
- {'$filter': {'input': '$access_tokens', 'as': 'token',
- 'cond': {'$gt': ['$$token.expires_at', current_time]}}},
- {'$literal': [{'digest': access_digest, 'expires_at': timestamp(access_expires)}]},
- ]}}}],
+ {
+ 'refresh_digest': refresh_digest,
+ 'expires_at': {'$gt': current_time},
+ },
+ [
+ {
+ '$set': {
+ 'access_tokens': {
+ '$concatArrays': [
+ # Prune old access tokens while retaining
+ # tokens used by other tabs.
+ {
+ '$filter': {
+ 'input': '$access_tokens',
+ 'as': 'token',
+ 'cond': {
+ '$gt': [
+ '$$token.expires_at',
+ current_time,
+ ]
+ },
+ }
+ },
+ {
+ '$literal': [
+ {
+ 'digest': access_digest,
+ 'expires_at': timestamp(
+ access_expires
+ ),
+ }
+ ]
+ },
+ ]
+ }
+ }
+ }
+ ],
projection={'user_id': 1},
return_document=ReturnDocument.AFTER,
)
@@ -85,13 +146,20 @@ class Database:
async def get_access_token(self, digest, now):
await self.initialize()
current_time = timestamp(now)
- # TTL cleanup is asynchronous, so authorization must check both deadlines.
- session = await self.db.sessions.find_one({
- 'expires_at': {'$gt': current_time},
- 'access_tokens': {'$elemMatch': {
- 'digest': digest, 'expires_at': {'$gt': current_time},
- }},
- }, {'user_id': 1})
+ # TTL cleanup is asynchronous, so authorization must check both
+ # deadlines.
+ session = await self.db.sessions.find_one(
+ {
+ 'expires_at': {'$gt': current_time},
+ 'access_tokens': {
+ '$elemMatch': {
+ 'digest': digest,
+ 'expires_at': {'$gt': current_time},
+ }
+ },
+ },
+ {'user_id': 1},
+ )
if session:
return {'user_id': session['user_id'], 'login_id': session['_id']}
@@ -103,31 +171,60 @@ class Database:
await self.initialize()
query = {'owner': owner}
if search:
- # Store Python-casefolded fields to preserve Unicode substring search.
- # $literal keeps search text such as "$title" from becoming an expression.
- query['$expr'] = {'$gte': [{'$indexOfCP': [
- {'$concat': ['$_title_search', '\n', '$_body_search']}, {'$literal': search},
- ]}, 0]}
+ # Store Python-casefolded fields to preserve Unicode substring
+ # search.
+ # $literal keeps search text such as "$title" from becoming an
+ # expression.
+ query['$expr'] = {
+ '$gte': [
+ {
+ '$indexOfCP': [
+ {
+ '$concat': [
+ '$_title_search',
+ '\n',
+ '$_body_search',
+ ]
+ },
+ {'$literal': search},
+ ]
+ },
+ 0,
+ ]
+ }
if pinned:
query['pinned'] = pinned == 'true'
- # MongoDB skip accepts signed 64-bit integers. Larger offsets are empty.
- items = ([{'$match': {'_id': {'$exists': False}}}] if offset >= 2**63 else
- [{'$skip': offset}, {'$limit': limit}, {'$project': NOTE_FIELDS}])
+ # MongoDB skip accepts signed 64-bit integers. Larger offsets are
+ # empty.
+ items = (
+ [{'$match': {'_id': {'$exists': False}}}]
+ if offset >= 2**63
+ else [
+ {'$skip': offset},
+ {'$limit': limit},
+ {'$project': NOTE_FIELDS},
+ ]
+ )
cursor = await self.db.notes.aggregate([
{'$match': query},
{'$sort': {'pinned': -1, 'updated_at': -1, 'id': -1}},
{'$facet': {'items': items, 'count': [{'$count': 'total'}]}},
])
result = (await cursor.to_list(length=1))[0]
- return {'items': result['items'],
- 'total': result['count'][0]['total'] if result['count'] else 0,
- 'limit': limit, 'offset': offset}
+ return {
+ 'items': result['items'],
+ 'total': result['count'][0]['total'] if result['count'] else 0,
+ 'limit': limit,
+ 'offset': offset,
+ }
async def create_note(self, note):
await self.initialize()
await self.db.notes.insert_one({
- **note, '_id': note['id'],
- '_title_search': note['title'].casefold(), '_body_search': note['body'].casefold(),
+ **note,
+ '_id': note['id'],
+ '_title_search': note['title'].casefold(),
+ '_body_search': note['body'].casefold(),
})
async def get_note(self, note_id):
@@ -136,22 +233,34 @@ class Database:
async def update_note(self, note_id, owner, data):
await self.initialize()
- fields = {key: data[key] for key in ('title', 'body', 'pinned', 'updated_at') if key in data}
+ fields = {
+ key: data[key]
+ for key in ('title', 'body', 'pinned', 'updated_at')
+ if key in data
+ }
for key in ('title', 'body'):
if key in fields:
fields[f'_{key}_search'] = fields[key].casefold()
return await self.db.notes.find_one_and_update(
- {'_id': note_id, 'owner': owner}, {'$set': fields},
- projection=NOTE_FIELDS, return_document=ReturnDocument.AFTER,
+ {'_id': note_id, 'owner': owner},
+ {'$set': fields},
+ projection=NOTE_FIELDS,
+ return_document=ReturnDocument.AFTER,
)
async def delete_note(self, note_id, owner):
await self.initialize()
- result = await self.db.notes.delete_one({'_id': note_id, 'owner': owner})
+ result = await self.db.notes.delete_one(
+ {'_id': note_id, 'owner': owner}
+ )
return result.deleted_count == 1
async def export_notes(self, owner):
await self.initialize()
- return await self.db.notes.find({'owner': owner}, NOTE_FIELDS).sort(
- [('created_at', 1), ('id', 1)],
- ).to_list(length=None)
+ return (
+ await self.db.notes.find({'owner': owner}, NOTE_FIELDS)
+ .sort(
+ [('created_at', 1), ('id', 1)],
+ )
+ .to_list(length=None)
+ )
diff --git a/pyproject.toml b/pyproject.toml
index ff19cd8..b7abe96 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -12,3 +12,9 @@ license = "BSD-3-Clause"
license-files = ["LICENSE"]
authors = [{ name = "Harrison Erd", email = "[email protected]" }]
dependencies = ["orjson", "weft-server"]
+
+[tool.ruff]
+line-length = 79
+
+[tool.ruff.lint]
+select = ["E", "W", "N"]
diff --git a/tests/test_example_app.py b/tests/test_example_app.py
new file mode 100644
index 0000000..1f63187
--- /dev/null
+++ b/tests/test_example_app.py
@@ -0,0 +1,67 @@
+# -*- coding: utf-8 -*-
+"""
+ tests.test_example_app
+ ~~~~~~~~~~~~~~~~~~~~~~
+
+ Exercise example application construction and resource ownership.
+
+ :copyright: (c) 2026 by Harrison Erd.
+ :license: BSD-3-Clause, see LICENSE for more details.
+"""
+
+import subprocess
+import sys
+import unittest
+from pathlib import Path
+
+
+class AppFactoryTests(unittest.IsolatedAsyncioTestCase):
+ def test_import_factory_does_not_construct_a_database(self):
+ script = """
+from unittest.mock import patch
+
+with patch('pymongo.AsyncMongoClient', side_effect=RuntimeError(
+ 'Importing the factory must not construct a database client'
+)):
+ from example_app.app import create_app
+"""
+ result = subprocess.run(
+ [sys.executable, '-c', script],
+ cwd=Path(__file__).resolve().parents[1],
+ capture_output=True,
+ timeout=10,
+ )
+ self.assertEqual(result.returncode, 0, result.stderr.decode())
+
+ async def test_injected_database_follows_application_lifespan(self):
+ from example_app.app import create_app
+
+ events = []
+
+ class Database:
+ async def initialize(self):
+ events.append('initialize')
+
+ async def close(self):
+ events.append('close')
+
+ app = create_app(database=Database())
+ self.assertEqual(events, [])
+ messages = iter([
+ {'type': 'lifespan.startup'},
+ {'type': 'lifespan.shutdown'},
+ ])
+
+ async def receive():
+ return next(messages)
+
+ async def send(message):
+ events.append(message['type'])
+
+ await app({'type': 'lifespan'}, receive, send)
+ self.assertEqual(events, [
+ 'initialize',
+ 'lifespan.startup.complete',
+ 'close',
+ 'lifespan.shutdown.complete',
+ ])
diff --git a/tests/test_notebook.py b/tests/test_notebook.py
index 753f6e1..efcd2ee 100644
--- a/tests/test_notebook.py
+++ b/tests/test_notebook.py
@@ -1,36 +1,47 @@
-"""Exercise the example's API contracts through the actual Wred ASGI app."""
-import importlib
+# -*- coding: utf-8 -*-
+"""
+ tests.test_notebook
+ ~~~~~~~~~~~~~~~~~~~
+
+ Exercise the example's API contracts through the actual Wred ASGI app.
+
+ :copyright: (c) 2026 by Harrison Erd.
+ :license: BSD-3-Clause, see LICENSE for more details.
+"""
+
import asyncio
import os
-from pathlib import Path
import subprocess
import sys
-from http.cookies import SimpleCookie
import unittest
-from urllib.parse import unquote, urlsplit
+from http.cookies import SimpleCookie
+from pathlib import Path
from unittest.mock import patch
+from urllib.parse import unquote, urlsplit
from uuid import uuid4
import orjson
-from tests.test_wred import call_app
from example_app import auth, rate_limit
+from example_app.app import create_app
+from tests.test_wred import call_app
mongodb_uri = os.environ.get('WRED_TEST_MONGODB_URI')
-notebook = importlib.import_module('example_app.app') if mongodb_uri else None
[email protected](mongodb_uri, 'Set WRED_TEST_MONGODB_URI to run the MongoDB integration tests')
[email protected](
+ mongodb_uri,
+ 'Set WRED_TEST_MONGODB_URI to run the MongoDB integration tests',
+)
class NotebookTests(unittest.IsolatedAsyncioTestCase):
def setUp(self):
self.database_name = 'wred_test_' + uuid4().hex
self.apps = []
self.app = self.new_app()
self.addAsyncCleanup(self.clean_database)
- rate_limit.rate_buckets.clear()
def new_app(self):
- app = notebook.create_app(mongodb_uri, self.database_name)
+ app = create_app(mongodb_uri, self.database_name)
self.apps.append(app)
return app
@@ -41,16 +52,23 @@ class NotebookTests(unittest.IsolatedAsyncioTestCase):
for app in self.apps:
await app.database.close()
- async def request(self, path, method='GET', data=None, token=None, authorization=None,
- cookie=None, client='127.0.0.1', csrf_header=True, extra_headers=(), scheme='http', app=None):
+ async def request(self, path, method='GET', data=None, token=None,
+ authorization=None, cookie=None, client='127.0.0.1',
+ csrf_header=True, extra_headers=(), scheme='http',
+ app=None):
url = urlsplit(path)
async def receive():
if method == 'GET':
- raise AssertionError('GET handlers should not consume the request body')
+ raise AssertionError(
+ 'GET handlers should not consume the request body'
+ )
return {'type': 'http.request', 'body': orjson.dumps(data)}
- headers = [(b'content-type', b'application/json'), (b'host', b'notebook.test')]
+ headers = [
+ (b'content-type', b'application/json'),
+ (b'host', b'notebook.test'),
+ ]
if csrf_header:
headers.append((b'x-notebook-request', b'1'))
if cookie is not None:
@@ -60,11 +78,19 @@ class NotebookTests(unittest.IsolatedAsyncioTestCase):
headers.append((b'authorization', f'Bearer {token}'.encode()))
if authorization is not None:
headers.append((b'authorization', authorization))
- messages = await call_app(self.app if app is None else app, {
- 'type': 'http', 'path': unquote(url.path), 'method': method,
- 'query_string': url.query.encode(), 'headers': headers,
- 'scheme': scheme, 'client': (client, 12345),
- }, receive)
+ messages = await call_app(
+ self.app if app is None else app,
+ {
+ 'type': 'http',
+ 'path': unquote(url.path),
+ 'method': method,
+ 'query_string': url.query.encode(),
+ 'headers': headers,
+ 'scheme': scheme,
+ 'client': (client, 12345),
+ },
+ receive,
+ )
self.assertEqual(len(messages), 2)
self.assertEqual(messages[0]['type'], 'http.response.start')
self.assertEqual(messages[1]['type'], 'http.response.body')
@@ -75,10 +101,19 @@ class NotebookTests(unittest.IsolatedAsyncioTestCase):
return messages[0]['status'], body, response_headers
async def account(self, username='alice'):
- credentials = {'username': username, 'password': 'a long example passphrase'}
- status, user, _ = await self.request('/api/signup', 'POST', credentials)
- self.assertEqual((status, user), (201, {'id': username.strip().casefold()}))
- status, login, headers = await self.request('/api/login', 'POST', credentials)
+ credentials = {
+ 'username': username,
+ 'password': 'a long example passphrase',
+ }
+ status, user, _ = await self.request(
+ '/api/signup', 'POST', credentials
+ )
+ self.assertEqual(
+ (status, user), (201, {'id': username.strip().casefold()})
+ )
+ status, login, headers = await self.request(
+ '/api/login', 'POST', credentials
+ )
self.assertEqual(status, 200)
self.assertEqual(headers[b'cache-control'], b'no-store')
self.assertEqual(login['token_type'], 'bearer')
@@ -86,9 +121,13 @@ class NotebookTests(unittest.IsolatedAsyncioTestCase):
return login['access_token']
async def create(self, token, **data):
- status, note, headers = await self.request('/api/notes', 'POST', data, token)
+ status, note, headers = await self.request(
+ '/api/notes', 'POST', data, token
+ )
self.assertEqual(status, 201)
- self.assertEqual(headers[b'location'], f"/api/notes/{note['id']}".encode())
+ self.assertEqual(
+ headers[b'location'], f"/api/notes/{note['id']}".encode()
+ )
self.assertEqual(headers[b'cache-control'], b'no-store')
return note
@@ -105,264 +144,517 @@ class NotebookTests(unittest.IsolatedAsyncioTestCase):
self.assertEqual(headers[b'cache-control'], b'no-cache')
self.assertIn(marker, body)
status, body, headers = await self.request('/ui')
- self.assertEqual((status, body, headers[b'location']), (303, b'', b'/'))
- self.assertEqual((await self.request('/api/health'))[:2], (200, {'status': 'ok'}))
+ self.assertEqual(
+ (status, body, headers[b'location']), (303, b'', b'/')
+ )
+ self.assertEqual(
+ (await self.request('/api/health'))[:2], (200, {'status': 'ok'})
+ )
self.assertEqual((await self.request('/static/missing.js'))[0], 404)
async def test_lifespan_starts_and_stops_the_database(self):
worker = self.new_app()
- messages = iter([{'type': 'lifespan.startup'}, {'type': 'lifespan.shutdown'}])
+ messages = iter(
+ [{'type': 'lifespan.startup'}, {'type': 'lifespan.shutdown'}]
+ )
async def receive():
return next(messages)
responses = await call_app(worker, {'type': 'lifespan'}, receive)
- self.assertEqual(responses, [
- {'type': 'lifespan.startup.complete'}, {'type': 'lifespan.shutdown.complete'},
- ])
+ self.assertEqual(
+ responses,
+ [
+ {'type': 'lifespan.startup.complete'},
+ {'type': 'lifespan.shutdown.complete'},
+ ],
+ )
async def test_signup_login_and_independent_login_revocation(self):
token = await self.account(' Alice ')
status, profile, headers = await self.request('/api/me', token=token)
self.assertEqual((status, profile), (200, {'id': 'alice'}))
self.assertEqual(headers[b'cache-control'], b'no-store')
- credentials = {'username': 'alice', 'password': 'a long example passphrase'}
- self.assertEqual((await self.request('/api/signup', 'POST', credentials))[0], 409)
- for data in [dict(credentials, username='unknown'), dict(credentials, password='an incorrect long passphrase')]:
- self.assertEqual((await self.request('/api/login', 'POST', data))[:2],
- (401, {'error': 'Invalid credentials'}))
+ credentials = {
+ 'username': 'alice',
+ 'password': 'a long example passphrase',
+ }
+ self.assertEqual(
+ (await self.request('/api/signup', 'POST', credentials))[0], 409
+ )
+ for data in [
+ dict(credentials, username='unknown'),
+ dict(credentials, password='an incorrect long passphrase'),
+ ]:
+ self.assertEqual(
+ (await self.request('/api/login', 'POST', data))[:2],
+ (401, {'error': 'Invalid credentials'}),
+ )
_, login, _ = await self.request('/api/login', 'POST', credentials)
second_token = login['access_token']
self.assertNotEqual(token, second_token)
- self.assertEqual((await self.request('/api/logout', 'POST', token=token))[:2],
- (200, {'logged_out': True}))
+ self.assertEqual(
+ (await self.request('/api/logout', 'POST', token=token))[:2],
+ (200, {'logged_out': True}),
+ )
self.assertEqual((await self.request('/api/me', token=token))[0], 401)
- self.assertEqual((await self.request('/api/me', token=second_token))[:2],
- (200, {'id': 'alice'}))
+ self.assertEqual(
+ (await self.request('/api/me', token=second_token))[:2],
+ (200, {'id': 'alice'}),
+ )
expired_time = auth.time() + 3601
with patch.object(auth, 'time', return_value=expired_time):
- self.assertEqual((await self.request('/api/me', token=second_token))[0], 401)
+ self.assertEqual(
+ (await self.request('/api/me', token=second_token))[0], 401
+ )
async def test_accounts_notes_and_sessions_survive_a_new_process(self):
token = await self.account()
note = await self.create(token, title='Remember me', pinned=True)
- credentials = {'username': 'alice', 'password': 'a long example passphrase'}
- _, login, headers = await self.request('/api/login', 'POST', credentials)
- morsel = SimpleCookie(headers[b'set-cookie'].decode())[auth.refresh_cookie_name]
- cookie = f'{auth.refresh_cookie_name}={morsel.value}'
+ credentials = {
+ 'username': 'alice',
+ 'password': 'a long example passphrase',
+ }
+ _, login, headers = await self.request(
+ '/api/login', 'POST', credentials
+ )
+ morsel = SimpleCookie(headers[b'set-cookie'].decode())[
+ auth.REFRESH_COOKIE_NAME
+ ]
+ cookie = f'{auth.REFRESH_COOKIE_NAME}={morsel.value}'
script = """
import asyncio
import sys
import orjson
-from example_app.app import app
+from example_app.asgi import app
from tests.test_wred import call_app
+
async def main():
data = orjson.loads(sys.stdin.buffer.read())
results = []
for path, method, headers in [
- ('/api/notes', 'GET', [(b'authorization', ('Bearer ' + data['token']).encode())]),
- ('/api/refresh', 'POST', [(b'cookie', data['cookie'].encode()),
- (b'x-notebook-request', b'1')]),
+ (
+ '/api/notes',
+ 'GET',
+ [(b'authorization', ('Bearer ' + data['token']).encode())],
+ ),
+ (
+ '/api/refresh',
+ 'POST',
+ [
+ (b'cookie', data['cookie'].encode()),
+ (b'x-notebook-request', b'1'),
+ ],
+ ),
('/api/login', 'POST', [(b'x-notebook-request', b'1')]),
]:
+
async def receive():
- return {'type': 'http.request', 'body': orjson.dumps(data['credentials'])}
- messages = await call_app(app, {'type': 'http', 'path': path, 'method': method,
- 'headers': headers}, receive)
- results.append([messages[0]['status'], orjson.loads(messages[1]['body'])])
+ return {
+ 'type': 'http.request',
+ 'body': orjson.dumps(data['credentials']),
+ }
+
+ messages = await call_app(
+ app,
+ {
+ 'type': 'http',
+ 'path': path,
+ 'method': method,
+ 'headers': headers,
+ },
+ receive,
+ )
+ results.append(
+ [messages[0]['status'], orjson.loads(messages[1]['body'])]
+ )
await app.database.close()
sys.stdout.buffer.write(orjson.dumps(results))
+
asyncio.run(main())
"""
result = await asyncio.to_thread(
- subprocess.run, [sys.executable, '-c', script],
- input=orjson.dumps({'token': login['access_token'], 'cookie': cookie,
- 'credentials': credentials}),
- env={**os.environ, 'MONGODB_URI': mongodb_uri,
- 'MONGODB_DATABASE': self.database_name},
- cwd=Path(__file__).resolve().parents[1], capture_output=True, timeout=20,
+ subprocess.run,
+ [sys.executable, '-c', script],
+ input=orjson.dumps({
+ 'token': login['access_token'],
+ 'cookie': cookie,
+ 'credentials': credentials,
+ }),
+ env={
+ **os.environ,
+ 'MONGODB_URI': mongodb_uri,
+ 'MONGODB_DATABASE': self.database_name,
+ },
+ cwd=Path(__file__).resolve().parents[1],
+ capture_output=True,
+ timeout=20,
)
self.assertEqual(result.returncode, 0, result.stderr.decode())
listed, refreshed, logged_in = orjson.loads(result.stdout)
self.assertEqual((listed[0], listed[1]['items']), (200, [note]))
- self.assertEqual((refreshed[0], refreshed[1]['user']), (200, {'id': 'alice'}))
- self.assertEqual((logged_in[0], logged_in[1]['user']), (200, {'id': 'alice'}))
- # Tokens issued in the new process must be usable and revocable in the original.
+ self.assertEqual(
+ (refreshed[0], refreshed[1]['user']), (200, {'id': 'alice'})
+ )
+ self.assertEqual(
+ (logged_in[0], logged_in[1]['user']), (200, {'id': 'alice'})
+ )
+ # Tokens issued in the new process must be usable and revocable in the
+ # original.
renewed = refreshed[1]['access_token']
- self.assertEqual((await self.request('/api/me', token=renewed))[0], 200)
+ self.assertEqual(
+ (await self.request('/api/me', token=renewed))[0], 200
+ )
await self.request('/api/logout', 'POST', token=renewed)
restarted = self.new_app()
- self.assertEqual((await self.request('/api/me', token=login['access_token'], app=restarted))[0], 401)
- self.assertEqual((await self.request('/api/refresh', 'POST', cookie=cookie, app=restarted))[0], 401)
- self.assertEqual((await self.request('/api/me', token=token, app=restarted))[0], 200)
+ self.assertEqual(
+ (
+ await self.request(
+ '/api/me', token=login['access_token'], app=restarted
+ )
+ )[0],
+ 401,
+ )
+ self.assertEqual(
+ (
+ await self.request(
+ '/api/refresh', 'POST', cookie=cookie, app=restarted
+ )
+ )[0],
+ 401,
+ )
+ self.assertEqual(
+ (await self.request('/api/me', token=token, app=restarted))[0], 200
+ )
- async def test_workers_resolve_duplicate_signups_and_preserve_concurrent_patches(self):
+ async def test_workers_handle_duplicate_signups_and_concurrent_patches(
+ self
+ ):
worker = self.new_app()
- credentials = {'username': 'alice', 'password': 'a long example passphrase'}
+ credentials = {
+ 'username': 'alice',
+ 'password': 'a long example passphrase',
+ }
responses = await asyncio.gather(
self.request('/api/signup', 'POST', credentials),
self.request('/api/signup', 'POST', credentials, app=worker),
)
- self.assertEqual(sorted(response[0] for response in responses), [201, 409])
- _, login, _ = await self.request('/api/login', 'POST', credentials, app=worker)
+ self.assertEqual(
+ sorted(response[0] for response in responses), [201, 409]
+ )
+ _, login, _ = await self.request(
+ '/api/login', 'POST', credentials, app=worker
+ )
token = login['access_token']
note = await self.create(token, title='Original', body='Original body')
path = f"/api/notes/{note['id']}"
responses = await asyncio.gather(
self.request(path, 'PATCH', {'title': 'New title'}, token),
- self.request(path, 'PATCH', {'body': 'New body'}, token, app=worker),
+ self.request(
+ path, 'PATCH', {'body': 'New body'}, token, app=worker
+ ),
)
self.assertEqual([response[0] for response in responses], [200, 200])
_, edited, _ = await self.request(path, token=token, app=worker)
- self.assertEqual((edited['title'], edited['body']), ('New title', 'New body'))
+ self.assertEqual(
+ (edited['title'], edited['body']), ('New title', 'New body')
+ )
for search in ('New+title', 'New+body'):
- _, page, _ = await self.request('/api/notes?q=' + search, token=token, app=worker)
+ _, page, _ = await self.request(
+ '/api/notes?q=' + search, token=token, app=worker
+ )
self.assertEqual(page['items'], [edited])
async def test_invalid_credentials_and_bearer_headers(self):
- for data in [[], {}, {'username': 1, 'password': 'a long passphrase'},
- {'username': 'alice', 'password': False},
- {'username': ' ', 'password': 'a long passphrase'},
- {'username': 'alice', 'password': 'short'},
- {'username': 'x' * 65, 'password': 'a long passphrase'},
- {'username': 'alice', 'password': 'x' * 129}]:
+ for data in [
+ [],
+ {},
+ {'username': 1, 'password': 'a long passphrase'},
+ {'username': 'alice', 'password': False},
+ {'username': ' ', 'password': 'a long passphrase'},
+ {'username': 'alice', 'password': 'short'},
+ {'username': 'x' * 65, 'password': 'a long passphrase'},
+ {'username': 'alice', 'password': 'x' * 129},
+ ]:
with self.subTest(data=data):
- self.assertEqual((await self.request('/api/signup', 'POST', data))[0], 400)
+ self.assertEqual(
+ (await self.request('/api/signup', 'POST', data))[0], 400
+ )
for header in [b'', b'Basic example', b'Bearer ', b'Bearer unknown']:
with self.subTest(header=header):
- status, error, headers = await self.request('/api/notes', authorization=header)
- self.assertEqual((status, error), (401, {'error': 'Unauthorized'}))
+ status, error, headers = await self.request(
+ '/api/notes', authorization=header
+ )
+ self.assertEqual(
+ (status, error), (401, {'error': 'Unauthorized'})
+ )
self.assertEqual(headers[b'www-authenticate'], b'Bearer')
async def test_create_read_patch_replace_delete_and_export(self):
token = await self.account()
- note = await self.create(token, title=' First idea ', body='Some details', pinned=True)
- self.assertEqual((note['title'], note['body'], note['pinned'], note['owner']),
- ('First idea', 'Some details', True, 'alice'))
+ note = await self.create(
+ token, title=' First idea ', body='Some details', pinned=True
+ )
+ self.assertEqual(
+ (note['title'], note['body'], note['pinned'], note['owner']),
+ ('First idea', 'Some details', True, 'alice'),
+ )
path = f"/api/notes/{note['id']}"
- self.assertEqual((await self.request(path, token=token))[:2], (200, note))
- status, edited, _ = await self.request(path, 'PATCH', {'body': 'Edited details'}, token)
+ self.assertEqual(
+ (await self.request(path, token=token))[:2], (200, note)
+ )
+ status, edited, _ = await self.request(
+ path, 'PATCH', {'body': 'Edited details'}, token
+ )
self.assertEqual(status, 200)
- self.assertEqual((edited['title'], edited['body'], edited['pinned']),
- ('First idea', 'Edited details', True))
+ self.assertEqual(
+ (edited['title'], edited['body'], edited['pinned']),
+ ('First idea', 'Edited details', True),
+ )
self.assertEqual(edited['created_at'], note['created_at'])
- status, replaced, _ = await self.request(path, 'PUT', {'title': 'Replacement'}, token)
+ status, replaced, _ = await self.request(
+ path, 'PUT', {'title': 'Replacement'}, token
+ )
self.assertEqual(status, 200)
- self.assertEqual((replaced['title'], replaced['body'], replaced['pinned']),
- ('Replacement', '', False))
- status, exported, headers = await self.request('/api/notes/export', token=token)
+ self.assertEqual(
+ (replaced['title'], replaced['body'], replaced['pinned']),
+ ('Replacement', '', False),
+ )
+ status, exported, headers = await self.request(
+ '/api/notes/export', token=token
+ )
self.assertEqual((status, exported), (200, [replaced]))
- self.assertEqual(headers[b'content-disposition'], b'attachment; filename="notes.json"')
+ self.assertEqual(
+ headers[b'content-disposition'],
+ b'attachment; filename="notes.json"',
+ )
self.assertEqual(headers[b'cache-control'], b'no-store')
status, body, _ = await self.request(path, 'DELETE', token=token)
self.assertEqual((status, body), (204, b''))
self.assertEqual((await self.request(path, token=token))[0], 404)
- self.assertEqual((await self.request('/api/notes/export', token=token))[:2], (200, []))
+ self.assertEqual(
+ (await self.request('/api/notes/export', token=token))[:2],
+ (200, []),
+ )
- async def test_accounts_cannot_read_modify_delete_or_export_each_others_notes(self):
+ async def test_accounts_cannot_access_each_others_notes(self):
alice = await self.account('alice')
bob = await self.account('bob')
- note = await self.create(alice, title='Private', body='Only Alice can read this')
+ note = await self.create(
+ alice, title='Private', body='Only Alice can read this'
+ )
bob_note = await self.create(bob, title='Bob only')
path = f"/api/notes/{note['id']}"
- for method, data in [('GET', None), ('PATCH', {'title': 'Stolen'}),
- ('PUT', {'title': 'Stolen'}), ('DELETE', None)]:
+ for method, data in [
+ ('GET', None),
+ ('PATCH', {'title': 'Stolen'}),
+ ('PUT', {'title': 'Stolen'}),
+ ('DELETE', None),
+ ]:
with self.subTest(method=method):
- self.assertEqual((await self.request(path, method, data, bob))[0], 403)
- self.assertEqual((await self.request(path, token=alice))[:2], (200, note))
+ self.assertEqual(
+ (await self.request(path, method, data, bob))[0], 403
+ )
+ self.assertEqual(
+ (await self.request(path, token=alice))[:2], (200, note)
+ )
_, page, _ = await self.request('/api/notes', token=bob)
self.assertEqual((page['items'], page['total']), ([bob_note], 1))
- self.assertEqual((await self.request('/api/notes/export', token=bob))[:2], (200, [bob_note]))
+ self.assertEqual(
+ (await self.request('/api/notes/export', token=bob))[:2],
+ (200, [bob_note]),
+ )
for method in ['GET', 'PATCH', 'DELETE']:
- self.assertEqual((await self.request('/api/notes/missing', method, {'title': 'Missing'}, bob))[0], 404)
+ self.assertEqual(
+ (
+ await self.request(
+ '/api/notes/missing', method, {'title': 'Missing'}, bob
+ )
+ )[0],
+ 404,
+ )
async def test_filtering_pagination_and_invalid_queries(self):
token = await self.account()
- first = await self.create(token, title='Coffee idea', body='Learn C++; Straße; 100%_done; $title; [.*]', pinned=True)
- second = await self.create(token, title='Second idea', body='Coffee later')
+ first = await self.create(
+ token,
+ title='Coffee idea',
+ body='Learn C++; Straße; 100%_done; $title; [.*]',
+ pinned=True,
+ )
+ second = await self.create(
+ token, title='Second idea', body='Coffee later'
+ )
await self.create(token, title='A separate thought')
- _, page, _ = await self.request('/api/notes?q=COFFEE&limit=1&offset=0', token=token)
- self.assertEqual((page['items'], page['total'], page['limit'], page['offset']), ([first], 2, 1, 0))
- _, page, _ = await self.request('/api/notes?q=coffee&limit=1&offset=1', token=token)
+ _, page, _ = await self.request(
+ '/api/notes?q=COFFEE&limit=1&offset=0', token=token
+ )
+ self.assertEqual(
+ (page['items'], page['total'], page['limit'], page['offset']),
+ ([first], 2, 1, 0),
+ )
+ _, page, _ = await self.request(
+ '/api/notes?q=coffee&limit=1&offset=1', token=token
+ )
self.assertEqual(page['items'], [second])
- _, page, _ = await self.request('/api/notes?q=C%2B%2B&pinned=true', token=token)
+ _, page, _ = await self.request(
+ '/api/notes?q=C%2B%2B&pinned=true', token=token
+ )
self.assertEqual(page['items'], [first])
for search in ['STRASSE', '100%25_done', '$title', '%5B.*%5D']:
- _, page, _ = await self.request('/api/notes?q=' + search, token=token)
+ _, page, _ = await self.request(
+ '/api/notes?q=' + search, token=token
+ )
self.assertEqual(page['items'], [first])
- _, page, _ = await self.request('/api/notes?offset=9223372036854775808', token=token)
+ _, page, _ = await self.request(
+ '/api/notes?offset=9223372036854775808', token=token
+ )
self.assertEqual((page['items'], page['total']), ([], 3))
_, page, _ = await self.request('/api/notes?pinned=false', token=token)
self.assertEqual(page['total'], 2)
- for query in ['limit=abc', 'offset=abc', 'limit=0', 'limit=51', 'offset=-1',
- 'pinned=maybe', '&'.join(f'x={i}' for i in range(21))]:
+ for query in [
+ 'limit=abc',
+ 'offset=abc',
+ 'limit=0',
+ 'limit=51',
+ 'offset=-1',
+ 'pinned=maybe',
+ '&'.join(f'x={i}' for i in range(21)),
+ ]:
with self.subTest(query=query):
- self.assertEqual((await self.request('/api/notes?' + query, token=token))[0], 400)
-
- async def test_note_validation_prevents_invalid_or_unowned_field_updates(self):
+ self.assertEqual(
+ (await self.request('/api/notes?' + query, token=token))[
+ 0
+ ],
+ 400,
+ )
+
+ async def test_notes_reject_invalid_or_unowned_field_updates(self):
token = await self.account()
- for data in [[], {}, {'body': 'No title'}, {'title': 7}, {'title': ' '},
- {'title': 'x' * 101}, {'title': 'Valid', 'body': 3},
- {'title': 'Valid', 'body': 'x' * 10_001},
- {'title': 'Valid', 'pinned': 'true'}, {'title': 'Valid', 'owner': 'bob'}]:
+ for data in [
+ [],
+ {},
+ {'body': 'No title'},
+ {'title': 7},
+ {'title': ' '},
+ {'title': 'x' * 101},
+ {'title': 'Valid', 'body': 3},
+ {'title': 'Valid', 'body': 'x' * 10_001},
+ {'title': 'Valid', 'pinned': 'true'},
+ {'title': 'Valid', 'owner': 'bob'},
+ ]:
with self.subTest(data=data):
- self.assertEqual((await self.request('/api/notes', 'POST', data, token))[0], 400)
+ self.assertEqual(
+ (await self.request('/api/notes', 'POST', data, token))[0],
+ 400,
+ )
note = await self.create(token, title='Keep me')
path = f"/api/notes/{note['id']}"
for data in [{}, {'owner': 'bob'}, {'title': ''}, {'pinned': 1}]:
- self.assertEqual((await self.request(path, 'PATCH', data, token))[0], 400)
- self.assertEqual((await self.request(path, token=token))[:2], (200, note))
- self.assertEqual((await self.request(path, 'PUT', {'body': 'Missing title'}, token))[0], 400)
- status, pinned, _ = await self.request(path, 'PATCH', {'pinned': True}, token)
- self.assertEqual((status, pinned['pinned'], pinned['title']), (200, True, 'Keep me'))
-
- async def test_cookie_restores_login_and_logout_revokes_the_whole_login(self):
- credentials = {'username': 'alice', 'password': 'a long example passphrase'}
- self.assertEqual((await self.request('/api/signup', 'POST', credentials))[0], 201)
- status, login, headers = await self.request('/api/login', 'POST', credentials, scheme='https')
+ self.assertEqual(
+ (await self.request(path, 'PATCH', data, token))[0], 400
+ )
+ self.assertEqual(
+ (await self.request(path, token=token))[:2], (200, note)
+ )
+ self.assertEqual(
+ (
+ await self.request(
+ path, 'PUT', {'body': 'Missing title'}, token
+ )
+ )[0],
+ 400,
+ )
+ status, pinned, _ = await self.request(
+ path, 'PATCH', {'pinned': True}, token
+ )
+ self.assertEqual(
+ (status, pinned['pinned'], pinned['title']), (200, True, 'Keep me')
+ )
+
+ async def test_cookie_restores_login_and_logout_revokes_login(self):
+ credentials = {
+ 'username': 'alice',
+ 'password': 'a long example passphrase',
+ }
+ self.assertEqual(
+ (await self.request('/api/signup', 'POST', credentials))[0], 201
+ )
+ status, login, headers = await self.request(
+ '/api/login', 'POST', credentials, scheme='https'
+ )
self.assertEqual(status, 200)
cookies = SimpleCookie(headers[b'set-cookie'].decode())
- morsel = cookies[auth.refresh_cookie_name]
+ morsel = cookies[auth.REFRESH_COOKIE_NAME]
self.assertTrue(morsel['httponly'])
self.assertTrue(morsel['secure'])
self.assertEqual(morsel['samesite'], 'Strict')
self.assertEqual(morsel['path'], '/api')
self.assertEqual(int(morsel['max-age']), 7 * 24 * 3600)
self.assertEqual(morsel['domain'], '')
- cookie = f'{auth.refresh_cookie_name}={morsel.value}'
- status, restored, headers = await self.request('/api/refresh', 'POST', cookie=cookie)
+ cookie = f'{auth.REFRESH_COOKIE_NAME}={morsel.value}'
+ status, restored, headers = await self.request(
+ '/api/refresh', 'POST', cookie=cookie
+ )
self.assertEqual(status, 200)
self.assertNotEqual(restored['access_token'], login['access_token'])
self.assertEqual(restored['user'], {'id': 'alice'})
self.assertEqual(headers[b'cache-control'], b'no-store')
- self.assertEqual((await self.request('/api/me', token=restored['access_token']))[:2],
- (200, {'id': 'alice'}))
- # Cookies alone must never authorize notes endpoints, keeping bearer auth separate.
- self.assertEqual((await self.request('/api/notes', cookie=cookie))[0], 401)
+ self.assertEqual(
+ (await self.request('/api/me', token=restored['access_token']))[
+ :2
+ ],
+ (200, {'id': 'alice'}),
+ )
+ # Cookies alone must never authorize notes endpoints, keeping bearer
+ # auth separate.
+ self.assertEqual(
+ (await self.request('/api/notes', cookie=cookie))[0], 401
+ )
refreshed, logged_out = await asyncio.gather(
self.request('/api/refresh', 'POST', cookie=cookie),
- self.request('/api/logout', 'POST', token=restored['access_token']),
+ self.request(
+ '/api/logout', 'POST', token=restored['access_token']
+ ),
)
self.assertEqual(logged_out[0], 200)
self.assertIn(refreshed[0], (200, 401))
if refreshed[0] == 200:
- self.assertEqual((await self.request('/api/me', token=refreshed[1]['access_token']))[0], 401)
+ self.assertEqual(
+ (
+ await self.request(
+ '/api/me', token=refreshed[1]['access_token']
+ )
+ )[0],
+ 401,
+ )
_, _, headers = logged_out
- cleared = SimpleCookie(headers[b'set-cookie'].decode())[auth.refresh_cookie_name]
+ cleared = SimpleCookie(headers[b'set-cookie'].decode())[
+ auth.REFRESH_COOKIE_NAME
+ ]
self.assertEqual(cleared['max-age'], '0')
self.assertEqual(cleared['path'], '/api')
for access in [login['access_token'], restored['access_token']]:
- self.assertEqual((await self.request('/api/me', token=access))[0], 401)
- self.assertEqual((await self.request('/api/refresh', 'POST', cookie=cookie))[0], 401)
+ self.assertEqual(
+ (await self.request('/api/me', token=access))[0], 401
+ )
+ self.assertEqual(
+ (await self.request('/api/refresh', 'POST', cookie=cookie))[0], 401
+ )
- async def test_refresh_rejects_cross_origin_requests_without_revoking_valid_login(self):
- credentials = {'username': 'alice', 'password': 'a long example passphrase'}
+ async def test_cross_origin_refresh_preserves_valid_login(self):
+ credentials = {
+ 'username': 'alice',
+ 'password': 'a long example passphrase',
+ }
await self.request('/api/signup', 'POST', credentials)
_, _, headers = await self.request('/api/login', 'POST', credentials)
- cookie = SimpleCookie(headers[b'set-cookie'].decode())[auth.refresh_cookie_name]
- cookie = f'{auth.refresh_cookie_name}={cookie.value}'
+ cookie = SimpleCookie(headers[b'set-cookie'].decode())[
+ auth.REFRESH_COOKIE_NAME
+ ]
+ cookie = f'{auth.REFRESH_COOKIE_NAME}={cookie.value}'
for options in [
{'csrf_header': False},
{'extra_headers': [(b'origin', b'https://attacker.test')]},
@@ -371,65 +663,167 @@ asyncio.run(main())
{'extra_headers': [(b'sec-fetch-site', b'same-site')]},
]:
with self.subTest(options=options):
- self.assertEqual((await self.request('/api/refresh', 'POST', cookie=cookie, **options))[0], 403)
- self.assertEqual((await self.request('/api/refresh', 'POST', cookie=cookie,
- extra_headers=[(b'origin', b'http://notebook.test')]))[0], 200)
- self.assertEqual((await self.request('/api/login', 'POST', credentials, csrf_header=False))[0], 403)
+ self.assertEqual(
+ (
+ await self.request(
+ '/api/refresh', 'POST', cookie=cookie, **options
+ )
+ )[0],
+ 403,
+ )
+ self.assertEqual(
+ (
+ await self.request(
+ '/api/refresh',
+ 'POST',
+ cookie=cookie,
+ extra_headers=[(b'origin', b'http://notebook.test')],
+ )
+ )[0],
+ 200,
+ )
+ self.assertEqual(
+ (
+ await self.request(
+ '/api/login', 'POST', credentials, csrf_header=False
+ )
+ )[0],
+ 403,
+ )
- async def test_expired_and_invalid_refresh_cookies_clear_browser_state(self):
- credentials = {'username': 'alice', 'password': 'a long example passphrase'}
+ async def test_invalid_refresh_cookies_clear_browser_state(self):
+ credentials = {
+ 'username': 'alice',
+ 'password': 'a long example passphrase',
+ }
await self.request('/api/signup', 'POST', credentials)
- _, login, headers = await self.request('/api/login', 'POST', credentials)
- morsel = SimpleCookie(headers[b'set-cookie'].decode())[auth.refresh_cookie_name]
- self.assertFalse(morsel['secure'], 'Local HTTP demo cookies should work without TLS')
- cookie = f'{auth.refresh_cookie_name}={morsel.value}'
- # Access expiry is recoverable while the independent refresh credential is valid.
+ _, login, headers = await self.request(
+ '/api/login', 'POST', credentials
+ )
+ morsel = SimpleCookie(headers[b'set-cookie'].decode())[
+ auth.REFRESH_COOKIE_NAME
+ ]
+ self.assertFalse(
+ morsel['secure'], 'Local HTTP demo cookies should work without TLS'
+ )
+ cookie = f'{auth.REFRESH_COOKIE_NAME}={morsel.value}'
+ # Access expiry is recoverable while the independent refresh credential
+ # is valid.
now = auth.time()
with patch.object(auth, 'time', return_value=now + 3601):
- self.assertEqual((await self.request('/api/me', token=login['access_token']))[0], 401)
- self.assertEqual((await self.request('/api/refresh', 'POST', cookie=cookie))[0], 200)
+ self.assertEqual(
+ (await self.request('/api/me', token=login['access_token']))[
+ 0
+ ],
+ 401,
+ )
+ self.assertEqual(
+ (await self.request('/api/refresh', 'POST', cookie=cookie))[0],
+ 200,
+ )
with patch.object(auth, 'time', return_value=now + 7 * 24 * 3600 + 1):
- status, _, headers = await self.request('/api/refresh', 'POST', cookie=cookie)
+ status, _, headers = await self.request(
+ '/api/refresh', 'POST', cookie=cookie
+ )
self.assertEqual(status, 401)
- self.assertEqual(SimpleCookie(headers[b'set-cookie'].decode())[auth.refresh_cookie_name]['max-age'], '0')
+ self.assertEqual(
+ SimpleCookie(headers[b'set-cookie'].decode())[
+ auth.REFRESH_COOKIE_NAME
+ ]['max-age'],
+ '0',
+ )
for invalid in [None, 'notebook_refresh=unknown', 'malformed; cookie']:
- self.assertEqual((await self.request('/api/refresh', 'POST', cookie=invalid))[0], 401)
-
- async def test_rate_limits_slide_and_are_isolated_by_client_and_endpoint(self):
+ self.assertEqual(
+ (await self.request('/api/refresh', 'POST', cookie=invalid))[
+ 0
+ ],
+ 401,
+ )
+
+ async def test_rate_limits_slide_and_isolate_clients_and_endpoints(self):
with patch.object(rate_limit, 'monotonic', return_value=100):
- self.assertEqual((await self.request('/api/rate-limit-demo'))[0], 200)
+ self.assertEqual(
+ (await self.request('/api/rate-limit-demo'))[0], 200
+ )
with patch.object(rate_limit, 'monotonic', return_value=105):
- self.assertEqual((await self.request('/api/rate-limit-demo'))[0], 200)
+ self.assertEqual(
+ (await self.request('/api/rate-limit-demo'))[0], 200
+ )
with patch.object(rate_limit, 'monotonic', return_value=106):
- self.assertEqual((await self.request('/api/rate-limit-demo'))[0], 200)
+ self.assertEqual(
+ (await self.request('/api/rate-limit-demo'))[0], 200
+ )
with patch.object(rate_limit, 'monotonic', return_value=109):
status, error, headers = await self.request('/api/rate-limit-demo')
self.assertEqual(status, 429)
self.assertEqual(error['retry_after'], 1)
self.assertEqual(headers[b'retry-after'], b'1')
self.assertEqual(headers[b'cache-control'], b'no-store')
- self.assertEqual((await self.request('/api/rate-limit-demo', client='127.0.0.2'))[0], 200)
+ worker = self.new_app()
+ self.assertEqual(
+ (await self.request('/api/rate-limit-demo', app=worker))[0],
+ 200,
+ )
+ self.assertEqual(
+ (await self.request('/api/rate-limit-demo'))[0], 429
+ )
+ self.assertEqual(
+ (
+ await self.request(
+ '/api/rate-limit-demo', client='127.0.0.2'
+ )
+ )[0],
+ 200,
+ )
self.assertEqual((await self.request('/api/health'))[0], 200)
- self.assertEqual((await self.request('/api/rate-limit-demo',
- extra_headers=[(b'x-forwarded-for', b'127.0.0.3')]))[0], 429)
+ self.assertEqual(
+ (
+ await self.request(
+ '/api/rate-limit-demo',
+ extra_headers=[(b'x-forwarded-for', b'127.0.0.3')],
+ )
+ )[0],
+ 429,
+ )
with patch.object(rate_limit, 'monotonic', return_value=110):
- self.assertEqual((await self.request('/api/rate-limit-demo'))[0], 200)
+ self.assertEqual(
+ (await self.request('/api/rate-limit-demo'))[0], 200
+ )
with patch.object(rate_limit, 'monotonic', return_value=200):
- self.assertEqual((await self.request('/api/rate-limit-demo'))[0], 200)
+ self.assertEqual(
+ (await self.request('/api/rate-limit-demo'))[0], 200
+ )
async def test_concurrent_login_attempts_cannot_bypass_rate_limit(self):
- credentials = {'username': 'unknown', 'password': 'a long example passphrase'}
+ credentials = {
+ 'username': 'unknown',
+ 'password': 'a long example passphrase',
+ }
with patch.object(rate_limit, 'monotonic', return_value=100):
- responses = await asyncio.gather(*[
- self.request('/api/login', 'POST', credentials) for _ in range(6)
- ])
- self.assertEqual(sorted(status for status, _, _ in responses), [401] * 5 + [429])
- limited = next(response for response in responses if response[0] == 429)
+ responses = await asyncio.gather(
+ *[
+ self.request('/api/login', 'POST', credentials)
+ for _ in range(6)
+ ]
+ )
+ self.assertEqual(
+ sorted(status for status, _, _ in responses), [401] * 5 + [429]
+ )
+ limited = next(
+ response for response in responses if response[0] == 429
+ )
self.assertEqual(limited[2][b'retry-after'], b'60')
with patch.object(rate_limit, 'monotonic', return_value=110):
- self.assertEqual((await self.request('/api/login', 'POST', credentials))[2][b'retry-after'], b'50')
+ self.assertEqual(
+ (await self.request('/api/login', 'POST', credentials))[2][
+ b'retry-after'
+ ],
+ b'50',
+ )
with patch.object(rate_limit, 'monotonic', return_value=160):
- self.assertEqual((await self.request('/api/login', 'POST', credentials))[0], 401)
+ self.assertEqual(
+ (await self.request('/api/login', 'POST', credentials))[0], 401
+ )
if __name__ == '__main__':
diff --git a/tests/test_wred.py b/tests/test_wred.py
index 9c8b482..413d585 100644
--- a/tests/test_wred.py
+++ b/tests/test_wred.py
@@ -1,3 +1,14 @@
+# -*- coding: utf-8 -*-
+"""
+ tests.test_wred
+ ~~~~~~~~~~~~~~~
+
+ Exercise Wred request handling and ASGI response behavior.
+
+ :copyright: (c) 2026 by Harrison Erd.
+ :license: BSD-3-Clause, see LICENSE for more details.
+"""
+
import tempfile
import unittest
from contextlib import chdir
@@ -21,18 +32,29 @@ async def call_app(app, scope, receive):
class WredTests(unittest.IsolatedAsyncioTestCase):
- async def test_callable_responses_control_asgi_delivery_on_fixed_and_dynamic_routes(self):
+ async def test_callable_responses_on_fixed_and_dynamic_routes(self):
class Response:
def __init__(self, name):
self.name = name
async def __call__(self, scope, receive, send):
message = await receive()
- await send({'type': 'http.response.start', 'status': 202,
- 'headers': [(b'content-type', b'text/plain'), (b'x-response', b'asgi')]})
- await send({'type': 'http.response.body',
- 'body': f'{scope["path"]}:{self.name}:'.encode(), 'more_body': True})
- await send({'type': 'http.response.body', 'body': message['body']})
+ await send({
+ 'type': 'http.response.start',
+ 'status': 202,
+ 'headers': [
+ (b'content-type', b'text/plain'),
+ (b'x-response', b'asgi'),
+ ],
+ })
+ await send({
+ 'type': 'http.response.body',
+ 'body': f'{scope["path"]}:{self.name}:'.encode(),
+ 'more_body': True,
+ })
+ await send(
+ {'type': 'http.response.body', 'body': message['body']}
+ )
app = App()
@@ -51,31 +73,60 @@ class WredTests(unittest.IsolatedAsyncioTestCase):
('/api/dynamic/alice', '/api', b'/api/dynamic/alice:alice:'),
]:
with self.subTest(path=path):
+
async def receive():
return {'type': 'http.request', 'body': b'payload'}
- messages = await call_app(app, {
- 'type': 'http', 'path': path, 'root_path': root_path, 'method': 'POST',
- }, receive)
- self.assertEqual(messages, [
- {'type': 'http.response.start', 'status': 202,
- 'headers': [(b'content-type', b'text/plain'), (b'x-response', b'asgi')]},
- {'type': 'http.response.body', 'body': prefix, 'more_body': True},
- {'type': 'http.response.body', 'body': b'payload'},
- ])
-
- async def test_mounted_routing_uses_complete_root_path_without_changing_scope(self):
+ messages = await call_app(
+ app,
+ {
+ 'type': 'http',
+ 'path': path,
+ 'root_path': root_path,
+ 'method': 'POST',
+ },
+ receive,
+ )
+ self.assertEqual(
+ messages,
+ [
+ {
+ 'type': 'http.response.start',
+ 'status': 202,
+ 'headers': [
+ (b'content-type', b'text/plain'),
+ (b'x-response', b'asgi'),
+ ],
+ },
+ {
+ 'type': 'http.response.body',
+ 'body': prefix,
+ 'more_body': True,
+ },
+ {'type': 'http.response.body', 'body': b'payload'},
+ ],
+ )
+
+ async def test_mounted_routing_preserves_scope(self):
app = App()
@app.route('/items/{name}')
async def item(request, name):
- return {'route': name, 'path': request.path, 'scope_path': request.scope['path']}
+ return {
+ 'route': name,
+ 'path': request.path,
+ 'scope_path': request.scope['path'],
+ }
@app.route('/items/special')
@app.route('/')
@app.route('/apiary/items')
async def fixed(request):
- return {'route': 'fixed', 'path': request.path, 'scope_path': request.scope['path']}
+ return {
+ 'route': 'fixed',
+ 'path': request.path,
+ 'scope_path': request.scope['path'],
+ }
cases = [
('/items/alice', None, 'alice'),
@@ -92,6 +143,7 @@ class WredTests(unittest.IsolatedAsyncioTestCase):
]
for path, root_path, route in cases:
with self.subTest(path=path, root_path=root_path):
+
async def receive():
raise AssertionError('GET handlers should not read a body')
@@ -101,14 +153,20 @@ class WredTests(unittest.IsolatedAsyncioTestCase):
original = scope.copy()
messages = await call_app(app, scope, receive)
self.assertEqual(messages[0]['status'], 200)
- self.assertEqual(orjson.loads(messages[1]['body']), {
- 'route': route, 'path': path, 'scope_path': path,
- })
+ self.assertEqual(
+ orjson.loads(messages[1]['body']),
+ {
+ 'route': route,
+ 'path': path,
+ 'scope_path': path,
+ },
+ )
self.assertEqual(scope, original)
async def test_file_delivery_and_root_confinement(self):
# Exercise the public helper through the actual ASGI response boundary.
- # This guards MIME/status delivery and accidental exposure outside root.
+ # This guards MIME/status delivery and accidental exposure outside
+ # root.
with tempfile.TemporaryDirectory() as directory:
base = Path(directory)
root = base / 'static'
@@ -130,32 +188,102 @@ class WredTests(unittest.IsolatedAsyncioTestCase):
cases = [
('index.html', 200, b'text/html', b'<p>Hello</p>'),
- ('asset.unknownextension', 200, b'application/octet-stream', b'\x00\xff'),
- ('missing.txt', 404, b'application/json', b'{"error":"Not Found"}'),
+ (
+ 'asset.unknownextension',
+ 200,
+ b'application/octet-stream',
+ b'\x00\xff',
+ ),
+ (
+ 'missing.txt',
+ 404,
+ b'application/json',
+ b'{"error":"Not Found"}',
+ ),
('.', 404, b'application/json', b'{"error":"Not Found"}'),
- ('../secret.txt', 404, b'application/json', b'{"error":"Not Found"}'),
- (str(base / 'secret.txt'), 404, b'application/json', b'{"error":"Not Found"}'),
- ('escape.txt', 404, b'application/json', b'{"error":"Not Found"}'),
- ('invalid\x00.txt', 404, b'application/json', b'{"error":"Not Found"}'),
+ (
+ '../secret.txt',
+ 404,
+ b'application/json',
+ b'{"error":"Not Found"}',
+ ),
+ (
+ str(base / 'secret.txt'),
+ 404,
+ b'application/json',
+ b'{"error":"Not Found"}',
+ ),
+ (
+ 'escape.txt',
+ 404,
+ b'application/json',
+ b'{"error":"Not Found"}',
+ ),
+ (
+ 'invalid\x00.txt',
+ 404,
+ b'application/json',
+ b'{"error":"Not Found"}',
+ ),
]
cases = [
(options, *case)
- for options in ({}, {'root': ''}, {'root': None}, {'root': root}, {'root': str(root)})
+ for options in (
+ {},
+ {'root': ''},
+ {'root': None},
+ {'root': root},
+ {'root': str(root)},
+ )
for case in cases
]
- cases.append(({'root': custom_root}, 'index.html', 200, b'text/html', b'<p>Custom root</p>'))
+ cases.append(
+ (
+ {'root': custom_root},
+ 'index.html',
+ 200,
+ b'text/html',
+ b'<p>Custom root</p>',
+ )
+ )
with chdir(base):
- for root_options, filename, status, content_type, body in cases:
+ for (
+ root_options,
+ filename,
+ status,
+ content_type,
+ body,
+ ) in cases:
with self.subTest(root=root_options, filename=filename):
- async def receive():
- return {'type': 'http.request', 'body': orjson.dumps({'filename': filename})}
- messages = await call_app(app, {'type': 'http', 'path': '/file', 'method': 'POST'}, receive)
- self.assertEqual(messages, [
- {'type': 'http.response.start', 'status': status,
- 'headers': [(b'content-type', content_type)]},
- {'type': 'http.response.body', 'body': body},
- ])
+ async def receive():
+ return {
+ 'type': 'http.request',
+ 'body': orjson.dumps({'filename': filename}),
+ }
+
+ messages = await call_app(
+ app,
+ {
+ 'type': 'http',
+ 'path': '/file',
+ 'method': 'POST',
+ },
+ receive,
+ )
+ self.assertEqual(
+ messages,
+ [
+ {
+ 'type': 'http.response.start',
+ 'status': status,
+ 'headers': [
+ (b'content-type', content_type)
+ ],
+ },
+ {'type': 'http.response.body', 'body': body},
+ ],
+ )
async def test_dictionary_and_list_responses_are_json(self):
app = App()
@@ -174,12 +302,24 @@ class WredTests(unittest.IsolatedAsyncioTestCase):
]
for value, body in cases:
with self.subTest(value=value):
- messages = await call_app(app, {'type': 'http', 'path': '/', 'method': 'GET'}, receive)
- self.assertEqual(messages, [
- {'type': 'http.response.start', 'status': 200,
- 'headers': [(b'content-type', b'application/json')]},
- {'type': 'http.response.body', 'body': body},
- ])
+ messages = await call_app(
+ app,
+ {'type': 'http', 'path': '/', 'method': 'GET'},
+ receive,
+ )
+ self.assertEqual(
+ messages,
+ [
+ {
+ 'type': 'http.response.start',
+ 'status': 200,
+ 'headers': [
+ (b'content-type', b'application/json')
+ ],
+ },
+ {'type': 'http.response.body', 'body': body},
+ ],
+ )
async def test_custom_status_and_explicit_json_errors(self):
app = App()
@@ -205,15 +345,28 @@ class WredTests(unittest.IsolatedAsyncioTestCase):
]
for method, path, request_body, status, body in cases:
with self.subTest(path=path):
+
async def receive():
return {'type': 'http.request', 'body': request_body}
- messages = await call_app(app, {'type': 'http', 'path': path, 'method': method}, receive)
- self.assertEqual(messages, [
- {'type': 'http.response.start', 'status': status,
- 'headers': [(b'content-type', b'application/json')]},
- {'type': 'http.response.body', 'body': body},
- ])
+ messages = await call_app(
+ app,
+ {'type': 'http', 'path': path, 'method': method},
+ receive,
+ )
+ self.assertEqual(
+ messages,
+ [
+ {
+ 'type': 'http.response.start',
+ 'status': status,
+ 'headers': [
+ (b'content-type', b'application/json')
+ ],
+ },
+ {'type': 'http.response.body', 'body': body},
+ ],
+ )
async def test_parsing_and_response_errors_propagate_without_sending(self):
app = App()
@@ -239,7 +392,9 @@ class WredTests(unittest.IsolatedAsyncioTestCase):
('GET', '/invalid', b'', b'', (b'bad', 'text/plain'), ValueError),
]
for method, path, body, query_string, response, error in cases:
- with self.subTest(path=path, query_string=query_string, response=response):
+ with self.subTest(
+ path=path, query_string=query_string, response=response
+ ):
messages = []
async def receive():
@@ -249,19 +404,34 @@ class WredTests(unittest.IsolatedAsyncioTestCase):
messages.append(message)
with self.assertRaises(error):
- await app({'type': 'http', 'path': path, 'method': method,
- 'query_string': query_string}, receive, send)
+ await app(
+ {
+ 'type': 'http',
+ 'path': path,
+ 'method': method,
+ 'query_string': query_string,
+ },
+ receive,
+ send,
+ )
self.assertEqual(messages, [])
- async def test_callable_response_errors_do_not_replace_a_started_stream(self):
+ async def test_callable_response_errors_preserve_started_stream(self):
app = App()
@app.route('/stream', methods=['POST'])
async def stream(request):
async def response(scope, receive, send):
- await send({'type': 'http.response.start', 'status': 200,
- 'headers': [(b'content-type', b'text/plain')]})
- await send({'type': 'http.response.body', 'body': b'hello\n', 'more_body': True})
+ await send({
+ 'type': 'http.response.start',
+ 'status': 200,
+ 'headers': [(b'content-type', b'text/plain')],
+ })
+ await send({
+ 'type': 'http.response.body',
+ 'body': b'hello\n',
+ 'more_body': True,
+ })
await request.json()
return response
@@ -275,12 +445,26 @@ class WredTests(unittest.IsolatedAsyncioTestCase):
messages.append(message)
with self.assertRaises(orjson.JSONDecodeError):
- await app({'type': 'http', 'path': '/stream', 'method': 'POST'}, receive, send)
- self.assertEqual(messages, [
- {'type': 'http.response.start', 'status': 200,
- 'headers': [(b'content-type', b'text/plain')]},
- {'type': 'http.response.body', 'body': b'hello\n', 'more_body': True},
- ])
+ await app(
+ {'type': 'http', 'path': '/stream', 'method': 'POST'},
+ receive,
+ send,
+ )
+ self.assertEqual(
+ messages,
+ [
+ {
+ 'type': 'http.response.start',
+ 'status': 200,
+ 'headers': [(b'content-type', b'text/plain')],
+ },
+ {
+ 'type': 'http.response.body',
+ 'body': b'hello\n',
+ 'more_body': True,
+ },
+ ],
+ )
async def test_route_priority_and_method_registration(self):
app = App()
@@ -310,34 +494,60 @@ class WredTests(unittest.IsolatedAsyncioTestCase):
]
for method, path, body in cases:
with self.subTest(method=method, path=path):
+
async def receive():
raise AssertionError('These handlers do not read the body')
- messages = await call_app(app, {'type': 'http', 'path': path, 'method': method}, receive)
+ messages = await call_app(
+ app,
+ {'type': 'http', 'path': path, 'method': method},
+ receive,
+ )
self.assertEqual(messages[0]['status'], 200)
self.assertEqual(messages[1]['body'], body)
- async def test_query_parameters_decode_to_strings_without_reading_body(self):
+ async def test_query_decodes_to_strings_without_reading_body(self):
app = App()
@app.route('/search')
async def search(request):
- return {'params': request.query(), 'q': request.query().get('q', '')}
+ return {
+ 'params': request.query(),
+ 'q': request.query().get('q', ''),
+ }
cases = [
(None, 200, {'params': {}, 'q': ''}),
(b'', 200, {'params': {}, 'q': ''}),
- (b'q=coffee+ideas&limit=20', 200,
- {'params': {'q': 'coffee ideas', 'limit': '20'}, 'q': 'coffee ideas'}),
- (b'q=C%2B%2B&label=caf%C3%A9&x%26y=a%3Db', 200,
- {'params': {'q': 'C++', 'label': 'café', 'x&y': 'a=b'}, 'q': 'C++'}),
- (b'q=&bare&tag=first&tag=last', 200,
- {'params': {'q': '', 'bare': '', 'tag': 'last'}, 'q': ''}),
+ (
+ b'q=coffee+ideas&limit=20',
+ 200,
+ {
+ 'params': {'q': 'coffee ideas', 'limit': '20'},
+ 'q': 'coffee ideas',
+ },
+ ),
+ (
+ b'q=C%2B%2B&label=caf%C3%A9&x%26y=a%3Db',
+ 200,
+ {
+ 'params': {'q': 'C++', 'label': 'café', 'x&y': 'a=b'},
+ 'q': 'C++',
+ },
+ ),
+ (
+ b'q=&bare&tag=first&tag=last',
+ 200,
+ {'params': {'q': '', 'bare': '', 'tag': 'last'}, 'q': ''},
+ ),
]
for query_string, status, body in cases:
with self.subTest(query_string=query_string):
+
async def receive():
- raise AssertionError('Query parsing should not read the request body')
+ raise AssertionError(
+ 'Query parsing should not read the request body'
+ )
scope = {'type': 'http', 'path': '/search', 'method': 'GET'}
if query_string is not None:
@@ -350,9 +560,17 @@ class WredTests(unittest.IsolatedAsyncioTestCase):
app = App()
reads = 0
chunks = iter([
- {'type': 'http.request', 'body': b'{"name":', 'more_body': True},
+ {
+ 'type': 'http.request',
+ 'body': b'{"name":',
+ 'more_body': True,
+ },
{'type': 'http.request', 'more_body': True},
- {'type': 'http.request', 'body': b'"Alice"}', 'more_body': False},
+ {
+ 'type': 'http.request',
+ 'body': b'"Alice"}',
+ 'more_body': False,
+ },
])
@app.route('/stream', methods=['POST'])
@@ -360,28 +578,47 @@ class WredTests(unittest.IsolatedAsyncioTestCase):
first = await request.body()
parsed = await request.json()
cached = await request.body()
- return {'raw': first.decode(), 'cached': cached.decode(),
- 'parsed': parsed, 'header': request.headers['x-example'],
- 'cookie': request.headers['cookie']}
+ return {
+ 'raw': first.decode(),
+ 'cached': cached.decode(),
+ 'parsed': parsed,
+ 'header': request.headers['x-example'],
+ 'cookie': request.headers['cookie'],
+ }
async def receive():
nonlocal reads
reads += 1
return next(chunks)
- messages = await call_app(app, {
- 'type': 'http', 'http_version': '2', 'path': '/stream', 'method': 'POST',
- 'headers': [(b'X-Example', b'caf\xe9'),
- (b'Cookie', b'notebook_refresh=valid-token'),
- (b'cookie', b'theme=dark'), (b'COOKIE', b'lang=en')],
- }, receive)
+ messages = await call_app(
+ app,
+ {
+ 'type': 'http',
+ 'http_version': '2',
+ 'path': '/stream',
+ 'method': 'POST',
+ 'headers': [
+ (b'X-Example', b'caf\xe9'),
+ (b'Cookie', b'notebook_refresh=valid-token'),
+ (b'cookie', b'theme=dark'),
+ (b'COOKIE', b'lang=en'),
+ ],
+ },
+ receive,
+ )
self.assertEqual(reads, 3)
self.assertEqual(messages[0]['status'], 200)
- self.assertEqual(orjson.loads(messages[1]['body']), {
- 'raw': '{"name":"Alice"}', 'cached': '{"name":"Alice"}',
- 'parsed': {'name': 'Alice'}, 'header': 'café',
- 'cookie': 'notebook_refresh=valid-token; theme=dark; lang=en',
- })
+ self.assertEqual(
+ orjson.loads(messages[1]['body']),
+ {
+ 'raw': '{"name":"Alice"}',
+ 'cached': '{"name":"Alice"}',
+ 'parsed': {'name': 'Alice'},
+ 'header': 'café',
+ 'cookie': 'notebook_refresh=valid-token; theme=dark; lang=en',
+ },
+ )
async def test_disconnect_stops_body_reading(self):
app = App()
@@ -404,11 +641,15 @@ class WredTests(unittest.IsolatedAsyncioTestCase):
messages.append(message)
with self.assertRaisesRegex(ConnectionError, 'Client disconnected'):
- await app({'type': 'http', 'path': '/input', 'method': 'POST'}, receive, send)
+ await app(
+ {'type': 'http', 'path': '/input', 'method': 'POST'},
+ receive,
+ send,
+ )
self.assertEqual(messages, [])
self.assertEqual(reads, 1)
- async def test_unsupported_protocols_raise_without_receiving_or_sending(self):
+ async def test_unsupported_protocols_raise_without_io(self):
app = App()
async def receive():
@@ -421,7 +662,9 @@ class WredTests(unittest.IsolatedAsyncioTestCase):
async def send(message):
messages.append(message)
- with self.assertRaisesRegex(RuntimeError, 'Unsupported ASGI scope type'):
+ with self.assertRaisesRegex(
+ RuntimeError, 'Unsupported ASGI scope type'
+ ):
await app({'type': protocol}, receive, send)
self.assertEqual(messages, [])
@@ -435,16 +678,34 @@ class WredTests(unittest.IsolatedAsyncioTestCase):
return await app.static('private.txt', root=directory)
async def receive():
- raise AssertionError('File serving should not read the request body')
-
- # Simulate the OS refusing a real file read, including when tests run as root.
- with patch.object(Path, 'read_bytes', side_effect=PermissionError('denied')):
- messages = await call_app(app, {'type': 'http', 'path': '/', 'method': 'GET'}, receive)
- self.assertEqual(messages, [
- {'type': 'http.response.start', 'status': 404,
- 'headers': [(b'content-type', b'application/json')]},
- {'type': 'http.response.body', 'body': b'{"error":"Not Found"}'},
- ])
+ raise AssertionError(
+ 'File serving should not read the request body'
+ )
+
+ # Simulate the OS refusing a real file read, including when tests
+ # run as root.
+ with patch.object(
+ Path, 'read_bytes', side_effect=PermissionError('denied')
+ ):
+ messages = await call_app(
+ app,
+ {'type': 'http', 'path': '/', 'method': 'GET'},
+ receive,
+ )
+ self.assertEqual(
+ messages,
+ [
+ {
+ 'type': 'http.response.start',
+ 'status': 404,
+ 'headers': [(b'content-type', b'application/json')],
+ },
+ {
+ 'type': 'http.response.body',
+ 'body': b'{"error":"Not Found"}',
+ },
+ ],
+ )
async def test_raw_and_custom_response_bodies(self):
app = App()
@@ -455,32 +716,93 @@ class WredTests(unittest.IsolatedAsyncioTestCase):
cases = [
(b'\x00\xff', 200, b'application/octet-stream', b'\x00\xff', []),
- ((b'<p>Hello</p>', 'text/html; charset=utf-8', 202),
- 202, b'text/html; charset=utf-8', b'<p>Hello</p>', []),
- (([], 'application/json', 201), 201, b'application/json', b'[]', []),
- (({'error': 'Unauthorized'}, 'application/json', 401,
- [('WWW-Authenticate', 'Bearer'), ('Cache-Control', 'no-store')]),
- 401, b'application/json', b'{"error":"Unauthorized"}',
- [(b'www-authenticate', b'Bearer'), (b'cache-control', b'no-store')]),
- ((b'', 'text/plain', 303,
- [('Location', '/ui'), ('X-Label', 'caf\u00e9'),
- ('Set-Cookie', 'a=1; Secure'), ('Set-Cookie', 'b=2; Secure')]),
- 303, b'text/plain', b'',
- [(b'location', b'/ui'), (b'x-label', b'caf\xe9'),
- (b'set-cookie', b'a=1; Secure'), (b'set-cookie', b'b=2; Secure')]),
- (([], 'application/json', 200, []), 200, b'application/json', b'[]', []),
+ (
+ (b'<p>Hello</p>', 'text/html; charset=utf-8', 202),
+ 202,
+ b'text/html; charset=utf-8',
+ b'<p>Hello</p>',
+ [],
+ ),
+ (
+ ([], 'application/json', 201),
+ 201,
+ b'application/json',
+ b'[]',
+ [],
+ ),
+ (
+ (
+ {'error': 'Unauthorized'},
+ 'application/json',
+ 401,
+ [
+ ('WWW-Authenticate', 'Bearer'),
+ ('Cache-Control', 'no-store'),
+ ],
+ ),
+ 401,
+ b'application/json',
+ b'{"error":"Unauthorized"}',
+ [
+ (b'www-authenticate', b'Bearer'),
+ (b'cache-control', b'no-store'),
+ ],
+ ),
+ (
+ (
+ b'',
+ 'text/plain',
+ 303,
+ [
+ ('Location', '/ui'),
+ ('X-Label', 'caf\u00e9'),
+ ('Set-Cookie', 'a=1; Secure'),
+ ('Set-Cookie', 'b=2; Secure'),
+ ],
+ ),
+ 303,
+ b'text/plain',
+ b'',
+ [
+ (b'location', b'/ui'),
+ (b'x-label', b'caf\xe9'),
+ (b'set-cookie', b'a=1; Secure'),
+ (b'set-cookie', b'b=2; Secure'),
+ ],
+ ),
+ (
+ ([], 'application/json', 200, []),
+ 200,
+ b'application/json',
+ b'[]',
+ [],
+ ),
]
for response, status, content_type, body, headers in cases:
with self.subTest(response=response):
+
async def receive():
- raise AssertionError('These responses do not use a request body')
-
- messages = await call_app(app, {'type': 'http', 'path': '/', 'method': 'GET'}, receive)
- self.assertEqual(messages, [
- {'type': 'http.response.start', 'status': status,
- 'headers': [(b'content-type', content_type)] + headers},
- {'type': 'http.response.body', 'body': body},
- ])
+ raise AssertionError(
+ 'These responses do not use a request body'
+ )
+
+ messages = await call_app(
+ app,
+ {'type': 'http', 'path': '/', 'method': 'GET'},
+ receive,
+ )
+ self.assertEqual(
+ messages,
+ [
+ {
+ 'type': 'http.response.start',
+ 'status': status,
+ 'headers': [(b'content-type', content_type)]
+ + headers,
+ },
+ {'type': 'http.response.body', 'body': body},
+ ],
+ )
if __name__ == '__main__':
diff --git a/wred.py b/wred.py
index e7d31fe..cfa7110 100644
--- a/wred.py
+++ b/wred.py
@@ -38,9 +38,11 @@ class Request:
chunks = []
while True:
message = await self.receive()
- if message['type'] != 'http.request': raise ConnectionError("Client disconnected")
+ if message['type'] != 'http.request':
+ raise ConnectionError("Client disconnected")
chunks.append(message.get('body', b''))
- if not message.get('more_body'): break
+ if not message.get('more_body'):
+ break
self._body = b''.join(chunks)
return self._body
@@ -58,14 +60,16 @@ class App:
async def __call__(self, scope: dict[str, Any], receive: Callable[[], Awaitable[dict[str, Any]]], send: Callable[[dict[str, Any]], Awaitable[None]]) -> None:
"""Handle ASGI requests."""
- if scope['type'] != 'http': raise RuntimeError(f"Unsupported ASGI scope type: {scope['type']}")
+ if scope['type'] != 'http':
+ raise RuntimeError(f"Unsupported ASGI scope type: {scope['type']}")
await self.handle_http(Request(scope, receive, send))
async def handle_http(self, request: Request) -> None:
"""Handle HTTP requests with routing and JSON response support."""
path = request.path
root = request.scope.get('root_path', '').rstrip('/')
- if root and (path == root or path.startswith(root + '/')): path = path[len(root):] or '/'
+ if root and (path == root or path.startswith(root + '/')):
+ path = path[len(root):] or '/'
path = path.rstrip('/') or '/'
method = request.method
if path in self.routes and method in self.routes[path]:
@@ -78,17 +82,22 @@ class App:
async def dispatch_response(self, request: Request, response: Any) -> None:
"""Invoke an ASGI response or serialize an ordinary handler result."""
- if callable(response): await response(request.scope, request.receive, request.send)
- else: await self.send_response(request.send, response)
+ if callable(response):
+ await response(request.scope, request.receive, request.send)
+ else:
+ await self.send_response(request.send, response)
async def send_response(self, send: Callable[[dict[str, Any]], Awaitable[None]], response: Any, status: int = 200) -> None:
"""Send JSON or (body, content_type, status[, headers]) with string header pairs."""
body, content_type, headers = response, 'application/octet-stream', []
if isinstance(response, tuple):
body, content_type, status = response[:3]
- if len(response) == 4: headers = [(n.lower().encode('ascii'), v.encode('latin-1')) for n, v in response[3]]
- if isinstance(body, (dict, list)): body, content_type = orjson.dumps(body), 'application/json'
- elif not isinstance(body, bytes): raise TypeError("Response must be dict, list, bytes, or tuple")
+ if len(response) == 4:
+ headers = [(n.lower().encode('ascii'), v.encode('latin-1')) for n, v in response[3]]
+ if isinstance(body, (dict, list)):
+ body, content_type = orjson.dumps(body), 'application/json'
+ elif not isinstance(body, bytes):
+ raise TypeError("Response must be dict, list, bytes, or tuple")
await send({'type': 'http.response.start', 'status': status, 'headers': [(b'content-type', content_type.encode('latin-1'))] + headers})
await send({'type': 'http.response.body', 'body': body})
@@ -97,8 +106,10 @@ class App:
if len(pattern) != len(parts): return None
params = {}
for p, part in zip(pattern, parts):
- if p.startswith('{') and p.endswith('}'): params[p[1:-1]] = part
- elif p != part: return None
+ if p.startswith('{') and p.endswith('}'):
+ params[p[1:-1]] = part
+ elif p != part:
+ return None
return params
async def static(self, filename: str | Path, *, root: str | Path | None = 'static') -> tuple[bytes, str, int]:
@@ -107,9 +118,11 @@ class App:
try:
directory = Path(root or 'static').resolve()
path = (directory / filename).resolve()
- if not path.is_relative_to(directory) or not path.is_file(): return b'{"error":"Not Found"}', 'application/json', 404
+ if not path.is_relative_to(directory) or not path.is_file():
+ return b'{"error":"Not Found"}', 'application/json', 404
return path.read_bytes(), guess_type(path.name)[0] or 'application/octet-stream', 200
- except (OSError, ValueError, RuntimeError): return b'{"error":"Not Found"}', 'application/json', 404
+ except (OSError, ValueError, RuntimeError):
+ return b'{"error":"Not Found"}', 'application/json', 404
return await to_thread(read)
def route(self, path: str, methods: list[str] | tuple[str, ...] = ('GET',)) -> Callable: