From 2b11718c438953fcb27c67395e75571dedd8dc6a Mon Sep 17 00:00:00 2001 From: Chris Coutinho Date: Mon, 13 Oct 2025 18:07:48 +0200 Subject: [PATCH] test: continue working on oauth client --- nextcloud_mcp_server/client/__init__.py | 2 +- tests/conftest.py | 43 +++++++++------------ tests/integration/test_oauth_interactive.py | 39 ++++++++++++------- 3 files changed, 44 insertions(+), 40 deletions(-) diff --git a/nextcloud_mcp_server/client/__init__.py b/nextcloud_mcp_server/client/__init__.py index 27c1de1..621a379 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/apps/notifications/api/v2/notifications", + "/ocs/v2.php/cloud/capabilities", headers={"OCS-APIRequest": "true", "Accept": "application/json"}, ) response.raise_for_status() diff --git a/tests/conftest.py b/tests/conftest.py index 9a9b294..745c033 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -556,7 +556,9 @@ async def oauth_token() -> str: @pytest.fixture(scope="session") -async def nc_oauth_client(oauth_token: str) -> AsyncGenerator[NextcloudClient, Any]: +async def nc_oauth_client( + interactive_oauth_token: str, +) -> AsyncGenerator[NextcloudClient, Any]: """ Fixture to create a NextcloudClient instance using OAuth authentication. Uses the oauth_token fixture to get an access token. @@ -570,7 +572,7 @@ async def nc_oauth_client(oauth_token: str) -> AsyncGenerator[NextcloudClient, A logger.info(f"Creating OAuth NextcloudClient for user: {username}") client = NextcloudClient.from_token( base_url=nextcloud_host, - token=oauth_token, + token=interactive_oauth_token, username=username, ) @@ -587,19 +589,22 @@ async def nc_oauth_client(oauth_token: str) -> AsyncGenerator[NextcloudClient, A @pytest.fixture(scope="session") -async def nc_mcp_oauth_client_interactive() -> AsyncGenerator[ClientSession, Any]: +async def interactive_oauth_token() -> str: """ - Fixture to create an MCP client session for interactive OAuth integration tests. - Performs an interactive OAuth flow to obtain an access token. + Fixture to obtain an OAuth access token for integration tests. + + This uses the interactive OAuth flow to get a token. """ + import webbrowser from http.server import BaseHTTPRequestHandler, HTTPServer import threading from urllib.parse import urlparse, parse_qs - import time auth_code = None + httpd = None + server_thread = None class OAuthCallbackHandler(BaseHTTPRequestHandler): def do_GET(self): @@ -639,7 +644,6 @@ async def nc_mcp_oauth_client_interactive() -> AsyncGenerator[ClientSession, Any 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, @@ -650,7 +654,6 @@ async def nc_mcp_oauth_client_interactive() -> AsyncGenerator[ClientSession, Any 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) @@ -667,23 +670,15 @@ async def nc_mcp_oauth_client_interactive() -> AsyncGenerator[ClientSession, Any ) 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: - yield session - finally: - # Shut down the server - await http_client.get("http://localhost:8081/shutdown") + # Shut down the server + + await http_client.get("http://localhost:8081/shutdown") + if httpd: + httpd.server_close() + if server_thread: + server_thread.join(timeout=1) + return access_token diff --git a/tests/integration/test_oauth_interactive.py b/tests/integration/test_oauth_interactive.py index 1701993..09f991a 100644 --- a/tests/integration/test_oauth_interactive.py +++ b/tests/integration/test_oauth_interactive.py @@ -12,21 +12,30 @@ 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) + async def test_oauth_client_with_interactive_flow(self, nc_oauth_client): + """Test that OAuth client created via interactive flow can access Nextcloud APIs.""" + # Test 1: Check capabilities + capabilities = await nc_oauth_client.capabilities() + assert capabilities is not None + logger.info("OAuth client (interactive) successfully fetched capabilities") + # Test 2: List notes + notes = await nc_oauth_client.notes.get_all_notes() + assert isinstance(notes, list) logger.info( - f"Successfully executed 'nc_notes_list' tool on OAuth MCP server and got {len(notes_list)} notes." + f"OAuth client (interactive) successfully listed {len(notes)} notes" ) + + # Test 3: Create and delete a note + test_note = await nc_oauth_client.notes.create_note( + title="OAuth Interactive Test Note", + content="This note was created during OAuth interactive testing", + ) + assert test_note is not None + assert test_note.get("id") is not None + note_id = test_note["id"] + logger.info(f"OAuth client (interactive) successfully created note {note_id}") + + # Clean up + await nc_oauth_client.notes.delete_note(note_id=note_id) + logger.info(f"OAuth client (interactive) successfully deleted note {note_id}")