🐛 fix: fix proxy settings from environement variables

osc_sdk_python/authentication.py

@@ -1,7 +1,9 @@
+from requests.structures import CaseInsensitiveDict
 import datetime
 import hashlib
 import hmac
 import base64
+import requests
 
 from .version import get_version
 from .credentials import Profile
@@ -174,3 +176,17 @@ def get_basic_auth_header(self):
             "X-Osc-Date": date_iso,
             "Authorization": "Basic " + b64_creds,
         }
+
+    def __call__(self, request: requests.PreparedRequest) -> requests.PreparedRequest:
+        if self.is_basic_auth_configured():
+            auth_headers = self.get_basic_auth_header()
+        else:
+            auth_headers = self.forge_headers_signed(request.path_url, request.body or "")
+
+        if request.headers is None:
+            request.headers = CaseInsensitiveDict()
+
+        for head in auth_headers:
+            request.headers[head] = auth_headers[head]
+
+        return request

osc_sdk_python/call.py

@@ -21,7 +21,6 @@ def __init__(self, logger=None, limiter=None, **kwargs):
         self.limiter: RateLimiter | None = limiter
         self.retry_kwargs = {}
         self.session = Session()
-        self.session.trust_env = False
 
         kwargs = self.update_limiter(**kwargs)
         kwargs = self.update_retry(**kwargs)
@@ -88,7 +87,7 @@ def api(self, action, service="api", **data):
                 self.logger.do_log(
                     "uri: " + uri + "\npayload:\n" + json.dumps(data, indent=2)
                 )
-            return requester.send(uri, json.dumps(data))
+            return requester.send(json.dumps(data))
         except Exception as err:
             raise err
 

osc_sdk_python/requester.py

@@ -1,28 +1,21 @@
 from .retry import Retry
 
-
 class Requester:
     def __init__(self, session, auth, endpoint, **kwargs):
         self.session = session
         self.auth = auth
         self.endpoint = endpoint
         self.request_kwargs = kwargs
 
-    def send(self, uri, payload):
-        headers = None
-        if self.auth.is_basic_auth_configured():
-            headers = self.auth.get_basic_auth_header()
-        else:
-            headers = self.auth.forge_headers_signed(uri, payload)
-
+    def send(self, payload):
         if self.auth.x509_client_cert is not None:
             cert_file = self.auth.x509_client_cert
         else:
             cert_file = None
 
         retry_kwargs = self.request_kwargs.copy()
         retry_kwargs.update(
-            {"data": payload, "headers": headers, "verify": True, "cert": cert_file}
+            {"data": payload, "auth": self.auth, "verify": True, "cert": cert_file}
         )
 
         response = Retry(self.session, "post", self.endpoint, **retry_kwargs)

tests/test_proxy.py

@@ -0,0 +1,78 @@
+import os
+import unittest
+import unittest.mock
+
+from osc_sdk_python.call import Call
+from osc_sdk_python.credentials import Profile
+
+
+class TestProxySupport(unittest.TestCase):
+    def test_proxy_env_passed_to_session(self):
+        with unittest.mock.patch.object(
+            Profile, "from_standard_configuration"
+        ) as mock_from_std:
+            mock_from_std.return_value = Profile(region="eu-west-2", protocol="https")
+
+            with unittest.mock.patch.dict(
+                os.environ, {"HTTPS_PROXY": "http://proxy.example.com:8080"}
+            ):
+                call = Call(access_key="fake-ak", secret_key="fake-sk")
+
+                captured_proxies = []
+
+                class FakeAdapter:
+                    def send(self, request, **kwargs):
+                        captured_proxies.append(kwargs.get("proxies"))
+                        import requests
+
+                        resp = requests.Response()
+                        resp.status_code = 200
+                        resp._content = b'{"status": "ok"}'
+                        resp.request = request
+                        resp.url = request.url
+                        return resp
+
+                call.session.mount("https://", FakeAdapter())
+
+                result = call.api("ReadVms")
+
+                self.assertEqual(len(captured_proxies), 1)
+                self.assertEqual(
+                    captured_proxies[0],
+                    {"https": "http://proxy.example.com:8080"},
+                )
+                self.assertEqual(result, {"status": "ok"})
+
+    def test_no_proxy_env_no_proxies_kwarg(self):
+        with unittest.mock.patch.object(
+            Profile, "from_standard_configuration"
+        ) as mock_from_std:
+            mock_from_std.return_value = Profile(region="eu-west-2", protocol="https")
+
+            with unittest.mock.patch.dict(os.environ, {}, clear=True):
+                call = Call(access_key="fake-ak", secret_key="fake-sk")
+
+                captured_proxies = []
+
+                class FakeAdapter:
+                    def send(self, request, **kwargs):
+                        captured_proxies.append(kwargs.get("proxies"))
+                        import requests
+
+                        resp = requests.Response()
+                        resp.status_code = 200
+                        resp._content = b'{"status": "ok"}'
+                        resp.request = request
+                        resp.url = request.url
+                        return resp
+
+                call.session.mount("https://", FakeAdapter())
+
+                call.api("ReadVms")
+
+                self.assertEqual(len(captured_proxies), 1)
+                self.assertEqual(captured_proxies[0], {})
+
+
+if __name__ == "__main__":
+    unittest.main()