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

test: Setup interactive browser test

Browse Source
This commit is contained in:
Chris Coutinho
2025-10-13 18:07:47 +02:00
parent 4d7e4b9a4b
commit 33b962a7fc
9 changed files with 162 additions and 63 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.gitignore
+1
View File
@@ -4,3 +4,4 @@ __pycache__/
*.env
.env.local
.env.*.local
.nextcloud_oauth_test_client.json
nextcloud_mcp_server/app.py
+11 -9
View File
@@ -274,18 +274,20 @@ def get_app(transport: str = "sse", enabled_apps: list[str] | None = None):
# Determine authentication mode
oauth_enabled = is_oauth_mode()
# WARNING: This is a synchronous function but OAuth setup requires async
# For now, OAuth configuration will be handled differently
# We'll need to restructure this or use a factory pattern
if oauth_enabled:
logger.info("Configuring MCP server for OAuth mode")
logger.warning(
"OAuth mode requires async initialization - use factory pattern or separate setup"
# Asynchronously get the OAuth configuration
import asyncio
nextcloud_host, token_verifier, auth_settings = asyncio.run(
setup_oauth_config()
)
mcp = FastMCP(
"Nextcloud MCP",
lifespan=app_lifespan_oauth,
token_verifier=token_verifier,
auth=auth_settings,
)
# For now, fall back to a simplified OAuth setup
# TODO: This needs to be restructured to support async initialization
mcp = FastMCP("Nextcloud MCP", lifespan=app_lifespan_oauth)
else:
logger.info("Configuring MCP server for BasicAuth mode")
mcp = FastMCP("Nextcloud MCP", lifespan=app_lifespan_basic)
nextcloud_mcp_server/auth/client_registration.py
+1 -1
View File
@@ -211,7 +211,7 @@ async def load_or_register_client(
storage_path: str | Path,
client_name: str = "Nextcloud MCP Server",
redirect_uris: list[str] | None = None,
force_register: bool = False,
force_register: bool = True,
) -> ClientInfo:
"""
Load client from storage or register a new one if not found/expired.
nextcloud_mcp_server/auth/context_helper.py
+1
View File
@@ -30,6 +30,7 @@ def get_client_from_context(ctx: Context, base_url: str) -> NextcloudClient:
ValueError: If username cannot be extracted from token
"""
try:
logger.info(f"Inspecting session object: {dir(ctx.request_context.session)}")
# Get AccessToken from MCP session (set by TokenVerifier)
access_token: AccessToken = ctx.request_context.session.access_token
nextcloud_mcp_server/client/__init__.py
+1 -1
View File
@@ -104,7 +104,7 @@ class NextcloudClient:
async def capabilities(self):
response = await self._client.get(
"/ocs/v2.php/cloud/capabilities",
"/ocs/v2.php/apps/notifications/api/v2/notifications",
headers={"OCS-APIRequest": "true", "Accept": "application/json"},
)
response.raise_for_status()
pyproject.toml
+2 -1
View File
@@ -25,7 +25,8 @@ log_cli = 1
log_cli_level = "INFO"
log_level = "INFO"
markers = [
"integration: marks tests as slow (deselect with '-m \"not slow\"')"
"integration: marks tests as slow (deselect with '-m \"not slow\"')",
"interactive: marks tests as interactive (deselect with '-m \"not interactive\"')"
]
[tool.commitizen]
tests/conftest.py
+96 -39
View File
@@ -454,7 +454,7 @@ async def temporary_board_with_card(
async def get_oauth_token(nextcloud_url: str, username: str, password: str) -> str:
"""
Get an OAuth access token from Nextcloud OIDC using Resource Owner Password flow.
Get an OAuth access token from Nextcloud OIDC using Client Credentials flow.
This is a helper function for testing only - it bypasses the normal OAuth flow
to directly obtain a token for automated testing.
@@ -501,16 +501,13 @@ async def get_oauth_token(nextcloud_url: str, username: str, password: str) -> s
redirect_uris=["http://localhost:8000/oauth/callback"],
)
# Use client credentials to get a token via password grant
# Note: This requires the OIDC app to support Resource Owner Password flow
# Use client credentials to get a token via client_credentials grant
token_response = await http_client.post(
token_endpoint,
data={
"grant_type": "password",
"grant_type": "client_credentials",
"client_id": client_info.client_id,
"client_secret": client_info.client_secret,
"username": username,
"password": password,
"scope": "openid profile email",
},
)
@@ -590,43 +587,103 @@ async def nc_oauth_client(oauth_token: str) -> AsyncGenerator[NextcloudClient, A
@pytest.fixture(scope="session")
async def nc_mcp_oauth_client() -> AsyncGenerator[ClientSession, Any]:
async def nc_mcp_oauth_client_interactive() -> AsyncGenerator[ClientSession, Any]:
"""
Fixture to create an MCP client session for OAuth integration tests.
Connects to the OAuth-enabled MCP server on port 8001.
Fixture to create an MCP client session for interactive OAuth integration tests.
Performs an interactive OAuth flow to obtain an access token.
"""
logger.info("Creating Streamable HTTP client for OAuth MCP server")
streamable_context = streamablehttp_client("http://127.0.0.1:8001/mcp")
session_context = None
import webbrowser
from http.server import BaseHTTPRequestHandler, HTTPServer
import threading
from urllib.parse import urlparse, parse_qs
try:
read_stream, write_stream, _ = await streamable_context.__aenter__()
session_context = ClientSession(read_stream, write_stream)
session = await session_context.__aenter__()
await session.initialize()
logger.info("OAuth MCP client session initialized successfully")
import time
yield session
auth_code = None
finally:
# Clean up in reverse order, ignoring task scope issues
if session_context is not None:
try:
await session_context.__aexit__(None, None, None)
except RuntimeError as e:
if "cancel scope" in str(e):
logger.debug(f"Ignoring cancel scope teardown issue: {e}")
else:
logger.warning(f"Error closing OAuth session: {e}")
except Exception as e:
logger.warning(f"Error closing OAuth session: {e}")
class OAuthCallbackHandler(BaseHTTPRequestHandler):
def do_GET(self):
nonlocal auth_code
if self.path.startswith("/shutdown"):
self.send_response(200)
self.send_header("Content-type", "text/html")
self.end_headers()
self.wfile.write(
b"<html><body><h1>Server shutting down...</h1></body></html>"
)
threading.Thread(target=httpd.shutdown).start()
return
parsed_path = urlparse(self.path)
query = parse_qs(parsed_path.query)
auth_code = query.get("code", [None])[0]
self.send_response(200)
self.send_header("Content-type", "text/html")
self.end_headers()
self.wfile.write(
b"<html><body><h1>Authentication successful!</h1><p>You can close this window.</p></body></html>"
)
httpd = HTTPServer(("localhost", 8081), OAuthCallbackHandler)
server_thread = threading.Thread(target=httpd.serve_forever)
server_thread.daemon = True
server_thread.start()
from nextcloud_mcp_server.auth.client_registration import load_or_register_client
nextcloud_host = os.getenv("NEXTCLOUD_HOST")
async with httpx.AsyncClient() as http_client:
discovery_url = f"{nextcloud_host}/.well-known/openid-configuration"
discovery_response = await http_client.get(discovery_url)
oidc_config = discovery_response.json()
token_endpoint = oidc_config.get("token_endpoint")
registration_endpoint = oidc_config.get("registration_endpoint")
authorization_endpoint = oidc_config.get("authorization_endpoint")
client_info = await load_or_register_client(
nextcloud_url=nextcloud_host,
registration_endpoint=registration_endpoint,
storage_path=".nextcloud_oauth_test_client.json",
redirect_uris=["http://localhost:8081"],
force_register=True,
)
auth_url = f"{authorization_endpoint}?response_type=code&client_id={client_info.client_id}&redirect_uri=http://localhost:8081&scope=openid%20profile%20email"
webbrowser.open(auth_url)
while not auth_code:
logger.info("Sleeping until auth_code available")
time.sleep(1)
token_response = await http_client.post(
token_endpoint,
data={
"grant_type": "authorization_code",
"code": auth_code,
"redirect_uri": "http://localhost:8081",
"client_id": client_info.client_id,
"client_secret": client_info.client_secret,
},
)
logger.info(f"Token response: {token_response.text}")
# Shut down the server
token_data = token_response.json()
logger.info(f"Token data: {token_data}")
access_token = token_data.get("access_token")
headers = {"Authorization": f"Bearer {access_token}"}
logger.info(f"Headers: {headers}")
async with streamablehttp_client("http://127.0.0.1:8001/mcp", headers=headers) as (
read_stream,
write_stream,
_,
):
async with ClientSession(read_stream, write_stream) as session:
await session.initialize()
try:
await streamable_context.__aexit__(None, None, None)
except RuntimeError as e:
if "cancel scope" in str(e):
logger.debug(f"Ignoring cancel scope teardown issue: {e}")
else:
logger.warning(f"Error closing OAuth streamable HTTP client: {e}")
except Exception as e:
logger.warning(f"Error closing OAuth streamable HTTP client: {e}")
yield session
finally:
# Shut down the server
await http_client.get("http://localhost:8081/shutdown")
tests/integration/test_oauth.py
+17 -12
View File
@@ -105,22 +105,27 @@ class TestOAuthTokenValidation:
class TestOAuthMCPIntegration:
"""Test OAuth integration with MCP server."""
@pytest.mark.skip(
reason="OAuth MCP server integration requires full OAuth flow implementation"
)
async def test_mcp_oauth_server_connection(self, nc_mcp_oauth_client):
"""Test connection to OAuth-enabled MCP server."""
# This test is currently skipped because the OAuth MCP server
# requires the full OAuth authorization flow to be implemented
# in the MCP SDK and app.py
# Once implemented, this test should:
# 1. Connect to the OAuth MCP server
# 2. Verify tools are available
# 3. Call a tool and verify it works with OAuth auth
result = await nc_mcp_oauth_client.list_tools()
assert result is not None
assert len(result.tools) > 0
logger.info(f"OAuth MCP server has {len(result.tools)} tools available")
async def test_mcp_oauth_tool_execution(self, nc_mcp_oauth_client):
"""Test executing a tool on the OAuth-enabled MCP server."""
import json
# Example: Execute the 'nc_tables_list_tables' tool
result = await nc_mcp_oauth_client.call_tool("nc_tables_list_tables")
assert result.isError is False, f"Tool execution failed: {result.content}"
assert result.content is not None
notes_list = json.loads(result.content[0].text)
assert isinstance(notes_list, list)
logger.info(
f"Successfully executed 'nc_tables_list_tables' tool on OAuth MCP server and got {len(notes_list)} notes."
)
tests/integration/test_oauth_interactive.py
+32
View File
@@ -0,0 +1,32 @@
"""Interactive integration tests for OAuth authentication."""
import logging
import pytest
logger = logging.getLogger(__name__)
pytestmark = [pytest.mark.integration, pytest.mark.interactive]
class TestOAuthInteractive:
"""Test interactive OAuth authentication."""
async def test_mcp_oauth_tool_execution_interactive(
self, nc_mcp_oauth_client_interactive
):
"""Test executing a tool on the OAuth-enabled MCP server with an interactive token."""
# Example: Execute the 'nc_notes_list' tool
result = await nc_mcp_oauth_client_interactive.call_tool("nc_tables_list")
assert result.isError is False, f"Tool execution failed: {result.content}"
assert result.content is not None
import json
notes_list = json.loads(result.content[0].text)
assert isinstance(notes_list, list)
logger.info(
f"Successfully executed 'nc_notes_list' tool on OAuth MCP server and got {len(notes_list)} notes."
)