test: Fix oauth2 token extract from starlette requests

nextcloud_mcp_server/auth/context_helper.py

@@ -30,9 +30,19 @@ def get_client_from_context(ctx: Context, base_url: str) -> NextcloudClient:
         ValueError: If username cannot be extracted from token
     """
     try:
-        logger.info(f"Inspecting session object: {dir(ctx.request_context.session)}")
+        # In Starlette with FastMCP OAuth, the authenticated user info is stored in request.user
-        # Get AccessToken from MCP session (set by TokenVerifier)
+        # The FastMCP auth middleware sets request.user to an AuthenticatedUser object
-        access_token: AccessToken = ctx.request_context.session.access_token
+        # which contains the access_token
+        if hasattr(ctx.request_context.request, "user") and hasattr(
+            ctx.request_context.request.user, "access_token"
+        ):
+            access_token: AccessToken = ctx.request_context.request.user.access_token
+            logger.debug("Retrieved access token from request.user for OAuth request")
+        else:
+            logger.error(
+                "OAuth authentication failed: No access token found in request"
+            )
+            raise AttributeError("No access token found in OAuth request context")
 
         # Extract username from resource field (RFC 8707)
         # We stored the username here during token verification

tests/conftest.py

@@ -136,13 +136,20 @@ async def nc_mcp_client() -> AsyncGenerator[ClientSession, Any]:
 
 
 @pytest.fixture(scope="session")
-async def nc_mcp_oauth_client() -> AsyncGenerator[ClientSession, Any]:
+async def nc_mcp_oauth_client(
+    interactive_oauth_token: str,
+) -> 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.
+    Connects to the OAuth-enabled MCP server on port 8001 with OAuth authentication.
     """
     logger.info("Creating Streamable HTTP client for OAuth MCP server")
-    streamable_context = streamablehttp_client("http://127.0.0.1:8001/mcp")
+
+    # Pass OAuth token as Bearer token in headers
+    headers = {"Authorization": f"Bearer {interactive_oauth_token}"}
+    streamable_context = streamablehttp_client(
+        "http://127.0.0.1:8001/mcp", headers=headers
+    )
     session_context = None
 
     try:

tests/integration/test_oauth.py

@@ -119,15 +119,19 @@ async def test_mcp_oauth_tool_execution(nc_mcp_oauth_client):
     """Test executing a tool on the OAuth-enabled MCP server."""
     import json
 
-    # Example: Execute the 'nc_tables_list_tables' tool
+    # Example: Execute the 'nc_notes_search_notes' tool
-    result = await nc_mcp_oauth_client.call_tool("nc_tables_list_tables")
+    result = await nc_mcp_oauth_client.call_tool(
+        "nc_notes_search_notes", arguments={"query": ""}
+    )
 
     assert result.isError is False, f"Tool execution failed: {result.content}"
     assert result.content is not None
-    notes_list = json.loads(result.content[0].text)
+    response_data = json.loads(result.content[0].text)
 
-    assert isinstance(notes_list, list)
+    # The search response should have a 'results' field containing the list
+    assert "results" in response_data
+    assert isinstance(response_data["results"], list)
 
     logger.info(
-        f"Successfully executed 'nc_tables_list_tables' tool on OAuth MCP server and got {len(notes_list)} notes."
+        f"Successfully executed 'nc_notes_search_notes' tool on OAuth MCP server and got {len(response_data['results'])} notes."
     )