Allow trust env parameter to be defined in extra options of HTTP Conn…

…ection (apache#39161)

* refactor: Allow trust env parameter to be defined in extra options field of HTTP Connection

* refactor: Added missing comma

* refactor: Added special case when trust_env set to False

---------

Co-authored-by: David Blain <david.blain@infrabel.be>

airflow/providers/http/hooks/http.py

@@ -132,6 +132,7 @@ def get_conn(self, headers: dict[Any, Any] | None = None) -> requests.Session:
  session.verify = extra.pop("verify", extra.pop("verify_ssl", True))
  session.cert = extra.pop("cert", None)
  session.max_redirects = extra.pop("max_redirects", DEFAULT_REDIRECT_LIMIT)
+ session.trust_env = extra.pop("trust_env", True)
 
  try:
  session.headers.update(extra)
@@ -425,6 +426,7 @@ def _process_extra_options_from_connection(cls, conn: Connection, extra_options:
  verify_ssl = extra.pop("verify", extra.pop("verify_ssl", None))
  allow_redirects = extra.pop("allow_redirects", None)
  max_redirects = extra.pop("max_redirects", None)
+ trust_env = extra.pop("trust_env", None)
 
  if proxies is not None and "proxy" not in extra_options:
  extra_options["proxy"] = proxies
@@ -436,6 +438,8 @@ def _process_extra_options_from_connection(cls, conn: Connection, extra_options:
  extra_options["allow_redirects"] = allow_redirects
  if max_redirects is not None and "max_redirects" not in extra_options:
  extra_options["max_redirects"] = max_redirects
+ if trust_env is not None and "trust_env" not in extra_options:
+ extra_options["trust_env"] = trust_env
  return extra
 
  def _retryable_error_async(self, exception: ClientResponseError) -> bool:

tests/providers/http/hooks/test_http.py

@@ -138,6 +138,7 @@ def test_hook_ignore_max_redirects_from_extra_field_as_header(self):
  assert conn.verify is True
  assert conn.cert is None
  assert conn.max_redirects == 3
+ assert conn.trust_env is True
 
  def test_hook_ignore_proxies_from_extra_field_as_header(self):
  airflow_connection = get_airflow_connection_with_extra(
@@ -154,6 +155,7 @@ def test_hook_ignore_proxies_from_extra_field_as_header(self):
  assert conn.verify is True
  assert conn.cert is None
  assert conn.max_redirects == DEFAULT_REDIRECT_LIMIT
+ assert conn.trust_env is True
 
  def test_hook_ignore_verify_from_extra_field_as_header(self):
  airflow_connection = get_airflow_connection_with_extra(extra={"bearer": "test", "verify": False})
@@ -168,6 +170,7 @@ def test_hook_ignore_verify_from_extra_field_as_header(self):
  assert conn.verify is False
  assert conn.cert is None
  assert conn.max_redirects == DEFAULT_REDIRECT_LIMIT
+ assert conn.trust_env is True
 
  def test_hook_ignore_cert_from_extra_field_as_header(self):
  airflow_connection = get_airflow_connection_with_extra(extra={"bearer": "test", "cert": "cert.crt"})
@@ -182,6 +185,22 @@ def test_hook_ignore_cert_from_extra_field_as_header(self):
  assert conn.verify is True
  assert conn.cert == "cert.crt"
  assert conn.max_redirects == DEFAULT_REDIRECT_LIMIT
+ assert conn.trust_env is True
+
+ def test_hook_ignore_trust_env_from_extra_field_as_header(self):
+ airflow_connection = get_airflow_connection_with_extra(extra={"bearer": "test", "trust_env": False})
+ with mock.patch("airflow.hooks.base.BaseHook.get_connection", side_effect=airflow_connection):
+ expected_conn = airflow_connection()
+ conn = self.get_hook.get_conn()
+ assert dict(conn.headers, **json.loads(expected_conn.extra)) != conn.headers
+ assert conn.headers.get("bearer") == "test"
+ assert conn.headers.get("cert") is None
+ assert conn.proxies == {}
+ assert conn.stream is False
+ assert conn.verify is True
+ assert conn.cert is None
+ assert conn.max_redirects == DEFAULT_REDIRECT_LIMIT
+ assert conn.trust_env is False
 
  @mock.patch("requests.Request")
  def test_hook_with_method_in_lowercase(self, mock_requests):
@@ -616,6 +635,7 @@ async def test_async_request_uses_connection_extra_with_requests_parameters(self
  "verify": False,
  "allow_redirects": False,
  "max_redirects": 3,
+ "trust_env": False,
  },
  }
  )
@@ -633,6 +653,7 @@ async def test_async_request_uses_connection_extra_with_requests_parameters(self
  assert mocked_function.call_args.kwargs.get("verify_ssl") is False
  assert mocked_function.call_args.kwargs.get("allow_redirects") is False
  assert mocked_function.call_args.kwargs.get("max_redirects") == 3
+ assert mocked_function.call_args.kwargs.get("trust_env") is False
 
  def test_process_extra_options_from_connection(self):
  extra_options = {}
@@ -647,6 +668,7 @@ def test_process_extra_options_from_connection(self):
  "verify": False,
  "allow_redirects": False,
  "max_redirects": 3,
+ "trust_env": False,
  }
  )()
 
@@ -658,6 +680,7 @@ def test_process_extra_options_from_connection(self):
  "verify_ssl": False,
  "allow_redirects": False,
  "max_redirects": 3,
+ "trust_env": False,
  }
  assert actual == {"bearer": "test"}