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/__init__.py

@@ -11,12 +11,9 @@
 import logging
 import os
 import re
-from typing import Any, Callable, Dict, List, Optional, Tuple
-from urllib.parse import urlparse
-
+from typing import Callable, Dict, List, Optional, Tuple
 from cloudinit import subp, util
 from cloudinit.net.netops.iproute2 import Iproute2
-from cloudinit.url_helper import UrlError, readurl
 
 LOG = logging.getLogger(__name__)
 SYS_CLASS_NET = "/sys/class/net/"
@@ -1146,45 +1143,6 @@ def get_ib_hwaddrs_by_interface():
     return ret
 
 
-def has_url_connectivity(url_data: Dict[str, Any]) -> bool:
-    """Return true when the instance has access to the provided URL.
-
-    Logs a warning if url is not the expected format.
-
-    url_data is a dictionary of kwargs to send to readurl. E.g.:
-
-    has_url_connectivity({
-        "url": "http://example.invalid",
-        "headers": {"some": "header"},
-        "timeout": 10
-    })
-    """
-    if "url" not in url_data:
-        LOG.warning(
-            "Ignoring connectivity check. No 'url' to check in %s", url_data
-        )
-        return False
-    url = url_data["url"]
-    try:
-        result = urlparse(url)
-        if not any([result.scheme == "http", result.scheme == "https"]):
-            LOG.warning(
-                "Ignoring connectivity check. Invalid URL scheme %s",
-                url.scheme,
-            )
-            return False
-    except ValueError as err:
-        LOG.warning("Ignoring connectivity check. Invalid URL %s", err)
-        return False
-    if "timeout" not in url_data:
-        url_data["timeout"] = 5
-    try:
-        readurl(**url_data)
-    except UrlError:
-        return False
-    return True
-
-
 def maybe_get_address(convert_to_address: Callable, address: str, **kwargs):
     """Use a function to return an address. If conversion throws a ValueError
     exception return False.

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

@@ -7,7 +7,6 @@
 from textwrap import dedent
 
 import pytest
-import responses
 
 from cloudinit.distros import alpine, amazon, centos, debian, freebsd, rhel
 from cloudinit.distros.ubuntu import Distro
@@ -815,13 +814,18 @@ 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 +837,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 +848,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

@@ -10,20 +10,13 @@
 from unittest import mock
 
 import pytest
-import requests
-import responses
 
 import cloudinit.net as net
 from cloudinit import subp
 from cloudinit.net.ephemeral import EphemeralIPv4Network, EphemeralIPv6Network
 from cloudinit.subp import ProcessExecutionError
 from cloudinit.util import ensure_file, write_file
-from tests.unittests.helpers import (
-    CiTestCase,
-    ResponsesTestCase,
-    example_netdev,
-    random_string,
-)
+from tests.unittests.helpers import CiTestCase, example_netdev, random_string
 from tests.unittests.util import MockDistro
 
 
@@ -1201,44 +1194,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 = {}