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

docs: remove old docs

Browse Source
This commit is contained in:
Chris Coutinho
2025-10-18 17:07:55 +02:00
parent 056b6fc9d6
commit 644c59bf78
1 changed files with 0 additions and 712 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/load/INTEGRATION_GUIDE.md
-712
View File
@@ -1,712 +0,0 @@
# 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.