diff options
| -rwxr-xr-x | cloudinit/cmd/devel/net_convert.py | 3 | ||||
| -rwxr-xr-x | cloudinit/distros/__init__.py | 39 | ||||
| -rw-r--r-- | cloudinit/distros/alpine.py | 13 | ||||
| -rw-r--r-- | cloudinit/distros/arch.py | 10 | ||||
| -rw-r--r-- | cloudinit/distros/bsd.py | 3 | ||||
| -rw-r--r-- | cloudinit/distros/debian.py | 14 | ||||
| -rw-r--r-- | cloudinit/distros/opensuse.py | 9 | ||||
| -rw-r--r-- | cloudinit/distros/photon.py | 3 | ||||
| -rw-r--r-- | cloudinit/distros/rhel.py | 9 | ||||
| -rw-r--r-- | cloudinit/net/activators.py | 156 | ||||
| -rw-r--r-- | cloudinit/net/netplan.py | 10 | ||||
| -rw-r--r-- | cloudinit/net/network_state.py | 59 | ||||
| -rw-r--r-- | cloudinit/net/renderer.py | 2 | ||||
| -rw-r--r-- | cloudinit/net/renderers.py | 13 | ||||
| -rw-r--r-- | cloudinit/net/sysconfig.py | 11 | ||||
| -rw-r--r-- | cloudinit/net/tests/test_network_state.py | 6 | ||||
| -rw-r--r-- | tests/unittests/test_net.py | 2 | ||||
| -rw-r--r-- | tests/unittests/test_net_activators.py | 177 |
18 files changed, 413 insertions, 126 deletions
diff --git a/cloudinit/cmd/devel/net_convert.py b/cloudinit/cmd/devel/net_convert.py index 5c649fd0..f4a98e5e 100755 --- a/cloudinit/cmd/devel/net_convert.py +++ b/cloudinit/cmd/devel/net_convert.py @@ -96,9 +96,6 @@ def handle_args(name, args): pre_ns = ovf.get_network_config_from_conf(config, False) ns = network_state.parse_net_config_data(pre_ns) - if not ns: - raise RuntimeError("No valid network_state object created from" - " input data") if args.debug: sys.stderr.write('\n'.join( diff --git a/cloudinit/distros/__init__.py b/cloudinit/distros/__init__.py index 4991f42b..2caa8bc2 100755 --- a/cloudinit/distros/__init__.py +++ b/cloudinit/distros/__init__.py @@ -16,13 +16,16 @@ import stat import string import urllib.parse from io import StringIO +from typing import Any, Mapping from cloudinit import importer from cloudinit import log as logging from cloudinit import net +from cloudinit.net import activators from cloudinit.net import eni from cloudinit.net import network_state from cloudinit.net import renderers +from cloudinit.net.network_state import parse_net_config_data from cloudinit import persistence from cloudinit import ssh_util from cloudinit import type_utils @@ -72,7 +75,7 @@ class Distro(persistence.CloudInitPickleMixin, metaclass=abc.ABCMeta): hostname_conf_fn = "/etc/hostname" tz_zone_dir = "/usr/share/zoneinfo" init_cmd = ['service'] # systemctl, service etc - renderer_configs = {} + renderer_configs = {} # type: Mapping[str, Mapping[str, Any]] _preferred_ntp_clients = None networking_cls = LinuxNetworking # This is used by self.shutdown_command(), and can be overridden in @@ -106,14 +109,12 @@ class Distro(persistence.CloudInitPickleMixin, metaclass=abc.ABCMeta): raise NotImplementedError() def _write_network(self, settings): - raise RuntimeError( + """Deprecated. Remove if/when arch and gentoo support renderers.""" + raise NotImplementedError( "Legacy function '_write_network' was called in distro '%s'.\n" "_write_network_config needs implementation.\n" % self.name) - def _write_network_config(self, settings): - raise NotImplementedError() - - def _supported_write_network_config(self, network_config): + def _write_network_state(self, network_state): priority = util.get_cfg_by_path( self._cfg, ('network', 'renderers'), None) @@ -121,8 +122,7 @@ class Distro(persistence.CloudInitPickleMixin, metaclass=abc.ABCMeta): LOG.debug("Selected renderer '%s' from priority list: %s", name, priority) renderer = render_cls(config=self.renderer_configs.get(name)) - renderer.render_network_config(network_config) - return [] + renderer.render_network_state(network_state) def _find_tz_file(self, tz): tz_file = os.path.join(self.tz_zone_dir, str(tz)) @@ -174,6 +174,7 @@ class Distro(persistence.CloudInitPickleMixin, metaclass=abc.ABCMeta): mirror_info=arch_info) def apply_network(self, settings, bring_up=True): + """Deprecated. Remove if/when arch and gentoo support renderers.""" # this applies network where 'settings' is interfaces(5) style # it is obsolete compared to apply_network_config # Write it out @@ -188,6 +189,7 @@ class Distro(persistence.CloudInitPickleMixin, metaclass=abc.ABCMeta): return False def _apply_network_from_network_config(self, netconfig, bring_up=True): + """Deprecated. Remove if/when arch and gentoo support renderers.""" distro = self.__class__ LOG.warning("apply_network_config is not currently implemented " "for distribution '%s'. Attempting to use apply_network", @@ -208,8 +210,9 @@ class Distro(persistence.CloudInitPickleMixin, metaclass=abc.ABCMeta): # apply network config netconfig # This method is preferred to apply_network which only takes # a much less complete network config format (interfaces(5)). + network_state = parse_net_config_data(netconfig) try: - dev_names = self._write_network_config(netconfig) + self._write_network_state(network_state) except NotImplementedError: # backwards compat until all distros have apply_network_config return self._apply_network_from_network_config( @@ -217,7 +220,8 @@ class Distro(persistence.CloudInitPickleMixin, metaclass=abc.ABCMeta): # Now try to bring them up if bring_up: - return self._bring_up_interfaces(dev_names) + network_activator = activators.select_activator() + network_activator.bring_up_all_interfaces(network_state) return False def apply_network_config_names(self, netconfig): @@ -393,20 +397,11 @@ class Distro(persistence.CloudInitPickleMixin, metaclass=abc.ABCMeta): return self._preferred_ntp_clients def _bring_up_interface(self, device_name): - cmd = ['ifup', device_name] - LOG.debug("Attempting to run bring up interface %s using command %s", - device_name, cmd) - try: - (_out, err) = subp.subp(cmd) - if len(err): - LOG.warning("Running %s resulted in stderr output: %s", - cmd, err) - return True - except subp.ProcessExecutionError: - util.logexc(LOG, "Running interface command %s failed", cmd) - return False + """Deprecated. Remove if/when arch and gentoo support renderers.""" + raise NotImplementedError def _bring_up_interfaces(self, device_names): + """Deprecated. Remove if/when arch and gentoo support renderers.""" am_failed = 0 for d in device_names: if not self._bring_up_interface(d): diff --git a/cloudinit/distros/alpine.py b/cloudinit/distros/alpine.py index ca5bfe80..e4bed5a2 100644 --- a/cloudinit/distros/alpine.py +++ b/cloudinit/distros/alpine.py @@ -73,19 +73,6 @@ class Distro(distros.Distro): self.update_package_sources() self.package_command('add', pkgs=pkglist) - def _write_network_config(self, netconfig): - return self._supported_write_network_config(netconfig) - - def _bring_up_interfaces(self, device_names): - use_all = False - for d in device_names: - if d == 'all': - use_all = True - if use_all: - return distros.Distro._bring_up_interface(self, '-a') - else: - return distros.Distro._bring_up_interfaces(self, device_names) - def _write_hostname(self, your_hostname, out_fn): conf = None try: diff --git a/cloudinit/distros/arch.py b/cloudinit/distros/arch.py index 246e6fe7..c9acb11f 100644 --- a/cloudinit/distros/arch.py +++ b/cloudinit/distros/arch.py @@ -61,9 +61,9 @@ class Distro(distros.Distro): self.update_package_sources() self.package_command('', pkgs=pkglist) - def _write_network_config(self, netconfig): + def _write_network_state(self, network_state): try: - return self._supported_write_network_config(netconfig) + super()._write_network_state(network_state) except RendererNotFoundError as e: # Fall back to old _write_network raise NotImplementedError from e @@ -101,12 +101,6 @@ class Distro(distros.Distro): util.logexc(LOG, "Running interface command %s failed", cmd) return False - def _bring_up_interfaces(self, device_names): - for d in device_names: - if not self._bring_up_interface(d): - return False - return True - def _write_hostname(self, your_hostname, out_fn): conf = None try: diff --git a/cloudinit/distros/bsd.py b/cloudinit/distros/bsd.py index f717a667..c2fc1e0b 100644 --- a/cloudinit/distros/bsd.py +++ b/cloudinit/distros/bsd.py @@ -120,9 +120,6 @@ class BSD(distros.Distro): # Allow the output of this to flow outwards (ie not be captured) subp.subp(cmd, env=self._get_pkg_cmd_environ(), capture=False) - def _write_network_config(self, netconfig): - return self._supported_write_network_config(netconfig) - def set_timezone(self, tz): distros.set_etc_timezone(tz=tz, tz_file=self._find_tz_file(tz)) diff --git a/cloudinit/distros/debian.py b/cloudinit/distros/debian.py index 844aaf21..089e0c3e 100644 --- a/cloudinit/distros/debian.py +++ b/cloudinit/distros/debian.py @@ -111,19 +111,9 @@ class Distro(distros.Distro): self.update_package_sources() self.package_command('install', pkgs=pkglist) - def _write_network_config(self, netconfig): + def _write_network_state(self, network_state): _maybe_remove_legacy_eth0() - return self._supported_write_network_config(netconfig) - - def _bring_up_interfaces(self, device_names): - use_all = False - for d in device_names: - if d == 'all': - use_all = True - if use_all: - return distros.Distro._bring_up_interface(self, '--all') - else: - return distros.Distro._bring_up_interfaces(self, device_names) + return super()._write_network_state(network_state) def _write_hostname(self, your_hostname, out_fn): conf = None diff --git a/cloudinit/distros/opensuse.py b/cloudinit/distros/opensuse.py index 270cc189..b4193ac2 100644 --- a/cloudinit/distros/opensuse.py +++ b/cloudinit/distros/opensuse.py @@ -116,12 +116,6 @@ class Distro(distros.Distro): self._runner.run("update-sources", self.package_command, ['refresh'], freq=PER_INSTANCE) - def _bring_up_interfaces(self, device_names): - if device_names and 'all' in device_names: - raise RuntimeError(('Distro %s can not translate ' - 'the device name "all"') % (self.name)) - return distros.Distro._bring_up_interfaces(self, device_names) - def _read_hostname(self, filename, default=None): if self.uses_systemd() and filename.endswith('/previous-hostname'): return util.load_file(filename).strip() @@ -174,9 +168,6 @@ class Distro(distros.Distro): conf.set_hostname(hostname) util.write_file(out_fn, str(conf), 0o644) - def _write_network_config(self, netconfig): - return self._supported_write_network_config(netconfig) - @property def preferred_ntp_clients(self): """The preferred ntp client is dependent on the version.""" diff --git a/cloudinit/distros/photon.py b/cloudinit/distros/photon.py index 45125be7..0ced7b5f 100644 --- a/cloudinit/distros/photon.py +++ b/cloudinit/distros/photon.py @@ -76,9 +76,6 @@ class Distro(distros.Distro): # self.update_package_sources() self.package_command('install', pkgs=pkglist) - def _write_network_config(self, netconfig): - return self._supported_write_network_config(netconfig) - def _bring_up_interfaces(self, device_names): cmd = ['systemctl', 'restart', 'systemd-networkd', 'systemd-resolved'] LOG.debug('Attempting to run bring up interfaces using command %s', diff --git a/cloudinit/distros/rhel.py b/cloudinit/distros/rhel.py index 80a6f1d8..be5b3d24 100644 --- a/cloudinit/distros/rhel.py +++ b/cloudinit/distros/rhel.py @@ -65,9 +65,6 @@ class Distro(distros.Distro): def install_packages(self, pkglist): self.package_command('install', pkgs=pkglist) - def _write_network_config(self, netconfig): - return self._supported_write_network_config(netconfig) - def apply_locale(self, locale, out_fn=None): if self.uses_systemd(): if not out_fn: @@ -117,12 +114,6 @@ class Distro(distros.Distro): else: return default - def _bring_up_interfaces(self, device_names): - if device_names and 'all' in device_names: - raise RuntimeError(('Distro %s can not translate ' - 'the device name "all"') % (self.name)) - return distros.Distro._bring_up_interfaces(self, device_names) - def set_timezone(self, tz): tz_file = self._find_tz_file(tz) if self.uses_systemd(): diff --git a/cloudinit/net/activators.py b/cloudinit/net/activators.py new file mode 100644 index 00000000..34fee3bf --- /dev/null +++ b/cloudinit/net/activators.py @@ -0,0 +1,156 @@ +# This file is part of cloud-init. See LICENSE file for license information. +import logging +import os +from abc import ABC, abstractmethod +from typing import Iterable, List, Type + +from cloudinit import subp +from cloudinit import util +from cloudinit.net.eni import available as eni_available +from cloudinit.net.netplan import available as netplan_available +from cloudinit.net.network_state import NetworkState +from cloudinit.net.sysconfig import NM_CFG_FILE + + +LOG = logging.getLogger(__name__) + + +class NetworkActivator(ABC): + @staticmethod + @abstractmethod + def available() -> bool: + raise NotImplementedError() + + @staticmethod + @abstractmethod + def bring_up_interface(device_name: str) -> bool: + raise NotImplementedError() + + @classmethod + def bring_up_interfaces(cls, device_names: Iterable[str]) -> bool: + all_succeeded = True + for device in device_names: + if not cls.bring_up_interface(device): + all_succeeded = False + return all_succeeded + + @classmethod + def bring_up_all_interfaces(cls, network_state: NetworkState) -> bool: + return cls.bring_up_interfaces( + [i['name'] for i in network_state.iter_interfaces()] + ) + + +class IfUpDownActivator(NetworkActivator): + # Note that we're not overriding bring_up_interfaces to pass something + # like ifup --all because it isn't supported everywhere. + # E.g., NetworkManager has a ifupdown plugin that requires the name + # of a specific connection. + @staticmethod + def available(target=None) -> bool: + """Return true if ifupdown can be used on this system.""" + return eni_available(target=target) + + @staticmethod + def bring_up_interface(device_name: str) -> bool: + """Bring up interface using ifup.""" + cmd = ['ifup', device_name] + LOG.debug("Attempting to run bring up interface %s using command %s", + device_name, cmd) + try: + (_out, err) = subp.subp(cmd) + if len(err): + LOG.warning("Running %s resulted in stderr output: %s", + cmd, err) + return True + except subp.ProcessExecutionError: + util.logexc(LOG, "Running interface command %s failed", cmd) + return False + + +class NetworkManagerActivator(NetworkActivator): + @staticmethod + def available(target=None) -> bool: + config_present = os.path.isfile( + subp.target_path(target, path=NM_CFG_FILE) + ) + nmcli_present = subp.which('nmcli', target=target) + return config_present and bool(nmcli_present) + + @staticmethod + def bring_up_interface(device_name: str) -> bool: + try: + subp.subp(['nmcli', 'connection', 'up', device_name]) + except subp.ProcessExecutionError: + util.logexc(LOG, "nmcli failed to bring up {}".format(device_name)) + return False + return True + + +class NetplanActivator(NetworkActivator): + @staticmethod + def available(target=None) -> bool: + return netplan_available(target=target) + + @staticmethod + def _apply_netplan(): + LOG.debug('Applying current netplan config') + try: + subp.subp(['netplan', 'apply'], capture=True) + except subp.ProcessExecutionError: + util.logexc(LOG, "netplan apply failed") + return False + return True + + @staticmethod + def bring_up_interface(device_name: str) -> bool: + LOG.debug("Calling 'netplan apply' rather than " + "bringing up individual interfaces") + return NetplanActivator._apply_netplan() + + @staticmethod + def bring_up_interfaces(device_names: Iterable[str]) -> bool: + LOG.debug("Calling 'netplan apply' rather than " + "bringing up individual interfaces") + return NetplanActivator._apply_netplan() + + @staticmethod + def bring_up_all_interfaces(network_state: NetworkState) -> bool: + return NetplanActivator._apply_netplan() + + +# This section is mostly copied and pasted from renderers.py. An abstract +# version to encompass both seems overkill at this point +DEFAULT_PRIORITY = [ + IfUpDownActivator, + NetworkManagerActivator, + NetplanActivator, +] + + +def search_activator( + priority=None, target=None +) -> List[Type[NetworkActivator]]: + if priority is None: + priority = DEFAULT_PRIORITY + + unknown = [i for i in priority if i not in DEFAULT_PRIORITY] + if unknown: + raise ValueError( + "Unknown activators provided in priority list: %s" % unknown) + + return [activator for activator in priority if activator.available(target)] + + +def select_activator(priority=None, target=None) -> Type[NetworkActivator]: + found = search_activator(priority, target) + if not found: + if priority is None: + priority = DEFAULT_PRIORITY + tmsg = "" + if target and target != "/": + tmsg = " in target=%s" % target + raise RuntimeError( + "No available network activators found%s. Searched " + "through list: %s" % (tmsg, priority)) + return found[0] diff --git a/cloudinit/net/netplan.py b/cloudinit/net/netplan.py index 53347c83..41acf963 100644 --- a/cloudinit/net/netplan.py +++ b/cloudinit/net/netplan.py @@ -4,7 +4,12 @@ import copy import os from . import renderer -from .network_state import subnet_is_ipv6, NET_CONFIG_TO_V2, IPV6_DYNAMIC_TYPES +from .network_state import ( + NetworkState, + subnet_is_ipv6, + NET_CONFIG_TO_V2, + IPV6_DYNAMIC_TYPES, +) from cloudinit import log as logging from cloudinit import util @@ -256,7 +261,7 @@ class Renderer(renderer.Renderer): os.path.islink(SYS_CLASS_NET + iface)]: subp.subp(cmd, capture=True) - def _render_content(self, network_state): + def _render_content(self, network_state: NetworkState): # if content already in netplan format, pass it back if network_state.version == 2: @@ -426,4 +431,5 @@ def network_state_to_netplan(network_state, header=None): contents = renderer._render_content(network_state) return header + contents + # vi: ts=4 expandtab diff --git a/cloudinit/net/network_state.py b/cloudinit/net/network_state.py index 8018cfb9..95b064f0 100644 --- a/cloudinit/net/network_state.py +++ b/cloudinit/net/network_state.py @@ -58,38 +58,6 @@ NET_CONFIG_TO_V2 = { 'bridge_waitport': None}} -def parse_net_config_data(net_config, skip_broken=True): - """Parses the config, returns NetworkState object - - :param net_config: curtin network config dict - """ - state = None - version = net_config.get('version') - config = net_config.get('config') - if version == 2: - # v2 does not have explicit 'config' key so we - # pass the whole net-config as-is - config = net_config - - if version and config is not None: - nsi = NetworkStateInterpreter(version=version, config=config) - nsi.parse_config(skip_broken=skip_broken) - state = nsi.get_network_state() - - return state - - -def parse_net_config(path, skip_broken=True): - """Parses a curtin network configuration file and - return network state""" - ns = None - net_config = util.read_conf(path) - if 'network' in net_config: - ns = parse_net_config_data(net_config.get('network'), - skip_broken=skip_broken) - return ns - - def from_state_file(state_file): state = util.read_conf(state_file) nsi = NetworkStateInterpreter() @@ -1088,4 +1056,31 @@ def mask_and_ipv4_to_bcast_addr(mask, ip): return bcast_str +def parse_net_config_data(net_config, skip_broken=True) -> NetworkState: + """Parses the config, returns NetworkState object + + :param net_config: curtin network config dict + """ + state = None + version = net_config.get('version') + config = net_config.get('config') + if version == 2: + # v2 does not have explicit 'config' key so we + # pass the whole net-config as-is + config = net_config + + if version and config is not None: + nsi = NetworkStateInterpreter(version=version, config=config) + nsi.parse_config(skip_broken=skip_broken) + state = nsi.get_network_state() + + if not state: + raise RuntimeError( + "No valid network_state object created from network config. " + "Did you specify the correct version?" + ) + + return state + + # vi: ts=4 expandtab diff --git a/cloudinit/net/renderer.py b/cloudinit/net/renderer.py index 2a61a7a8..27447bc2 100644 --- a/cloudinit/net/renderer.py +++ b/cloudinit/net/renderer.py @@ -28,6 +28,8 @@ filter_by_physical = filter_by_type('physical') class Renderer(object): + def __init__(self, config=None): + pass @staticmethod def _render_persistent_net(network_state): diff --git a/cloudinit/net/renderers.py b/cloudinit/net/renderers.py index c3931a98..822b45de 100644 --- a/cloudinit/net/renderers.py +++ b/cloudinit/net/renderers.py @@ -1,10 +1,13 @@ # This file is part of cloud-init. See LICENSE file for license information. +from typing import List, Tuple, Type + from . import eni from . import freebsd from . import netbsd from . import netplan from . import networkd +from . import renderer from . import RendererNotFoundError from . import openbsd from . import sysconfig @@ -23,7 +26,9 @@ DEFAULT_PRIORITY = ["eni", "sysconfig", "netplan", "freebsd", "netbsd", "openbsd", "networkd"] -def search(priority=None, target=None, first=False): +def search( + priority=None, target=None, first=False +) -> List[Tuple[str, Type[renderer.Renderer]]]: if priority is None: priority = DEFAULT_PRIORITY @@ -40,13 +45,13 @@ def search(priority=None, target=None, first=False): if render_mod.available(target): cur = (name, render_mod.Renderer) if first: - return cur + return [cur] found.append(cur) return found -def select(priority=None, target=None): +def select(priority=None, target=None) -> Tuple[str, Type[renderer.Renderer]]: found = search(priority, target=target, first=True) if not found: if priority is None: @@ -57,6 +62,6 @@ def select(priority=None, target=None): raise RendererNotFoundError( "No available network renderers found%s. Searched " "through list: %s" % (tmsg, priority)) - return found + return found[0] # vi: ts=4 expandtab diff --git a/cloudinit/net/sysconfig.py b/cloudinit/net/sysconfig.py index 3a433c99..8031cd3a 100644 --- a/cloudinit/net/sysconfig.py +++ b/cloudinit/net/sysconfig.py @@ -18,8 +18,8 @@ from .network_state import ( is_ipv6_addr, net_prefix_to_ipv4_mask, subnet_is_ipv6, IPV6_DYNAMIC_TYPES) LOG = logging.getLogger(__name__) -NM_CFG_FILE = "/etc/NetworkManager/NetworkManager.conf" KNOWN_DISTROS = ['almalinux', 'centos', 'fedora', 'rhel', 'rocky', 'suse'] +NM_CFG_FILE = "/etc/NetworkManager/NetworkManager.conf" def _make_header(sep='#'): @@ -931,7 +931,9 @@ class Renderer(renderer.Renderer): netrules_path = subp.target_path(target, self.netrules_path) util.write_file(netrules_path, netrules_content, file_mode) if available_nm(target=target): - enable_ifcfg_rh(subp.target_path(target, path=NM_CFG_FILE)) + enable_ifcfg_rh(subp.target_path( + target, path=NM_CFG_FILE + )) sysconfig_path = subp.target_path(target, templates.get('control')) # Distros configuring /etc/sysconfig/network as a file e.g. Centos @@ -978,7 +980,10 @@ def available_sysconfig(target=None): def available_nm(target=None): - if not os.path.isfile(subp.target_path(target, path=NM_CFG_FILE)): + if not os.path.isfile(subp.target_path( + target, + path=NM_CFG_FILE + )): return False return True diff --git a/cloudinit/net/tests/test_network_state.py b/cloudinit/net/tests/test_network_state.py index fc4724a1..84e8308a 100644 --- a/cloudinit/net/tests/test_network_state.py +++ b/cloudinit/net/tests/test_network_state.py @@ -67,11 +67,13 @@ class TestNetworkStateParseConfig(CiTestCase): def test_missing_version_returns_none(self): ncfg = {} - self.assertEqual(None, network_state.parse_net_config_data(ncfg)) + with self.assertRaises(RuntimeError): + network_state.parse_net_config_data(ncfg) def test_unknown_versions_returns_none(self): ncfg = {'version': 13.2} - self.assertEqual(None, network_state.parse_net_config_data(ncfg)) + with self.assertRaises(RuntimeError): + network_state.parse_net_config_data(ncfg) def test_version_2_passes_self_as_config(self): ncfg = {'version': 2, 'otherconfig': {}, 'somemore': [1, 2, 3]} diff --git a/tests/unittests/test_net.py b/tests/unittests/test_net.py index 1aab51ee..43e209c1 100644 --- a/tests/unittests/test_net.py +++ b/tests/unittests/test_net.py @@ -5277,7 +5277,7 @@ class TestNetRenderers(CiTestCase): # available should only be called until one is found. m_eni_avail.return_value = True m_sysc_avail.side_effect = Exception("Should not call me") - found = renderers.search(priority=['eni', 'sysconfig'], first=True) + found = renderers.search(priority=['eni', 'sysconfig'], first=True)[0] self.assertEqual(['eni'], [found[0]]) @mock.patch("cloudinit.net.renderers.sysconfig.available") diff --git a/tests/unittests/test_net_activators.py b/tests/unittests/test_net_activators.py new file mode 100644 index 00000000..f11486ff --- /dev/null +++ b/tests/unittests/test_net_activators.py @@ -0,0 +1,177 @@ +from collections import namedtuple +from unittest.mock import patch + +import pytest + +from cloudinit.net.activators import ( + DEFAULT_PRIORITY, + search_activator, + select_activator, +) +from cloudinit.net.activators import ( + IfUpDownActivator, + NetplanActivator, + NetworkManagerActivator +) +from cloudinit.net.network_state import parse_net_config_data +from cloudinit.safeyaml import load + + +V1_CONFIG = """\ +version: 1 +config: +- type: physical + name: eth0 +- type: physical + name: eth1 +""" + +V2_CONFIG = """\ +version: 2 +ethernets: + eth0: + dhcp4: true + eth1: + dhcp4: true +""" + +IF_UP_DOWN_AVAILABLE_CALLS = [ + (('ifquery',), {'search': ['/sbin', '/usr/sbin'], 'target': None}), + (('ifup',), {'search': ['/sbin', '/usr/sbin'], 'target': None}), + (('ifdown',), {'search': ['/sbin', '/usr/sbin'], 'target': None}), +] + +IF_UP_DOWN_CALL_LIST = [ + ((['ifup', 'eth0'], ), {}), + ((['ifup', 'eth1'], ), {}), +] + +NETPLAN_AVAILABLE_CALLS = [ + (('netplan',), {'search': ['/usr/sbin', '/sbin'], 'target': None}), +] + +NETPLAN_CALL_LIST = [ + ((['netplan', 'apply'], ), {'capture': True}), +] + +NETWORK_MANAGER_AVAILABLE_CALLS = [ + (('nmcli',), {'target': None}), +] + +NETWORK_MANAGER_CALL_LIST = [ + ((['nmcli', 'connection', 'up', 'eth0'], ), {}), + ((['nmcli', 'connection', 'up', 'eth1'], ), {}), +] + + +@pytest.yield_fixture +def available_mocks(): + mocks = namedtuple('Mocks', 'm_which, m_file') + with patch('cloudinit.subp.which', return_value=True) as m_which: + with patch('os.path.isfile', return_value=True) as m_file: + yield mocks(m_which, m_file) + + +@pytest.yield_fixture +def unavailable_mocks(): + mocks = namedtuple('Mocks', 'm_which, m_file') + with patch('cloudinit.subp.which', return_value=False) as m_which: + with patch('os.path.isfile', return_value=False) as m_file: + yield mocks(m_which, m_file) + + +class TestSearchAndSelect: + def test_defaults(self, available_mocks): + resp = search_activator() + assert resp == DEFAULT_PRIORITY + + activator = select_activator() + assert activator == DEFAULT_PRIORITY[0] + + def test_priority(self, available_mocks): + new_order = [NetplanActivator, NetworkManagerActivator] + resp = search_activator(priority=new_order) + assert resp == new_order + + activator = select_activator(priority=new_order) + assert activator == new_order[0] + + def test_target(self, available_mocks): + search_activator(target='/tmp') + assert '/tmp' == available_mocks.m_which.call_args[1]['target'] + + select_activator(target='/tmp') + assert '/tmp' == available_mocks.m_which.call_args[1]['target'] + + @patch('cloudinit.net.activators.IfUpDownActivator.available', + return_value=False) + def test_first_not_available(self, m_available, available_mocks): + resp = search_activator() + assert resp == DEFAULT_PRIORITY[1:] + + resp = select_activator() + assert resp == DEFAULT_PRIORITY[1] + + def test_priority_not_exist(self, available_mocks): + with pytest.raises(ValueError): + search_activator(priority=['spam', 'eggs']) + with pytest.raises(ValueError): + select_activator(priority=['spam', 'eggs']) + + def test_none_available(self, unavailable_mocks): + resp = search_activator() + assert resp == [] + + with pytest.raises(RuntimeError): + select_activator() + + +@pytest.mark.parametrize('activator, available_calls, expected_call_list', [ + (IfUpDownActivator, IF_UP_DOWN_AVAILABLE_CALLS, IF_UP_DOWN_CALL_LIST), + (NetplanActivator, NETPLAN_AVAILABLE_CALLS, NETPLAN_CALL_LIST), + (NetworkManagerActivator, NETWORK_MANAGER_AVAILABLE_CALLS, + NETWORK_MANAGER_CALL_LIST), +]) +class TestIfUpDownActivator: + def test_available( + self, activator, available_calls, expected_call_list, available_mocks + ): + activator.available() + assert available_mocks.m_which.call_args_list == available_calls + + @patch('cloudinit.subp.subp', return_value=('', '')) + def test_bring_up_interface( + self, m_subp, activator, available_calls, expected_call_list, + available_mocks + ): + activator.bring_up_interface('eth0') + assert len(m_subp.call_args_list) == 1 + assert m_subp.call_args_list[0] == expected_call_list[0] + + @patch('cloudinit.subp.subp', return_value=('', '')) + def test_bring_up_interfaces( + self, m_subp, activator, available_calls, expected_call_list, + available_mocks + ): + activator.bring_up_interfaces(['eth0', 'eth1']) + assert expected_call_list == m_subp.call_args_list + + @patch('cloudinit.subp.subp', return_value=('', '')) + def test_bring_up_all_interfaces_v1( + self, m_subp, activator, available_calls, expected_call_list, + available_mocks + ): + network_state = parse_net_config_data(load(V1_CONFIG)) + activator.bring_up_all_interfaces(network_state) + for call in m_subp.call_args_list: + assert call in expected_call_list + + @patch('cloudinit.subp.subp', return_value=('', '')) + def test_bring_up_all_interfaces_v2( + self, m_subp, activator, available_calls, expected_call_list, + available_mocks + ): + network_state = parse_net_config_data(load(V2_CONFIG)) + activator.bring_up_all_interfaces(network_state) + for call in m_subp.call_args_list: + assert call in expected_call_list |