Enable extra_options even when extra not defined for HttpHook.run

providers/http/src/airflow/providers/http/hooks/http.py

@@ -159,7 +159,7 @@ def get_conn(
         connection = self.get_connection(self.http_conn_id)
         self._set_base_url(connection)
         session = self._configure_session_from_auth(session, connection)
-        if connection.extra:
+        if connection.extra or extra_options:
             session = self._configure_session_from_extra(session, connection, extra_options)
         session = self._configure_session_from_mount_adapters(session)
         if self.default_headers:
@@ -298,10 +298,10 @@ def run_and_check(
 
         settings = session.merge_environment_settings(
             prepped_request.url,
-            proxies=extra_options.get("proxies", {}),
-            stream=extra_options.get("stream", False),
-            verify=extra_options.get("verify"),
-            cert=extra_options.get("cert"),
+            proxies=session.proxies,
+            stream=session.stream,
+            verify=session.verify,
+            cert=session.cert,
         )
 
         # Send the request.

providers/http/tests/unit/http/hooks/test_http.py

@@ -48,6 +48,10 @@ def aioresponse():
         yield async_response
 
 
+def get_airflow_dummy_connection(conn_id: str = "http_default"):
+    return Connection(conn_id=conn_id, conn_type="http", host="test:8080/")
+
+
 def get_airflow_connection(conn_id: str = "http_default"):
     return Connection(conn_id=conn_id, conn_type="http", host="test:8080/", extra='{"bearer": "test"}')
 
@@ -294,6 +298,31 @@ def test_post_request_do_not_raise_for_status_if_check_response_is_false(self, r
             resp = self.post_hook.run("v1/test", extra_options={"check_response": False})
             assert resp.status_code == 418
 
+    def test_post_request_raises_error_when_redirects_with_max_redirects_set_to_0(self, requests_mock):
+        requests_mock.post(
+            "http://test:8080/v1/test",
+            status_code=302,
+            headers={"Location": "http://test:8080/v1/redirected"},
+        )
+
+        requests_mock.post(
+            "http://test:8080/v1/redirected",
+            status_code=200,
+            text='{"message": "OK"}',
+        )
+
+        with mock.patch(
+            "airflow.hooks.base.BaseHook.get_connection", side_effect=get_airflow_dummy_connection
+        ):
+            with pytest.raises(requests.exceptions.TooManyRedirects) as err:
+                self.post_hook.run("v1/test", extra_options={"max_redirects": 0})
+
+            assert str(err.value) == "Exceeded 0 redirects."
+            history = requests_mock.request_history
+            assert len(history) == 1
+            assert history[0].url == "http://test:8080/v1/test"
+            assert history[0].method == "POST"
+
     def test_post_request_do_not_raise_for_status_if_check_response_is_false_within_extra(
         self, requests_mock
     ):

providers/http/tests/unit/http/operators/test_http.py

@@ -142,6 +142,7 @@ def test_pagination(
         a dictionary that override previous' call parameters.
         """
         is_second_call: bool = False
+        extra_options_verify = extra_options["verify"]
 
         def pagination_function(response: Response) -> dict | None:
             """Paginated function which returns None at the second call."""
@@ -175,7 +176,7 @@ def pagination_function(response: Response) -> dict | None:
         first_call = first_endpoint.request_history[0]
         assert first_call.headers.items() >= headers.items()
         assert first_call.body == RequestEncodingMixin._encode_params(data)
-        assert first_call.verify is extra_options["verify"]
+        assert first_call.verify == extra_options_verify
 
         # Ensure the second - paginated - call is made with parameters merged from the pagination function
         second_call = second_endpoint.request_history[0]

providers/http/tests/unit/http/sensors/test_http.py

@@ -265,6 +265,10 @@ def __init__(self):
         self.response = requests.Response()
         self.response.status_code = 200
         self.response._content = "apache/airflow".encode("ascii", "ignore")
+        self.proxies = None
+        self.stream = None
+        self.verify = False
+        self.cert = None
 
     def send(self, *args, **kwargs):
         return self.response