Strava MCP Server

"""Tests for the Strava authentication module.""" import asyncio from unittest.mock import AsyncMock, MagicMock, patch import pytest from fastapi import FastAPI from httpx import Response from strava_mcp.auth import StravaAuthenticator, get_strava_refresh_token @pytest.fixture def client_credentials(): """Fixture for client credentials.""" return { "client_id": "test_client_id", "client_secret": "test_client_secret", } @pytest.fixture def mock_token_response(): """Fixture for token response.""" return { "access_token": "test_access_token", "refresh_token": "test_refresh_token", "expires_at": 1609459200, "expires_in": 21600, "token_type": "Bearer", } @pytest.fixture def fastapi_app(): """Fixture for FastAPI app.""" return FastAPI() @pytest.fixture def authenticator(client_credentials, fastapi_app): """Fixture for StravaAuthenticator.""" return StravaAuthenticator( client_id=client_credentials["client_id"], client_secret=client_credentials["client_secret"], app=fastapi_app, ) def test_get_authorization_url(authenticator): """Test getting the authorization URL.""" url = authenticator.get_authorization_url() # Check that the URL contains the expected parameters assert "https://www.strava.com/oauth/authorize" in url assert f"client_id={authenticator.client_id}" in url # URL is encoded, so we need to check the non-encoded parts assert "redirect_uri=http%3A%2F%2F127.0.0.1%3A3008%2Fexchange_token" in url assert "response_type=code" in url assert "scope=" in url def test_setup_routes(authenticator, fastapi_app): """Test setting up routes.""" authenticator.setup_routes(fastapi_app) # Check that the routes were added routes = [route.path for route in fastapi_app.routes] assert authenticator.redirect_path in routes assert "/auth" in routes def test_setup_routes_no_app(authenticator): """Test setting up routes with no app.""" authenticator.app = None with pytest.raises(ValueError, match="No FastAPI app provided"): authenticator.setup_routes() @pytest.mark.asyncio async def test_exchange_token_success(authenticator, mock_token_response): """Test exchanging token successfully.""" # Setup mock with patch("httpx.AsyncClient") as mock_client: mock_response = MagicMock(spec=Response) mock_response.status_code = 200 mock_response.json.return_value = mock_token_response mock_client.return_value.__aenter__.return_value.post.return_value = mock_response # Set up a future to receive the token authenticator.token_future = asyncio.Future() # Call the handler response = await authenticator.exchange_token(code="test_code") # Check response assert response.status_code == 200 assert "Authorization successful" in response.body.decode() # Check token future assert authenticator.token_future.done() assert await authenticator.token_future == "test_refresh_token" # Check token was saved assert authenticator.refresh_token == "test_refresh_token" # Verify correct API call mock_client.return_value.__aenter__.return_value.post.assert_called_once() args, kwargs = mock_client.return_value.__aenter__.return_value.post.call_args assert args[0] == "https://www.strava.com/oauth/token" assert kwargs["data"]["client_id"] == authenticator.client_id assert kwargs["data"]["client_secret"] == authenticator.client_secret assert kwargs["data"]["code"] == "test_code" assert kwargs["data"]["grant_type"] == "authorization_code" @pytest.mark.asyncio async def test_exchange_token_failure(authenticator): """Test exchanging token with failure.""" # Setup mock with patch("httpx.AsyncClient") as mock_client: mock_response = MagicMock(spec=Response) mock_response.status_code = 400 mock_response.text = "Invalid code" mock_client.return_value.__aenter__.return_value.post.return_value = mock_response # Set up a future to receive the token authenticator.token_future = asyncio.Future() # Call the handler response = await authenticator.exchange_token(code="invalid_code") # Check response assert response.status_code == 200 assert "Authorization failed" in response.body.decode() # Check token future assert authenticator.token_future.done() # We expect a specific exception here, so using pytest.raises is appropriate with pytest.raises(Exception): # noqa: B017 await authenticator.token_future @pytest.mark.asyncio async def test_start_auth_flow(authenticator): """Test starting auth flow.""" with patch.object(authenticator, "get_authorization_url", return_value="https://example.com/auth"): response = await authenticator.start_auth_flow() assert response.status_code == 307 assert response.headers["location"] == "https://example.com/auth" @pytest.mark.asyncio async def test_get_refresh_token(authenticator): """Test getting refresh token.""" # Mock the webbrowser.open call with patch("webbrowser.open", return_value=True) as mock_open: with patch.object(authenticator, "get_authorization_url", return_value="https://example.com/auth"): # Set the future result after a delay authenticator.token_future = None # Reset it so a new one is created # Start the token request in background task = asyncio.create_task(authenticator.get_refresh_token()) # Wait a bit and set the result await asyncio.sleep(0.1) # Initialize the token_future before setting result if not authenticator.token_future: authenticator.token_future = asyncio.Future() authenticator.token_future.set_result("test_refresh_token") # Get the result token = await task # Verify assert token == "test_refresh_token" mock_open.assert_called_once_with("https://example.com/auth") @pytest.mark.asyncio async def test_get_refresh_token_no_browser(authenticator): """Test getting refresh token without opening browser.""" with patch("webbrowser.open") as mock_open: with patch.object(authenticator, "get_authorization_url", return_value="https://example.com/auth"): # Set the future result after a delay authenticator.token_future = None # Reset it so a new one is created # Start the token request in background task = asyncio.create_task(authenticator.get_refresh_token(open_browser=False)) # Wait a bit and set the result await asyncio.sleep(0.1) # Initialize the token_future before setting result if not authenticator.token_future: authenticator.token_future = asyncio.Future() authenticator.token_future.set_result("test_refresh_token") # Get the result token = await task # Verify assert token == "test_refresh_token" mock_open.assert_not_called() @pytest.mark.asyncio async def test_get_refresh_token_browser_fails(authenticator): """Test getting refresh token with browser opening failing.""" with patch("webbrowser.open", return_value=False) as mock_open: with patch.object(authenticator, "get_authorization_url", return_value="https://example.com/auth"): # Set the future result after a delay authenticator.token_future = None # Reset it so a new one is created # Start the token request in background task = asyncio.create_task(authenticator.get_refresh_token()) # Wait a bit and set the result await asyncio.sleep(0.1) # Initialize the token_future before setting result if not authenticator.token_future: authenticator.token_future = asyncio.Future() authenticator.token_future.set_result("test_refresh_token") # Get the result token = await task # Verify assert token == "test_refresh_token" mock_open.assert_called_once_with("https://example.com/auth") @pytest.mark.asyncio async def test_get_strava_refresh_token(client_credentials): """Test get_strava_refresh_token function.""" with patch("strava_mcp.auth.StravaAuthenticator") as mock_authenticator_class: # Setup mock mock_authenticator = MagicMock() mock_authenticator.get_refresh_token = AsyncMock(return_value="test_refresh_token") mock_authenticator.setup_routes = MagicMock() mock_authenticator_class.return_value = mock_authenticator # Test without app token = await get_strava_refresh_token(client_credentials["client_id"], client_credentials["client_secret"]) # Verify assert token == "test_refresh_token" mock_authenticator_class.assert_called_once_with( client_credentials["client_id"], client_credentials["client_secret"], None ) mock_authenticator.setup_routes.assert_not_called() # Reset mocks mock_authenticator_class.reset_mock() mock_authenticator.get_refresh_token.reset_mock() mock_authenticator.setup_routes.reset_mock() # Test with app app = FastAPI() token = await get_strava_refresh_token( client_credentials["client_id"], client_credentials["client_secret"], app ) # Verify assert token == "test_refresh_token" mock_authenticator_class.assert_called_once_with( client_credentials["client_id"], client_credentials["client_secret"], app ) mock_authenticator.setup_routes.assert_called_once_with(app)