diff options
Diffstat (limited to 'cloudinit/net')
| -rw-r--r-- | cloudinit/net/__init__.py | 37 | ||||
| -rw-r--r-- | cloudinit/net/dhcp.py | 20 | ||||
| -rw-r--r-- | cloudinit/net/tests/test_dhcp.py | 8 | ||||
| -rw-r--r-- | cloudinit/net/tests/test_init.py | 20 |
4 files changed, 59 insertions, 26 deletions
diff --git a/cloudinit/net/__init__.py b/cloudinit/net/__init__.py index 017c50c5..7558745f 100644 --- a/cloudinit/net/__init__.py +++ b/cloudinit/net/__init__.py @@ -11,6 +11,7 @@ import ipaddress import logging import os import re +from typing import Any, Dict from cloudinit import subp from cloudinit import util @@ -971,18 +972,33 @@ def get_ib_hwaddrs_by_interface(): return ret -def has_url_connectivity(url): - """Return true when the instance has access to the provided URL +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'] if not any([url.startswith('http://'), url.startswith('https://')]): LOG.warning( "Ignoring connectivity check. Expected URL beginning with http*://" " received '%s'", url) return False + if 'timeout' not in url_data: + url_data['timeout'] = 5 try: - readurl(url, timeout=5) + readurl(**url_data) except UrlError: return False return True @@ -1025,14 +1041,15 @@ class EphemeralIPv4Network(object): No operations are performed if the provided interface already has the specified configuration. - This can be verified with the connectivity_url. + This can be verified with the connectivity_url_data. If unconnected, bring up the interface with valid ip, prefix and broadcast. If router is provided setup a default route for that interface. Upon context exit, clean up the interface leaving no configuration behind. """ def __init__(self, interface, ip, prefix_or_mask, broadcast, router=None, - connectivity_url=None, static_routes=None): + connectivity_url_data: Dict[str, Any] = None, + static_routes=None): """Setup context manager and validate call signature. @param interface: Name of the network interface to bring up. @@ -1041,7 +1058,7 @@ class EphemeralIPv4Network(object): prefix. @param broadcast: Broadcast address for the IPv4 network. @param router: Optionally the default gateway IP. - @param connectivity_url: Optionally, a URL to verify if a usable + @param connectivity_url_data: Optionally, a URL to verify if a usable connection already exists. @param static_routes: Optionally a list of static routes from DHCP """ @@ -1056,7 +1073,7 @@ class EphemeralIPv4Network(object): 'Cannot setup network: {0}'.format(e) ) from e - self.connectivity_url = connectivity_url + self.connectivity_url_data = connectivity_url_data self.interface = interface self.ip = ip self.broadcast = broadcast @@ -1066,11 +1083,11 @@ class EphemeralIPv4Network(object): def __enter__(self): """Perform ephemeral network setup if interface is not connected.""" - if self.connectivity_url: - if has_url_connectivity(self.connectivity_url): + if self.connectivity_url_data: + if has_url_connectivity(self.connectivity_url_data): LOG.debug( 'Skip ephemeral network setup, instance has connectivity' - ' to %s', self.connectivity_url) + ' to %s', self.connectivity_url_data['url']) return self._bringup_device() diff --git a/cloudinit/net/dhcp.py b/cloudinit/net/dhcp.py index 9b94c9a0..3f4b0418 100644 --- a/cloudinit/net/dhcp.py +++ b/cloudinit/net/dhcp.py @@ -4,6 +4,7 @@ # # This file is part of cloud-init. See LICENSE file for license information. +from typing import Dict, Any import configobj import logging import os @@ -38,21 +39,26 @@ class NoDHCPLeaseError(Exception): class EphemeralDHCPv4(object): - def __init__(self, iface=None, connectivity_url=None, dhcp_log_func=None): + def __init__( + self, + iface=None, + connectivity_url_data: Dict[str, Any] = None, + dhcp_log_func=None + ): self.iface = iface self._ephipv4 = None self.lease = None self.dhcp_log_func = dhcp_log_func - self.connectivity_url = connectivity_url + self.connectivity_url_data = connectivity_url_data def __enter__(self): """Setup sandboxed dhcp context, unless connectivity_url can already be reached.""" - if self.connectivity_url: - if has_url_connectivity(self.connectivity_url): + if self.connectivity_url_data: + if has_url_connectivity(self.connectivity_url_data): LOG.debug( 'Skip ephemeral DHCP setup, instance has connectivity' - ' to %s', self.connectivity_url) + ' to %s', self.connectivity_url_data) return return self.obtain_lease() @@ -104,8 +110,8 @@ class EphemeralDHCPv4(object): if kwargs['static_routes']: kwargs['static_routes'] = ( parse_static_routes(kwargs['static_routes'])) - if self.connectivity_url: - kwargs['connectivity_url'] = self.connectivity_url + if self.connectivity_url_data: + kwargs['connectivity_url_data'] = self.connectivity_url_data ephipv4 = EphemeralIPv4Network(**kwargs) ephipv4.__enter__() self._ephipv4 = ephipv4 diff --git a/cloudinit/net/tests/test_dhcp.py b/cloudinit/net/tests/test_dhcp.py index 5ae048e2..28b4ecf7 100644 --- a/cloudinit/net/tests/test_dhcp.py +++ b/cloudinit/net/tests/test_dhcp.py @@ -617,7 +617,9 @@ class TestEphemeralDhcpNoNetworkSetup(HttprettyTestCase): url = 'http://example.org/index.html' httpretty.register_uri(httpretty.GET, url) - with net.dhcp.EphemeralDHCPv4(connectivity_url=url) as lease: + with net.dhcp.EphemeralDHCPv4( + connectivity_url_data={'url': url}, + ) as lease: self.assertIsNone(lease) # Ensure that no teardown happens: m_dhcp.assert_not_called() @@ -635,7 +637,9 @@ class TestEphemeralDhcpNoNetworkSetup(HttprettyTestCase): m_subp.return_value = ('', '') httpretty.register_uri(httpretty.GET, url, body={}, status=404) - with net.dhcp.EphemeralDHCPv4(connectivity_url=url) as lease: + with net.dhcp.EphemeralDHCPv4( + connectivity_url_data={'url': url}, + ) as lease: self.assertEqual(fake_lease, lease) # Ensure that dhcp discovery occurs m_dhcp.called_once_with() diff --git a/cloudinit/net/tests/test_init.py b/cloudinit/net/tests/test_init.py index ad9c90ff..f9102f7b 100644 --- a/cloudinit/net/tests/test_init.py +++ b/cloudinit/net/tests/test_init.py @@ -622,11 +622,14 @@ class TestEphemeralIPV4Network(CiTestCase): params = { 'interface': 'eth0', 'ip': '192.168.2.2', 'prefix_or_mask': '255.255.255.0', 'broadcast': '192.168.2.255', - 'connectivity_url': 'http://example.org/index.html'} + 'connectivity_url_data': {'url': 'http://example.org/index.html'} + } with net.EphemeralIPv4Network(**params): - self.assertEqual([mock.call('http://example.org/index.html', - timeout=5)], m_readurl.call_args_list) + self.assertEqual( + [mock.call(url='http://example.org/index.html', timeout=5)], + m_readurl.call_args_list + ) # Ensure that no teardown happens: m_subp.assert_has_calls([]) @@ -850,25 +853,28 @@ class TestHasURLConnectivity(HttprettyTestCase): 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(self.url), 'Expected True on url connect') + net.has_url_connectivity({'url': self.url}), + 'Expected True on url connect') def test_true_on_url_connectivity_success(self): httpretty.register_uri(httpretty.GET, self.url) self.assertTrue( - net.has_url_connectivity(self.url), 'Expected True on url connect') + 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(self.url), + net.has_url_connectivity({'url': self.url}), 'Expected False on url timeout') def test_true_on_url_connectivity_failure(self): httpretty.register_uri(httpretty.GET, self.url, body={}, status=404) self.assertFalse( - net.has_url_connectivity(self.url), 'Expected False on url fail') + net.has_url_connectivity({'url': self.url}), + 'Expected False on url fail') def _mk_v1_phys(mac, name, driver, device_id): |