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

test: continue working on oauth client

Browse Source
This commit is contained in:
Chris Coutinho
2025-10-13 18:07:48 +02:00
parent 33b962a7fc
commit 2b11718c43
3 changed files with 44 additions and 40 deletions
Download Patch File Download Diff File Expand all files Collapse all files
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/apps/notifications/api/v2/notifications",
"/ocs/v2.php/cloud/capabilities",
headers={"OCS-APIRequest": "true", "Accept": "application/json"},
)
response.raise_for_status()
tests/conftest.py
+19 -24
View File
@@ -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
tests/integration/test_oauth_interactive.py
+24 -15
View File
@@ -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}")