refactor: use global proxy config instead of drilling proxy args

uaclient/actions.py

@@ -289,7 +289,7 @@ def get_cloud_instance(
 ) -> AutoAttachCloudInstance:
     instance = None  # type: Optional[AutoAttachCloudInstance]
     try:
-        instance = identity.cloud_instance_factory(cfg.proxies)
+        instance = identity.cloud_instance_factory()
     except exceptions.CloudFactoryError as e:
         if isinstance(e, exceptions.CloudFactoryNoCloudError):
             raise exceptions.UserFacingError(

uaclient/api/u/pro/attach/auto/should_auto_attach/v1.py

@@ -22,7 +22,7 @@ def should_auto_attach() -> ShouldAutoAttachResult:
 
 def _should_auto_attach(cfg: UAConfig) -> ShouldAutoAttachResult:
     try:
-        cloud_instance_factory(cfg.proxies)
+        cloud_instance_factory()
     except exceptions.CloudFactoryError:
         return ShouldAutoAttachResult(
             should_auto_attach=False,

uaclient/clouds/__init__.py

@@ -1,11 +1,8 @@
 import abc
-from typing import Any, Dict, Optional
+from typing import Any, Dict
 
 
 class AutoAttachCloudInstance(metaclass=abc.ABCMeta):
-    def __init__(self, proxies: Dict[str, Optional[str]]):
-        self.proxies = proxies
-
     @property
     @abc.abstractmethod
     def identity_doc(self) -> Dict[str, Any]:

uaclient/clouds/aws.py

@@ -29,10 +29,7 @@ class UAAutoAttachAWSInstance(AutoAttachCloudInstance):
     def _get_imds_url_response(self):
         headers = self._request_imds_v2_token_headers()
         response = http.readurl(
-            IMDS_URL.format(self._ip_address),
-            headers=headers,
-            timeout=1,
-            proxies=self.proxies,
+            IMDS_URL.format(self._ip_address), headers=headers, timeout=1
         )
         if response.code == 200:
             return response.body
@@ -80,7 +77,6 @@ def _get_imds_v2_token_headers(self, ip_address):
             method="PUT",
             headers={AWS_TOKEN_REQ_HEADER: AWS_TOKEN_TTL_SECONDS},
             timeout=1,
-            proxies=self.proxies,
         )
         if response.code == 200:
             self._api_token = response.body

uaclient/clouds/azure.py

@@ -30,10 +30,7 @@ def identity_doc(self) -> Dict[str, Any]:
         responses = {}
         for key, url in sorted(IMDS_URLS.items()):
             response = http.readurl(
-                url,
-                headers={"Metadata": "true"},
-                timeout=1,
-                proxies=self.proxies,
+                url, headers={"Metadata": "true"}, timeout=1
             )
             if response.code != 200:
                 raise exceptions.CloudMetadataError(
@@ -68,9 +65,7 @@ def is_pro_license_present(self, *, wait_for_change: bool) -> bool:
 
         url = IMDS_URLS.get("compute", "")
         try:
-            response = http.readurl(
-                url, headers={"Metadata": "true"}, proxies=self.proxies
-            )
+            response = http.readurl(url, headers={"Metadata": "true"})
         except OSError as e:
             LOG.error(e)
             raise exceptions.CancelProLicensePolling()

uaclient/clouds/gcp.py

@@ -33,8 +33,7 @@
 
 
 class UAAutoAttachGCPInstance(AutoAttachCloudInstance):
-    def __init__(self, proxies: Dict[str, Optional[str]]):
-        super().__init__(proxies)
+    def __init__(self):
         # store ETAG
         # https://cloud.google.com/compute/docs/metadata/querying-metadata#etags  # noqa
         self.etag = None  # type: Optional[str]
@@ -45,10 +44,7 @@ def __init__(self, proxies: Dict[str, Optional[str]]):
     @util.retry(exceptions.GCPProAccountError, retry_sleeps=[0.5, 1, 1])
     def identity_doc(self) -> Dict[str, Any]:
         response = http.readurl(
-            TOKEN_URL,
-            headers={"Metadata-Flavor": "Google"},
-            timeout=1,
-            proxies=self.proxies,
+            TOKEN_URL, headers={"Metadata-Flavor": "Google"}, timeout=1
         )
         if response.code == 200:
             return {"identityToken": response.body}
@@ -111,11 +107,7 @@ def is_pro_license_present(self, *, wait_for_change: bool) -> bool:
             if self.etag:
                 url += LAST_ETAG.format(etag=self.etag)
 
-        response = http.readurl(
-            url,
-            headers={"Metadata-Flavor": "Google"},
-            proxies=self.proxies,
-        )
+        response = http.readurl(url, headers={"Metadata-Flavor": "Google"})
         if response.code == 200:
             license_ids = [license["id"] for license in response.json_list]
             self.etag = response.headers.get("etag")

uaclient/clouds/identity.py

@@ -51,9 +51,7 @@ def get_cloud_type() -> Tuple[Optional[str], Optional[NoCloudTypeReason]]:
     return (None, NoCloudTypeReason.NO_CLOUD_DETECTED)
 
 
-def cloud_instance_factory(
-    proxies: Dict[str, Optional[str]]
-) -> clouds.AutoAttachCloudInstance:
+def cloud_instance_factory() -> clouds.AutoAttachCloudInstance:
     """
     :raises CloudFactoryError: if no cloud instance object can be constructed
     :raises CloudFactoryNoCloudError: if no cloud instance object can be
@@ -79,7 +77,7 @@ def cloud_instance_factory(
     cls = cloud_instance_map.get(cloud_type)
     if not cls:
         raise exceptions.CloudFactoryUnsupportedCloudError(cloud_type)
-    instance = cls(proxies)
+    instance = cls()
     if not instance.is_viable:
         raise exceptions.CloudFactoryNonViableCloudError(cloud_type)
     return instance

uaclient/clouds/tests/test_aws.py

@@ -21,7 +21,7 @@
 
 class TestUAAutoAttachAWSInstance:
     def test_cloud_type(self):
-        instance = UAAutoAttachAWSInstance({})
+        instance = UAAutoAttachAWSInstance()
         assert "aws" == instance.cloud_type
 
     @mock.patch(M_PATH + "http.readurl")
@@ -34,7 +34,7 @@ def test__get_imds_v2_token_headers_none_on_404(self, readurl):
             json_dict={},
             json_list=[],
         )
-        instance = UAAutoAttachAWSInstance({})
+        instance = UAAutoAttachAWSInstance()
         assert None is instance._get_imds_v2_token_headers(
             ip_address=IMDS_IPV4_ADDRESS
         )
@@ -46,7 +46,7 @@ def test__get_imds_v2_token_headers_none_on_404(self, readurl):
     @mock.patch(M_PATH + "http.readurl")
     def test__get_imds_v2_token_headers_caches_response(self, readurl):
         """Return API token headers for IMDSv2 access. Response is cached."""
-        instance = UAAutoAttachAWSInstance({})
+        instance = UAAutoAttachAWSInstance()
         url = "http://169.254.169.254/latest/api/token"
         readurl.return_value = http.HTTPResponse(
             code=200,
@@ -66,7 +66,6 @@ def test__get_imds_v2_token_headers_caches_response(self, readurl):
                 method="PUT",
                 headers={AWS_TOKEN_REQ_HEADER: AWS_TOKEN_TTL_SECONDS},
                 timeout=1,
-                proxies={},
             )
         ] == readurl.call_args_list
 
@@ -79,9 +78,7 @@ def test_retry_backoff_on__get_imds_v2_token_headers_caches_response(
     ):
         """Retry backoff before failing _get_imds_v2_token_headers."""
 
-        def fake_someurlerrors(
-            url, method=None, headers=None, timeout=1, proxies={}
-        ):
+        def fake_someurlerrors(url, method=None, headers=None, timeout=1):
             if readurl.call_count <= fail_count:
                 return http.HTTPResponse(
                     code=700 + readurl.call_count,
@@ -99,7 +96,7 @@ def fake_someurlerrors(
             )
 
         readurl.side_effect = fake_someurlerrors
-        instance = UAAutoAttachAWSInstance({})
+        instance = UAAutoAttachAWSInstance()
         if exception:
             with pytest.raises(exceptions.CloudMetadataError):
                 instance._get_imds_v2_token_headers(
@@ -133,7 +130,7 @@ def test_identity_doc_from_aws_url_pkcs7(self, readurl):
             json_dict={},
             json_list=[],
         )
-        instance = UAAutoAttachAWSInstance({})
+        instance = UAAutoAttachAWSInstance()
         assert {"pkcs7": "pkcs7WOOT!=="} == instance.identity_doc
         url = "http://169.254.169.254/latest/dynamic/instance-identity/pkcs7"
         token_url = "http://169.254.169.254/latest/api/token"
@@ -143,13 +140,9 @@ def test_identity_doc_from_aws_url_pkcs7(self, readurl):
                 method="PUT",
                 headers={AWS_TOKEN_REQ_HEADER: AWS_TOKEN_TTL_SECONDS},
                 timeout=1,
-                proxies={},
             ),
             mock.call(
-                url,
-                headers={AWS_TOKEN_PUT_HEADER: "pkcs7WOOT!=="},
-                timeout=1,
-                proxies={},
+                url, headers={AWS_TOKEN_PUT_HEADER: "pkcs7WOOT!=="}, timeout=1
             ),
         ] == readurl.call_args_list
 
@@ -162,9 +155,7 @@ def test_retry_backoff_on_failed_identity_doc(
     ):
         """Retry backoff is attempted before failing to get AWS.identity_doc"""
 
-        def fake_someurlerrors(
-            url, method=None, headers=None, timeout=1, proxies={}
-        ):
+        def fake_someurlerrors(url, method=None, headers=None, timeout=1):
             # due to _get_imds_v2_token_headers
             if "latest/api/token" in url:
                 return http.HTTPResponse(
@@ -191,7 +182,7 @@ def fake_someurlerrors(
             )
 
         readurl.side_effect = fake_someurlerrors
-        instance = UAAutoAttachAWSInstance({})
+        instance = UAAutoAttachAWSInstance()
         if exception:
             with pytest.raises(exceptions.CloudMetadataError) as excinfo:
                 instance.identity_doc
@@ -215,7 +206,7 @@ def fake_someurlerrors(
     def test_is_viable_based_on_sys_hypervisor_uuid(self, load_file, uuid):
         """Viable ec2 platform is determined by /sys/hypervisor/uuid prefix"""
         load_file.return_value = uuid
-        instance = UAAutoAttachAWSInstance({})
+        instance = UAAutoAttachAWSInstance()
         assert True is instance.is_viable
 
     @pytest.mark.parametrize(
@@ -249,21 +240,19 @@ def fake_load_file(f_name):
             raise AssertionError("Invalid load_file of {}".format(f_name))
 
         load_file.side_effect = fake_load_file
-        instance = UAAutoAttachAWSInstance({})
+        instance = UAAutoAttachAWSInstance()
         assert viable is instance.is_viable
 
     @pytest.mark.parametrize("caplog_text", [logging.DEBUG], indirect=True)
     @mock.patch(M_PATH + "http.readurl")
     def test_identity_doc_default_to_ipv6_if_ipv4_fail(
         self, readurl, caplog_text
     ):
-        instance = UAAutoAttachAWSInstance({})
+        instance = UAAutoAttachAWSInstance()
         ipv4_address = IMDS_IPV4_ADDRESS
         ipv6_address = IMDS_IPV6_ADDRESS
 
-        def fake_someurlerrors(
-            url, method=None, headers=None, timeout=1, proxies={}
-        ):
+        def fake_someurlerrors(url, method=None, headers=None, timeout=1):
             if ipv4_address in url:
                 raise Exception("IPv4 exception")
 
@@ -293,20 +282,17 @@ def fake_someurlerrors(
                 method="PUT",
                 headers={AWS_TOKEN_REQ_HEADER: AWS_TOKEN_TTL_SECONDS},
                 timeout=1,
-                proxies={},
             ),
             mock.call(
                 IMDS_V2_TOKEN_URL.format(ipv6_address),
                 method="PUT",
                 headers={AWS_TOKEN_REQ_HEADER: AWS_TOKEN_TTL_SECONDS},
                 timeout=1,
-                proxies={},
             ),
             mock.call(
                 IMDS_URL.format(ipv6_address),
                 headers={AWS_TOKEN_PUT_HEADER: "base64token=="},
                 timeout=1,
-                proxies={},
             ),
         ]
 
@@ -321,7 +307,7 @@ def test_identity_doc_logs_error_if_both_ipv4_and_ipv6_fails(
         self, readurl, caplog_text
     ):
 
-        instance = UAAutoAttachAWSInstance({})
+        instance = UAAutoAttachAWSInstance()
         ipv4_address = IMDS_IPV4_ADDRESS
         ipv6_address = IMDS_IPV6_ADDRESS
 
@@ -342,14 +328,12 @@ def test_identity_doc_logs_error_if_both_ipv4_and_ipv6_fails(
                 method="PUT",
                 headers={AWS_TOKEN_REQ_HEADER: AWS_TOKEN_TTL_SECONDS},
                 timeout=1,
-                proxies={},
             ),
             mock.call(
                 IMDS_V2_TOKEN_URL.format(ipv6_address),
                 method="PUT",
                 headers={AWS_TOKEN_REQ_HEADER: AWS_TOKEN_TTL_SECONDS},
                 timeout=1,
-                proxies={},
             ),
         ]
         assert expected == readurl.call_args_list
@@ -364,11 +348,11 @@ def test_identity_doc_logs_error_if_both_ipv4_and_ipv6_fails(
 
     def test_unsupported_should_poll_for_pro_license(self):
         """Unsupported"""
-        instance = UAAutoAttachAWSInstance({})
+        instance = UAAutoAttachAWSInstance()
         assert not instance.should_poll_for_pro_license()
 
     def test_unsupported_is_pro_license_present(self):
         """Unsupported"""
-        instance = UAAutoAttachAWSInstance({})
+        instance = UAAutoAttachAWSInstance()
         with pytest.raises(exceptions.InPlaceUpgradeNotSupportedError):
             instance.is_pro_license_present(wait_for_change=False)