Liniting issues fixed

Author: yurikunash

src/mcp/client/auth.py

@@ -207,36 +207,36 @@ def __init__(
     def _extract_resource_metadata_from_www_auth(self, init_response: httpx.Response) -> str | None:
         """
         Extract protected resource metadata URL from WWW-Authenticate header as per RFC9728.
-            
+
         Returns:
             Resource metadata URL if found in WWW-Authenticate header, None otherwise
         """
         if not init_response or init_response.status_code != 401:
             return None
-            
+
         www_auth_header = init_response.headers.get("WWW-Authenticate")
         if not www_auth_header:
             return None
-        
+
         # Pattern matches: resource_metadata="url" or resource_metadata=url (unquoted)
         pattern = r'resource_metadata=(?:"([^"]+)"|([^\s,]+))'
         match = re.search(pattern, www_auth_header)
-        
+
         if match:
             # Return quoted value if present, otherwise unquoted value
             return match.group(1) or match.group(2)
-        
+
         return None
 
-    async def _discover_protected_resource(self, init_response: httpx.Response | None = None) -> httpx.Request:
+    async def _discover_protected_resource(self, init_response: httpx.Response) -> httpx.Request:
         # RFC9728: Try to extract resource_metadata URL from WWW-Authenticate header of the initial response
-        url = self._extract_resource_metadata_from_www_auth(init_response) if init_response else None
-        
+        url = self._extract_resource_metadata_from_www_auth(init_response)
+
         if not url:
             # Fallback to well-known discovery
             auth_base_url = self.context.get_authorization_base_url(self.context.server_url)
             url = urljoin(auth_base_url, "/.well-known/oauth-protected-resource")
-        
+
         return httpx.Request("GET", url, headers={MCP_PROTOCOL_VERSION: LATEST_PROTOCOL_VERSION})
 
     async def _handle_protected_resource_response(self, response: httpx.Response) -> None:
@@ -521,7 +521,7 @@ async def async_auth_flow(self, request: httpx.Request) -> AsyncGenerator[httpx.
 
             if self.context.is_token_valid():
                 self._add_auth_header(request)
-            
+
             response = yield request
 
             if response.status_code == 401:

tests/client/test_auth.py

@@ -198,6 +198,7 @@ class TestOAuthFlow:
     @pytest.mark.anyio
     async def test_discover_protected_resource_request(self, client_metadata, mock_storage):
         """Test protected resource discovery request building maintains backward compatibility."""
+
         async def redirect_handler(url: str) -> None:
             pass
 
@@ -211,30 +212,24 @@ async def callback_handler() -> tuple[str, str | None]:
             redirect_handler=redirect_handler,
             callback_handler=callback_handler,
         )
-        
-        # Test without response (backward compatibility)
-        request = await provider._discover_protected_resource()
-        assert request.method == "GET"
-        assert str(request.url) == "https://api.example.com/.well-known/oauth-protected-resource"
-        assert "mcp-protocol-version" in request.headers
-        
-        # Test with response but no WWW-Authenticate (fallback)
+
+        # Test without WWW-Authenticate (fallback)
         init_response = httpx.Response(
-            status_code=401,
-            headers={},
-            request=httpx.Request("GET", "https://request-api.example.com")
+            status_code=401, headers={}, request=httpx.Request("GET", "https://request-api.example.com")
         )
 
         request = await provider._discover_protected_resource(init_response)
-        assert request.method == "GET" 
+        assert request.method == "GET"
         assert str(request.url) == "https://api.example.com/.well-known/oauth-protected-resource"
         assert "mcp-protocol-version" in request.headers
 
         # Test with WWW-Authenticate header
-        init_response.headers["WWW-Authenticate"] = 'Bearer resource_metadata="https://prm.example.com/.well-known/oauth-protected-resource/path"'
-        
+        init_response.headers["WWW-Authenticate"] = (
+            'Bearer resource_metadata="https://prm.example.com/.well-known/oauth-protected-resource/path"'
+        )
+
         request = await provider._discover_protected_resource(init_response)
-        assert request.method == "GET" 
+        assert request.method == "GET"
         assert str(request.url) == "https://prm.example.com/.well-known/oauth-protected-resource/path"
         assert "mcp-protocol-version" in request.headers
 
@@ -580,28 +575,42 @@ async def test_auth_flow_with_valid_tokens(self, oauth_provider, mock_storage, v
             pass  # Expected
 
 
-class TestRFC9728WWWAuthenticate:
-    """Test RFC9728 WWW-Authenticate header parsing functionality."""
-
-    @pytest.mark.parametrize("www_auth_header,expected_url", [
-        # Quoted URL
-        ('Bearer resource_metadata="https://api.example.com/.well-known/oauth-protected-resource"', 
-         "https://api.example.com/.well-known/oauth-protected-resource"),
-        # Unquoted URL
-        ("Bearer resource_metadata=https://api.example.com/.well-known/oauth-protected-resource", 
-         "https://api.example.com/.well-known/oauth-protected-resource"),
-        # Complex header with multiple parameters
-        ('Bearer realm="api", resource_metadata="https://api.example.com/.well-known/oauth-protected-resource", error="insufficient_scope"', 
-         "https://api.example.com/.well-known/oauth-protected-resource"),
-        # Different URL format
-        ('Bearer resource_metadata="https://custom.domain.com/metadata"', 
-         "https://custom.domain.com/metadata"),
-        # With path and query params
-        ('Bearer resource_metadata="https://api.example.com/auth/metadata?version=1"', 
-         "https://api.example.com/auth/metadata?version=1"),
-    ])
-    def test_extract_resource_metadata_from_www_auth_valid_cases(self, client_metadata, mock_storage, www_auth_header, expected_url):
+class TestProtectedResourceWWWAuthenticate:
+    """Test RFC9728 WWW-Authenticate header parsing functionality for protected resource."""
+
+    @pytest.mark.parametrize(
+        "www_auth_header,expected_url",
+        [
+            # Quoted URL
+            (
+                'Bearer resource_metadata="https://api.example.com/.well-known/oauth-protected-resource"',
+                "https://api.example.com/.well-known/oauth-protected-resource",
+            ),
+            # Unquoted URL
+            (
+                "Bearer resource_metadata=https://api.example.com/.well-known/oauth-protected-resource",
+                "https://api.example.com/.well-known/oauth-protected-resource",
+            ),
+            # Complex header with multiple parameters
+            (
+                'Bearer realm="api", resource_metadata="https://api.example.com/.well-known/oauth-protected-resource", '
+                'error="insufficient_scope"',
+                "https://api.example.com/.well-known/oauth-protected-resource",
+            ),
+            # Different URL format
+            ('Bearer resource_metadata="https://custom.domain.com/metadata"', "https://custom.domain.com/metadata"),
+            # With path and query params
+            (
+                'Bearer resource_metadata="https://api.example.com/auth/metadata?version=1"',
+                "https://api.example.com/auth/metadata?version=1",
+            ),
+        ],
+    )
+    def test_extract_resource_metadata_from_www_auth_valid_cases(
+        self, client_metadata, mock_storage, www_auth_header, expected_url
+    ):
         """Test extraction of resource_metadata URL from various valid WWW-Authenticate headers."""
+
         async def redirect_handler(url: str) -> None:
             pass
 
@@ -615,31 +624,45 @@ async def callback_handler() -> tuple[str, str | None]:
             redirect_handler=redirect_handler,
             callback_handler=callback_handler,
         )
-        
+
         init_response = httpx.Response(
             status_code=401,
             headers={"WWW-Authenticate": www_auth_header},
-            request=httpx.Request("GET", "https://api.example.com/test")
+            request=httpx.Request("GET", "https://api.example.com/test"),
         )
-        
+
         result = provider._extract_resource_metadata_from_www_auth(init_response)
         assert result == expected_url
 
-    @pytest.mark.parametrize("status_code,www_auth_header,description", [
-        # No header
-        (401, None, "no WWW-Authenticate header"),
-        # Empty header  
-        (401, "", "empty WWW-Authenticate header"),
-        # Header without resource_metadata
-        (401, 'Bearer realm="api", error="insufficient_scope"', "no resource_metadata parameter"),
-        # Malformed header
-        (401, "Bearer resource_metadata=", "malformed resource_metadata parameter"),
-        # Non-401 status code
-        (200, 'Bearer resource_metadata="https://api.example.com/.well-known/oauth-protected-resource"', "200 OK response"),
-        (500, 'Bearer resource_metadata="https://api.example.com/.well-known/oauth-protected-resource"', "500 error response"),
-    ])
-    def test_extract_resource_metadata_from_www_auth_invalid_cases(self, client_metadata, mock_storage, status_code, www_auth_header, description):
+    @pytest.mark.parametrize(
+        "status_code,www_auth_header,description",
+        [
+            # No header
+            (401, None, "no WWW-Authenticate header"),
+            # Empty header
+            (401, "", "empty WWW-Authenticate header"),
+            # Header without resource_metadata
+            (401, 'Bearer realm="api", error="insufficient_scope"', "no resource_metadata parameter"),
+            # Malformed header
+            (401, "Bearer resource_metadata=", "malformed resource_metadata parameter"),
+            # Non-401 status code
+            (
+                200,
+                'Bearer resource_metadata="https://api.example.com/.well-known/oauth-protected-resource"',
+                "200 OK response",
+            ),
+            (
+                500,
+                'Bearer resource_metadata="https://api.example.com/.well-known/oauth-protected-resource"',
+                "500 error response",
+            ),
+        ],
+    )
+    def test_extract_resource_metadata_from_www_auth_invalid_cases(
+        self, client_metadata, mock_storage, status_code, www_auth_header, description
+    ):
         """Test extraction returns None for invalid cases."""
+
         async def redirect_handler(url: str) -> None:
             pass
 
@@ -653,33 +676,11 @@ async def callback_handler() -> tuple[str, str | None]:
             redirect_handler=redirect_handler,
             callback_handler=callback_handler,
         )
-        
+
         headers = {"WWW-Authenticate": www_auth_header} if www_auth_header is not None else {}
         init_response = httpx.Response(
-            status_code=status_code,
-            headers=headers,
-            request=httpx.Request("GET", "https://api.example.com/test")
+            status_code=status_code, headers=headers, request=httpx.Request("GET", "https://api.example.com/test")
         )
-        
+
         result = provider._extract_resource_metadata_from_www_auth(init_response)
         assert result is None, f"Should return None for {description}"
-
-    def test_extract_resource_metadata_from_www_auth_none_response(self, client_metadata, mock_storage):
-        """Test extraction with None response returns None."""
-        async def redirect_handler(url: str) -> None:
-            pass
-
-        async def callback_handler() -> tuple[str, str | None]:
-            return "test_auth_code", "test_state"
-
-        provider = OAuthClientProvider(
-            server_url="https://api.example.com/v1/mcp",
-            client_metadata=client_metadata,
-            storage=mock_storage,
-            redirect_handler=redirect_handler,
-            callback_handler=callback_handler,
-        )
-        
-        result = provider._extract_resource_metadata_from_www_auth(None)
-        assert result is None
-