nextcloud-mcp-server/tests/server/oauth/test_login_elicitation.py

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