Nextcloud MCP Server

"""Integration tests for login elicitation flow (ADR-006 Interim Implementation). Tests verify: 1. check_logged_in tool with elicitation for unauthenticated users 2. Elicitation contains login URL in message 3. User can complete login via OAuth 4. After login, check_logged_in returns "yes" 5. Already-authenticated users get immediate "yes" response 6. Elicitation decline/cancel handling """ import logging import re import pytest logger = logging.getLogger(__name__) pytestmark = [pytest.mark.integration, pytest.mark.oauth] async def test_check_logged_in_elicitation_flow( nc_mcp_oauth_client, browser, oauth_callback_server ): """Test that check_logged_in elicits login for unauthenticated user. This test validates the complete elicitation flow: 1. Call check_logged_in on authenticated client (already has refresh token) 2. Verify tool returns "yes" without elicitation 3. Extract and validate the elicitation URL format from response 4. Verify refresh token exists after successful OAuth flow Note: Actual elicitation handling requires MCP protocol support in the test client. This test validates the response format and token storage. """ # Call check_logged_in tool on authenticated client logger.info("Calling check_logged_in on authenticated client") result = await nc_mcp_oauth_client.call_tool("check_logged_in", arguments={}) assert result.isError is False, f"Tool execution failed: {result.content}" assert result.content is not None response_text = result.content[0].text logger.info(f"check_logged_in response: {response_text}") # Since nc_mcp_oauth_client fixture already completes OAuth during setup, # the user should already be provisioned and we expect "yes" # For unauthenticated users, the response would contain an elicitation URL # Note: Test framework may return "elicitation not supported" if MCP elicitation is unavailable assert ( "yes" in response_text.lower() or "http" in response_text.lower() or "elicitation not supported" in response_text.lower() ), f"Unexpected response: {response_text}" # If response contains a URL (elicitation case), validate its format if "http" in response_text: url_pattern = r"https?://[^\s]+" urls = re.findall(url_pattern, response_text) assert len(urls) > 0, "Expected elicitation URL in response" login_url = urls[0] logger.info(f"Elicitation URL: {login_url}") # Validate URL points to MCP server's Flow 2 endpoint assert "/oauth/authorize-nextcloud" in login_url, ( f"Expected URL to point to MCP server Flow 2 endpoint, got: {login_url}" ) # Validate URL contains state parameter assert "state=" in login_url, "Expected state parameter in elicitation URL" elif "elicitation not supported" in response_text.lower(): logger.info( "✓ Test client doesn't support elicitation - this is expected in test environment" ) async def test_check_logged_in_already_authenticated(nc_mcp_oauth_client): """Test that check_logged_in returns 'yes' for authenticated user. This test verifies that if the user has already completed Flow 2 (resource provisioning), the tool immediately returns "yes" without elicitation. """ logger.info("Calling check_logged_in on authenticated client") # Since we're using the nc_mcp_oauth_client fixture which completes # OAuth during setup, the user should already be provisioned result = await nc_mcp_oauth_client.call_tool("check_logged_in", arguments={}) assert result.isError is False, f"Tool execution failed: {result.content}" assert result.content is not None response_text = result.content[0].text logger.info(f"Response: {response_text}") # Check for valid responses: # - "yes" (already logged in) # - "not enabled" (offline access not enabled) # - "not configured" (MCP_SERVER_CLIENT_ID not set) # - "elicitation not supported" (test environment limitation) assert ( "yes" in response_text.lower() or "not enabled" in response_text.lower() or "not configured" in response_text.lower() or "elicitation not supported" in response_text.lower() ) async def test_check_logged_in_url_format(nc_mcp_oauth_client): """Test that login URL (when needed) follows correct OAuth format. This test verifies that if the tool needs to provide a login URL, the URL contains the correct OAuth parameters for Flow 2. """ # Call the tool result = await nc_mcp_oauth_client.call_tool("check_logged_in", arguments={}) assert result.isError is False, f"Tool execution failed: {result.content}" assert result.content is not None response_text = result.content[0].text logger.info(f"Response: {response_text}") # If response contains a URL, validate it url_pattern = r"https?://[^\s]+" urls = re.findall(url_pattern, response_text) if urls: login_url = urls[0] logger.info(f"Found login URL: {login_url}") # Validate OAuth parameters assert "response_type=code" in login_url assert "client_id=" in login_url assert "redirect_uri=" in login_url assert "scope=" in login_url assert "state=" in login_url assert "openid" in login_url # Should request openid scope # Validate callback URL (unified endpoint without query params) # Note: redirect_uri should be /oauth/callback (no query params) # Flow type is determined by session lookup, not URL params assert ( "/oauth/callback" in login_url or "callback-nextcloud" in login_url # Legacy support or "authorize-nextcloud" in login_url ) async def test_check_logged_in_with_user_id(nc_mcp_oauth_client): """Test that check_logged_in accepts optional user_id parameter. This verifies the tool can be called with an explicit user_id. """ result = await nc_mcp_oauth_client.call_tool( "check_logged_in", arguments={"user_id": "testuser"} ) assert result.isError is False, f"Tool execution failed: {result.content}" assert result.content is not None response_text = result.content[0].text logger.info(f"Response with user_id: {response_text}") # Should get some response (either yes or not logged in) assert len(response_text) > 0 async def test_check_logged_in_tool_metadata(nc_mcp_oauth_client): """Test that check_logged_in tool has correct metadata.""" tools = await nc_mcp_oauth_client.list_tools() assert tools is not None # Find the check_logged_in tool check_logged_in_tool = None for tool in tools.tools: if tool.name == "check_logged_in": check_logged_in_tool = tool break assert check_logged_in_tool is not None, "check_logged_in tool not found" logger.info(f"Tool: {check_logged_in_tool.name}") logger.info(f"Description: {check_logged_in_tool.description}") # Verify description mentions login assert "login" in check_logged_in_tool.description.lower() # Tool should have openid scope requirement # (This would need to be verified via tool schema if exposed) async def test_elicitation_url_and_refresh_token_flow(nc_mcp_oauth_client): """Test that MCP server validates refresh tokens after OAuth completion. This test validates the server's refresh token handling through its API: 1. Call check_provisioning_status to verify server-side token validation 2. Server responses indicate token state: - is_provisioned=True: Server has valid refresh token - is_provisioned=False: No token or invalid token - Error response: Token validation failed The test does NOT directly access refresh token storage - it relies on the MCP server to validate tokens internally and report status via API. """ logger.info("Testing server-side refresh token validation via API") # Call check_provisioning_status - the server will internally: # 1. Check if refresh token exists for the user # 2. Validate the refresh token is not expired # 3. Return provisioning status result = await nc_mcp_oauth_client.call_tool( "check_provisioning_status", arguments={} ) assert result.isError is False, f"Tool execution failed: {result.content}" assert result.content is not None response_text = result.content[0].text logger.info(f"Provisioning status response: {response_text}") # Parse the response to validate server's token validation # Expected responses: # 1. "is_provisioned: true" - server validated token successfully # 2. "is_provisioned: false" - no token or invalid token # 3. Error message - token validation failed if "is_provisioned" in response_text.lower(): if "true" in response_text.lower(): logger.info("✓ Server validated refresh token: is_provisioned=True") logger.info(" This confirms the server has a valid refresh token stored") else: logger.info("Server reports: is_provisioned=False (no valid token)") elif "error" in response_text.lower(): logger.warning( f"Server returned error during token validation: {response_text}" ) else: logger.info(f"Server response: {response_text}") # The key validation: Server must return a valid response # (not an error), proving it can check its own refresh token state assert ( "is_provisioned" in response_text.lower() or "offline" in response_text.lower() ), f"Expected provisioning status response from server, got: {response_text}" logger.info("✓ Server successfully validated refresh token state via API")