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

test: Fix oauth interactive browser tests

Browse Source
This commit is contained in:
Chris Coutinho
2025-10-14 01:23:32 +02:00
parent e42cabb6ed
commit b26ff4f9bc
2 changed files with 174 additions and 121 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/conftest.py
+59 -8
View File
@@ -135,6 +135,49 @@ async def nc_mcp_client() -> AsyncGenerator[ClientSession, Any]:
logger.warning(f"Error closing streamable HTTP client: {e}") logger.warning(f"Error closing streamable HTTP client: {e}")
@pytest.fixture(scope="session")
async def nc_mcp_oauth_client() -> AsyncGenerator[ClientSession, Any]:
"""
Fixture to create an MCP client session for OAuth integration tests using streamable-http.
Connects to the OAuth-enabled MCP server on port 8001.
"""
logger.info("Creating Streamable HTTP client for OAuth MCP server")
streamable_context = streamablehttp_client("http://127.0.0.1:8001/mcp")
session_context = None
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")
yield session
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}")
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}")
@pytest.fixture @pytest.fixture
async def temporary_note(nc_client: NextcloudClient): async def temporary_note(nc_client: NextcloudClient):
""" """
@@ -613,29 +656,37 @@ async def interactive_oauth_token() -> str:
pass pass
def do_GET(self): def do_GET(self):
if self.path.startswith("/shutdown"): # Ignore subsequent requests if we already have a code
# (this is a session-scoped fixture, so only process the first auth code)
if auth_state["code"] is not None:
self.send_response(200) self.send_response(200)
self.send_header("Content-type", "text/html") self.send_header("Content-type", "text/html")
self.end_headers() self.end_headers()
self.wfile.write( self.wfile.write(
b"<html><body><h1>Server shutting down...</h1></body></html>" b"<html><body><h1>Authentication already completed</h1></body></html>"
) )
threading.Thread(target=httpd.shutdown).start()
return return
# Parse the callback request
parsed_path = urlparse(self.path) parsed_path = urlparse(self.path)
query = parse_qs(parsed_path.query) query = parse_qs(parsed_path.query)
code = query.get("code", [None])[0] code = query.get("code", [None])[0]
# Only process if we have a valid code
if code:
auth_state["code"] = code auth_state["code"] = code
logger.info( logger.info(f"OAuth callback received. Code: {code[:20]}...")
f"OAuth callback received. Code: {code[:20] if code else 'None'}..."
)
self.send_response(200) self.send_response(200)
self.send_header("Content-type", "text/html") self.send_header("Content-type", "text/html")
self.end_headers() self.end_headers()
self.wfile.write( self.wfile.write(
b"<html><body><h1>Authentication successful!</h1><p>You can close this window.</p></body></html>" b"<html><body><h1>Authentication successful!</h1><p>You can close this window.</p></body></html>"
) )
else:
# Ignore requests without a code (e.g., favicon requests)
logger.debug(f"Ignoring request without auth code: {self.path}")
self.send_response(404)
self.end_headers()
httpd = HTTPServer(("localhost", 8081), OAuthCallbackHandler) httpd = HTTPServer(("localhost", 8081), OAuthCallbackHandler)
server_thread = threading.Thread(target=httpd.serve_forever) server_thread = threading.Thread(target=httpd.serve_forever)
@@ -704,9 +755,9 @@ async def interactive_oauth_token() -> str:
access_token = token_data.get("access_token") access_token = token_data.get("access_token")
# Shut down the server # Shut down the server
# Call shutdown directly instead of via HTTP to avoid race conditions
await http_client.get("http://localhost:8081/shutdown")
if httpd: if httpd:
httpd.shutdown()
httpd.server_close() httpd.server_close()
if server_thread: if server_thread:
server_thread.join(timeout=1) server_thread.join(timeout=1)
tests/integration/test_oauth.py
+24 -22
View File
@@ -1,9 +1,12 @@
"""Integration tests for OAuth authentication.""" """Integration tests for OAuth authentication."""
import logging import logging
import os
import pytest import pytest
from httpx import HTTPStatusError
from nextcloud_mcp_server.auth import BearerAuth
from nextcloud_mcp_server.client import NextcloudClient from nextcloud_mcp_server.client import NextcloudClient
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
@@ -11,10 +14,10 @@ logger = logging.getLogger(__name__)
pytestmark = [pytest.mark.integration, pytest.mark.oauth] pytestmark = [pytest.mark.integration, pytest.mark.oauth]
class TestOAuthClient: # OAuth Client Tests
"""Test OAuth-authenticated NextcloudClient."""
async def test_oauth_client_capabilities(self, nc_oauth_client: NextcloudClient):
async def test_oauth_client_capabilities(nc_oauth_client: NextcloudClient):
"""Test that OAuth client can fetch capabilities.""" """Test that OAuth client can fetch capabilities."""
capabilities = await nc_oauth_client.capabilities() capabilities = await nc_oauth_client.capabilities()
@@ -24,14 +27,16 @@ class TestOAuthClient:
f"OAuth client successfully fetched capabilities: {capabilities.get('ocs').get('meta')}" f"OAuth client successfully fetched capabilities: {capabilities.get('ocs').get('meta')}"
) )
async def test_oauth_client_notes_list(self, nc_oauth_client: NextcloudClient):
async def test_oauth_client_notes_list(nc_oauth_client: NextcloudClient):
"""Test that OAuth client can list notes.""" """Test that OAuth client can list notes."""
notes = await nc_oauth_client.notes.get_all_notes() notes = await nc_oauth_client.notes.get_all_notes()
assert isinstance(notes, list) assert isinstance(notes, list)
logger.info(f"OAuth client successfully listed {len(notes)} notes") logger.info(f"OAuth client successfully listed {len(notes)} notes")
async def test_oauth_client_create_note(self, nc_oauth_client: NextcloudClient):
async def test_oauth_client_create_note(nc_oauth_client: NextcloudClient):
"""Test that OAuth client can create and delete a note.""" """Test that OAuth client can create and delete a note."""
# Create note # Create note
note_title = "OAuth Test Note" note_title = "OAuth Test Note"
@@ -57,15 +62,15 @@ class TestOAuthClient:
raise raise
class TestOAuthTokenValidation: # OAuth Token Validation Tests
"""Test OAuth token validation and bearer auth."""
async def test_token_in_request_headers( async def test_token_in_request_headers(
self, nc_oauth_client: NextcloudClient, oauth_token: str nc_oauth_client: NextcloudClient, interactive_oauth_token: str
): ):
"""Verify that bearer token is being used in requests.""" """Verify that bearer token is being used in requests."""
# The client should be using BearerAuth # The client should be using BearerAuth
assert nc_oauth_client._auth is not None assert nc_oauth_client._client.auth is not None
# Make a request and verify it works # Make a request and verify it works
capabilities = await nc_oauth_client.capabilities() capabilities = await nc_oauth_client.capabilities()
@@ -73,12 +78,9 @@ class TestOAuthTokenValidation:
logger.info("OAuth bearer token is correctly included in requests") logger.info("OAuth bearer token is correctly included in requests")
async def test_invalid_token_fails(self):
async def test_invalid_token_fails():
"""Test that an invalid token results in authentication failure.""" """Test that an invalid token results in authentication failure."""
import os
from nextcloud_mcp_server.auth import BearerAuth
nextcloud_host = os.getenv("NEXTCLOUD_HOST") nextcloud_host = os.getenv("NEXTCLOUD_HOST")
if not nextcloud_host: if not nextcloud_host:
pytest.skip("NEXTCLOUD_HOST not set") pytest.skip("NEXTCLOUD_HOST not set")
@@ -90,11 +92,10 @@ class TestOAuthTokenValidation:
auth=BearerAuth("invalid_token_12345"), auth=BearerAuth("invalid_token_12345"),
) )
# Attempt to use the client should fail with 401 # Attempt to use a protected endpoint - should fail with 401
from httpx import HTTPStatusError # Note: capabilities endpoint is public and doesn't require auth
with pytest.raises(HTTPStatusError) as exc_info: with pytest.raises(HTTPStatusError) as exc_info:
await invalid_client.capabilities() await invalid_client.notes.get_all_notes()
assert exc_info.value.response.status_code == 401 assert exc_info.value.response.status_code == 401
@@ -102,10 +103,10 @@ class TestOAuthTokenValidation:
logger.info("Invalid OAuth token correctly rejected") logger.info("Invalid OAuth token correctly rejected")
class TestOAuthMCPIntegration: # OAuth MCP Integration Tests
"""Test OAuth integration with MCP server."""
async def test_mcp_oauth_server_connection(self, nc_mcp_oauth_client):
async def test_mcp_oauth_server_connection(nc_mcp_oauth_client):
"""Test connection to OAuth-enabled MCP server.""" """Test connection to OAuth-enabled MCP server."""
result = await nc_mcp_oauth_client.list_tools() result = await nc_mcp_oauth_client.list_tools()
assert result is not None assert result is not None
@@ -113,7 +114,8 @@ class TestOAuthMCPIntegration:
logger.info(f"OAuth MCP server has {len(result.tools)} tools available") logger.info(f"OAuth MCP server has {len(result.tools)} tools available")
async def test_mcp_oauth_tool_execution(self, nc_mcp_oauth_client):
async def test_mcp_oauth_tool_execution(nc_mcp_oauth_client):
"""Test executing a tool on the OAuth-enabled MCP server.""" """Test executing a tool on the OAuth-enabled MCP server."""
import json import json