Merge pull request #814 from HardMax71/fix/strict-typing-other-handlers

fix(http): pass mypy --strict in the other middleware handlers and fix four handler bugs

Author: baywet

packages/http/httpx/kiota_http/middleware/headers_inspection_handler.py

@@ -70,7 +70,7 @@ def _get_current_options(self, request: httpx.Request) -> HeadersInspectionHandl
         Returns:
             HeadersInspectionHandlerOption: The options to be used.
         """
-        current_options = None
+        current_options: Optional[HeadersInspectionHandlerOption] = None
         request_options = request.extensions.get(REQUEST_OPTIONS_KEY)
         if request_options:
             current_options = request_options.get(HeadersInspectionHandlerOption.get_key(), None)

packages/http/httpx/kiota_http/middleware/parameters_name_decoding_handler.py

@@ -1,3 +1,5 @@
+from typing import Optional
+
 from kiota_abstractions.request_option import RequestOption
 
 import httpx
@@ -10,19 +12,16 @@
 
 class ParametersNameDecodingHandler(BaseMiddleware):
 
-    def __init__(
-        self,
-        options: ParametersNameDecodingHandlerOption = ParametersNameDecodingHandlerOption(),
-    ):
+    def __init__(self, options: Optional[ParametersNameDecodingHandlerOption] = None) -> None:
         """Create an instance of ParametersNameDecodingHandler
 
         Args:
             options (ParametersNameDecodingHandlerOption, optional): The parameters name
-            decoding handler options value.
-            Defaults to ParametersNameDecodingHandlerOption
+            decoding handler options value. A new ParametersNameDecodingHandlerOption
+            per handler when not provided.
         """
         super().__init__()
-        self.options = options
+        self.options = options if options is not None else ParametersNameDecodingHandlerOption()
 
     async def send(
         self, request: httpx.Request, transport: httpx.AsyncBaseTransport
@@ -72,7 +71,7 @@ def _get_current_options(self, request: httpx.Request) -> ParametersNameDecoding
         """
         request_options = request.extensions.get(REQUEST_OPTIONS_KEY)
         if request_options:
-            current_options = request_options.get(
+            current_options: ParametersNameDecodingHandlerOption = request_options.get(
                 ParametersNameDecodingHandlerOption.get_key(), self.options
             )
             return current_options

packages/http/httpx/kiota_http/middleware/redirect_handler.py

@@ -27,12 +27,14 @@ class RedirectHandler(BaseMiddleware):
     STATUS_CODE_SEE_OTHER: int = 303
     LOCATION_HEADER: str = "Location"
 
-    def __init__(self, options: RedirectHandlerOption = RedirectHandlerOption()) -> None:
+    def __init__(self, options: typing.Optional[RedirectHandlerOption] = None) -> None:
         super().__init__()
-        self.options = options
+        self.options = options if options is not None else RedirectHandlerOption()
         self.redirect_on_status_codes: set[int] = self.DEFAULT_REDIRECT_STATUS_CODES
 
-    def increment(self, response, max_redirect, history) -> bool:
+    def increment(
+        self, response: httpx.Response, max_redirect: int, history: list[httpx.Request]
+    ) -> bool:
         """Increment the redirect attempts for this request.
         Args
             response(httpx.Response): A httpx response object.
@@ -53,7 +55,8 @@ def get_redirect_location(self, response: httpx.Response) -> typing.Union[str, b
             location. ``False`` if not a redirect status code.
         """
         if response.status_code in self.redirect_on_status_codes:
-            return response.headers.get('location')
+            location: typing.Optional[str] = response.headers.get('location')
+            return location
         return None
 
     async def send(
@@ -114,8 +117,9 @@ def _get_current_options(self, request: httpx.Request) -> RedirectHandlerOption:
         """
         request_options = request.extensions.get(REQUEST_OPTIONS_KEY)
         if request_options:
-            current_options = request_options.get( # type:ignore
-                RedirectHandlerOption.get_key(), self.options)
+            current_options: RedirectHandlerOption = request_options.get(
+                RedirectHandlerOption.get_key(), self.options
+            )
             return current_options
         return self.options
 
@@ -124,7 +128,7 @@ def _build_redirect_request(
         request: httpx.Request,
         response: httpx.Response,
         options: RedirectHandlerOption,
-        request_options: typing.Optional[dict] = None,
+        request_options: typing.Optional[dict[str, object]] = None,
     ) -> httpx.Request:
         """
         Given a request and a redirect response, return a new request that

packages/http/httpx/kiota_http/middleware/url_replace_handler.py

@@ -1,3 +1,5 @@
+from typing import Optional
+
 from kiota_abstractions.request_option import RequestOption
 from opentelemetry.semconv.attributes.url_attributes import URL_FULL
 
@@ -9,16 +11,16 @@
 
 class UrlReplaceHandler(BaseMiddleware):
 
-    def __init__(self, options: UrlReplaceHandlerOption = UrlReplaceHandlerOption()):
+    def __init__(self, options: Optional[UrlReplaceHandlerOption] = None) -> None:
         """Create an instance of UrlReplaceHandler
 
         Args:
             options (UrlReplaceHandlerOption, optional): The url replacement
-            options to pass to the handler.
-            Defaults to UrlReplaceHandlerOption
+            options to pass to the handler. A new UrlReplaceHandlerOption
+            per handler when not provided.
         """
         super().__init__()
-        self.options = options
+        self.options = options if options is not None else UrlReplaceHandlerOption()
 
     async def send(
         self, request: httpx.Request, transport: httpx.AsyncBaseTransport
@@ -32,19 +34,19 @@ async def send(
         Returns:
             Response: The response object.
         """
-        response = None
         _enable_span = self._create_observability_span(request, "UrlReplaceHandler_send")
-        if self.options and self.options.is_enabled:
-            _enable_span.set_attribute("com.microsoft.kiota.handler.url_replacer.enable", True)
+        try:
             current_options = self._get_current_options(request)
+            if current_options.is_enabled:
+                _enable_span.set_attribute("com.microsoft.kiota.handler.url_replacer.enable", True)
 
-            url_string: str = str(request.url)
-            url_string = self.replace_url_segment(url_string, current_options)
-            request.url = httpx.URL(url_string)
-            _enable_span.set_attribute(URL_FULL, str(request.url))
-        response = await super().send(request, transport)
-        _enable_span.end()
-        return response
+                url_string: str = str(request.url)
+                url_string = self.replace_url_segment(url_string, current_options)
+                request.url = httpx.URL(url_string)
+                _enable_span.set_attribute(URL_FULL, str(request.url))
+            return await super().send(request, transport)
+        finally:
+            _enable_span.end()
 
     def _get_current_options(self, request: httpx.Request) -> UrlReplaceHandlerOption:
         """Returns the options to use for the request.Overries default options if
@@ -58,7 +60,9 @@ def _get_current_options(self, request: httpx.Request) -> UrlReplaceHandlerOptio
         """
         request_options = request.extensions.get(REQUEST_OPTIONS_KEY)
         if request_options:
-            current_options = request_options.get(UrlReplaceHandlerOption.get_key(), self.options)
+            current_options: UrlReplaceHandlerOption = request_options.get(
+                UrlReplaceHandlerOption.get_key(), self.options
+            )
             return current_options
         return self.options
 

packages/http/httpx/kiota_http/middleware/user_agent_handler.py

@@ -1,3 +1,5 @@
+from typing import Optional
+
 from kiota_abstractions.request_option import RequestOption
 
 from httpx import AsyncBaseTransport, Request, Response
@@ -11,8 +13,8 @@ class UserAgentHandler(BaseMiddleware):
     Middleware handler for User Agent.
     """
 
-    def __init__(self, options: UserAgentHandlerOption = UserAgentHandlerOption(), **kwargs):
-        super().__init__(**kwargs)
+    def __init__(self, options: Optional[UserAgentHandlerOption] = None) -> None:
+        super().__init__()
         self.options = UserAgentHandlerOption() if options is None else options
 
     async def send(self, request: Request, transport: AsyncBaseTransport) -> Response:
@@ -24,7 +26,7 @@ async def send(self, request: Request, transport: AsyncBaseTransport) -> Respons
         _span = self._create_observability_span(request, "UserAgentHandler_send")
         if current_options and current_options.is_enabled:
             _span.set_attribute("com.microsoft.kiota.handler.useragent.enable", True)
-            value = f"{self.options.product_name}/{self.options.product_version}"
+            value = f"{current_options.product_name}/{current_options.product_version}"
             self._update_user_agent(request, value)
         _span.end()
         return await super().send(request, transport)
@@ -41,11 +43,13 @@ def _get_current_options(self, request: Request) -> UserAgentHandlerOption:
         """
         request_options = request.extensions.get(REQUEST_OPTIONS_KEY)
         if request_options:
-            current_options = request_options.get(UserAgentHandlerOption.get_key(), self.options)
+            current_options: UserAgentHandlerOption = request_options.get(
+                UserAgentHandlerOption.get_key(), self.options
+            )
             return current_options
         return self.options
 
-    def _update_user_agent(self, request: Request, value: str):
+    def _update_user_agent(self, request: Request, value: str) -> None:
         """Updates the values of the User-Agent header."""
         user_agent = request.headers.get("User-Agent", "")
         if not user_agent:

packages/http/httpx/tests/middleware_tests/test_url_replace_handler.py

@@ -1,7 +1,7 @@
 import httpx
 import pytest
 
-from kiota_http.middleware import UrlReplaceHandler
+from kiota_http.middleware import REQUEST_OPTIONS_KEY, UrlReplaceHandler
 from kiota_http.middleware.options import UrlReplaceHandlerOption
 
 ORIGINAL_URL = "https://graph.microsoft.com/users/user-id-to-replace/messages"
@@ -53,3 +53,34 @@ def test_replace_url_segment():
 def test_default_replacement_pairs_are_not_shared():
     UrlReplaceHandlerOption().replacement_pairs["/users/user-id-to-replace"] = "/me"
     assert UrlReplaceHandlerOption().replacement_pairs == {}
+
+
+@pytest.mark.asyncio
+async def test_request_option_enables_replacement():
+    option = UrlReplaceHandlerOption(
+        enabled=True, replacement_pairs={"/users/user-id-to-replace": "/me"}
+    )
+    request = httpx.Request(
+        "GET", ORIGINAL_URL, extensions={REQUEST_OPTIONS_KEY: {option.get_key(): option}}
+    )
+    handler = UrlReplaceHandler(options=UrlReplaceHandlerOption(enabled=False))
+
+    async with httpx.MockTransport(lambda request: httpx.Response(200)) as transport:
+        response = await handler.send(request, transport)
+
+    assert str(response.request.url) == REPLACED_URL
+
+
+@pytest.mark.asyncio
+async def test_span_ends_on_transport_failure(span_exporter):
+
+    def request_handler(request):
+        raise httpx.ReadError("request interrupted")
+
+    handler = UrlReplaceHandler()
+    request = httpx.Request("GET", ORIGINAL_URL)
+    async with httpx.MockTransport(request_handler) as transport:
+        with pytest.raises(httpx.ReadError):
+            await handler.send(request, transport)
+
+    assert [span.name for span in span_exporter.get_finished_spans()] == ["UrlReplaceHandler_send"]

packages/http/httpx/tests/middleware_tests/test_user_agent_handler.py

@@ -1,6 +1,8 @@
+import httpx
 import pytest
 
 from kiota_http._version import VERSION
+from kiota_http.middleware import REQUEST_OPTIONS_KEY, UserAgentHandler
 from kiota_http.middleware.options.user_agent_handler_option import UserAgentHandlerOption
 
 
@@ -27,3 +29,16 @@ def test_is_enabled_setter():
     options = UserAgentHandlerOption()
     options.is_enabled = False
     assert not options.is_enabled
+
+
+@pytest.mark.asyncio
+async def test_request_option_sets_the_product():
+    option = UserAgentHandlerOption(product_name="custom", product_version="9")
+    request = httpx.Request(
+        "GET", "https://example.com", extensions={REQUEST_OPTIONS_KEY: {option.get_key(): option}}
+    )
+
+    async with httpx.MockTransport(lambda request: httpx.Response(200)) as transport:
+        response = await UserAgentHandler().send(request, transport)
+
+    assert response.request.headers["User-Agent"] == "custom/9"

packages/http/httpx/tests/test_kiota_client_factory.py

@@ -141,6 +141,14 @@ def test_get_default_middleware():
     assert isinstance(middleware[6], BodyInspectionHandler)
 
 
+def test_default_middleware_does_not_share_options():
+    first = KiotaClientFactory.get_default_middleware(None)
+    second = KiotaClientFactory.get_default_middleware(None)
+
+    for first_handler, second_handler in zip(first, second):
+        assert first_handler.options is not second_handler.options
+
+
 def test_get_default_middleware_with_options():
     """Test fetching of default middleware with custom options passed"""
     retry_options = RetryHandlerOption(max_retries=7)