patx/wred
Persist notebook example data with async MongoDB
Commit 5265e03 · patx · 2026-10-15T14:21:53-04:00
Comments
No comments yet.
Diff
diff --git a/.gitignore b/.gitignore
index 536cf1b..65c99e5 100644
--- a/.gitignore
+++ b/.gitignore
@@ -23,3 +23,6 @@ coverage.xml
.env.*
!.env.example
.aws/
+
+# Local example database
+.data/
diff --git a/README.md b/README.md
index 93edebe..31b09a7 100644
--- a/README.md
+++ b/README.md
@@ -54,6 +54,58 @@ Unsupported protocols raise `RuntimeError` without sending HTTP messages,
following [ASGI guidance](https://asgi.readthedocs.io/en/latest/specs/main.html#protocol-specifications).
You can also export a middleware wrapper as `app`.
+## Notebook example
+
+The repository includes a complete notebook app with signup, login, private notes,
+search, pagination, and JSON export. It uses MongoDB with PyMongo's
+[native async client](https://www.mongodb.com/docs/languages/python/pymongo-driver/current/reference/migration/).
+Install the example's optional dependency from the repository root:
+
+```bash
+uv sync --extra example
+```
+
+Start a local MongoDB server, or use an existing MongoDB deployment. For example,
+with Docker installed:
+
+```bash
+docker run -d --name wred-mongo -p 127.0.0.1:27017:27017 -v wred-mongo-data:/data/db mongo:8
+```
+
+Then start the app and open <http://127.0.0.1:8000>:
+
+```bash
+uv run --extra example weft example_app.app:app
+```
+
+The defaults are `MONGODB_URI=mongodb://127.0.0.1:27017` and
+`MONGODB_DATABASE=wred_notebook`. Set these environment variables to choose another
+server or database. Collections and indexes are created automatically. Accounts,
+notes, and login sessions survive app restarts as long as the MongoDB data is
+preserved. Passwords are hashed; only digests of access and refresh tokens are
+stored.
+
+Database calls use `await` with one connection pool per app worker. The example
+wraps Wred with an ASGI lifespan handler to initialize indexes on startup and close
+the client on shutdown; Wred's core stays HTTP-only. HTTP-only servers initialize
+the database on the first database request. Each login is one document, keeping
+refresh and logout atomic on a standalone MongoDB server. Expired sessions are
+removed by a TTL index; authorization also checks expiry immediately. Note search,
+filtering, sorting, and pagination run in MongoDB. Rate-limit counters remain per
+process and reset on restart; use shared rate limiting for multiple workers and
+HTTPS outside localhost.
+
+Run the full test suite against a running MongoDB server:
+
+```bash
+WRED_TEST_MONGODB_URI=mongodb://127.0.0.1:27017 uv run --extra example python -m unittest discover -v
+```
+
+Integration tests create and remove separate databases named `wred_test_<random>`;
+the test account needs permission to create indexes and drop those databases.
+Without `WRED_TEST_MONGODB_URI`, the core tests run and the MongoDB integration
+tests are skipped.
+
## Routes
Handlers are async and receive a request. Methods default to `['GET']`;
diff --git a/example_app/app.py b/example_app/app.py
index 1d271ff..62d3498 100644
--- a/example_app/app.py
+++ b/example_app/app.py
@@ -1,18 +1,56 @@
-"""Run the notebook demo from the repository root:
+"""Run the notebook demo with MongoDB listening on localhost:27017:
- uv sync
- uv run weft example_app.app:app
+ uv sync --extra example
+ uv run --extra example weft example_app.app:app
-Open http://127.0.0.1:8000. Users, tokens, and notes disappear on restart.
-Login is restored with an HttpOnly refresh cookie; API calls use bearer tokens.
-Storage and rate limits are process-local. Production needs shared storage and HTTPS.
+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.
"""
+import os
+
from wred import App
from . import auth, notes, pages
+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)
+ self.http = App()
+ pages.register_routes(self.http)
+ auth.register_routes(self.http, self.database)
+ notes.register_routes(self.http, self.database)
+
+ async def __call__(self, scope, receive, send):
+ if scope['type'] != 'lifespan':
+ return await self.http(scope, receive, send)
+ while True:
+ message = await receive()
+ if message['type'] == 'lifespan.startup':
+ try:
+ await self.database.initialize()
+ except Exception as error:
+ await self.database.close()
+ await send({'type': 'lifespan.startup.failed', 'message': str(error)})
+ return
+ await send({'type': 'lifespan.startup.complete'})
+ elif message['type'] == 'lifespan.shutdown':
+ await self.database.close()
+ await send({'type': 'lifespan.shutdown.complete'})
+ 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'),
+ )
+
-app = App()
-pages.register_routes(app)
-auth.register_routes(app)
-notes.register_routes(app)
+app = create_app()
diff --git a/example_app/auth.py b/example_app/auth.py
index a8df158..9b08300 100644
--- a/example_app/auth.py
+++ b/example_app/auth.py
@@ -5,11 +5,10 @@ from functools import wraps
from hashlib import pbkdf2_hmac, sha256
from http.cookies import CookieError, SimpleCookie
from secrets import compare_digest, token_bytes, token_urlsafe
-from time import monotonic
+from time import time
from wred import WredRequestError
from .rate_limit import rate_limit
-from .storage import users, tokens, refresh_tokens
token_lifetime = 3600
refresh_lifetime = 7 * 24 * 3600
@@ -52,16 +51,7 @@ def read_refresh_cookie(request):
return morsel.value if morsel else ''
-def issue_access_token(username, login_id):
- token = token_urlsafe(32)
- # Store a digest so the server does not retain the raw bearer credential.
- digest = sha256(token.encode('utf-8')).hexdigest()
- # Tag tokens with their login ID so logout can revoke all tokens from that login.
- tokens[digest] = {
- 'user_id': username,
- 'login_id': login_id,
- 'expires_at': monotonic() + token_lifetime,
- }
+def token_response(token, username):
return {
'access_token': token,
'token_type': 'bearer',
@@ -70,13 +60,6 @@ def issue_access_token(username, login_id):
}
-def revoke_login(login_id):
- for store in (tokens, refresh_tokens):
- for digest, record in list(store.items()):
- if record['login_id'] == login_id:
- del store[digest]
-
-
async def read_credentials(request):
data = await request.json()
if not isinstance(data, dict):
@@ -94,46 +77,39 @@ async def hash_password(password, salt):
return await to_thread(pbkdf2_hmac, 'sha256', password.encode('utf-8'), salt, 600_000)
-async def validate_access_token(token):
- digest = sha256(token.encode('utf-8')).hexdigest()
- record = tokens.get(digest)
- if record is None:
- return None
- if record['expires_at'] <= monotonic():
- tokens.pop(digest, None)
- return None
- return {'id': record['user_id']}
-
-
-def protected(handler):
- @wraps(handler)
- async def wrapper(request, *args, **kwargs):
- header = request.headers.get('authorization', '')
- scheme, _, token = header.partition(' ')
- user = None
- if scheme.lower() == 'bearer' and token:
- user = await validate_access_token(token)
- if user is None:
- return (
- {'error': 'Unauthorized'},
- 'application/json',
- 401,
- [('WWW-Authenticate', 'Bearer')],
- )
- request.user = user
- request.access_token = token
- response = await handler(request, *args, **kwargs)
- if isinstance(response, tuple):
- 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')]
-
- return wrapper
-
-
-def register_routes(app):
+def protected(database):
+ def decorator(handler):
+ @wraps(handler)
+ async def wrapper(request, *args, **kwargs):
+ header = request.headers.get('authorization', '')
+ scheme, _, token = header.partition(' ')
+ user = None
+ if scheme.lower() == 'bearer' and token:
+ digest = sha256(token.encode('utf-8')).hexdigest()
+ user = await database.get_access_token(digest, time())
+ if user is None:
+ return (
+ {'error': 'Unauthorized'},
+ 'application/json',
+ 401,
+ [('WWW-Authenticate', 'Bearer')],
+ )
+ request.user = {'id': user['user_id']}
+ request.login_id = user['login_id']
+ response = await handler(request, *args, **kwargs)
+ if isinstance(response, tuple):
+ 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')]
+
+ return wrapper
+
+ return decorator
+
+
+def register_routes(app, database):
"""Register authentication endpoints on the supplied Wred app."""
@app.route('/api/signup', methods=['POST'])
@@ -142,10 +118,9 @@ def register_routes(app):
username, password = await read_credentials(request)
salt = token_bytes(16)
password_hash = await hash_password(password, salt)
- # Check after the await so simultaneous signups cannot overwrite the same account.
- if username in users:
+ # 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
- users[username] = {'salt': salt, 'password_hash': password_hash}
return {'id': username}, 'application/json', 201
@app.route('/api/login', methods=['POST'])
@@ -159,20 +134,22 @@ def register_routes(app):
)
username, password = await read_credentials(request)
# Use a dummy record for unknown users so failed logins still perform password hashing.
- record = users.get(username, {'salt': bytes(16), 'password_hash': bytes(32)})
- password_hash = await hash_password(password, record['salt'])
- if username not in users or not compare_digest(password_hash, record['password_hash']):
+ 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']):
return {'error': 'Invalid credentials'}, 'application/json', 401
login_id = token_urlsafe(16)
refresh = token_urlsafe(32)
- # Store only its digest and a fixed absolute expiry; refreshes do not extend this deadline.
- refresh_tokens[sha256(refresh.encode()).hexdigest()] = {
- 'user_id': username,
- 'login_id': login_id,
- 'expires_at': monotonic() + refresh_lifetime,
- }
+ access = token_urlsafe(32)
+ 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,
+ )
return (
- issue_access_token(username, login_id),
+ token_response(access, username),
'application/json',
200,
[('Cache-Control', 'no-store'), refresh_cookie(request, refresh, refresh_lifetime)],
@@ -188,33 +165,33 @@ def register_routes(app):
403,
)
digest = sha256(read_refresh_cookie(request).encode()).hexdigest()
- record = refresh_tokens.get(digest)
- if record is None or record['expires_at'] <= monotonic():
- if record is not None:
- revoke_login(record['login_id'])
+ access = token_urlsafe(32)
+ now = time()
+ username = await database.refresh_login(
+ 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)],
)
- # Reuse the cookie while issuing a fresh memory-only access token; concurrent tabs work.
+ # Reuse the cookie while issuing a fresh access token; concurrent tabs work.
return (
- issue_access_token(record['user_id'], record['login_id']),
+ token_response(access, username),
'application/json',
200,
[('Cache-Control', 'no-store')],
)
@app.route('/api/me')
- @protected
+ @protected(database)
async def me(request):
return request.user
@app.route('/api/logout', methods=['POST'])
- @protected
+ @protected(database)
async def logout(request):
- digest = sha256(request.access_token.encode('utf-8')).hexdigest()
- login_id = tokens[digest]['login_id']
- revoke_login(login_id)
+ await database.revoke_login(request.login_id)
return {'logged_out': True}, 'application/json', 200, [refresh_cookie(request, '', 0)]
diff --git a/example_app/notes.py b/example_app/notes.py
index 1f8efaa..66542b5 100644
--- a/example_app/notes.py
+++ b/example_app/notes.py
@@ -5,7 +5,6 @@ from secrets import token_urlsafe
from wred import WredRequestError
from .auth import protected
-from .storage import notes
async def read_note(request, partial=False):
@@ -27,8 +26,8 @@ async def read_note(request, partial=False):
return data if partial else {'body': '', 'pinned': False, **data}
-def owned_note(note_id, user):
- note = notes.get(note_id)
+async def owned_note(database, note_id, user):
+ note = await database.get_note(note_id)
if note is None:
return {'error': 'Note not found'}, 'application/json', 404
if note['owner'] != user['id']:
@@ -36,11 +35,11 @@ def owned_note(note_id, user):
return note
-def register_routes(app):
+def register_routes(app, database):
"""Register note endpoints; each handler requires a valid bearer token."""
@app.route('/api/notes', methods=['GET'])
- @protected
+ @protected(database)
async def list_notes(request):
# Keep this demo's query budget at 20 fields, including repeated keys.
if request.scope.get('query_string', b'').count(b'&') >= 20:
@@ -57,24 +56,10 @@ def register_routes(app):
pinned = query.get('pinned', '')
if pinned not in ('', 'true', 'false'):
raise WredRequestError('Pinned filter must be true or false')
- # Apply ownership before searching or counting so other users' data never leaks.
- items = [note for note in notes.values() if note['owner'] == request.user['id']]
- items = [
- note for note in items if search in (note['title'] + '\n' + note['body']).casefold()
- ]
- if pinned:
- items = [note for note in items if note['pinned'] == (pinned == 'true')]
- # Show pinned notes first, then most recently edited notes within each group.
- items.sort(key=lambda note: (note['pinned'], note['updated_at'], note['id']), reverse=True)
- return {
- 'items': items[offset : offset + limit],
- 'total': len(items),
- 'limit': limit,
- 'offset': offset,
- }
+ return await database.list_notes(request.user['id'], search, pinned, limit, offset)
@app.route('/api/notes', methods=['POST'])
- @protected
+ @protected(database)
async def create_note(request):
data = await read_note(request)
note_id = token_urlsafe(9)
@@ -86,39 +71,42 @@ def register_routes(app):
'created_at': now,
'updated_at': now,
}
- notes[note_id] = note
+ await database.create_note(note)
return note, 'application/json', 201, [('Location', f'/api/notes/{note_id}')]
@app.route('/api/notes/{note_id}', methods=['GET'])
- @protected
+ @protected(database)
async def get_note(request, note_id):
- return owned_note(note_id, request.user)
+ return await owned_note(database, note_id, request.user)
@app.route('/api/notes/{note_id}', methods=['PUT', 'PATCH'])
- @protected
+ @protected(database)
async def update_note(request, note_id):
- note = owned_note(note_id, request.user)
+ note = await owned_note(database, note_id, request.user)
if isinstance(note, tuple):
return note
data = await read_note(request, partial=request.method == 'PATCH')
- note.update(data)
- note['updated_at'] = datetime.now(UTC).isoformat(timespec='microseconds')
- return note
+ 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
+ return updated
@app.route('/api/notes/{note_id}', methods=['DELETE'])
- @protected
+ @protected(database)
async def delete_note(request, note_id):
- note = owned_note(note_id, request.user)
+ note = await owned_note(database, note_id, request.user)
if isinstance(note, tuple):
return note
- del notes[note_id]
+ if not await database.delete_note(note_id, request.user['id']):
+ 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.
@app.route('/api/notes/export', methods=['GET'])
- @protected
+ @protected(database)
async def export_notes(request):
- items = [note for note in notes.values() if note['owner'] == request.user['id']]
+ items = await database.export_notes(request.user['id'])
return (
items,
'application/json',
diff --git a/example_app/static/index.html b/example_app/static/index.html
index 63aebcc..4293b30 100644
--- a/example_app/static/index.html
+++ b/example_app/static/index.html
@@ -100,7 +100,7 @@
<button id="rate-demo" class="quiet" type="button">Send 5 requests</button>
<p id="rate-result" role="status" aria-live="polite"></p>
</details>
- <footer class="app-footer">This is a demo. Your login survives reloads for up to seven days; data and logins reset when the server restarts.</footer>
+ <footer class="app-footer">This is a demo. Your notes are saved. You stay logged in for up to seven days, including after a server restart.</footer>
</div>
</body>
</html>
diff --git a/example_app/storage.py b/example_app/storage.py
index 6f9c8d9..e61dfb4 100644
--- a/example_app/storage.py
+++ b/example_app/storage.py
@@ -1,7 +1,157 @@
-"""Shared in-memory demo data; all records disappear when the process restarts."""
+"""Application-owned MongoDB storage using PyMongo's native async client."""
-# Real deployments need persistent shared storage and shared rate-limit counters.
-users = {} # Username -> salt and password hash; never store plaintext passwords.
-tokens = {} # Token digest -> user ID and expiry; never store raw bearer tokens.
-refresh_tokens = {} # Cookie digest -> user ID, login ID, and absolute expiry.
-notes = {} # Note ID -> note, including the owning username.
+from asyncio import Lock
+from datetime import UTC, datetime
+
+from pymongo import AsyncMongoClient, ReturnDocument
+from pymongo.errors import DuplicateKeyError
+
+
+NOTE_FIELDS = {'_id': 0, '_title_search': 0, '_body_search': 0}
+
+
+def timestamp(seconds):
+ return datetime.fromtimestamp(seconds, UTC)
+
+
+class Database:
+ def __init__(self, uri, name):
+ # One client/pool per app worker and event loop; connection is lazy.
+ self.client = AsyncMongoClient(uri, serverSelectionTimeoutMS=5000)
+ self.db = self.client[name]
+ self._initialized = False
+ self._init_lock = Lock()
+
+ async def initialize(self):
+ if self._initialized:
+ return
+ 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.notes.create_index(
+ [('owner', 1), ('pinned', -1), ('updated_at', -1), ('id', -1)],
+ )
+ self._initialized = True
+
+ async def close(self):
+ await self.client.close()
+
+ async def create_user(self, username, salt, password_hash):
+ await self.initialize()
+ try:
+ await self.db.users.insert_one(
+ {'_id': username, 'salt': salt, 'password_hash': password_hash},
+ )
+ except DuplicateKeyError:
+ return False
+ return True
+
+ async def get_user(self, username):
+ 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):
+ await self.initialize()
+ # Keep a login and its access credentials in one document. Refresh and
+ # logout are atomic without requiring replica-set transactions.
+ await self.db.sessions.insert_one({
+ '_id': login_id,
+ 'user_id': username,
+ 'refresh_digest': refresh_digest,
+ 'expires_at': timestamp(refresh_expires),
+ 'access_tokens': [{'digest': access_digest, 'expires_at': timestamp(access_expires)}],
+ })
+
+ 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)}]},
+ ]}}}],
+ projection={'user_id': 1},
+ return_document=ReturnDocument.AFTER,
+ )
+ return session['user_id'] if session else None
+
+ 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})
+ if session:
+ return {'user_id': session['user_id'], 'login_id': session['_id']}
+
+ async def revoke_login(self, login_id):
+ await self.initialize()
+ await self.db.sessions.delete_one({'_id': login_id})
+
+ async def list_notes(self, owner, search, pinned, limit, offset):
+ 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]}
+ 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}])
+ 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}
+
+ 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(),
+ })
+
+ async def get_note(self, note_id):
+ await self.initialize()
+ return await self.db.notes.find_one({'_id': note_id}, NOTE_FIELDS)
+
+ 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}
+ 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,
+ )
+
+ async def delete_note(self, note_id, owner):
+ await self.initialize()
+ 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)
diff --git a/pyproject.toml b/pyproject.toml
index 7d60272..5988bca 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -13,3 +13,5 @@ license-files = ["LICENSE"]
authors = [{ name = "Harrison Erd", email = "[email protected]" }]
dependencies = ["orjson", "weft-server"]
+[project.optional-dependencies]
+example = ["pymongo>=4.13,<5"]
diff --git a/tests/test_notebook.py b/tests/test_notebook.py
index 93d6c4d..753f6e1 100644
--- a/tests/test_notebook.py
+++ b/tests/test_notebook.py
@@ -1,29 +1,48 @@
"""Exercise the example's API contracts through the actual Wred ASGI app."""
import importlib
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 unittest.mock import patch
+from uuid import uuid4
import orjson
from tests.test_wred import call_app
-from example_app import auth, storage, rate_limit
+from example_app import auth, rate_limit
-notebook = importlib.import_module('example_app.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')
class NotebookTests(unittest.IsolatedAsyncioTestCase):
def setUp(self):
- storage.users.clear()
- storage.tokens.clear()
- storage.refresh_tokens.clear()
+ self.database_name = 'wred_test_' + uuid4().hex
+ self.apps = []
+ self.app = self.new_app()
+ self.addAsyncCleanup(self.clean_database)
rate_limit.rate_buckets.clear()
- storage.notes.clear()
+
+ def new_app(self):
+ app = notebook.create_app(mongodb_uri, self.database_name)
+ self.apps.append(app)
+ return app
+
+ async def clean_database(self):
+ try:
+ await self.app.database.client.drop_database(self.database_name)
+ finally:
+ 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'):
+ cookie=None, client='127.0.0.1', csrf_header=True, extra_headers=(), scheme='http', app=None):
url = urlsplit(path)
async def receive():
@@ -41,7 +60,7 @@ 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(notebook.app, {
+ 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),
@@ -90,6 +109,18 @@ class NotebookTests(unittest.IsolatedAsyncioTestCase):
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'}])
+
+ 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'},
+ ])
+
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)
@@ -108,10 +139,88 @@ class NotebookTests(unittest.IsolatedAsyncioTestCase):
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'}))
- expired_time = auth.monotonic() + 3601
- with patch.object(auth, 'monotonic', return_value=expired_time):
+ 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)
+ 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}'
+ script = """
+import asyncio
+import sys
+import orjson
+from example_app.app 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/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'])])
+ 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,
+ )
+ 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.
+ renewed = refreshed[1]['access_token']
+ 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)
+
+ async def test_workers_resolve_duplicate_signups_and_preserve_concurrent_patches(self):
+ worker = self.new_app()
+ 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)
+ 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.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'))
+ for search in ('New+title', 'New+body'):
+ _, 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},
@@ -171,7 +280,7 @@ class NotebookTests(unittest.IsolatedAsyncioTestCase):
async def test_filtering_pagination_and_invalid_queries(self):
token = await self.account()
- first = await self.create(token, title='Coffee idea', body='Learn C++', pinned=True)
+ 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)
@@ -180,6 +289,11 @@ class NotebookTests(unittest.IsolatedAsyncioTestCase):
self.assertEqual(page['items'], [second])
_, 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)
+ self.assertEqual(page['items'], [first])
+ _, 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',
@@ -227,7 +341,15 @@ class NotebookTests(unittest.IsolatedAsyncioTestCase):
(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)
- _, _, headers = await self.request('/api/logout', 'POST', token=restored['access_token'])
+ refreshed, logged_out = await asyncio.gather(
+ self.request('/api/refresh', 'POST', cookie=cookie),
+ 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)
+ _, _, headers = logged_out
cleared = SimpleCookie(headers[b'set-cookie'].decode())[auth.refresh_cookie_name]
self.assertEqual(cleared['max-age'], '0')
self.assertEqual(cleared['path'], '/api')
@@ -262,11 +384,11 @@ class NotebookTests(unittest.IsolatedAsyncioTestCase):
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.monotonic()
- with patch.object(auth, 'monotonic', return_value=now + 3601):
+ 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)
- with patch.object(auth, 'monotonic', return_value=now + 7 * 24 * 3600 + 1):
+ with patch.object(auth, 'time', return_value=now + 7 * 24 * 3600 + 1):
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')