feat(ephemeral): replace old has_url_connectivity() with new _check_c…

…onnectivity_to_imds()

WIP-ONGOING-REFACTORIZATION.rst

@@ -267,7 +267,6 @@ categories:
 
   * ``ParserError``
   * ``RendererNotFoundError``
-  * ``has_url_connectivity``
   * ``is_ip_address``
   * ``is_ipv4_address``
   * ``natural_sort_key``

cloudinit/net/ephemeral.py

@@ -296,14 +296,16 @@ def __init__(
     def __enter__(self):
         """Setup sandboxed dhcp context, unless connectivity_url can already be
         reached."""
-        for url_data in self.connectivity_urls_data:
-            if net.has_url_connectivity(url_data):
-                LOG.debug(
-                    "Skip ephemeral DHCP setup, instance has connectivity"
-                    " to %s",
-                    url_data,
-                )
-                return
+        if imds_reached_at_url := _check_connectivity_to_imds(
+            self.connectivity_urls_data
+        ):
+            LOG.debug(
+                "Skip ephemeral DHCP setup, instance has connectivity"
+                " to %s",
+                imds_reached_at_url,
+            )
+            return None
+        # If we don't have connectivity, perform dhcp discovery
         return self.obtain_lease()
 
     def __exit__(self, excp_type, excp_value, excp_traceback):

tests/unittests/net/test_dhcp.py

@@ -815,13 +815,16 @@ def test_multiple_files(self):
 
 
 @pytest.mark.usefixtures("disable_netdev_info")
+@mock.patch("cloudinit.net.ephemeral._check_connectivity_to_imds")
 class TestEphemeralDhcpNoNetworkSetup(ResponsesTestCase):
     @mock.patch("cloudinit.net.dhcp.maybe_perform_dhcp_discovery")
-    def test_ephemeral_dhcp_no_network_if_url_connectivity(self, m_dhcp):
+    def test_ephemeral_dhcp_no_network_if_url_connectivity(self, m_dhcp, m_imds):
         """No EphemeralDhcp4 network setup when connectivity_url succeeds."""
         url = "http://example.org/index.html"
 
-        self.responses.add(responses.GET, url)
+        # need to return a valid url to indicate that we have connectivity
+        m_imds.return_value = url
+
         with EphemeralDHCPv4(
             MockDistro(),
             connectivity_urls_data=[{"url": url}],
@@ -833,7 +836,7 @@ def test_ephemeral_dhcp_no_network_if_url_connectivity(self, m_dhcp):
     @mock.patch("cloudinit.net.dhcp.subp.subp")
     @mock.patch("cloudinit.net.ephemeral.maybe_perform_dhcp_discovery")
     def test_ephemeral_dhcp_setup_network_if_url_connectivity(
-        self, m_dhcp, m_subp
+        self, m_dhcp, m_subp, m_imds
     ):
         """No EphemeralDhcp4 network setup when connectivity_url succeeds."""
         url = "http://example.org/index.html"
@@ -844,7 +847,9 @@ def test_ephemeral_dhcp_setup_network_if_url_connectivity(
         }
         m_subp.return_value = ("", "")
 
-        self.responses.add(responses.GET, url, body=b"", status=404)
+        # None indicates that we do NOT have connectivity to the IMDS
+        # therefore, DHCP discovery will be performed
+        m_imds.return_value = None
         with EphemeralDHCPv4(
             MockDistro(),
             connectivity_urls_data=[{"url": url}],

tests/unittests/net/test_init.py

@@ -1201,44 +1201,6 @@ def test_ephemeral_ipv6_network_performs_setup(self, m_subp, _):
             assert expected_setup_calls == m_subp.call_args_list
 
 
-class TestHasURLConnectivity(ResponsesTestCase):
-    def setUp(self):
-        super(TestHasURLConnectivity, self).setUp()
-        self.url = "http://fake/"
-        self.kwargs = {"allow_redirects": True, "timeout": 5.0}
-
-    @mock.patch("cloudinit.net.readurl")
-    def test_url_timeout_on_connectivity_check(self, m_readurl):
-        """A timeout of 5 seconds is provided when reading a url."""
-        self.assertTrue(
-            net.has_url_connectivity({"url": self.url}),
-            "Expected True on url connect",
-        )
-
-    def test_true_on_url_connectivity_success(self):
-        self.responses.add(responses.GET, self.url)
-        self.assertTrue(
-            net.has_url_connectivity({"url": self.url}),
-            "Expected True on url connect",
-        )
-
-    @mock.patch("requests.Session.request")
-    def test_true_on_url_connectivity_timeout(self, m_request):
-        """A timeout raised accessing the url will return False."""
-        m_request.side_effect = requests.Timeout("Fake Connection Timeout")
-        self.assertFalse(
-            net.has_url_connectivity({"url": self.url}),
-            "Expected False on url timeout",
-        )
-
-    def test_true_on_url_connectivity_failure(self):
-        self.responses.add(responses.GET, self.url, body=b"", status=404)
-        self.assertFalse(
-            net.has_url_connectivity({"url": self.url}),
-            "Expected False on url fail",
-        )
-
-
 def _mk_v1_phys(mac, name, driver, device_id):
     v1_cfg = {"type": "physical", "name": name, "mac_address": mac}
     params = {}