From 33b962a7fc41e78562a73091812bedadfcb371b9 Mon Sep 17 00:00:00 2001 From: Chris Coutinho Date: Mon, 13 Oct 2025 18:07:47 +0200 Subject: [PATCH] test: Setup interactive browser test --- .gitignore | 1 + nextcloud_mcp_server/app.py | 20 +-- .../auth/client_registration.py | 2 +- nextcloud_mcp_server/auth/context_helper.py | 1 + nextcloud_mcp_server/client/__init__.py | 2 +- pyproject.toml | 3 +- tests/conftest.py | 135 +++++++++++++----- tests/integration/test_oauth.py | 29 ++-- tests/integration/test_oauth_interactive.py | 32 +++++ 9 files changed, 162 insertions(+), 63 deletions(-) create mode 100644 tests/integration/test_oauth_interactive.py diff --git a/.gitignore b/.gitignore index 85bf658..fcc442a 100644 --- a/.gitignore +++ b/.gitignore @@ -4,3 +4,4 @@ __pycache__/ *.env .env.local .env.*.local +.nextcloud_oauth_test_client.json diff --git a/nextcloud_mcp_server/app.py b/nextcloud_mcp_server/app.py index f63ad08..c694bef 100644 --- a/nextcloud_mcp_server/app.py +++ b/nextcloud_mcp_server/app.py @@ -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) diff --git a/nextcloud_mcp_server/auth/client_registration.py b/nextcloud_mcp_server/auth/client_registration.py index 7ae9d28..2e2943d 100644 --- a/nextcloud_mcp_server/auth/client_registration.py +++ b/nextcloud_mcp_server/auth/client_registration.py @@ -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. diff --git a/nextcloud_mcp_server/auth/context_helper.py b/nextcloud_mcp_server/auth/context_helper.py index 1c160ce..c081f84 100644 --- a/nextcloud_mcp_server/auth/context_helper.py +++ b/nextcloud_mcp_server/auth/context_helper.py @@ -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 diff --git a/nextcloud_mcp_server/client/__init__.py b/nextcloud_mcp_server/client/__init__.py index 621a379..27c1de1 100644 --- a/nextcloud_mcp_server/client/__init__.py +++ b/nextcloud_mcp_server/client/__init__.py @@ -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() diff --git a/pyproject.toml b/pyproject.toml index 3cc71d2..bc6e08c 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -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] diff --git a/tests/conftest.py b/tests/conftest.py index 0d6a3f1..9a9b294 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -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"

Server shutting down...

" + ) + 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"

Authentication successful!

You can close this window.

" + ) + + 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") diff --git a/tests/integration/test_oauth.py b/tests/integration/test_oauth.py index 0cc35a0..fc8bbd6 100644 --- a/tests/integration/test_oauth.py +++ b/tests/integration/test_oauth.py @@ -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." + ) diff --git a/tests/integration/test_oauth_interactive.py b/tests/integration/test_oauth_interactive.py new file mode 100644 index 0000000..1701993 --- /dev/null +++ b/tests/integration/test_oauth_interactive.py @@ -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." + )