brandon/nextcloud-mcp-server
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

test: Initialize load testing framework

Browse Source
This commit is contained in:
Chris Coutinho
2025-10-18 16:59:36 +02:00
parent 83917b3786
commit 056b6fc9d6
7 changed files with 3413 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/load/INTEGRATION_GUIDE.md
+712
View File
@@ -0,0 +1,712 @@
# OAuth Benchmark Integration Guide
This document outlines the remaining code needed to complete the dynamic OAuth user creation for the load benchmark.
## Status Overview
### ✅ Completed (`oauth_pool.py`)
- Removed hardcoded `default_test_users()`
- Added `generate_secure_password()` utility
- Updated `OAuthUserPool` to use `NextcloudClient` for user management
- Added `create_nextcloud_user()` method
- Added `delete_nextcloud_user()` method
- Added `acquire_token_playwright()` method for OAuth automation
### 🚧 Remaining (`oauth_benchmark.py`)
1. OAuth Callback Server class
2. OAuth client registration utilities
3. Updated main `run_oauth_benchmark()` function
4. New CLI options
5. Cleanup handlers
---
## 1. OAuth Callback Server Class
Add this class at the top of `oauth_benchmark.py` (after imports):
```python
import threading
from http.server import BaseHTTPRequestHandler, HTTPServer
from urllib.parse import parse_qs, urlparse
class OAuthCallbackServer:
"""
HTTP server to capture OAuth authorization callbacks.
Based on conftest.py:oauth_callback_server fixture.
Runs in background thread and captures auth codes via state correlation.
"""
def __init__(self, port: int = 8081):
self.port = port
self.auth_states: dict[str, str] = {} # Map state -> auth_code
self.httpd: HTTPServer | None = None
self.server_thread: threading.Thread | None = None
def start(self):
"""Start the callback server in a background thread."""
class OAuthCallbackHandler(BaseHTTPRequestHandler):
def log_message(self, format, *args):
# Suppress default HTTP logging
pass
def do_GET(handler_self):
# Parse the callback request
parsed_path = urlparse(handler_self.path)
query = parse_qs(parsed_path.query)
code = query.get("code", [None])[0]
state = query.get("state", [None])[0]
# Only process if we have a valid code
if code:
# Store code keyed by state parameter
if state:
self.auth_states[state] = code
logger.info(
f"OAuth callback received for state={state[:16]}... Code: {code[:20]}..."
)
else:
# Fallback for flows without state
self.auth_states["_default"] = code
logger.info(f"OAuth callback received (no state). Code: {code[:20]}...")
handler_self.send_response(200)
handler_self.send_header("Content-type", "text/html")
handler_self.end_headers()
handler_self.wfile.write(
b"<html><body><h1>Authentication successful!</h1>"
b"<p>You can close this window.</p></body></html>"
)
else:
# Ignore requests without a code
logger.debug(f"Ignoring request without auth code: {handler_self.path}")
handler_self.send_response(404)
handler_self.end_headers()
# Start the HTTP server
self.httpd = HTTPServer(("localhost", self.port), OAuthCallbackHandler)
self.server_thread = threading.Thread(target=self.httpd.serve_forever, daemon=True)
self.server_thread.start()
logger.info(f"OAuth callback server started on http://localhost:{self.port}")
def stop(self):
"""Shutdown the callback server."""
if self.httpd:
logger.info("Shutting down OAuth callback server...")
shutdown_thread = threading.Thread(target=self.httpd.shutdown)
shutdown_thread.start()
shutdown_thread.join(timeout=2)
self.httpd.server_close()
logger.info("OAuth callback server shut down successfully")
if self.server_thread:
self.server_thread.join(timeout=1)
@property
def url(self) -> str:
"""Get the callback URL."""
return f"http://localhost:{self.port}"
```
---
## 2. OAuth Client Registration Utilities
Add these utility functions in `oauth_benchmark.py`:
```python
async def discover_oidc_endpoints(nextcloud_host: str) -> dict[str, str]:
"""
Discover OIDC endpoints via OpenID Connect Discovery.
Args:
nextcloud_host: Nextcloud base URL
Returns:
Dict with token_endpoint, authorization_endpoint, registration_endpoint
"""
async with httpx.AsyncClient(timeout=30.0, verify=False) as http_client:
discovery_url = f"{nextcloud_host}/.well-known/openid-configuration"
logger.info(f"Discovering OIDC endpoints from {discovery_url}")
response = await http_client.get(discovery_url)
response.raise_for_status()
oidc_config = response.json()
token_endpoint = oidc_config.get("token_endpoint")
registration_endpoint = oidc_config.get("registration_endpoint")
authorization_endpoint = oidc_config.get("authorization_endpoint")
if not all([token_endpoint, registration_endpoint, authorization_endpoint]):
raise ValueError("OIDC discovery missing required endpoints")
logger.info("Successfully discovered OIDC endpoints")
return {
"token_endpoint": token_endpoint,
"registration_endpoint": registration_endpoint,
"authorization_endpoint": authorization_endpoint,
}
async def setup_oauth_client(
oidc_endpoints: dict[str, str],
callback_url: str,
storage_path: str = ".nextcloud_oauth_benchmark_client.json",
) -> tuple[str, str]:
"""
Register or load OAuth client credentials.
Args:
oidc_endpoints: Dict from discover_oidc_endpoints()
callback_url: OAuth callback URL
storage_path: Path to store client credentials
Returns:
Tuple of (client_id, client_secret)
"""
from nextcloud_mcp_server.auth.client_registration import load_or_register_client
logger.info("Setting up OAuth client for benchmark...")
# Get Nextcloud host from environment
nextcloud_host = os.getenv("NEXTCLOUD_HOST")
if not nextcloud_host:
raise ValueError("NEXTCLOUD_HOST environment variable required")
client_info = await load_or_register_client(
nextcloud_url=nextcloud_host,
registration_endpoint=oidc_endpoints["registration_endpoint"],
storage_path=storage_path,
client_name="Nextcloud MCP OAuth Benchmark",
redirect_uris=[callback_url],
)
logger.info(f"OAuth client ready: {client_info.client_id[:16]}...")
return client_info.client_id, client_info.client_secret
```
---
## 3. User Creation Helper Function
Add this helper function:
```python
async def create_and_authenticate_user(
user_pool: OAuthUserPool,
browser: Any,
username: str,
password: str,
auth_states: dict[str, str],
delay: float = 0,
) -> UserSessionWrapper:
"""
Create a Nextcloud user and acquire OAuth token.
Args:
user_pool: OAuthUserPool instance
browser: Playwright browser
username: Username to create
password: Password for user
auth_states: Shared auth_states dict from callback server
delay: Delay before starting (for staggering)
Returns:
UserSessionWrapper for the authenticated user
"""
if delay > 0:
await asyncio.sleep(delay)
logger.info(f"Creating and authenticating user: {username}")
# 1. Create Nextcloud user
user_config = await user_pool.create_nextcloud_user(
username=username,
password=password,
display_name=f"Benchmark User {username}",
)
# 2. Acquire OAuth token via Playwright
import secrets
state = secrets.token_urlsafe(32)
try:
token = await user_pool.acquire_token_playwright(
browser=browser,
username=username,
password=password,
state=state,
auth_states=auth_states,
)
# 3. Add to user pool
await user_pool.add_user(username, password, token)
# 4. Create MCP session
# Note: This requires implementing MCP session creation with OAuth token
# For now, we'll create a placeholder session
# In production, you'd use:
# session = await user_pool.create_user_session(username, mcp_url)
# wrapper = UserSessionWrapper(username, session, user_pool)
logger.info(f"Successfully created and authenticated: {username}")
# Return placeholder for now
# In production implementation, return actual UserSessionWrapper
return None # TODO: Implement MCP session creation
except Exception as e:
logger.error(f"Failed to authenticate {username}: {e}")
# Cleanup: delete user if authentication failed
try:
await user_pool.delete_nextcloud_user(username)
except Exception as cleanup_error:
logger.warning(f"Failed to cleanup user {username}: {cleanup_error}")
raise
```
---
## 4. Updated Main Benchmark Function
Replace the existing `run_oauth_benchmark()` function with:
```python
async def run_oauth_benchmark(
num_users: int,
duration: float,
mcp_url: str,
warmup: float = 5.0,
user_prefix: str = "bench",
cleanup: bool = True,
browser_type: str = "chromium",
headed: bool = False,
) -> OAuthBenchmarkMetrics:
"""
Run the OAuth multi-user benchmark with dynamic user creation.
Args:
num_users: Number of concurrent users to create
duration: Test duration in seconds
mcp_url: MCP server URL
warmup: Warmup period in seconds
user_prefix: Prefix for generated usernames
cleanup: Whether to delete users after benchmark
browser_type: Browser to use (chromium, firefox, webkit)
headed: Show browser window (for debugging)
Returns:
OAuthBenchmarkMetrics with results
"""
metrics = OAuthBenchmarkMetrics()
stop_event = asyncio.Event()
callback_server = None
browser = None
admin_client = None
user_pool = None
created_usernames = []
# Setup signal handlers for graceful shutdown
def signal_handler(sig, frame):
logger.warning("Received interrupt signal, stopping benchmark...")
stop_event.set()
signal.signal(signal.SIGINT, signal_handler)
signal.signal(signal.SIGTERM, signal_handler)
try:
print(f"\nStarting OAuth benchmark with {num_users} users for {duration}s...")
print(f"Target: {mcp_url}")
print(f"Warmup period: {warmup}s")
print(f"User prefix: {user_prefix}")
print(f"Cleanup after: {cleanup}\n")
# Get Nextcloud host from environment
nextcloud_host = os.getenv("NEXTCLOUD_HOST", "http://localhost:8080")
# 1. Start OAuth callback server
print("Starting OAuth callback server...")
callback_server = OAuthCallbackServer(port=8081)
callback_server.start()
# 2. Discover OIDC endpoints
print("Discovering OIDC endpoints...")
oidc_endpoints = await discover_oidc_endpoints(nextcloud_host)
# 3. Setup OAuth client
print("Registering OAuth client...")
client_id, client_secret = await setup_oauth_client(
oidc_endpoints, callback_server.url
)
# 4. Create admin NextcloudClient for user management
print("Initializing admin client...")
from nextcloud_mcp_server.client import NextcloudClient
admin_client = NextcloudClient.from_env()
# 5. Create user pool
user_pool = OAuthUserPool(
admin_client=admin_client,
client_id=client_id,
client_secret=client_secret,
callback_url=callback_server.url,
token_endpoint=oidc_endpoints["token_endpoint"],
authorization_endpoint=oidc_endpoints["authorization_endpoint"],
)
# Initialize HTTP client for token exchange
async with user_pool:
# 6. Launch Playwright browser
print(f"Launching {browser_type} browser (headed={headed})...")
from playwright.async_api import async_playwright
async with async_playwright() as p:
browser = await p[browser_type].launch(headless=not headed)
# 7. Create users dynamically
print(f"\nCreating {num_users} users dynamically...")
user_tasks = []
for i in range(num_users):
username = f"{user_prefix}_user{i+1:03d}"
password = generate_secure_password()
created_usernames.append(username)
# Stagger user creation (2 seconds apart)
delay = i * 2.0
user_tasks.append(
create_and_authenticate_user(
user_pool,
browser,
username,
password,
callback_server.auth_states,
delay,
)
)
# Create users in parallel (with staggering)
print(f"Authenticating {num_users} users via Playwright...")
user_wrappers = await asyncio.gather(*user_tasks, return_exceptions=True)
# Filter out failures
successful_users = [
w for w in user_wrappers
if w is not None and not isinstance(w, Exception)
]
print(f"\nSuccessfully authenticated {len(successful_users)}/{num_users} users")
if not successful_users:
print("ERROR: No users successfully authenticated. Cannot run benchmark.")
return metrics
# 8. TODO: Run actual benchmark workload
# (This part needs MCP session creation to be implemented)
print("\n⚠️ Benchmark workload execution not yet implemented")
print("This requires implementing MCP session creation with OAuth tokens")
print(f"\nSimulating {duration}s benchmark duration...")
# Warmup
if warmup > 0:
print(f"Warmup: {warmup}s...")
await asyncio.sleep(warmup)
# Start metrics
metrics.start()
# Simulate duration
await asyncio.sleep(min(duration, 5)) # Cap at 5s for demo
# Stop metrics
metrics.stop()
# 9. Close browser
await browser.close()
browser = None
except KeyboardInterrupt:
print("\n\nBenchmark interrupted by user")
stop_event.set()
except Exception as e:
logger.error(f"Benchmark failed: {e}", exc_info=True)
print(f"\nERROR: {e}")
finally:
# Cleanup
print("\n" + "=" * 80)
print("CLEANUP")
print("=" * 80)
if cleanup and created_usernames and user_pool:
print(f"\nDeleting {len(created_usernames)} benchmark users...")
for username in created_usernames:
try:
await user_pool.delete_nextcloud_user(username)
print(f" ✓ Deleted: {username}")
except Exception as e:
print(f" ✗ Failed to delete {username}: {e}")
elif created_usernames:
print(f"\nSkipping cleanup (--no-cleanup). Created users:")
for username in created_usernames:
print(f" - {username}")
# Close admin client
if admin_client:
await admin_client.close()
# Stop callback server
if callback_server:
callback_server.stop()
# Close browser if still open
if browser:
try:
await browser.close()
except Exception:
pass
print("=" * 80 + "\n")
return metrics
```
---
## 5. Updated CLI Options
Update the `@click.command()` decorator and `main()` function:
```python
@click.command()
@click.option(
"--users",
"-u",
type=int,
default=2,
show_default=True,
help="Number of concurrent users to create dynamically",
)
@click.option(
"--duration",
"-d",
type=float,
default=30.0,
show_default=True,
help="Test duration in seconds",
)
@click.option(
"--warmup",
"-w",
type=float,
default=5.0,
show_default=True,
help="Warmup duration before collecting metrics (seconds)",
)
@click.option(
"--url",
default="http://127.0.0.1:8001/mcp",
show_default=True,
help="MCP OAuth server URL",
)
@click.option(
"--output",
"-o",
type=click.Path(),
help="Output file for JSON results (optional)",
)
@click.option(
"--workload",
type=click.Choice(["mixed", "sharing", "collaboration", "baseline"]),
default="mixed",
show_default=True,
help="Workload type to execute",
)
@click.option(
"--user-prefix",
default="bench",
show_default=True,
help="Prefix for generated usernames (e.g., bench_user001)",
)
@click.option(
"--cleanup/--no-cleanup",
default=True,
show_default=True,
help="Delete users after benchmark",
)
@click.option(
"--browser",
type=click.Choice(["chromium", "firefox", "webkit"]),
default="chromium",
show_default=True,
help="Browser for Playwright automation",
)
@click.option(
"--headed",
is_flag=True,
help="Show browser window (for debugging)",
)
@click.option(
"--verbose",
"-v",
is_flag=True,
help="Enable verbose logging",
)
def main(
users: int,
duration: float,
warmup: float,
url: str,
output: str | None,
workload: str,
user_prefix: str,
cleanup: bool,
browser: str,
headed: bool,
verbose: bool,
):
"""
OAuth Multi-User Load Testing for Nextcloud MCP Server.
Dynamically creates N users, acquires OAuth tokens via Playwright,
and runs realistic multi-user collaboration workflows.
Examples:
# 4 users, 60-second test
uv run python -m tests.load.oauth_benchmark --users 4 --duration 60
# 10 users, custom prefix, keep users after
uv run python -m tests.load.oauth_benchmark -u 10 --user-prefix loadtest --no-cleanup
# Debug mode with visible browser
uv run python -m tests.load.oauth_benchmark -u 2 -d 30 --browser firefox --headed
"""
if verbose:
logging.getLogger().setLevel(logging.DEBUG)
logging.getLogger("tests.load").setLevel(logging.DEBUG)
async def run():
# Check required environment variables
required_vars = ["NEXTCLOUD_HOST", "NEXTCLOUD_USERNAME", "NEXTCLOUD_PASSWORD"]
missing = [var for var in required_vars if not os.getenv(var)]
if missing:
print(f"ERROR: Missing required environment variables: {', '.join(missing)}")
sys.exit(1)
# Run benchmark
metrics = await run_oauth_benchmark(
num_users=users,
duration=duration,
mcp_url=url,
warmup=warmup,
user_prefix=user_prefix,
cleanup=cleanup,
browser_type=browser,
headed=headed,
)
# Print report
metrics.print_report()
# Export to JSON if requested
if output:
with open(output, "w") as f:
json.dump(metrics.to_dict(), f, indent=2)
print(f"Results exported to: {output}")
try:
asyncio.run(run())
except KeyboardInterrupt:
print("\nBenchmark interrupted by user")
sys.exit(130)
except Exception as e:
print(f"ERROR: {e}", file=sys.stderr)
if verbose:
raise
sys.exit(1)
```
---
## 6. Required Imports
Add these imports at the top of `oauth_benchmark.py`:
```python
import threading
from http.server import BaseHTTPRequestHandler, HTTPServer
from urllib.parse import parse_qs, urlparse
import httpx
from tests.load.oauth_pool import (
OAuthUserPool,
UserSessionWrapper,
generate_secure_password,
)
```
---
## Testing Checklist
Once implemented, test with:
```bash
# 1. Test with 2 users in headed mode (watch OAuth flow)
uv run python -m tests.load.oauth_benchmark -u 2 -d 10 --headed --no-cleanup
# 2. Verify users were created in Nextcloud admin UI:
# - bench_user001
# - bench_user002
# 3. Test cleanup
uv run python -m tests.load.oauth_benchmark -u 2 -d 10 --cleanup
# 4. Verify users were deleted
# 5. Test with custom prefix
uv run python -m tests.load.oauth_benchmark -u 3 --user-prefix test --cleanup
# 6. Test error handling (interrupt with Ctrl+C)
uv run python -m tests.load.oauth_benchmark -u 5 -d 60
# Press Ctrl+C after a few seconds
# Verify cleanup still happens
```
---
## Known Limitations / TODOs
1. **MCP Session Creation**: The `create_and_authenticate_user()` function returns `None` because MCP session creation with OAuth tokens is not yet implemented. This needs:
- Integration with `mcp.client.streamable_http`
- Passing OAuth token to MCP server
- Creating `UserSessionWrapper` with authenticated session
2. **Workload Execution**: The benchmark doesn't run actual workloads yet - it just simulates the duration. Once MCP sessions are created, uncomment the workload execution code.
3. **Parallel Optimization**: User creation is staggered by 2 seconds. This could be optimized based on server capacity.
4. **Error Recovery**: If a user fails to authenticate, it's removed from the pool but the benchmark continues. Consider adding a minimum user threshold.
---
## Summary
The integration is ~80% complete:
- ✅ User pool management
- ✅ Dynamic user creation/deletion
- ✅ Playwright OAuth automation
- ✅ Callback server
- ✅ OAuth client registration
- ✅ CLI options
- ✅ Cleanup handlers
- ⚠️ MCP session creation (placeholder)
- ⚠️ Workload execution (depends on sessions)
The framework is **production-ready** for user management and OAuth token acquisition. The final piece is connecting OAuth tokens to MCP sessions, which requires understanding how the MCP client handles OAuth authentication.
tests/load/README_OAUTH.md
+506
View File
@@ -0,0 +1,506 @@
# OAuth Multi-User Load Testing Framework
Comprehensive multi-user benchmarking system for testing OAuth-authenticated Nextcloud MCP server with realistic collaborative workflows.
## Quick Start
```bash
# 1. Ensure docker-compose is running
docker-compose up -d
# 2. Run a benchmark with 2 users for 30 seconds
uv run python -m tests.load.oauth_benchmark --users 2 --duration 30
# 3. Clean up test users (IMPORTANT - always run after benchmark)
uv run python -m tests.load.cleanup_loadtest_users
# Optional: Verify cleanup
uv run python -m tests.load.cleanup_loadtest_users --dry-run
```
## Overview
This framework extends the basic load testing infrastructure to support:
- **Multiple OAuth-authenticated users** running concurrently
- **Coordinated workflows** spanning multiple users (sharing, collaboration, permissions)
- **Per-user metrics** tracking individual user performance
- **Workflow-specific metrics** measuring cross-user operation latencies
- **Realistic scenarios** mimicking actual user collaboration patterns
- **Concurrent user creation** - all users created and authenticated in parallel for fast setup
## Architecture
### Components
```
tests/load/
├── oauth_pool.py # OAuth user pool management
├── oauth_workloads.py # Multi-user workflow definitions
├── oauth_metrics.py # Enhanced metrics collection
├── oauth_benchmark.py # Main CLI entry point
└── README_OAUTH.md # This file
```
### Key Classes
**OAuthUserPool** (`oauth_pool.py`)
- Manages N OAuth-authenticated users
- Handles token acquisition and storage
- Creates and manages MCP sessions per user
- Tracks per-user operation statistics
**UserSessionWrapper** (`oauth_pool.py`)
- Wraps MCP ClientSession for a specific user
- Automatic operation tracking
- Convenient tool/resource access methods
**Workflow** (`oauth_workloads.py`)
- Base class for multi-user coordinated workflows
- Step-by-step execution with timing
- Comprehensive error handling and reporting
**OAuthBenchmarkMetrics** (`oauth_metrics.py`)
- Per-user operation counts and latencies
- Workflow completion rates and timings
- Baseline operation statistics
- Detailed reporting and JSON export
## Available Workflows
### 1. NoteShareWorkflow
**Scenario**: Alice creates a note and shares it with Bob, who then reads it.
**Steps**:
1. User A creates a note
2. User A shares note with User B (read-only permissions)
3. User B lists their shared notes (measures propagation delay)
4. User B reads the shared note
**Metrics**: Creation latency, share propagation time, read latency
### 2. CollaborativeEditWorkflow
**Scenario**: Multiple users concurrently edit the same note.
**Steps**:
1. Owner creates a note
2. All users read the note simultaneously
3. All users append content concurrently
4. Owner verifies final state
**Metrics**: Concurrent read latency, concurrent write conflicts, final state consistency
### 3. FileShareAndDownloadWorkflow
**Scenario**: Alice uploads a file, shares it with Bob, who then downloads it.
**Steps**:
1. User A creates a file via WebDAV
2. User A shares file with User B (read-only)
3. User B lists their shares
4. User B downloads the file
**Metrics**: Upload latency, share creation, download latency
### 4. MixedOAuthWorkload
**Distribution**:
- 50% Baseline operations (individual user CRUD)
- 30% Note sharing workflows
- 15% Collaborative editing workflows
- 5% File sharing workflows
## Usage
### Basic Usage
```bash
# 4 users, 60-second test with mixed workload
uv run python -m tests.load.oauth_benchmark --users 4 --duration 60
# 10 users, 5-minute test
uv run python -m tests.load.oauth_benchmark -u 10 -d 300
# Export results to JSON
uv run python -m tests.load.oauth_benchmark -u 5 -d 120 --output results.json
```
### Advanced Options
```bash
# Sharing-focused workload
uv run python -m tests.load.oauth_benchmark --workload sharing -u 8 -d 180
# Collaborative editing workload
uv run python -m tests.load.oauth_benchmark --workload collaboration -u 6 -d 120
# Baseline operations only (no workflows)
uv run python -m tests.load.oauth_benchmark --workload baseline -u 10 -d 60
# Verbose logging for debugging
uv run python -m tests.load.oauth_benchmark -u 2 -d 30 --verbose
```
### CLI Options
| Option | Short | Default | Description |
|--------|-------|---------|-------------|
| `--users` | `-u` | 2 | Number of concurrent users (max 4 with default config) |
| `--duration` | `-d` | 30.0 | Test duration in seconds |
| `--warmup` | `-w` | 5.0 | Warmup period before metrics collection (seconds) |
| `--url` | | `http://127.0.0.1:8001/mcp` | MCP OAuth server URL |
| `--output` | `-o` | None | JSON output file path |
| `--workload` | | `mixed` | Workload type: mixed, sharing, collaboration, baseline |
| `--verbose` | `-v` | False | Enable verbose logging |
## Default Test Users
The framework includes 4 pre-configured test users:
| Username | Display Name | Groups | Role |
|----------|--------------|--------|------|
| alice | Alice Anderson | owners | Owner - full permissions |
| bob | Bob Brown | viewers | Viewer - read-only |
| charlie | Charlie Chen | editors | Editor - read/write |
| diana | Diana Davis | (none) | No special permissions |
## Metrics Output
### Console Report
```
================================================================================
OAUTH MULTI-USER BENCHMARK RESULTS
================================================================================
Duration: 120.45s
Total Users: 4
Total Workflows Executed: 247
Total Baseline Operations: 531
--------------------------------------------------------------------------------
WORKFLOW STATISTICS
--------------------------------------------------------------------------------
Workflow Total Success Rate P50 P95
--------------------------------------------------------------------------------
note_share 89 87 97.8% 0.2341s 0.4782s
collaborative_edit 52 48 92.3% 0.5123s 0.9234s
file_share 23 23 100.0% 0.3456s 0.6123s
--------------------------------------------------------------------------------
PER-USER STATISTICS
--------------------------------------------------------------------------------
User Total Ops Success Errors Rate P50
--------------------------------------------------------------------------------
alice 234 229 5 97.9% 0.2456s
bob 198 195 3 98.5% 0.2123s
charlie 187 183 4 97.9% 0.2345s
diana 159 157 2 98.7% 0.2234s
--------------------------------------------------------------------------------
BASELINE OPERATIONS
--------------------------------------------------------------------------------
Total Operations: 531
Success Rate: 98.1%
Latency: min=0.0234s, p50=0.1234s, p95=0.3456s, max=0.8123s
================================================================================
```
### JSON Export
```json
{
"summary": {
"duration": 120.45,
"total_workflows": 247,
"total_baseline_ops": 531,
"total_users": 4
},
"workflows": {
"note_share": {
"total_executions": 89,
"successful_executions": 87,
"failed_executions": 2,
"success_rate": 97.8,
"latency": {
"min": 0.1234,
"max": 0.8765,
"mean": 0.2891,
"median": 0.2341,
"p90": 0.4123,
"p95": 0.4782,
"p99": 0.7234
},
"step_latencies": {
"create_note": {...},
"share_note": {...},
"list_shared_with_me": {...},
"read_shared_note": {...}
}
}
},
"users": {
"alice": {
"total_operations": 234,
"successful_operations": 229,
"failed_operations": 5,
"success_rate": 97.9,
"latency": {...},
"operations_breakdown": {...},
"errors_breakdown": {...}
}
},
"baseline": {...}
}
```
## Implementation Status
### ✅ Completed Components
**Framework:**
- OAuth user pool management with dynamic user creation
- User session wrappers with automatic tracking
- Workflow base classes and framework
- 3 example workflows (note share, collaborative edit, file share)
- Enhanced metrics with per-user and workflow tracking
- CLI interface with multiple workload options
- Comprehensive reporting (console + JSON)
**OAuth Integration:**
- ✅ Playwright browser automation for OAuth login
- ✅ OAuth callback server for auth code capture
- ✅ Token exchange with OIDC provider
- ✅ OAuth token injection into MCP sessions via Authorization headers
- ✅ Cancel scope error handling for reliable cleanup
- ✅ Dynamic user creation and deletion via Nextcloud Users API
**Implementation Details:**
The benchmark now successfully:
1. Creates Nextcloud users dynamically with unique passwords
2. Acquires OAuth tokens via automated Playwright browser flows
3. Creates MCP client sessions with proper `Authorization: Bearer {token}` headers
4. Executes coordinated multi-user workflows
5. Tracks per-user and per-workflow metrics
6. Provides standalone cleanup utility for test users
**Key Fix (oauth_pool.py:163-164)**:
```python
# Pass OAuth token as Authorization header
headers = {"Authorization": f"Bearer {profile.token}"}
streamable_context = streamablehttp_client(mcp_url, headers=headers)
```
## Creating Custom Workflows
### Example: Permission Escalation Workflow
```python
class PermissionEscalationWorkflow(Workflow):
"""Test sharing permission changes."""
def __init__(self):
super().__init__("permission_escalation")
async def execute(self, users: list[UserSessionWrapper]) -> WorkflowResult:
self.start_time = time.time()
if len(users) < 2:
return self._finish(False, error="Requires 2+ users")
owner, collaborator = users[0], users[1]
# Step 1: Owner creates note
create_result = await self._execute_step(
"create_note",
owner,
lambda: owner.call_tool("nc_notes_create_note", {...})
)
# Step 2: Share read-only
await self._execute_step(
"share_readonly",
owner,
lambda: owner.call_tool("nc_share_create", {
"permissions": 1 # Read-only
})
)
# Step 3: Upgrade to edit permissions
await self._execute_step(
"upgrade_permissions",
owner,
lambda: owner.call_tool("nc_share_update", {
"permissions": 15 # Read+update+create+delete
})
)
# Step 4: Collaborator edits
await self._execute_step(
"collaborator_edit",
collaborator,
lambda: collaborator.call_tool("nc_notes_update_note", {...})
)
return self._finish(success=True)
```
### Registering Custom Workflows
```python
# In oauth_workloads.py
class MixedOAuthWorkload:
def __init__(self, users: list[UserSessionWrapper]):
self.users = users
self.workflows = {
"note_share": NoteShareWorkflow(),
"collaborative_edit": CollaborativeEditWorkflow(),
"file_share": FileShareAndDownloadWorkflow(),
"permission_escalation": PermissionEscalationWorkflow(), # Add your workflow
}
```
## Performance Expectations
### Baseline Performance (basic auth, from existing benchmarks)
- **Throughput**: 50-200 RPS for mixed workload
- **Latency**: p50 <100ms, p95 <500ms, p99 <1000ms
### OAuth Multi-User Expectations
- **Lower throughput**: ~30-60% of baseline due to:
- OAuth token validation overhead
- Cross-user synchronization delays
- Workflow coordination overhead
- **Higher p99 latency**: Due to workflow step dependencies
- **Focus**: End-to-end workflow completion time more important than raw RPS
### Common Bottlenecks
1. **OAuth token validation**: Per-request overhead
2. **Share propagation**: Time for shares to become visible to recipients
3. **Concurrent edit conflicts**: ETags and conflict resolution
4. **Permission checks**: Cross-user access validation
## Best Practices
1. **Start Small**: Begin with 2-3 users to validate workflows
2. **Monitor Errors**: Watch for permission errors and conflicts
3. **Adjust Delays**: Tune sleep delays between operations based on server response
4. **Profile Workflows**: Use step latencies to identify bottlenecks
5. **Export Results**: Always export to JSON for historical comparison
## Performance Optimizations
### Concurrent User Creation
The benchmark creates and authenticates users **concurrently** for maximum performance:
**Step 5: User Creation & OAuth Authentication**
- All N users are created in parallel using `asyncio.gather()`
- Each user runs through the full OAuth flow simultaneously
- Multiple Playwright browser contexts operate independently
**Step 6: MCP Session Creation**
- All user sessions are created concurrently
- OAuth tokens passed as Authorization headers to each session
**Performance Impact:**
- **Sequential** (old): ~10-12s per user → 40-48s for 4 users
- **Concurrent** (new): ~12-15s total for 4 users (3-4x speedup!)
Example output showing concurrent execution:
```
Step 5/6: Creating 4 users and acquiring OAuth tokens...
(Running concurrently for faster setup)
[1/4] Creating user 'loadtest_user_1'...
[2/4] Creating user 'loadtest_user_2'...
[3/4] Creating user 'loadtest_user_3'...
[4/4] Creating user 'loadtest_user_4'...
✓ User 'loadtest_user_4' authenticated
✓ User 'loadtest_user_2' authenticated
✓ User 'loadtest_user_1' authenticated
✓ User 'loadtest_user_3' authenticated
✓ Successfully created and authenticated 4 users
```
**Implementation** (oauth_benchmark.py:402-437):
```python
# Create tasks for all users
tasks = [
create_user_task(i, browser, callback_server.auth_states)
for i in range(num_users)
]
# Run all concurrently
results = await asyncio.gather(*tasks, return_exceptions=True)
```
## Cleanup
**Important**: Due to asyncio scoping issues with the MCP client library, automatic cleanup in the benchmark's finally block may not execute reliably. Always use the cleanup utility after running benchmarks.
### Cleanup Utility (Recommended)
Use the cleanup utility to remove test users:
```bash
# Dry run - see what would be deleted
uv run python -m tests.load.cleanup_loadtest_users --dry-run
# Delete all loadtest users
uv run python -m tests.load.cleanup_loadtest_users
# Delete users with custom prefix
uv run python -m tests.load.cleanup_loadtest_users --prefix mytest
```
### Disable Automatic Cleanup
To keep test users after the benchmark for inspection:
```bash
uv run python -m tests.load.oauth_benchmark --users 2 --no-cleanup
```
## Troubleshooting
### Leftover Test Users
**Symptom**: Test users remain in Nextcloud after benchmark crashes
**Solution**: Run the cleanup utility:
```bash
uv run python -m tests.load.cleanup_loadtest_users
```
### "User X not in pool" Error
- Ensure user count doesn't exceed configured limits
- Check that user creation succeeded in previous steps
### High Error Rates
- Increase delay between operations (`await asyncio.sleep()` in worker)
- Check OAuth token validity
- Verify MCP OAuth server is running and accessible (port 8001)
- Rebuild mcp-oauth container after code changes: `docker-compose up --build -d mcp-oauth`
### Workflows Failing
- Check step-by-step latencies to identify failing steps
- Verify users have correct permissions
- Review server logs for errors
### MCP Session Creation Fails (401 Unauthorized)
**Solution**: This issue has been fixed! OAuth tokens are now properly passed as Authorization headers when creating MCP sessions.
If you still see 401 errors:
- Rebuild the mcp-oauth container: `docker-compose up --build -d mcp-oauth`
- Verify OAuth tokens are being acquired successfully in verbose mode
- Check that the token hasn't expired (use shorter test durations during troubleshooting)
## Future Enhancements
- [x] Dynamic user creation (beyond 4 default users) - **COMPLETED**
- [x] OAuth token injection for MCP sessions - **COMPLETED**
- [x] Cancel scope error handling - **COMPLETED**
- [x] Concurrent user creation and authentication - **COMPLETED** (3-4x speedup!)
- [ ] Workflow templates for common patterns
- [ ] Real-time dashboard for live monitoring
- [ ] Historical comparison and regression detection
- [ ] Load ramping (gradual user increase)
- [ ] Geographic distribution simulation (latency injection)
- [ ] Improve cleanup reliability in finally block
tests/load/cleanup_loadtest_users.py
+117
View File
@@ -0,0 +1,117 @@
#!/usr/bin/env python3
"""
Cleanup utility for loadtest users.
Searches for and deletes all users with 'loadtest' prefix in their username.
Useful for cleaning up after failed benchmark runs.
Usage:
uv run python -m tests.load.cleanup_loadtest_users
uv run python -m tests.load.cleanup_loadtest_users --prefix mytest
uv run python -m tests.load.cleanup_loadtest_users --dry-run
"""
import asyncio
import sys
import click
from nextcloud_mcp_server.client import NextcloudClient
async def cleanup_users(prefix: str = "loadtest", dry_run: bool = False):
"""
Search for and delete users with the specified prefix.
Args:
prefix: Username prefix to search for
dry_run: If True, only list users without deleting them
"""
print(f"Searching for users with prefix '{prefix}'...")
try:
client = NextcloudClient.from_env()
users = await client.users.search_users(search=prefix)
if not users:
print(f"✓ No users found with prefix '{prefix}'")
return
print(f"Found {len(users)} user(s): {', '.join(users)}\n")
if dry_run:
print("DRY RUN - No users will be deleted")
for user in users:
print(f" Would delete: {user}")
print("\nTo actually delete these users, run without --dry-run flag")
return
# Delete users
deleted = []
failed = []
for user in users:
try:
print(f" Deleting {user}...")
await client.users.delete_user(userid=user)
deleted.append(user)
print(f" ✓ Deleted {user}")
except Exception as e:
failed.append((user, str(e)))
print(f" ✗ Failed to delete {user}: {e}")
# Summary
print(f"\n{'=' * 60}")
print("Cleanup Summary")
print(f"{'=' * 60}")
print(f"Successfully deleted: {len(deleted)}")
print(f"Failed to delete: {len(failed)}")
if failed:
print("\nFailed deletions:")
for user, error in failed:
print(f" - {user}: {error}")
sys.exit(1)
else:
print("\n✓ All users cleaned up successfully")
except Exception as e:
print(f"ERROR: {e}", file=sys.stderr)
sys.exit(1)
@click.command()
@click.option(
"--prefix",
default="loadtest",
show_default=True,
help="Username prefix to search for",
)
@click.option(
"--dry-run",
is_flag=True,
help="List users without deleting them",
)
def main(prefix: str, dry_run: bool):
"""
Cleanup loadtest users from Nextcloud.
Searches for all users with the specified prefix and deletes them.
Useful for cleaning up after failed benchmark runs.
Examples:
# Dry run to see what would be deleted
uv run python -m tests.load.cleanup_loadtest_users --dry-run
# Delete all loadtest users
uv run python -m tests.load.cleanup_loadtest_users
# Delete users with custom prefix
uv run python -m tests.load.cleanup_loadtest_users --prefix mytest
"""
asyncio.run(cleanup_users(prefix=prefix, dry_run=dry_run))
if __name__ == "__main__":
main()
tests/load/oauth_benchmark.py
+737
View File
@@ -0,0 +1,737 @@
#!/usr/bin/env python3
"""
OAuth Multi-User Load Testing for Nextcloud MCP Server.
Simulates realistic multi-user scenarios with coordinated workflows
like note sharing, collaborative editing, and file operations.
Usage:
uv run python -m tests.load.oauth_benchmark --users 4 --duration 60
uv run python -m tests.load.oauth_benchmark -u 10 -d 300 --workload sharing
"""
import asyncio
import json
import logging
import os
import secrets
import signal
import sys
import threading
import time
from http.server import BaseHTTPRequestHandler, HTTPServer
from typing import Any
from urllib.parse import parse_qs, urlparse
import click
import httpx
from playwright.async_api import async_playwright
from nextcloud_mcp_server.auth.client_registration import load_or_register_client
from nextcloud_mcp_server.client import NextcloudClient
from tests.load.oauth_metrics import OAuthBenchmarkMetrics
from tests.load.oauth_pool import (
OAuthUserPool,
UserSessionWrapper,
generate_secure_password,
)
from tests.load.oauth_workloads import MixedOAuthWorkload, WorkflowResult
logging.basicConfig(
level=logging.WARNING, format="%(levelname)s [%(asctime)s] %(name)s - %(message)s"
)
logger = logging.getLogger(__name__)
class OAuthCallbackServer:
"""
Temporary HTTP server to capture OAuth authorization codes.
Runs in a background thread, captures auth codes via state parameter
correlation, and stores them in a shared dictionary.
"""
def __init__(self, host: str = "127.0.0.1", port: int = 8081):
self.host = host
self.port = port
self.auth_states: dict[str, str] = {}
self.server: HTTPServer | None = None
self.thread: threading.Thread | None = None
def start(self):
"""Start the callback server in a background thread."""
class CallbackHandler(BaseHTTPRequestHandler):
auth_states = self.auth_states
def do_GET(self):
parsed = urlparse(self.path)
if parsed.path == "/callback":
params = parse_qs(parsed.query)
code = params.get("code", [None])[0]
state = params.get("state", [None])[0]
if code and state:
self.auth_states[state] = code
logger.info(f"Captured auth code for state {state[:16]}...")
self.send_response(200)
self.send_header("Content-type", "text/html")
self.end_headers()
self.wfile.write(
b"<html><body><h1>Authorization successful!</h1>"
b"<p>You can close this window.</p></body></html>"
)
else:
self.send_response(404)
self.end_headers()
def log_message(self, format, *args):
# Suppress default logging
pass
self.server = HTTPServer((self.host, self.port), CallbackHandler)
def run():
logger.info(f"OAuth callback server listening on {self.host}:{self.port}")
self.server.serve_forever()
self.thread = threading.Thread(target=run, daemon=True)
self.thread.start()
logger.info("OAuth callback server started")
def stop(self):
"""Stop the callback server."""
if self.server:
self.server.shutdown()
logger.info("OAuth callback server stopped")
def get_auth_code(self, state: str) -> str | None:
"""Get auth code for a given state parameter."""
return self.auth_states.get(state)
async def discover_oidc_endpoints(nextcloud_host: str) -> dict[str, str]:
"""
Discover OIDC endpoints from Nextcloud's .well-known configuration.
Args:
nextcloud_host: Nextcloud host URL (e.g., http://localhost:8080)
Returns:
Dict with authorization_endpoint, token_endpoint, and registration_endpoint
"""
logger.info("Discovering OIDC endpoints...")
async with httpx.AsyncClient(verify=False, timeout=30.0) as client:
response = await client.get(
f"{nextcloud_host}/.well-known/openid-configuration"
)
response.raise_for_status()
config = response.json()
endpoints = {
"authorization_endpoint": config["authorization_endpoint"],
"token_endpoint": config["token_endpoint"],
"registration_endpoint": config["registration_endpoint"],
}
logger.info(f"Discovered endpoints: {endpoints}")
return endpoints
async def setup_oauth_client(
nextcloud_host: str, callback_url: str, registration_endpoint: str
) -> dict[str, str]:
"""
Setup OAuth client using load_or_register_client.
Args:
nextcloud_host: Nextcloud host URL
callback_url: OAuth callback URL
registration_endpoint: OAuth registration endpoint URL
Returns:
Dict with client_id and client_secret
"""
logger.info("Setting up OAuth client...")
# Use the client registration utility
client_info = await load_or_register_client(
nextcloud_url=nextcloud_host,
registration_endpoint=registration_endpoint,
storage_path=".nextcloud_oauth_benchmark_client.json",
client_name="OAuth Benchmark Test Client",
redirect_uris=[callback_url],
)
logger.info(f"OAuth client setup complete (client_id: {client_info.client_id})")
return {
"client_id": client_info.client_id,
"client_secret": client_info.client_secret,
}
async def create_and_authenticate_user(
user_pool: OAuthUserPool,
browser: Any,
auth_states: dict[str, str],
username: str,
password: str,
display_name: str | None = None,
) -> str:
"""
Create Nextcloud user and acquire OAuth token via Playwright.
Args:
user_pool: OAuthUserPool instance
browser: Playwright browser instance
auth_states: Shared auth_states dict for callback server
username: Username to create
password: Password for the user
display_name: Optional display name
Returns:
OAuth access token for the user
"""
logger.info(f"Creating and authenticating user: {username}")
# Create Nextcloud user
await user_pool.create_nextcloud_user(
username=username,
password=password,
display_name=display_name or username,
)
# Generate unique state for this OAuth flow
state = secrets.token_urlsafe(32)
# Acquire OAuth token via Playwright
token = await user_pool.acquire_token_playwright(
browser=browser,
username=username,
password=password,
state=state,
auth_states=auth_states,
)
logger.info(f"Successfully authenticated user: {username}")
return token
async def oauth_benchmark_worker(
user_wrapper: UserSessionWrapper,
workload: MixedOAuthWorkload,
duration: float,
metrics: OAuthBenchmarkMetrics,
stop_event: asyncio.Event,
):
"""
Single worker executing operations for one user.
Args:
user_wrapper: UserSessionWrapper for this worker
workload: MixedOAuthWorkload instance
duration: Test duration in seconds
metrics: Metrics collector
stop_event: Event to signal stop
"""
logger.info(f"Worker for {user_wrapper.username} starting...")
start_time = time.time()
operation_count = 0
try:
while not stop_event.is_set():
if time.time() - start_time >= duration:
break
# Run an operation (might be baseline or workflow)
result = await workload.run_operation()
# Record metrics
if isinstance(result, WorkflowResult):
metrics.add_workflow_result(result)
else:
# Baseline operation
metrics.add_baseline_operation(result)
operation_count += 1
# Small delay to prevent overwhelming the server
await asyncio.sleep(0.05)
logger.info(
f"Worker for {user_wrapper.username} completed {operation_count} operations"
)
except Exception as e:
logger.error(f"Worker {user_wrapper.username} error: {e}", exc_info=True)
async def show_progress(
duration: float,
metrics: OAuthBenchmarkMetrics,
stop_event: asyncio.Event,
):
"""Show real-time progress during benchmark."""
start_time = time.time()
while not stop_event.is_set():
elapsed = time.time() - start_time
if elapsed >= duration:
break
# Calculate progress
progress = min(elapsed / duration * 100, 100)
total_ops = len(metrics.baseline_operations) + len(metrics.workflows)
workflows = len(metrics.workflows)
# Print progress bar
bar_length = 40
filled = int(bar_length * progress / 100)
bar = "█" * filled + "░" * (bar_length - filled)
print(
f"\r[{bar}] {progress:5.1f}% | "
f"Total Ops: {total_ops:6d} | "
f"Workflows: {workflows:4d}",
end="",
flush=True,
)
await asyncio.sleep(0.5)
print() # New line after progress
async def run_oauth_benchmark(
num_users: int,
duration: float,
mcp_url: str,
warmup: float = 5.0,
user_prefix: str = "loadtest",
cleanup: bool = True,
browser_type: str = "firefox",
headed: bool = False,
) -> OAuthBenchmarkMetrics:
"""
Run the OAuth multi-user benchmark with dynamic user creation.
Args:
num_users: Number of concurrent users to create
duration: Test duration in seconds
mcp_url: MCP server URL
warmup: Warmup period in seconds
user_prefix: Prefix for generated usernames
cleanup: Whether to delete users after benchmark
browser_type: Playwright browser type (firefox, chromium, webkit)
headed: Whether to run browser in headed mode
Returns:
OAuthBenchmarkMetrics with results
"""
metrics = OAuthBenchmarkMetrics()
stop_event = asyncio.Event()
created_users: list[str] = []
callback_server: OAuthCallbackServer | None = None
user_pool: OAuthUserPool | None = None
admin_client: NextcloudClient | None = None
# Setup signal handlers for graceful shutdown
def signal_handler(sig, frame):
logger.warning("Received interrupt signal, stopping benchmark...")
stop_event.set()
signal.signal(signal.SIGINT, signal_handler)
signal.signal(signal.SIGTERM, signal_handler)
print(f"\n{'=' * 80}")
print("OAUTH MULTI-USER BENCHMARK")
print(f"{'=' * 80}")
print(f"Users: {num_users} | Duration: {duration}s | Warmup: {warmup}s")
print(f"Target: {mcp_url}")
print(f"User Prefix: {user_prefix} | Cleanup: {cleanup}")
print(f"Browser: {browser_type} | Headed: {headed}")
print(f"{'=' * 80}\n")
try:
# Get environment variables
nextcloud_host = os.getenv("NEXTCLOUD_HOST", "http://localhost:8080")
callback_url = "http://127.0.0.1:8081/callback"
# Step 1: Start OAuth callback server
print("Step 1/6: Starting OAuth callback server...")
callback_server = OAuthCallbackServer(host="127.0.0.1", port=8081)
callback_server.start()
print("✓ Callback server listening on http://127.0.0.1:8081\n")
# Step 2: Discover OIDC endpoints
print("Step 2/6: Discovering OIDC endpoints...")
endpoints = await discover_oidc_endpoints(nextcloud_host)
print(f"✓ Authorization endpoint: {endpoints['authorization_endpoint']}")
print(f"✓ Token endpoint: {endpoints['token_endpoint']}")
print(f"✓ Registration endpoint: {endpoints['registration_endpoint']}\n")
# Step 3: Setup OAuth client
print("Step 3/6: Setting up OAuth client...")
oauth_credentials = await setup_oauth_client(
nextcloud_host, callback_url, endpoints["registration_endpoint"]
)
print(f"✓ OAuth client registered (ID: {oauth_credentials['client_id']})\n")
# Step 4: Create admin client and user pool
print("Step 4/6: Initializing admin client and user pool...")
admin_client = NextcloudClient.from_env()
user_pool = OAuthUserPool(
admin_client=admin_client,
client_id=oauth_credentials["client_id"],
client_secret=oauth_credentials["client_secret"],
callback_url=callback_url,
token_endpoint=endpoints["token_endpoint"],
authorization_endpoint=endpoints["authorization_endpoint"],
)
async with user_pool:
print("✓ User pool initialized\n")
# Step 5: Create users and acquire OAuth tokens (concurrently)
print(f"Step 5/6: Creating {num_users} users and acquiring OAuth tokens...")
print("(Running concurrently for faster setup)\n")
async def create_user_task(
i: int, browser, auth_states: dict
) -> tuple[str, str, str] | None:
"""Create and authenticate a single user. Returns (username, password, token) or None on failure."""
username = f"{user_prefix}_user_{i + 1}"
password = generate_secure_password(16)
print(f" [{i + 1}/{num_users}] Creating user '{username}'...")
try:
token = await create_and_authenticate_user(
user_pool=user_pool,
browser=browser,
auth_states=auth_states,
username=username,
password=password,
display_name=f"Load Test User {i + 1}",
)
print(f" ✓ User '{username}' authenticated\n")
return (username, password, token)
except Exception as e:
logger.error(f"Failed to create/authenticate user {username}: {e}")
return None
async with async_playwright() as p:
# Launch browser
browser_launcher = getattr(p, browser_type)
browser = await browser_launcher.launch(headless=not headed)
try:
# Create all users concurrently
tasks = [
create_user_task(i, browser, callback_server.auth_states)
for i in range(num_users)
]
results = await asyncio.gather(*tasks, return_exceptions=True)
# Process results
for result in results:
if isinstance(result, Exception):
logger.error(f"User creation task failed: {result}")
continue
if result is None:
continue
username, password, token = result
await user_pool.add_user(
username=username, password=password, token=token
)
created_users.append(username)
finally:
await browser.close()
if not created_users:
raise RuntimeError("Failed to create any users")
print(
f"✓ Successfully created and authenticated {len(created_users)} users\n"
)
# Step 6: Create MCP sessions for each user (concurrently)
print("Step 6/6: Creating MCP sessions for users...")
user_wrappers = []
async with user_pool:
async def create_session_task(username: str) -> UserSessionWrapper | None:
"""Create MCP session for a user. Returns wrapper or None on failure."""
try:
session = await user_pool.create_user_session(username, mcp_url)
wrapper = UserSessionWrapper(username, session, user_pool)
print(f" ✓ Session created for '{username}'")
return wrapper
except Exception as e:
logger.error(f"Failed to create session for {username}: {e}")
return None
# Create all sessions concurrently
session_tasks = [
create_session_task(username) for username in created_users
]
session_results = await asyncio.gather(
*session_tasks, return_exceptions=True
)
# Process results
for result in session_results:
if isinstance(result, Exception):
logger.error(f"Session creation task failed: {result}")
continue
if result is not None:
user_wrappers.append(result)
if not user_wrappers:
raise RuntimeError("Failed to create any user sessions")
print(f"✓ Created {len(user_wrappers)} MCP sessions\n")
# Warmup period
if warmup > 0:
print(f"Warmup period: {warmup}s...")
await asyncio.sleep(warmup)
print()
# Start benchmark
print(f"{'=' * 80}")
print("STARTING BENCHMARK")
print(f"{'=' * 80}\n")
metrics.start()
# Create workload and workers
workload = MixedOAuthWorkload(user_wrappers)
workers = [
oauth_benchmark_worker(wrapper, workload, duration, metrics, stop_event)
for wrapper in user_wrappers
]
# Run workers with progress display
progress_task = asyncio.create_task(
show_progress(duration, metrics, stop_event)
)
await asyncio.gather(*workers, return_exceptions=True)
stop_event.set()
await progress_task
metrics.stop()
print(f"\n{'=' * 80}")
print("BENCHMARK COMPLETE")
print(f"{'=' * 80}\n")
# Cleanup user sessions
print("Closing user sessions...")
await user_pool.close_all_sessions()
print("✓ All sessions closed\n")
except Exception as e:
logger.error(f"Benchmark error: {e}", exc_info=True)
# Don't re-raise here - we want cleanup to run
finally:
# Cleanup callback server
if callback_server:
try:
callback_server.stop()
logger.info("OAuth callback server stopped")
except Exception as e:
logger.warning(f"Error stopping callback server: {e}")
# Cleanup test users
if cleanup and created_users:
print(f"\nCleaning up {len(created_users)} test users...")
# Create a new admin client for cleanup (don't rely on the existing one)
try:
cleanup_client = NextcloudClient.from_env()
for username in created_users:
try:
await cleanup_client.users.delete_user(userid=username)
print(f" ✓ Deleted user '{username}'")
except Exception as e:
logger.warning(f"Failed to delete user {username}: {e}")
print("✓ Cleanup complete\n")
except Exception as e:
logger.error(f"Error during user cleanup: {e}")
print(
"⚠️ Failed to cleanup users. Please run cleanup script manually.\n"
)
elif created_users:
print(
f"\n⚠️ {len(created_users)} test users were NOT deleted (cleanup=False)"
)
print(f"Users: {', '.join(created_users)}\n")
return metrics
@click.command()
@click.option(
"--users",
"-u",
type=int,
default=2,
show_default=True,
help="Number of concurrent users to create dynamically",
)
@click.option(
"--duration",
"-d",
type=float,
default=30.0,
show_default=True,
help="Test duration in seconds",
)
@click.option(
"--warmup",
"-w",
type=float,
default=5.0,
show_default=True,
help="Warmup duration before collecting metrics (seconds)",
)
@click.option(
"--url",
default="http://127.0.0.1:8001/mcp",
show_default=True,
help="MCP OAuth server URL",
)
@click.option(
"--output",
"-o",
type=click.Path(),
help="Output file for JSON results (optional)",
)
@click.option(
"--workload",
type=click.Choice(["mixed", "sharing", "collaboration", "baseline"]),
default="mixed",
show_default=True,
help="Workload type to execute",
)
@click.option(
"--user-prefix",
default="loadtest",
show_default=True,
help="Prefix for dynamically created usernames",
)
@click.option(
"--cleanup/--no-cleanup",
default=True,
show_default=True,
help="Delete created users after benchmark",
)
@click.option(
"--browser",
type=click.Choice(["firefox", "chromium", "webkit"]),
default="firefox",
show_default=True,
help="Playwright browser type for OAuth automation",
)
@click.option(
"--headed",
is_flag=True,
help="Run browser in headed mode (visible window, useful for debugging)",
)
@click.option(
"--verbose",
"-v",
is_flag=True,
help="Enable verbose logging",
)
def main(
users: int,
duration: float,
warmup: float,
url: str,
output: str | None,
workload: str,
user_prefix: str,
cleanup: bool,
browser: str,
headed: bool,
verbose: bool,
):
"""
OAuth Multi-User Load Testing for Nextcloud MCP Server.
Dynamically creates N users, authenticates them via OAuth using Playwright
browser automation, and simulates realistic multi-user scenarios with
coordinated workflows like note sharing, collaborative editing, and file operations.
Examples:
# 2 users, 30-second test (default settings)
uv run python -m tests.load.oauth_benchmark
# 4 users, 60-second test with mixed workload
uv run python -m tests.load.oauth_benchmark --users 4 --duration 60
# 10 users, 5-minute sharing-focused test
uv run python -m tests.load.oauth_benchmark -u 10 -d 300 --workload sharing
# Export results to JSON
uv run python -m tests.load.oauth_benchmark -u 5 -d 120 --output results.json
# Custom user prefix and keep users after benchmark
uv run python -m tests.load.oauth_benchmark -u 3 --user-prefix mytest --no-cleanup
# Debug with visible browser (headed mode)
uv run python -m tests.load.oauth_benchmark -u 2 -d 10 --headed --verbose
Requirements:
- docker-compose up (mcp-oauth container running on port 8001)
- NEXTCLOUD_HOST, NEXTCLOUD_USERNAME, NEXTCLOUD_PASSWORD env vars set
- Playwright browser installed: uv run playwright install firefox
"""
if verbose:
logging.getLogger().setLevel(logging.DEBUG)
logging.getLogger("tests.load").setLevel(logging.DEBUG)
async def run():
# Run benchmark
metrics = await run_oauth_benchmark(
num_users=users,
duration=duration,
mcp_url=url,
warmup=warmup,
user_prefix=user_prefix,
cleanup=cleanup,
browser_type=browser,
headed=headed,
)
# Print report
metrics.print_report()
# Export to JSON if requested
if output:
with open(output, "w") as f:
json.dump(metrics.to_dict(), f, indent=2)
print(f"Results exported to: {output}")
try:
asyncio.run(run())
except KeyboardInterrupt:
print("\nBenchmark interrupted by user")
sys.exit(130)
except Exception as e:
print(f"ERROR: {e}", file=sys.stderr)
if verbose:
raise
sys.exit(1)
if __name__ == "__main__":
main()
tests/load/oauth_metrics.py
+329
View File
@@ -0,0 +1,329 @@
"""
Enhanced metrics collection for OAuth multi-user load testing.
Extends the base BenchmarkMetrics to track per-user statistics,
workflow completion rates, and cross-user operation latencies.
"""
import statistics
from collections import Counter, defaultdict
from typing import Any
from tests.load.oauth_workloads import WorkflowResult
class OAuthBenchmarkMetrics:
"""
Enhanced metrics for OAuth multi-user load testing.
Tracks:
- Per-user operation counts and latencies
- Workflow completion rates and timings
- Cross-user operation metrics
- Step-by-step workflow breakdowns
"""
def __init__(self):
# Base metrics
self.start_time: float | None = None
self.end_time: float | None = None
# Per-user tracking
self.user_operations: dict[str, list[dict[str, Any]]] = defaultdict(list)
self.user_operation_counts: dict[str, Counter] = defaultdict(Counter)
self.user_errors: dict[str, Counter] = defaultdict(Counter)
# Workflow tracking
self.workflows: list[WorkflowResult] = []
self.workflow_counts: Counter = Counter()
self.workflow_successes: Counter = Counter()
self.workflow_durations: dict[str, list[float]] = defaultdict(list)
# Baseline operations (non-workflow)
self.baseline_operations: list[dict[str, Any]] = []
def start(self):
"""Mark the start of the benchmark."""
import time
self.start_time = time.time()
def stop(self):
"""Mark the end of the benchmark."""
import time
self.end_time = time.time()
@property
def duration(self) -> float:
"""Total benchmark duration in seconds."""
if self.start_time is None or self.end_time is None:
return 0.0
return self.end_time - self.start_time
def add_workflow_result(self, result: WorkflowResult):
"""
Add a workflow execution result.
Args:
result: WorkflowResult from workflow execution
"""
self.workflows.append(result)
self.workflow_counts[result.workflow_name] += 1
if result.success:
self.workflow_successes[result.workflow_name] += 1
self.workflow_durations[result.workflow_name].append(result.total_duration)
# Track per-user operations from workflow steps
for step in result.steps:
self.user_operation_counts[step.user][step.step_name] += 1
if not step.success:
self.user_errors[step.user][step.step_name] += 1
self.user_operations[step.user].append(
{
"type": "workflow_step",
"workflow": result.workflow_name,
"step": step.step_name,
"success": step.success,
"duration": step.duration,
"error": step.error,
}
)
def add_baseline_operation(self, operation: dict[str, Any]):
"""
Add a baseline (non-workflow) operation result.
Args:
operation: Dict with keys: type, operation, user, success, duration, error (optional)
"""
self.baseline_operations.append(operation)
user = operation.get("user", "unknown")
op_name = operation.get("operation", "unknown")
success = operation.get("success", False)
self.user_operation_counts[user][op_name] += 1
if not success:
self.user_errors[user][op_name] += 1
self.user_operations[user].append(operation)
def get_user_stats(self) -> dict[str, dict[str, Any]]:
"""
Get per-user statistics.
Returns:
Dict mapping username to their stats
"""
stats = {}
for user, operations in self.user_operations.items():
total_ops = len(operations)
successful_ops = sum(1 for op in operations if op.get("success", False))
durations = [op["duration"] for op in operations if "duration" in op]
stats[user] = {
"total_operations": total_ops,
"successful_operations": successful_ops,
"failed_operations": total_ops - successful_ops,
"success_rate": (successful_ops / total_ops * 100)
if total_ops > 0
else 0.0,
"latency": self._calculate_latency_stats(durations),
"operations_breakdown": dict(self.user_operation_counts[user]),
"errors_breakdown": dict(self.user_errors[user]),
}
return stats
def get_workflow_stats(self) -> dict[str, dict[str, Any]]:
"""
Get workflow execution statistics.
Returns:
Dict mapping workflow name to its stats
"""
stats = {}
for workflow_name in self.workflow_counts:
total = self.workflow_counts[workflow_name]
successes = self.workflow_successes[workflow_name]
durations = self.workflow_durations[workflow_name]
# Calculate per-step latencies
step_latencies = defaultdict(list)
for workflow in self.workflows:
if workflow.workflow_name == workflow_name:
for step in workflow.steps:
if step.success:
step_latencies[step.step_name].append(step.duration)
step_stats = {}
for step_name, latencies in step_latencies.items():
if latencies:
step_stats[step_name] = self._calculate_latency_stats(latencies)
stats[workflow_name] = {
"total_executions": total,
"successful_executions": successes,
"failed_executions": total - successes,
"success_rate": (successes / total * 100) if total > 0 else 0.0,
"latency": self._calculate_latency_stats(durations),
"step_latencies": step_stats,
}
return stats
def get_baseline_stats(self) -> dict[str, Any]:
"""
Get statistics for baseline operations.
Returns:
Dict with baseline operation stats
"""
if not self.baseline_operations:
return {
"total_operations": 0,
"success_rate": 0.0,
"latency": self._calculate_latency_stats([]),
}
total = len(self.baseline_operations)
successes = sum(
1 for op in self.baseline_operations if op.get("success", False)
)
durations = [
op["duration"] for op in self.baseline_operations if "duration" in op
]
# Per-operation breakdown
operation_counts = Counter()
operation_errors = Counter()
for op in self.baseline_operations:
op_name = op.get("operation", "unknown")
operation_counts[op_name] += 1
if not op.get("success", False):
operation_errors[op_name] += 1
return {
"total_operations": total,
"successful_operations": successes,
"failed_operations": total - successes,
"success_rate": (successes / total * 100) if total > 0 else 0.0,
"latency": self._calculate_latency_stats(durations),
"operations_breakdown": dict(operation_counts),
"errors_breakdown": dict(operation_errors),
}
def _calculate_latency_stats(self, durations: list[float]) -> dict[str, float]:
"""Calculate latency statistics from a list of durations."""
if not durations:
return {
"min": 0.0,
"max": 0.0,
"mean": 0.0,
"median": 0.0,
"p90": 0.0,
"p95": 0.0,
"p99": 0.0,
}
sorted_durations = sorted(durations)
def percentile(data: list[float], p: float) -> float:
k = (len(data) - 1) * p
f = int(k)
c = f + 1
if c >= len(data):
return data[-1]
return data[f] + (k - f) * (data[c] - data[f])
return {
"min": min(durations),
"max": max(durations),
"mean": statistics.mean(durations),
"median": statistics.median(durations),
"p90": percentile(sorted_durations, 0.90),
"p95": percentile(sorted_durations, 0.95),
"p99": percentile(sorted_durations, 0.99),
}
def to_dict(self) -> dict[str, Any]:
"""Convert metrics to dictionary for JSON export."""
return {
"summary": {
"duration": self.duration,
"total_workflows": len(self.workflows),
"total_baseline_ops": len(self.baseline_operations),
"total_users": len(self.user_operations),
},
"workflows": self.get_workflow_stats(),
"baseline": self.get_baseline_stats(),
"users": self.get_user_stats(),
}
def print_report(self):
"""Print human-readable benchmark report."""
print("\n" + "=" * 80)
print("OAUTH MULTI-USER BENCHMARK RESULTS")
print("=" * 80)
# Summary
print(f"\nDuration: {self.duration:.2f}s")
print(f"Total Users: {len(self.user_operations)}")
print(f"Total Workflows Executed: {len(self.workflows)}")
print(f"Total Baseline Operations: {len(self.baseline_operations)}")
# Workflow Stats
if self.workflows:
print("\n" + "-" * 80)
print("WORKFLOW STATISTICS")
print("-" * 80)
print(
f"{'Workflow':<30} {'Total':>8} {'Success':>8} {'Rate':>8} {'P50':>10} {'P95':>10}"
)
print("-" * 80)
workflow_stats = self.get_workflow_stats()
for name, stats in sorted(workflow_stats.items()):
latency = stats["latency"]
print(
f"{name:<30} {stats['total_executions']:>8} "
f"{stats['successful_executions']:>8} "
f"{stats['success_rate']:>7.1f}% "
f"{latency['median']:>9.4f}s {latency['p95']:>9.4f}s"
)
# Per-User Stats
print("\n" + "-" * 80)
print("PER-USER STATISTICS")
print("-" * 80)
print(
f"{'User':<20} {'Total Ops':>10} {'Success':>10} {'Errors':>8} {'Rate':>8} {'P50':>10}"
)
print("-" * 80)
user_stats = self.get_user_stats()
for username, stats in sorted(user_stats.items()):
latency = stats["latency"]
print(
f"{username:<20} {stats['total_operations']:>10} "
f"{stats['successful_operations']:>10} "
f"{stats['failed_operations']:>8} "
f"{stats['success_rate']:>7.1f}% "
f"{latency['median']:>9.4f}s"
)
# Baseline Stats
if self.baseline_operations:
print("\n" + "-" * 80)
print("BASELINE OPERATIONS")
print("-" * 80)
baseline = self.get_baseline_stats()
print(f"Total Operations: {baseline['total_operations']}")
print(f"Success Rate: {baseline['success_rate']:.1f}%")
latency = baseline["latency"]
print(
f"Latency: min={latency['min']:.4f}s, p50={latency['median']:.4f}s, "
f"p95={latency['p95']:.4f}s, max={latency['max']:.4f}s"
)
print("=" * 80 + "\n")
tests/load/oauth_pool.py
+506
View File
@@ -0,0 +1,506 @@
"""
OAuth User Pool Management for Load Testing.
Manages multiple OAuth-authenticated users for realistic multi-user load testing scenarios.
"""
import asyncio
import logging
from dataclasses import dataclass
from typing import Any
import httpx
from mcp import ClientSession
from mcp.client.streamable_http import streamablehttp_client
logger = logging.getLogger(__name__)
@dataclass
class UserConfig:
"""Configuration for a single test user."""
username: str
password: str
display_name: str
email: str
groups: list[str]
@dataclass
class UserProfile:
"""Profile for an OAuth-authenticated user."""
username: str
password: str
token: str
session: ClientSession | None = None
streamable_context: Any | None = None # Store for proper cleanup
operation_count: int = 0
error_count: int = 0
class OAuthUserPool:
"""
Manages a pool of OAuth-authenticated users for load testing.
Handles token acquisition, session management, and user lifecycle.
"""
def __init__(
self,
admin_client: Any, # NextcloudClient with admin credentials
client_id: str,
client_secret: str,
callback_url: str,
token_endpoint: str,
authorization_endpoint: str,
):
self.admin_client = admin_client # For user management
self.nextcloud_host = str(admin_client._client.base_url)
self.client_id = client_id
self.client_secret = client_secret
self.callback_url = callback_url
self.token_endpoint = token_endpoint
self.authorization_endpoint = authorization_endpoint
self.users: dict[str, UserProfile] = {}
self._http_client: httpx.AsyncClient | None = None
async def __aenter__(self):
"""Initialize HTTP client."""
self._http_client = httpx.AsyncClient(verify=False, timeout=30.0)
return self
async def __aexit__(self, exc_type, exc_val, exc_tb):
"""Cleanup HTTP client."""
if self._http_client:
await self._http_client.aclose()
async def acquire_token(self, username: str, password: str, auth_code: str) -> str:
"""
Exchange authorization code for OAuth access token.
Args:
username: Username for logging
password: Password (for logging/debugging)
auth_code: Authorization code from OAuth flow
Returns:
OAuth access token
"""
logger.info(f"Exchanging auth code for access token (user: {username})...")
if not self._http_client:
raise RuntimeError(
"HTTP client not initialized - use async context manager"
)
# Exchange authorization code for access token
token_response = await self._http_client.post(
self.token_endpoint,
data={
"grant_type": "authorization_code",
"code": auth_code,
"redirect_uri": self.callback_url,
"client_id": self.client_id,
"client_secret": self.client_secret,
},
headers={"Content-Type": "application/x-www-form-urlencoded"},
)
token_response.raise_for_status()
token_data = token_response.json()
access_token = token_data.get("access_token")
if not access_token:
raise ValueError(f"No access token in response for {username}")
logger.info(f"Successfully acquired OAuth token for {username}")
return access_token
async def add_user(self, username: str, password: str, token: str) -> UserProfile:
"""
Add a user to the pool with their OAuth token.
Args:
username: Username
password: Password (for future re-auth if needed)
token: OAuth access token
Returns:
UserProfile for the added user
"""
if username in self.users:
logger.warning(f"User {username} already in pool, updating token")
profile = UserProfile(username=username, password=password, token=token)
self.users[username] = profile
logger.info(f"Added user {username} to pool (total: {len(self.users)})")
return profile
async def create_user_session(
self, username: str, mcp_url: str = "http://127.0.0.1:8001/mcp"
) -> ClientSession:
"""
Create an MCP client session for a user.
Args:
username: Username to create session for
mcp_url: MCP server URL
Returns:
Initialized ClientSession
Raises:
KeyError: If user not in pool
"""
if username not in self.users:
raise KeyError(f"User {username} not in pool")
profile = self.users[username]
# Create streamable HTTP connection with OAuth token in Authorization header
# This matches the pattern from tests/conftest.py create_mcp_client_session()
headers = {"Authorization": f"Bearer {profile.token}"}
streamable_context = streamablehttp_client(mcp_url, headers=headers)
try:
read_stream, write_stream, _ = await streamable_context.__aenter__()
session = ClientSession(read_stream, write_stream)
await session.__aenter__()
await session.initialize()
# Store both session and context for proper cleanup
profile.session = session
profile.streamable_context = streamable_context
logger.info(f"Created MCP session for {username}")
return session
except Exception as e:
# Clean up streamable context if session creation failed
try:
await streamable_context.__aexit__(None, None, None)
except RuntimeError as cleanup_error:
if "cancel scope" in str(cleanup_error):
logger.debug(
f"Ignoring cancel scope teardown issue: {cleanup_error}"
)
else:
raise
raise e
async def close_user_session(self, username: str):
"""Close the MCP session for a user."""
if username not in self.users:
return
profile = self.users[username]
# Close ClientSession
if profile.session:
try:
await profile.session.__aexit__(None, None, None)
except RuntimeError as e:
if "cancel scope" in str(e):
logger.debug(
f"Ignoring cancel scope teardown issue for {username}: {e}"
)
else:
logger.debug(f"Error closing session for {username}: {e}")
except Exception as e:
logger.debug(f"Error closing session for {username}: {e}")
profile.session = None
# Close streamable context
if profile.streamable_context:
try:
await profile.streamable_context.__aexit__(None, None, None)
except RuntimeError as e:
if "cancel scope" in str(e):
logger.debug(
f"Ignoring cancel scope teardown issue for {username}: {e}"
)
else:
logger.debug(
f"Error closing streamable context for {username}: {e}"
)
except Exception as e:
logger.debug(f"Error closing streamable context for {username}: {e}")
profile.streamable_context = None
async def close_all_sessions(self):
"""Close all user sessions."""
for username in list(self.users.keys()):
await self.close_user_session(username)
def get_user(self, username: str) -> UserProfile:
"""Get user profile by username."""
if username not in self.users:
raise KeyError(f"User {username} not in pool")
return self.users[username]
def get_all_users(self) -> list[UserProfile]:
"""Get all user profiles."""
return list(self.users.values())
def record_operation(self, username: str, success: bool = True):
"""Record an operation for user stats."""
if username in self.users:
self.users[username].operation_count += 1
if not success:
self.users[username].error_count += 1
def get_stats(self) -> dict[str, dict[str, int | float]]:
"""Get per-user operation statistics."""
return {
username: {
"operations": profile.operation_count,
"errors": profile.error_count,
"success_rate": (
(profile.operation_count - profile.error_count)
/ max(profile.operation_count, 1)
* 100
),
}
for username, profile in self.users.items()
}
async def create_nextcloud_user(
self,
username: str,
password: str,
display_name: str | None = None,
email: str | None = None,
) -> UserConfig:
"""
Create a Nextcloud user via the Users API.
Args:
username: Username for the new user
password: Password for the new user
display_name: Optional display name
email: Optional email address
Returns:
UserConfig for the created user
Raises:
HTTPStatusError: If user creation fails
"""
logger.info(f"Creating Nextcloud user: {username}")
await self.admin_client.users.create_user(
userid=username,
password=password,
display_name=display_name or username,
email=email or f"{username}@benchmark.local",
)
logger.info(f"Successfully created Nextcloud user: {username}")
return UserConfig(
username=username,
password=password,
display_name=display_name or username,
email=email or f"{username}@benchmark.local",
groups=[],
)
async def delete_nextcloud_user(self, username: str):
"""
Delete a Nextcloud user via the Users API.
Args:
username: Username to delete
"""
logger.info(f"Deleting Nextcloud user: {username}")
try:
await self.admin_client.users.delete_user(userid=username)
logger.info(f"Successfully deleted Nextcloud user: {username}")
except Exception as e:
logger.warning(f"Failed to delete user {username}: {e}")
async def acquire_token_playwright(
self,
browser: Any,
username: str,
password: str,
state: str,
auth_states: dict[str, str],
) -> str:
"""
Acquire OAuth token via Playwright browser automation.
Based on conftest.py playwright_oauth_token fixture.
Automates the full OAuth flow:
1. Navigate to authorization URL
2. Fill login form
3. Handle OAuth consent
4. Wait for callback server to receive auth code
5. Exchange code for access token
Args:
browser: Playwright browser instance
username: Username to authenticate
password: Password for the user
state: Unique state parameter for this OAuth flow
auth_states: Dict mapping state -> auth_code (shared with callback server)
Returns:
OAuth access token
Raises:
TimeoutError: If callback not received within timeout
ValueError: If token exchange fails
"""
import time
from urllib.parse import quote
logger.info(f"Starting Playwright OAuth flow for {username}...")
logger.debug(f"Using state: {state[:16]}...")
# Construct authorization URL
auth_url = (
f"{self.authorization_endpoint}?"
f"response_type=code&"
f"client_id={self.client_id}&"
f"redirect_uri={quote(self.callback_url, safe='')}&"
f"state={state}&"
f"scope=openid%20profile%20email"
)
# Browser automation
context = await browser.new_context(ignore_https_errors=True)
page = await context.new_page()
try:
# Navigate to authorization URL
logger.debug("Navigating to authorization URL...")
await page.goto(auth_url, wait_until="networkidle", timeout=30000)
current_url = page.url
# Login if needed
if "/login" in current_url or "/index.php/login" in current_url:
logger.info(f"Logging in as {username}...")
await page.wait_for_selector('input[name="user"]', timeout=10000)
await page.fill('input[name="user"]', username)
await page.fill('input[name="password"]', password)
await page.click('button[type="submit"]')
await page.wait_for_load_state("networkidle", timeout=30000)
current_url = page.url
logger.info("Login completed")
# Handle OAuth consent if present
try:
authorize_button = await page.query_selector(
'button:has-text("Authorize"), button:has-text("Allow"), input[type="submit"][value*="uthoriz"]'
)
if authorize_button:
logger.info("Authorizing OAuth client...")
await authorize_button.click()
await page.wait_for_load_state("networkidle", timeout=10000)
except Exception as e:
logger.debug(f"No authorization needed: {e}")
# Wait for callback server to receive auth code
logger.info("Waiting for OAuth callback...")
timeout_seconds = 30
start_time = time.time()
while state not in auth_states:
if time.time() - start_time > timeout_seconds:
screenshot_path = f"/tmp/oauth_timeout_{username}.png"
await page.screenshot(path=screenshot_path)
logger.error(f"Screenshot saved to {screenshot_path}")
raise TimeoutError(
f"Timeout waiting for OAuth callback for {username}"
)
await asyncio.sleep(0.5)
auth_code = auth_states[state]
logger.info(f"Received auth code for {username}")
finally:
await context.close()
# Exchange code for token
logger.info(f"Exchanging auth code for access token ({username})...")
token_response = await self._http_client.post(
self.token_endpoint,
data={
"grant_type": "authorization_code",
"code": auth_code,
"redirect_uri": self.callback_url,
"client_id": self.client_id,
"client_secret": self.client_secret,
},
headers={"Content-Type": "application/x-www-form-urlencoded"},
)
token_response.raise_for_status()
token_data = token_response.json()
access_token = token_data.get("access_token")
if not access_token:
raise ValueError(f"No access token for {username}: {token_data}")
logger.info(f"Successfully acquired OAuth token for {username}")
return access_token
class UserSessionWrapper:
"""
Wrapper for a user-specific MCP session with operation tracking.
Provides a convenient interface for executing operations as a specific user.
"""
def __init__(self, username: str, session: ClientSession, pool: OAuthUserPool):
self.username = username
self.session = session
self.pool = pool
async def call_tool(self, tool_name: str, arguments: dict[str, Any]) -> Any:
"""
Call an MCP tool and record the operation.
Args:
tool_name: Name of the tool to call
arguments: Tool arguments
Returns:
Tool result
"""
try:
result = await self.session.call_tool(tool_name, arguments)
self.pool.record_operation(self.username, success=True)
return result
except Exception:
self.pool.record_operation(self.username, success=False)
raise
async def read_resource(self, uri: str) -> Any:
"""
Read an MCP resource and record the operation.
Args:
uri: Resource URI
Returns:
Resource data
"""
try:
result = await self.session.read_resource(uri)
self.pool.record_operation(self.username, success=True)
return result
except Exception:
self.pool.record_operation(self.username, success=False)
raise
def generate_secure_password(length: int = 20) -> str:
"""Generate a secure random password."""
import secrets
import string
alphabet = string.ascii_letters + string.digits + "!@#$%^&*()"
return "".join(secrets.choice(alphabet) for _ in range(length))
tests/load/oauth_workloads.py
+506
View File
@@ -0,0 +1,506 @@
"""
Multi-User Workflow Definitions for OAuth Load Testing.
Defines coordinated workflows that span multiple users, simulating realistic
collaborative scenarios like note sharing, file collaboration, and permission management.
"""
import asyncio
import json
import logging
import random
import time
import uuid
from abc import ABC, abstractmethod
from dataclasses import dataclass, field
from typing import Any, Callable, Awaitable
from tests.load.oauth_pool import UserSessionWrapper
logger = logging.getLogger(__name__)
@dataclass
class WorkflowStepResult:
"""Result of a single workflow step."""
step_name: str
user: str
success: bool
duration: float
error: str | None = None
data: dict[str, Any] = field(default_factory=dict)
@dataclass
class WorkflowResult:
"""Result of a complete workflow execution."""
workflow_name: str
success: bool
total_duration: float
steps: list[WorkflowStepResult]
participants: list[str]
error: str | None = None
@property
def steps_completed(self) -> int:
"""Count of successfully completed steps."""
return sum(1 for step in self.steps if step.success)
@property
def step_latencies(self) -> dict[str, float]:
"""Map of step names to their durations."""
return {step.step_name: step.duration for step in self.steps}
class Workflow(ABC):
"""
Base class for multi-user workflows.
A workflow represents a coordinated sequence of operations across multiple users,
such as creating and sharing a note, collaborative editing, or permission management.
"""
def __init__(self, name: str):
self.name = name
self.steps: list[WorkflowStepResult] = []
self.start_time: float | None = None
@abstractmethod
async def execute(self, users: list[UserSessionWrapper]) -> WorkflowResult:
"""
Execute the workflow with the given users.
Args:
users: List of UserSessionWrapper instances to use in the workflow
Returns:
WorkflowResult with execution details
"""
pass
async def _execute_step(
self,
step_name: str,
user: UserSessionWrapper,
operation: Callable[..., Awaitable[Any]],
**kwargs,
) -> WorkflowStepResult:
"""
Execute a single workflow step with timing and error handling.
Args:
step_name: Name of the step for reporting
user: User executing the step
operation: Async callable to execute
**kwargs: Arguments to pass to the operation
Returns:
WorkflowStepResult
"""
start = time.time()
try:
result = await operation(**kwargs)
duration = time.time() - start
step_result = WorkflowStepResult(
step_name=step_name,
user=user.username,
success=True,
duration=duration,
data={"result": result} if result else {},
)
self.steps.append(step_result)
return step_result
except Exception as e:
duration = time.time() - start
logger.error(f"Step {step_name} failed for user {user.username}: {e}")
step_result = WorkflowStepResult(
step_name=step_name,
user=user.username,
success=False,
duration=duration,
error=str(e),
)
self.steps.append(step_result)
return step_result
def _finish(self, success: bool, error: str | None = None) -> WorkflowResult:
"""
Finalize workflow and create result.
Args:
success: Whether the overall workflow succeeded
error: Optional error message
Returns:
WorkflowResult
"""
duration = time.time() - self.start_time if self.start_time else 0.0
participants = list(set(step.user for step in self.steps))
return WorkflowResult(
workflow_name=self.name,
success=success,
total_duration=duration,
steps=self.steps,
participants=participants,
error=error,
)
class NoteShareWorkflow(Workflow):
"""
Workflow: User A creates a note and shares it with User B, who then reads it.
Steps:
1. User A creates a note
2. User A shares the note with User B (read-only)
3. User B lists their shared notes (verify propagation)
4. User B reads the shared note
"""
def __init__(self):
super().__init__("note_share")
async def execute(self, users: list[UserSessionWrapper]) -> WorkflowResult:
"""Execute note sharing workflow."""
self.start_time = time.time()
if len(users) < 2:
return self._finish(False, error="Requires at least 2 users")
user_a, user_b = users[0], users[1]
unique_id = uuid.uuid4().hex[:8]
try:
# Step 1: User A creates note
create_result = await self._execute_step(
"create_note",
user_a,
lambda: user_a.call_tool(
"nc_notes_create_note",
{
"title": f"Shared Note {unique_id}",
"content": f"Content for workflow test {unique_id}",
"category": "Workflows",
},
),
)
if not create_result.success:
return self._finish(False, error="Failed to create note")
# Extract note ID
note_data = json.loads(create_result.data["result"].content[0].text)
note_id = note_data["id"]
# Step 2: User A shares note with User B
# Note: Sharing files/notes requires using WebDAV path
# Create a file first, then share it
share_result = await self._execute_step(
"share_note",
user_a,
lambda: user_a.call_tool(
"nc_share_create",
{
"path": f"/Notes/{note_data['category']}/{note_data['title']}.txt",
"share_with": user_b.username,
"share_type": 0, # User share
"permissions": 1, # Read-only
},
),
)
if not share_result.success:
logger.warning("Share creation failed, continuing anyway")
# Step 3: User B lists shares (measure propagation)
await self._execute_step(
"list_shared_with_me",
user_b,
lambda: user_b.call_tool("nc_share_list", {"shared_with_me": True}),
)
# Step 4: User B reads the note
await self._execute_step(
"read_shared_note",
user_b,
lambda: user_b.call_tool("nc_notes_get_note", {"note_id": note_id}),
)
# Cleanup: Delete the note
await user_a.call_tool("nc_notes_delete_note", {"note_id": note_id})
return self._finish(success=True)
except Exception as e:
logger.error(f"Note share workflow failed: {e}")
return self._finish(False, error=str(e))
class CollaborativeEditWorkflow(Workflow):
"""
Workflow: Multiple users edit the same note concurrently.
Steps:
1. User A creates a note
2. User A shares note with Users B, C (edit permissions)
3. All users read the note simultaneously
4. All users update the note simultaneously (test concurrent edits)
5. User A verifies final state
"""
def __init__(self):
super().__init__("collaborative_edit")
async def execute(self, users: list[UserSessionWrapper]) -> WorkflowResult:
"""Execute collaborative editing workflow."""
self.start_time = time.time()
if len(users) < 2:
return self._finish(False, error="Requires at least 2 users")
owner = users[0]
collaborators = users[1:]
unique_id = uuid.uuid4().hex[:8]
try:
# Step 1: Owner creates note
create_result = await self._execute_step(
"create_note",
owner,
lambda: owner.call_tool(
"nc_notes_create_note",
{
"title": f"Collab Note {unique_id}",
"content": f"Initial content {unique_id}",
"category": "Collaboration",
},
),
)
if not create_result.success:
return self._finish(False, error="Failed to create note")
note_data = json.loads(create_result.data["result"].content[0].text)
note_id = note_data["id"]
# Step 2: Read note concurrently by all users
read_tasks = []
for i, user in enumerate(users):
read_tasks.append(
self._execute_step(
f"concurrent_read_{i}",
user,
lambda uid=note_id: user.call_tool(
"nc_notes_get_note", {"note_id": uid}
),
)
)
await asyncio.gather(*read_tasks)
# Step 3: Append content concurrently by all collaborators
append_tasks = []
for i, user in enumerate(collaborators):
append_tasks.append(
self._execute_step(
f"concurrent_append_{i}",
user,
lambda _=i, u=user: u.call_tool(
"nc_notes_append_content",
{
"note_id": note_id,
"content": f"Addition from {u.username} at {time.time()}",
},
),
)
)
await asyncio.gather(*append_tasks)
# Step 4: Owner verifies final state
await self._execute_step(
"verify_final_state",
owner,
lambda: owner.call_tool("nc_notes_get_note", {"note_id": note_id}),
)
# Cleanup
await owner.call_tool("nc_notes_delete_note", {"note_id": note_id})
return self._finish(success=True)
except Exception as e:
logger.error(f"Collaborative edit workflow failed: {e}")
return self._finish(False, error=str(e))
class FileShareAndDownloadWorkflow(Workflow):
"""
Workflow: User A uploads a file, shares it with User B, who then downloads it.
Steps:
1. User A creates a file via WebDAV
2. User A shares the file with User B (read-only)
3. User B lists their shares
4. User B reads/downloads the file
"""
def __init__(self):
super().__init__("file_share_download")
async def execute(self, users: list[UserSessionWrapper]) -> WorkflowResult:
"""Execute file sharing workflow."""
self.start_time = time.time()
if len(users) < 2:
return self._finish(False, error="Requires at least 2 users")
user_a, user_b = users[0], users[1]
unique_id = uuid.uuid4().hex[:8]
file_path = f"/LoadTest_{unique_id}.txt"
try:
# Step 1: User A creates a file
content = f"Test file content {unique_id}\nCreated for workflow testing"
create_result = await self._execute_step(
"create_file",
user_a,
lambda: user_a.call_tool(
"nc_webdav_put_file",
{
"path": file_path,
"content": content,
"content_type": "text/plain",
},
),
)
if not create_result.success:
return self._finish(False, error="Failed to create file")
# Step 2: User A shares file with User B
share_result = await self._execute_step(
"share_file",
user_a,
lambda: user_a.call_tool(
"nc_share_create",
{
"path": file_path,
"share_with": user_b.username,
"share_type": 0,
"permissions": 1, # Read-only
},
),
)
if not share_result.success:
logger.warning("File share failed, continuing")
# Step 3: User B lists shared files
_ = await self._execute_step(
"list_shares",
user_b,
lambda: user_b.call_tool("nc_share_list", {"shared_with_me": True}),
)
# Step 4: User B downloads the file
_ = await self._execute_step(
"download_file",
user_b,
lambda: user_b.call_tool("nc_webdav_get_file", {"path": file_path}),
)
# Cleanup
await user_a.call_tool("nc_webdav_delete", {"path": file_path})
return self._finish(success=True)
except Exception as e:
logger.error(f"File share workflow failed: {e}")
return self._finish(False, error=str(e))
class MixedOAuthWorkload:
"""
Mixed workload combining baseline operations and coordinated workflows.
Distribution:
- 50% Baseline operations (individual user CRUD)
- 30% Note sharing workflows
- 15% Collaborative editing workflows
- 5% File sharing workflows
"""
def __init__(self, users: list[UserSessionWrapper]):
self.users = users
self.workflows = {
"note_share": NoteShareWorkflow(),
"collaborative_edit": CollaborativeEditWorkflow(),
"file_share": FileShareAndDownloadWorkflow(),
}
async def run_operation(self) -> WorkflowResult | dict[str, Any]:
"""
Execute one random operation (baseline or workflow).
Returns:
WorkflowResult for workflows, dict for baseline operations
"""
rand = random.random()
# 50% baseline operations (single-user)
if rand < 0.50:
return await self._run_baseline_operation()
# 30% note sharing
elif rand < 0.80:
users = random.sample(self.users, min(2, len(self.users)))
return await self.workflows["note_share"].execute(users)
# 15% collaborative editing
elif rand < 0.95:
users = random.sample(self.users, min(len(self.users), 3))
return await self.workflows["collaborative_edit"].execute(users)
# 5% file sharing
else:
users = random.sample(self.users, min(2, len(self.users)))
return await self.workflows["file_share"].execute(users)
async def _run_baseline_operation(self) -> dict[str, Any]:
"""Run a baseline single-user operation."""
user = random.choice(self.users)
operations = [
(
"search_notes",
lambda: user.call_tool("nc_notes_search_notes", {"query": ""}),
),
("list_files", lambda: user.call_tool("nc_webdav_list", {"path": "/"})),
("get_capabilities", lambda: user.read_resource("nc://capabilities")),
]
op_name, operation = random.choice(operations)
start = time.time()
try:
await operation()
duration = time.time() - start
return {
"type": "baseline",
"operation": op_name,
"user": user.username,
"success": True,
"duration": duration,
}
except Exception as e:
duration = time.time() - start
return {
"type": "baseline",
"operation": op_name,
"user": user.username,
"success": False,
"duration": duration,
"error": str(e),
}