cloud-init-dev team mailing list archive
-
cloud-init-dev team
-
Mailing list archive
-
Message #05436
[Merge] ~raharper/cloud-init:ubuntu-devel-newupstream-20180905 into cloud-init:ubuntu/devel
Ryan Harper has proposed merging ~raharper/cloud-init:ubuntu-devel-newupstream-20180905 into cloud-init:ubuntu/devel.
Commit message:
cloud-init (18.3-38-gd47d404e-0ubuntu1) cosmic; urgency=medium
* New upstream snapshot.
- tests: print failed testname instead of docstring upon failure
- tests: Disallow use of util.subp except for where needed.
- sysconfig: refactor sysconfig to accept distro specific templates paths
-- Ryan Harper <ryan.harper@xxxxxxxxxxxxx> Wed, 05 Sep 2018 16:03:46 -0500
Requested reviews:
cloud-init commiters (cloud-init-dev)
For more details, see:
https://code.launchpad.net/~raharper/cloud-init/+git/cloud-init/+merge/354360
--
Your team cloud-init commiters is requested to review the proposed merge of ~raharper/cloud-init:ubuntu-devel-newupstream-20180905 into cloud-init:ubuntu/devel.
diff --git a/cloudinit/analyze/tests/test_dump.py b/cloudinit/analyze/tests/test_dump.py
index f4c4284..db2a667 100644
--- a/cloudinit/analyze/tests/test_dump.py
+++ b/cloudinit/analyze/tests/test_dump.py
@@ -5,8 +5,8 @@ from textwrap import dedent
from cloudinit.analyze.dump import (
dump_events, parse_ci_logline, parse_timestamp)
-from cloudinit.util import subp, write_file
-from cloudinit.tests.helpers import CiTestCase
+from cloudinit.util import which, write_file
+from cloudinit.tests.helpers import CiTestCase, mock, skipIf
class TestParseTimestamp(CiTestCase):
@@ -15,21 +15,9 @@ class TestParseTimestamp(CiTestCase):
"""Logs with cloud-init detailed formats will be properly parsed."""
trusty_fmt = '%Y-%m-%d %H:%M:%S,%f'
trusty_stamp = '2016-09-12 14:39:20,839'
-
- parsed = parse_timestamp(trusty_stamp)
-
- # convert ourselves
dt = datetime.strptime(trusty_stamp, trusty_fmt)
- expected = float(dt.strftime('%s.%f'))
-
- # use date(1)
- out, _err = subp(['date', '+%s.%3N', '-d', trusty_stamp])
- timestamp = out.strip()
- date_ts = float(timestamp)
-
- self.assertEqual(expected, parsed)
- self.assertEqual(expected, date_ts)
- self.assertEqual(date_ts, parsed)
+ self.assertEqual(
+ float(dt.strftime('%s.%f')), parse_timestamp(trusty_stamp))
def test_parse_timestamp_handles_syslog_adding_year(self):
"""Syslog timestamps lack a year. Add year and properly parse."""
@@ -39,17 +27,9 @@ class TestParseTimestamp(CiTestCase):
# convert stamp ourselves by adding the missing year value
year = datetime.now().year
dt = datetime.strptime(syslog_stamp + " " + str(year), syslog_fmt)
- expected = float(dt.strftime('%s.%f'))
- parsed = parse_timestamp(syslog_stamp)
-
- # use date(1)
- out, _ = subp(['date', '+%s.%3N', '-d', syslog_stamp])
- timestamp = out.strip()
- date_ts = float(timestamp)
-
- self.assertEqual(expected, parsed)
- self.assertEqual(expected, date_ts)
- self.assertEqual(date_ts, parsed)
+ self.assertEqual(
+ float(dt.strftime('%s.%f')),
+ parse_timestamp(syslog_stamp))
def test_parse_timestamp_handles_journalctl_format_adding_year(self):
"""Journalctl precise timestamps lack a year. Add year and parse."""
@@ -59,37 +39,22 @@ class TestParseTimestamp(CiTestCase):
# convert stamp ourselves by adding the missing year value
year = datetime.now().year
dt = datetime.strptime(journal_stamp + " " + str(year), journal_fmt)
- expected = float(dt.strftime('%s.%f'))
- parsed = parse_timestamp(journal_stamp)
-
- # use date(1)
- out, _ = subp(['date', '+%s.%6N', '-d', journal_stamp])
- timestamp = out.strip()
- date_ts = float(timestamp)
-
- self.assertEqual(expected, parsed)
- self.assertEqual(expected, date_ts)
- self.assertEqual(date_ts, parsed)
+ self.assertEqual(
+ float(dt.strftime('%s.%f')), parse_timestamp(journal_stamp))
+ @skipIf(not which("date"), "'date' command not available.")
def test_parse_unexpected_timestamp_format_with_date_command(self):
- """Dump sends unexpected timestamp formats to data for processing."""
+ """Dump sends unexpected timestamp formats to date for processing."""
new_fmt = '%H:%M %m/%d %Y'
new_stamp = '17:15 08/08'
-
# convert stamp ourselves by adding the missing year value
year = datetime.now().year
dt = datetime.strptime(new_stamp + " " + str(year), new_fmt)
- expected = float(dt.strftime('%s.%f'))
- parsed = parse_timestamp(new_stamp)
# use date(1)
- out, _ = subp(['date', '+%s.%6N', '-d', new_stamp])
- timestamp = out.strip()
- date_ts = float(timestamp)
-
- self.assertEqual(expected, parsed)
- self.assertEqual(expected, date_ts)
- self.assertEqual(date_ts, parsed)
+ with self.allow_subp(["date"]):
+ self.assertEqual(
+ float(dt.strftime('%s.%f')), parse_timestamp(new_stamp))
class TestParseCILogLine(CiTestCase):
@@ -135,7 +100,9 @@ class TestParseCILogLine(CiTestCase):
'timestamp': timestamp}
self.assertEqual(expected, parse_ci_logline(line))
- def test_parse_logline_returns_event_for_finish_events(self):
+ @mock.patch("cloudinit.analyze.dump.parse_timestamp_from_date")
+ def test_parse_logline_returns_event_for_finish_events(self,
+ m_parse_from_date):
"""parse_ci_logline returns a finish event for a parsed log line."""
line = ('2016-08-30 21:53:25.972325+00:00 y1 [CLOUDINIT]'
' handlers.py[DEBUG]: finish: modules-final: SUCCESS: running'
@@ -147,7 +114,10 @@ class TestParseCILogLine(CiTestCase):
'origin': 'cloudinit',
'result': 'SUCCESS',
'timestamp': 1472594005.972}
+ m_parse_from_date.return_value = "1472594005.972"
self.assertEqual(expected, parse_ci_logline(line))
+ m_parse_from_date.assert_has_calls(
+ [mock.call("2016-08-30 21:53:25.972325+00:00")])
SAMPLE_LOGS = dedent("""\
@@ -162,10 +132,16 @@ Nov 03 06:51:06.074410 x2 cloud-init[106]: [CLOUDINIT] util.py[DEBUG]:\
class TestDumpEvents(CiTestCase):
maxDiff = None
- def test_dump_events_with_rawdata(self):
+ @mock.patch("cloudinit.analyze.dump.parse_timestamp_from_date")
+ def test_dump_events_with_rawdata(self, m_parse_from_date):
"""Rawdata is split and parsed into a tuple of events and data"""
+ m_parse_from_date.return_value = "1472594005.972"
events, data = dump_events(rawdata=SAMPLE_LOGS)
expected_data = SAMPLE_LOGS.splitlines()
+ self.assertEqual(
+ [mock.call("2016-08-30 21:53:25.972325+00:00")],
+ m_parse_from_date.call_args_list)
+ self.assertEqual(expected_data, data)
year = datetime.now().year
dt1 = datetime.strptime(
'Nov 03 06:51:06.074410 %d' % year, '%b %d %H:%M:%S.%f %Y')
@@ -183,12 +159,14 @@ class TestDumpEvents(CiTestCase):
'result': 'SUCCESS',
'timestamp': 1472594005.972}]
self.assertEqual(expected_events, events)
- self.assertEqual(expected_data, data)
- def test_dump_events_with_cisource(self):
+ @mock.patch("cloudinit.analyze.dump.parse_timestamp_from_date")
+ def test_dump_events_with_cisource(self, m_parse_from_date):
"""Cisource file is read and parsed into a tuple of events and data."""
tmpfile = self.tmp_path('logfile')
write_file(tmpfile, SAMPLE_LOGS)
+ m_parse_from_date.return_value = 1472594005.972
+
events, data = dump_events(cisource=open(tmpfile))
year = datetime.now().year
dt1 = datetime.strptime(
@@ -208,3 +186,5 @@ class TestDumpEvents(CiTestCase):
'timestamp': 1472594005.972}]
self.assertEqual(expected_events, events)
self.assertEqual(SAMPLE_LOGS.splitlines(), [d.strip() for d in data])
+ m_parse_from_date.assert_has_calls(
+ [mock.call("2016-08-30 21:53:25.972325+00:00")])
diff --git a/cloudinit/cmd/devel/net_convert.py b/cloudinit/cmd/devel/net_convert.py
index 271dc5e..a0f58a0 100755
--- a/cloudinit/cmd/devel/net_convert.py
+++ b/cloudinit/cmd/devel/net_convert.py
@@ -10,6 +10,7 @@ import yaml
from cloudinit.sources.helpers import openstack
from cloudinit.sources import DataSourceAzure as azure
+from cloudinit import distros
from cloudinit.net import eni, netplan, network_state, sysconfig
from cloudinit import log
@@ -36,6 +37,11 @@ def get_parser(parser=None):
metavar="PATH",
help="directory to place output in",
required=True)
+ parser.add_argument("-D", "--distro",
+ choices=[item for sublist in
+ distros.OSFAMILIES.values()
+ for item in sublist],
+ required=True)
parser.add_argument("-m", "--mac",
metavar="name,mac",
action='append',
@@ -96,14 +102,20 @@ def handle_args(name, args):
sys.stderr.write('\n'.join([
"", "Internal State",
yaml.dump(ns, default_flow_style=False, indent=4), ""]))
+ distro_cls = distros.fetch(args.distro)
+ distro = distro_cls(args.distro, {}, None)
+ config = {}
if args.output_kind == "eni":
r_cls = eni.Renderer
+ config = distro.renderer_configs.get('eni')
elif args.output_kind == "netplan":
r_cls = netplan.Renderer
+ config = distro.renderer_configs.get('netplan')
else:
r_cls = sysconfig.Renderer
+ config = distro.renderer_configs.get('sysconfig')
- r = r_cls()
+ r = r_cls(config=config)
sys.stderr.write(''.join([
"Read input format '%s' from '%s'.\n" % (
args.kind, args.network_data.name),
diff --git a/cloudinit/cmd/tests/test_status.py b/cloudinit/cmd/tests/test_status.py
index 37a8993..aded858 100644
--- a/cloudinit/cmd/tests/test_status.py
+++ b/cloudinit/cmd/tests/test_status.py
@@ -39,7 +39,8 @@ class TestStatus(CiTestCase):
ensure_file(self.disable_file) # Create the ignored disable file
(is_disabled, reason) = wrap_and_call(
'cloudinit.cmd.status',
- {'uses_systemd': False},
+ {'uses_systemd': False,
+ 'get_cmdline': "root=/dev/my-root not-important"},
status._is_cloudinit_disabled, self.disable_file, self.paths)
self.assertFalse(
is_disabled, 'expected enabled cloud-init on sysvinit')
@@ -50,7 +51,8 @@ class TestStatus(CiTestCase):
ensure_file(self.disable_file) # Create observed disable file
(is_disabled, reason) = wrap_and_call(
'cloudinit.cmd.status',
- {'uses_systemd': True},
+ {'uses_systemd': True,
+ 'get_cmdline': "root=/dev/my-root not-important"},
status._is_cloudinit_disabled, self.disable_file, self.paths)
self.assertTrue(is_disabled, 'expected disabled cloud-init')
self.assertEqual(
diff --git a/cloudinit/config/tests/test_snap.py b/cloudinit/config/tests/test_snap.py
index 34c80f1..3c47289 100644
--- a/cloudinit/config/tests/test_snap.py
+++ b/cloudinit/config/tests/test_snap.py
@@ -162,6 +162,7 @@ class TestAddAssertions(CiTestCase):
class TestRunCommands(CiTestCase):
with_logs = True
+ allowed_subp = [CiTestCase.SUBP_SHELL_TRUE]
def setUp(self):
super(TestRunCommands, self).setUp()
@@ -424,8 +425,10 @@ class TestHandle(CiTestCase):
'snap': {'commands': ['echo "HI" >> %s' % outfile,
'echo "MOM" >> %s' % outfile]}}
mock_path = 'cloudinit.config.cc_snap.sys.stderr'
- with mock.patch(mock_path, new_callable=StringIO):
- handle('snap', cfg=cfg, cloud=None, log=self.logger, args=None)
+ with self.allow_subp([CiTestCase.SUBP_SHELL_TRUE]):
+ with mock.patch(mock_path, new_callable=StringIO):
+ handle('snap', cfg=cfg, cloud=None, log=self.logger, args=None)
+
self.assertEqual('HI\nMOM\n', util.load_file(outfile))
@mock.patch('cloudinit.config.cc_snap.util.subp')
diff --git a/cloudinit/config/tests/test_ubuntu_advantage.py b/cloudinit/config/tests/test_ubuntu_advantage.py
index f1beeff..b7cf9be 100644
--- a/cloudinit/config/tests/test_ubuntu_advantage.py
+++ b/cloudinit/config/tests/test_ubuntu_advantage.py
@@ -23,6 +23,7 @@ class FakeCloud(object):
class TestRunCommands(CiTestCase):
with_logs = True
+ allowed_subp = [CiTestCase.SUBP_SHELL_TRUE]
def setUp(self):
super(TestRunCommands, self).setUp()
@@ -234,8 +235,10 @@ class TestHandle(CiTestCase):
'ubuntu-advantage': {'commands': ['echo "HI" >> %s' % outfile,
'echo "MOM" >> %s' % outfile]}}
mock_path = '%s.sys.stderr' % MPATH
- with mock.patch(mock_path, new_callable=StringIO):
- handle('nomatter', cfg=cfg, cloud=None, log=self.logger, args=None)
+ with self.allow_subp([CiTestCase.SUBP_SHELL_TRUE]):
+ with mock.patch(mock_path, new_callable=StringIO):
+ handle('nomatter', cfg=cfg, cloud=None, log=self.logger,
+ args=None)
self.assertEqual('HI\nMOM\n', util.load_file(outfile))
diff --git a/cloudinit/distros/__init__.py b/cloudinit/distros/__init__.py
index fde054e..d9101ce 100755
--- a/cloudinit/distros/__init__.py
+++ b/cloudinit/distros/__init__.py
@@ -91,7 +91,7 @@ class Distro(object):
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=network_config)
+ renderer.render_network_config(network_config)
return []
def _find_tz_file(self, tz):
diff --git a/cloudinit/distros/opensuse.py b/cloudinit/distros/opensuse.py
index 9f90e95..1fe896a 100644
--- a/cloudinit/distros/opensuse.py
+++ b/cloudinit/distros/opensuse.py
@@ -28,13 +28,23 @@ class Distro(distros.Distro):
hostname_conf_fn = '/etc/HOSTNAME'
init_cmd = ['service']
locale_conf_fn = '/etc/sysconfig/language'
- network_conf_fn = '/etc/sysconfig/network'
+ network_conf_fn = '/etc/sysconfig/network/config'
network_script_tpl = '/etc/sysconfig/network/ifcfg-%s'
resolve_conf_fn = '/etc/resolv.conf'
route_conf_tpl = '/etc/sysconfig/network/ifroute-%s'
systemd_hostname_conf_fn = '/etc/hostname'
systemd_locale_conf_fn = '/etc/locale.conf'
tz_local_fn = '/etc/localtime'
+ renderer_configs = {
+ 'sysconfig': {
+ 'control': 'etc/sysconfig/network/config',
+ 'iface_templates': '%(base)s/network/ifcfg-%(name)s',
+ 'route_templates': {
+ 'ipv4': '%(base)s/network/ifroute-%(name)s',
+ 'ipv6': '%(base)s/network/ifroute-%(name)s',
+ }
+ }
+ }
def __init__(self, name, cfg, paths):
distros.Distro.__init__(self, name, cfg, paths)
@@ -208,6 +218,9 @@ class Distro(distros.Distro):
nameservers, searchservers)
return dev_names
+ 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/rhel.py b/cloudinit/distros/rhel.py
index 1fecb61..ff51343 100644
--- a/cloudinit/distros/rhel.py
+++ b/cloudinit/distros/rhel.py
@@ -39,6 +39,16 @@ class Distro(distros.Distro):
resolve_conf_fn = "/etc/resolv.conf"
tz_local_fn = "/etc/localtime"
usr_lib_exec = "/usr/libexec"
+ renderer_configs = {
+ 'sysconfig': {
+ 'control': 'etc/sysconfig/network',
+ 'iface_templates': '%(base)s/network-scripts/ifcfg-%(name)s',
+ 'route_templates': {
+ 'ipv4': '%(base)s/network-scripts/route-%(name)s',
+ 'ipv6': '%(base)s/network-scripts/route6-%(name)s'
+ }
+ }
+ }
def __init__(self, name, cfg, paths):
distros.Distro.__init__(self, name, cfg, paths)
diff --git a/cloudinit/net/eni.py b/cloudinit/net/eni.py
index 80be242..c6f631a 100644
--- a/cloudinit/net/eni.py
+++ b/cloudinit/net/eni.py
@@ -480,7 +480,7 @@ class Renderer(renderer.Renderer):
return '\n\n'.join(['\n'.join(s) for s in sections]) + "\n"
- def render_network_state(self, network_state, target=None):
+ def render_network_state(self, network_state, templates=None, target=None):
fpeni = util.target_path(target, self.eni_path)
util.ensure_dir(os.path.dirname(fpeni))
header = self.eni_header if self.eni_header else ""
diff --git a/cloudinit/net/netplan.py b/cloudinit/net/netplan.py
index 6352e78..bc1087f 100644
--- a/cloudinit/net/netplan.py
+++ b/cloudinit/net/netplan.py
@@ -189,7 +189,7 @@ class Renderer(renderer.Renderer):
self._postcmds = config.get('postcmds', False)
self.clean_default = config.get('clean_default', True)
- def render_network_state(self, network_state, target):
+ def render_network_state(self, network_state, templates=None, target=None):
# check network state for version
# if v2, then extract network_state.config
# else render_v2_from_state
diff --git a/cloudinit/net/renderer.py b/cloudinit/net/renderer.py
index 57652e2..5f32e90 100644
--- a/cloudinit/net/renderer.py
+++ b/cloudinit/net/renderer.py
@@ -45,11 +45,14 @@ class Renderer(object):
return content.getvalue()
@abc.abstractmethod
- def render_network_state(self, network_state, target=None):
+ def render_network_state(self, network_state, templates=None,
+ target=None):
"""Render network state."""
- def render_network_config(self, network_config, target=None):
+ def render_network_config(self, network_config, templates=None,
+ target=None):
return self.render_network_state(
- network_state=parse_net_config_data(network_config), target=target)
+ network_state=parse_net_config_data(network_config),
+ templates=templates, target=target)
# vi: ts=4 expandtab
diff --git a/cloudinit/net/sysconfig.py b/cloudinit/net/sysconfig.py
index 3d71923..66e970e 100644
--- a/cloudinit/net/sysconfig.py
+++ b/cloudinit/net/sysconfig.py
@@ -91,19 +91,20 @@ class ConfigMap(object):
class Route(ConfigMap):
"""Represents a route configuration."""
- route_fn_tpl_ipv4 = '%(base)s/network-scripts/route-%(name)s'
- route_fn_tpl_ipv6 = '%(base)s/network-scripts/route6-%(name)s'
-
- def __init__(self, route_name, base_sysconf_dir):
+ def __init__(self, route_name, base_sysconf_dir,
+ ipv4_tpl, ipv6_tpl):
super(Route, self).__init__()
self.last_idx = 1
self.has_set_default_ipv4 = False
self.has_set_default_ipv6 = False
self._route_name = route_name
self._base_sysconf_dir = base_sysconf_dir
+ self.route_fn_tpl_ipv4 = ipv4_tpl
+ self.route_fn_tpl_ipv6 = ipv6_tpl
def copy(self):
- r = Route(self._route_name, self._base_sysconf_dir)
+ r = Route(self._route_name, self._base_sysconf_dir,
+ self.route_fn_tpl_ipv4, self.route_fn_tpl_ipv6)
r._conf = self._conf.copy()
r.last_idx = self.last_idx
r.has_set_default_ipv4 = self.has_set_default_ipv4
@@ -169,18 +170,22 @@ class Route(ConfigMap):
class NetInterface(ConfigMap):
"""Represents a sysconfig/networking-script (and its config + children)."""
- iface_fn_tpl = '%(base)s/network-scripts/ifcfg-%(name)s'
-
iface_types = {
'ethernet': 'Ethernet',
'bond': 'Bond',
'bridge': 'Bridge',
}
- def __init__(self, iface_name, base_sysconf_dir, kind='ethernet'):
+ def __init__(self, iface_name, base_sysconf_dir, templates,
+ kind='ethernet'):
super(NetInterface, self).__init__()
self.children = []
- self.routes = Route(iface_name, base_sysconf_dir)
+ self.templates = templates
+ route_tpl = self.templates.get('route_templates')
+ self.routes = Route(iface_name, base_sysconf_dir,
+ ipv4_tpl=route_tpl.get('ipv4'),
+ ipv6_tpl=route_tpl.get('ipv6'))
+ self.iface_fn_tpl = self.templates.get('iface_templates')
self.kind = kind
self._iface_name = iface_name
@@ -213,7 +218,8 @@ class NetInterface(ConfigMap):
'name': self.name})
def copy(self, copy_children=False, copy_routes=False):
- c = NetInterface(self.name, self._base_sysconf_dir, kind=self._kind)
+ c = NetInterface(self.name, self._base_sysconf_dir,
+ self.templates, kind=self._kind)
c._conf = self._conf.copy()
if copy_children:
c.children = list(self.children)
@@ -251,6 +257,8 @@ class Renderer(renderer.Renderer):
('bridge_bridgeprio', 'PRIO'),
])
+ templates = {}
+
def __init__(self, config=None):
if not config:
config = {}
@@ -261,6 +269,11 @@ class Renderer(renderer.Renderer):
nm_conf_path = 'etc/NetworkManager/conf.d/99-cloud-init.conf'
self.networkmanager_conf_path = config.get('networkmanager_conf_path',
nm_conf_path)
+ self.templates = {
+ 'control': config.get('control'),
+ 'iface_templates': config.get('iface_templates'),
+ 'route_templates': config.get('route_templates'),
+ }
@classmethod
def _render_iface_shared(cls, iface, iface_cfg):
@@ -512,7 +525,7 @@ class Renderer(renderer.Renderer):
return content_str
@staticmethod
- def _render_networkmanager_conf(network_state):
+ def _render_networkmanager_conf(network_state, templates=None):
content = networkmanager_conf.NetworkManagerConf("")
# If DNS server information is provided, configure
@@ -556,14 +569,17 @@ class Renderer(renderer.Renderer):
cls._render_subnet_routes(iface_cfg, route_cfg, iface_subnets)
@classmethod
- def _render_sysconfig(cls, base_sysconf_dir, network_state):
+ def _render_sysconfig(cls, base_sysconf_dir, network_state,
+ templates=None):
'''Given state, return /etc/sysconfig files + contents'''
+ if not templates:
+ templates = cls.templates
iface_contents = {}
for iface in network_state.iter_interfaces():
if iface['type'] == "loopback":
continue
iface_name = iface['name']
- iface_cfg = NetInterface(iface_name, base_sysconf_dir)
+ iface_cfg = NetInterface(iface_name, base_sysconf_dir, templates)
cls._render_iface_shared(iface, iface_cfg)
iface_contents[iface_name] = iface_cfg
cls._render_physical_interfaces(network_state, iface_contents)
@@ -578,17 +594,21 @@ class Renderer(renderer.Renderer):
if iface_cfg:
contents[iface_cfg.path] = iface_cfg.to_string()
if iface_cfg.routes:
- contents[iface_cfg.routes.path_ipv4] = \
- iface_cfg.routes.to_string("ipv4")
- contents[iface_cfg.routes.path_ipv6] = \
- iface_cfg.routes.to_string("ipv6")
+ for cpath, proto in zip([iface_cfg.routes.path_ipv4,
+ iface_cfg.routes.path_ipv6],
+ ["ipv4", "ipv6"]):
+ if cpath not in contents:
+ contents[cpath] = iface_cfg.routes.to_string(proto)
return contents
- def render_network_state(self, network_state, target=None):
+ def render_network_state(self, network_state, templates=None, target=None):
+ if not templates:
+ templates = self.templates
file_mode = 0o644
base_sysconf_dir = util.target_path(target, self.sysconf_dir)
for path, data in self._render_sysconfig(base_sysconf_dir,
- network_state).items():
+ network_state,
+ templates=templates).items():
util.write_file(path, data, file_mode)
if self.dns_path:
dns_path = util.target_path(target, self.dns_path)
@@ -598,7 +618,8 @@ class Renderer(renderer.Renderer):
if self.networkmanager_conf_path:
nm_conf_path = util.target_path(target,
self.networkmanager_conf_path)
- nm_conf_content = self._render_networkmanager_conf(network_state)
+ nm_conf_content = self._render_networkmanager_conf(network_state,
+ templates)
if nm_conf_content:
util.write_file(nm_conf_path, nm_conf_content, file_mode)
if self.netrules_path:
@@ -606,13 +627,16 @@ class Renderer(renderer.Renderer):
netrules_path = util.target_path(target, self.netrules_path)
util.write_file(netrules_path, netrules_content, file_mode)
- # always write /etc/sysconfig/network configuration
- sysconfig_path = util.target_path(target, "etc/sysconfig/network")
- netcfg = [_make_header(), 'NETWORKING=yes']
- if network_state.use_ipv6:
- netcfg.append('NETWORKING_IPV6=yes')
- netcfg.append('IPV6_AUTOCONF=no')
- util.write_file(sysconfig_path, "\n".join(netcfg) + "\n", file_mode)
+ sysconfig_path = util.target_path(target, templates.get('control'))
+ # Distros configuring /etc/sysconfig/network as a file e.g. Centos
+ if sysconfig_path.endswith('network'):
+ util.ensure_dir(os.path.dirname(sysconfig_path))
+ netcfg = [_make_header(), 'NETWORKING=yes']
+ if network_state.use_ipv6:
+ netcfg.append('NETWORKING_IPV6=yes')
+ netcfg.append('IPV6_AUTOCONF=no')
+ util.write_file(sysconfig_path,
+ "\n".join(netcfg) + "\n", file_mode)
def available(target=None):
diff --git a/cloudinit/net/tests/test_init.py b/cloudinit/net/tests/test_init.py
index 5c017d1..8b444f1 100644
--- a/cloudinit/net/tests/test_init.py
+++ b/cloudinit/net/tests/test_init.py
@@ -199,6 +199,8 @@ class TestGenerateFallbackConfig(CiTestCase):
self.sysdir = self.tmp_dir() + '/'
self.m_sys_path.return_value = self.sysdir
self.addCleanup(sys_mock.stop)
+ self.add_patch('cloudinit.net.util.is_container', 'm_is_container',
+ return_value=False)
self.add_patch('cloudinit.net.util.udevadm_settle', 'm_settle')
def test_generate_fallback_finds_connected_eth_with_mac(self):
diff --git a/cloudinit/sources/DataSourceAltCloud.py b/cloudinit/sources/DataSourceAltCloud.py
index 24fd65f..8cd312d 100644
--- a/cloudinit/sources/DataSourceAltCloud.py
+++ b/cloudinit/sources/DataSourceAltCloud.py
@@ -181,27 +181,18 @@ class DataSourceAltCloud(sources.DataSource):
# modprobe floppy
try:
- cmd = CMD_PROBE_FLOPPY
- (cmd_out, _err) = util.subp(cmd)
- LOG.debug('Command: %s\nOutput%s', ' '.join(cmd), cmd_out)
+ modprobe_floppy()
except ProcessExecutionError as e:
- util.logexc(LOG, 'Failed command: %s\n%s', ' '.join(cmd), e)
- return False
- except OSError as e:
- util.logexc(LOG, 'Failed command: %s\n%s', ' '.join(cmd), e)
+ util.logexc(LOG, 'Failed modprobe: %s', e)
return False
floppy_dev = '/dev/fd0'
# udevadm settle for floppy device
try:
- (cmd_out, _err) = util.udevadm_settle(exists=floppy_dev, timeout=5)
- LOG.debug('Command: %s\nOutput%s', ' '.join(cmd), cmd_out)
- except ProcessExecutionError as e:
- util.logexc(LOG, 'Failed command: %s\n%s', ' '.join(cmd), e)
- return False
- except OSError as e:
- util.logexc(LOG, 'Failed command: %s\n%s', ' '.join(cmd), e)
+ util.udevadm_settle(exists=floppy_dev, timeout=5)
+ except (ProcessExecutionError, OSError) as e:
+ util.logexc(LOG, 'Failed udevadm_settle: %s\n', e)
return False
try:
@@ -258,6 +249,11 @@ class DataSourceAltCloud(sources.DataSource):
return False
+def modprobe_floppy():
+ out, _err = util.subp(CMD_PROBE_FLOPPY)
+ LOG.debug('Command: %s\nOutput%s', ' '.join(CMD_PROBE_FLOPPY), out)
+
+
# Used to match classes to dependencies
# Source DataSourceAltCloud does not really depend on networking.
# In the future 'dsmode' like behavior can be added to offer user
diff --git a/cloudinit/sources/DataSourceSmartOS.py b/cloudinit/sources/DataSourceSmartOS.py
index ad8cfb9..593ac91 100644
--- a/cloudinit/sources/DataSourceSmartOS.py
+++ b/cloudinit/sources/DataSourceSmartOS.py
@@ -683,6 +683,18 @@ def jmc_client_factory(
raise ValueError("Unknown value for smartos_type: %s" % smartos_type)
+def identify_file(content_f):
+ cmd = ["file", "--brief", "--mime-type", content_f]
+ f_type = None
+ try:
+ (f_type, _err) = util.subp(cmd)
+ LOG.debug("script %s mime type is %s", content_f, f_type)
+ except util.ProcessExecutionError as e:
+ util.logexc(
+ LOG, ("Failed to identify script type for %s" % content_f, e))
+ return None if f_type is None else f_type.strip()
+
+
def write_boot_content(content, content_f, link=None, shebang=False,
mode=0o400):
"""
@@ -715,18 +727,11 @@ def write_boot_content(content, content_f, link=None, shebang=False,
util.write_file(content_f, content, mode=mode)
if shebang and not content.startswith("#!"):
- try:
- cmd = ["file", "--brief", "--mime-type", content_f]
- (f_type, _err) = util.subp(cmd)
- LOG.debug("script %s mime type is %s", content_f, f_type)
- if f_type.strip() == "text/plain":
- new_content = "\n".join(["#!/bin/bash", content])
- util.write_file(content_f, new_content, mode=mode)
- LOG.debug("added shebang to file %s", content_f)
-
- except Exception as e:
- util.logexc(LOG, ("Failed to identify script type for %s" %
- content_f, e))
+ f_type = identify_file(content_f)
+ if f_type == "text/plain":
+ util.write_file(
+ content_f, "\n".join(["#!/bin/bash", content]), mode=mode)
+ LOG.debug("added shebang to file %s", content_f)
if link:
try:
diff --git a/cloudinit/tests/helpers.py b/cloudinit/tests/helpers.py
index de24e25..42f56c2 100644
--- a/cloudinit/tests/helpers.py
+++ b/cloudinit/tests/helpers.py
@@ -14,11 +14,12 @@ import time
import mock
import six
import unittest2
+from unittest2.util import strclass
try:
- from contextlib import ExitStack
+ from contextlib import ExitStack, contextmanager
except ImportError:
- from contextlib2 import ExitStack
+ from contextlib2 import ExitStack, contextmanager
try:
from configparser import ConfigParser
@@ -33,6 +34,8 @@ from cloudinit import helpers as ch
from cloudinit.sources import DataSourceNone
from cloudinit import util
+_real_subp = util.subp
+
# Used for skipping tests
SkipTest = unittest2.SkipTest
skipIf = unittest2.skipIf
@@ -115,6 +118,9 @@ class TestCase(unittest2.TestCase):
super(TestCase, self).setUp()
self.reset_global_state()
+ def shortDescription(self):
+ return strclass(self.__class__) + '.' + self._testMethodName
+
def add_patch(self, target, attr, *args, **kwargs):
"""Patches specified target object and sets it as attr on test
instance also schedules cleanup"""
@@ -143,6 +149,17 @@ class CiTestCase(TestCase):
# Subclass overrides for specific test behavior
# Whether or not a unit test needs logfile setup
with_logs = False
+ allowed_subp = False
+ SUBP_SHELL_TRUE = "shell=true"
+
+ @contextmanager
+ def allow_subp(self, allowed_subp):
+ orig = self.allowed_subp
+ try:
+ self.allowed_subp = allowed_subp
+ yield
+ finally:
+ self.allowed_subp = orig
def setUp(self):
super(CiTestCase, self).setUp()
@@ -155,11 +172,41 @@ class CiTestCase(TestCase):
handler.setFormatter(formatter)
self.old_handlers = self.logger.handlers
self.logger.handlers = [handler]
+ if self.allowed_subp is True:
+ util.subp = _real_subp
+ else:
+ util.subp = self._fake_subp
+
+ def _fake_subp(self, *args, **kwargs):
+ if 'args' in kwargs:
+ cmd = kwargs['args']
+ else:
+ cmd = args[0]
+
+ if not isinstance(cmd, six.string_types):
+ cmd = cmd[0]
+ pass_through = False
+ if not isinstance(self.allowed_subp, (list, bool)):
+ raise TypeError("self.allowed_subp supports list or bool.")
+ if isinstance(self.allowed_subp, bool):
+ pass_through = self.allowed_subp
+ else:
+ pass_through = (
+ (cmd in self.allowed_subp) or
+ (self.SUBP_SHELL_TRUE in self.allowed_subp and
+ kwargs.get('shell')))
+ if pass_through:
+ return _real_subp(*args, **kwargs)
+ raise Exception(
+ "called subp. set self.allowed_subp=True to allow\n subp(%s)" %
+ ', '.join([str(repr(a)) for a in args] +
+ ["%s=%s" % (k, repr(v)) for k, v in kwargs.items()]))
def tearDown(self):
if self.with_logs:
# Remove the handler we setup
logging.getLogger().handlers = self.old_handlers
+ util.subp = _real_subp
super(CiTestCase, self).tearDown()
def tmp_dir(self, dir=None, cleanup=True):
@@ -326,6 +373,13 @@ class FilesystemMockingTestCase(ResourceUsingTestCase):
self.patchOS(root)
return root
+ @contextmanager
+ def reRooted(self, root=None):
+ try:
+ yield self.reRoot(root)
+ finally:
+ self.patched_funcs.close()
+
class HttprettyTestCase(CiTestCase):
# necessary as http_proxy gets in the way of httpretty
diff --git a/debian/changelog b/debian/changelog
index f303d70..5f738e9 100644
--- a/debian/changelog
+++ b/debian/changelog
@@ -1,3 +1,12 @@
+cloud-init (18.3-38-gd47d404e-0ubuntu1) cosmic; urgency=medium
+
+ * New upstream snapshot.
+ - tests: print failed testname instead of docstring upon failure
+ - tests: Disallow use of util.subp except for where needed.
+ - sysconfig: refactor sysconfig to accept distro specific templates paths
+
+ -- Ryan Harper <ryan.harper@xxxxxxxxxxxxx> Wed, 05 Sep 2018 16:03:46 -0500
+
cloud-init (18.3-35-g3f6d0972-0ubuntu1) cosmic; urgency=medium
* New upstream snapshot.
diff --git a/tests/unittests/test_datasource/test_altcloud.py b/tests/unittests/test_datasource/test_altcloud.py
index 3253f3a..ff35904 100644
--- a/tests/unittests/test_datasource/test_altcloud.py
+++ b/tests/unittests/test_datasource/test_altcloud.py
@@ -262,64 +262,56 @@ class TestUserDataRhevm(CiTestCase):
'''
Test to exercise method: DataSourceAltCloud.user_data_rhevm()
'''
- cmd_pass = ['true']
- cmd_fail = ['false']
- cmd_not_found = ['bogus bad command']
-
def setUp(self):
'''Set up.'''
self.paths = helpers.Paths({'cloud_dir': '/tmp'})
- self.mount_dir = tempfile.mkdtemp()
+ self.mount_dir = self.tmp_dir()
_write_user_data_files(self.mount_dir, 'test user data')
-
- def tearDown(self):
- # Reset
-
- _remove_user_data_files(self.mount_dir)
-
- # Attempt to remove the temp dir ignoring errors
- try:
- shutil.rmtree(self.mount_dir)
- except OSError:
- pass
-
- dsac.CLOUD_INFO_FILE = '/etc/sysconfig/cloud-info'
- dsac.CMD_PROBE_FLOPPY = ['modprobe', 'floppy']
- dsac.CMD_UDEVADM_SETTLE = ['udevadm', 'settle',
- '--quiet', '--timeout=5']
+ self.add_patch(
+ 'cloudinit.sources.DataSourceAltCloud.modprobe_floppy',
+ 'm_modprobe_floppy', return_value=None)
+ self.add_patch(
+ 'cloudinit.sources.DataSourceAltCloud.util.udevadm_settle',
+ 'm_udevadm_settle', return_value=('', ''))
+ self.add_patch(
+ 'cloudinit.sources.DataSourceAltCloud.util.mount_cb',
+ 'm_mount_cb')
def test_mount_cb_fails(self):
'''Test user_data_rhevm() where mount_cb fails.'''
- dsac.CMD_PROBE_FLOPPY = self.cmd_pass
+ self.m_mount_cb.side_effect = util.MountFailedError("Failed Mount")
dsrc = dsac.DataSourceAltCloud({}, None, self.paths)
self.assertEqual(False, dsrc.user_data_rhevm())
def test_modprobe_fails(self):
'''Test user_data_rhevm() where modprobe fails.'''
- dsac.CMD_PROBE_FLOPPY = self.cmd_fail
+ self.m_modprobe_floppy.side_effect = util.ProcessExecutionError(
+ "Failed modprobe")
dsrc = dsac.DataSourceAltCloud({}, None, self.paths)
self.assertEqual(False, dsrc.user_data_rhevm())
def test_no_modprobe_cmd(self):
'''Test user_data_rhevm() with no modprobe command.'''
- dsac.CMD_PROBE_FLOPPY = self.cmd_not_found
+ self.m_modprobe_floppy.side_effect = util.ProcessExecutionError(
+ "No such file or dir")
dsrc = dsac.DataSourceAltCloud({}, None, self.paths)
self.assertEqual(False, dsrc.user_data_rhevm())
def test_udevadm_fails(self):
'''Test user_data_rhevm() where udevadm fails.'''
- dsac.CMD_UDEVADM_SETTLE = self.cmd_fail
+ self.m_udevadm_settle.side_effect = util.ProcessExecutionError(
+ "Failed settle.")
dsrc = dsac.DataSourceAltCloud({}, None, self.paths)
self.assertEqual(False, dsrc.user_data_rhevm())
def test_no_udevadm_cmd(self):
'''Test user_data_rhevm() with no udevadm command.'''
- dsac.CMD_UDEVADM_SETTLE = self.cmd_not_found
+ self.m_udevadm_settle.side_effect = OSError("No such file or dir")
dsrc = dsac.DataSourceAltCloud({}, None, self.paths)
self.assertEqual(False, dsrc.user_data_rhevm())
diff --git a/tests/unittests/test_datasource/test_cloudsigma.py b/tests/unittests/test_datasource/test_cloudsigma.py
index f6a59b6..380ad1b 100644
--- a/tests/unittests/test_datasource/test_cloudsigma.py
+++ b/tests/unittests/test_datasource/test_cloudsigma.py
@@ -42,6 +42,9 @@ class CepkoMock(Cepko):
class DataSourceCloudSigmaTest(test_helpers.CiTestCase):
def setUp(self):
super(DataSourceCloudSigmaTest, self).setUp()
+ self.add_patch(
+ "cloudinit.sources.DataSourceCloudSigma.util.is_container",
+ "m_is_container", return_value=False)
self.paths = helpers.Paths({'run_dir': self.tmp_dir()})
self.datasource = DataSourceCloudSigma.DataSourceCloudSigma(
"", "", paths=self.paths)
diff --git a/tests/unittests/test_datasource/test_configdrive.py b/tests/unittests/test_datasource/test_configdrive.py
index 68400f2..b0abfc5 100644
--- a/tests/unittests/test_datasource/test_configdrive.py
+++ b/tests/unittests/test_datasource/test_configdrive.py
@@ -224,6 +224,9 @@ class TestConfigDriveDataSource(CiTestCase):
def setUp(self):
super(TestConfigDriveDataSource, self).setUp()
+ self.add_patch(
+ "cloudinit.sources.DataSourceConfigDrive.util.find_devs_with",
+ "m_find_devs_with", return_value=[])
self.tmp = self.tmp_dir()
def test_ec2_metadata(self):
@@ -642,7 +645,7 @@ class TestConvertNetworkData(CiTestCase):
routes)
eni_renderer = eni.Renderer()
eni_renderer.render_network_state(
- network_state.parse_net_config_data(ncfg), self.tmp)
+ network_state.parse_net_config_data(ncfg), target=self.tmp)
with open(os.path.join(self.tmp, "etc",
"network", "interfaces"), 'r') as f:
eni_rendering = f.read()
@@ -664,7 +667,7 @@ class TestConvertNetworkData(CiTestCase):
eni_renderer = eni.Renderer()
eni_renderer.render_network_state(
- network_state.parse_net_config_data(ncfg), self.tmp)
+ network_state.parse_net_config_data(ncfg), target=self.tmp)
with open(os.path.join(self.tmp, "etc",
"network", "interfaces"), 'r') as f:
eni_rendering = f.read()
@@ -695,7 +698,7 @@ class TestConvertNetworkData(CiTestCase):
known_macs=KNOWN_MACS)
eni_renderer = eni.Renderer()
eni_renderer.render_network_state(
- network_state.parse_net_config_data(ncfg), self.tmp)
+ network_state.parse_net_config_data(ncfg), target=self.tmp)
with open(os.path.join(self.tmp, "etc",
"network", "interfaces"), 'r') as f:
eni_rendering = f.read()
diff --git a/tests/unittests/test_datasource/test_nocloud.py b/tests/unittests/test_datasource/test_nocloud.py
index cdbd1e1..21931eb 100644
--- a/tests/unittests/test_datasource/test_nocloud.py
+++ b/tests/unittests/test_datasource/test_nocloud.py
@@ -25,6 +25,8 @@ class TestNoCloudDataSource(CiTestCase):
self.mocks.enter_context(
mock.patch.object(util, 'get_cmdline', return_value=self.cmdline))
+ self.mocks.enter_context(
+ mock.patch.object(util, 'read_dmi_data', return_value=None))
def test_nocloud_seed_dir(self):
md = {'instance-id': 'IID', 'dsmode': 'local'}
diff --git a/tests/unittests/test_datasource/test_opennebula.py b/tests/unittests/test_datasource/test_opennebula.py
index 36b4d77..6159101 100644
--- a/tests/unittests/test_datasource/test_opennebula.py
+++ b/tests/unittests/test_datasource/test_opennebula.py
@@ -43,6 +43,7 @@ DS_PATH = "cloudinit.sources.DataSourceOpenNebula"
class TestOpenNebulaDataSource(CiTestCase):
parsed_user = None
+ allowed_subp = ['bash']
def setUp(self):
super(TestOpenNebulaDataSource, self).setUp()
diff --git a/tests/unittests/test_datasource/test_ovf.py b/tests/unittests/test_datasource/test_ovf.py
index fc4eb36..9d52eb9 100644
--- a/tests/unittests/test_datasource/test_ovf.py
+++ b/tests/unittests/test_datasource/test_ovf.py
@@ -124,7 +124,9 @@ class TestDatasourceOVF(CiTestCase):
ds = self.datasource(sys_cfg={}, distro={}, paths=paths)
retcode = wrap_and_call(
'cloudinit.sources.DataSourceOVF',
- {'util.read_dmi_data': None},
+ {'util.read_dmi_data': None,
+ 'transport_iso9660': (False, None, None),
+ 'transport_vmware_guestd': (False, None, None)},
ds.get_data)
self.assertFalse(retcode, 'Expected False return from ds.get_data')
self.assertIn(
@@ -138,7 +140,9 @@ class TestDatasourceOVF(CiTestCase):
paths=paths)
retcode = wrap_and_call(
'cloudinit.sources.DataSourceOVF',
- {'util.read_dmi_data': 'vmware'},
+ {'util.read_dmi_data': 'vmware',
+ 'transport_iso9660': (False, None, None),
+ 'transport_vmware_guestd': (False, None, None)},
ds.get_data)
self.assertFalse(retcode, 'Expected False return from ds.get_data')
self.assertIn(
diff --git a/tests/unittests/test_datasource/test_smartos.py b/tests/unittests/test_datasource/test_smartos.py
index dca0b3d..46d67b9 100644
--- a/tests/unittests/test_datasource/test_smartos.py
+++ b/tests/unittests/test_datasource/test_smartos.py
@@ -20,10 +20,8 @@ import multiprocessing
import os
import os.path
import re
-import shutil
import signal
import stat
-import tempfile
import unittest2
import uuid
@@ -31,15 +29,27 @@ from cloudinit import serial
from cloudinit.sources import DataSourceSmartOS
from cloudinit.sources.DataSourceSmartOS import (
convert_smartos_network_data as convert_net,
- SMARTOS_ENV_KVM, SERIAL_DEVICE, get_smartos_environ)
+ SMARTOS_ENV_KVM, SERIAL_DEVICE, get_smartos_environ,
+ identify_file)
import six
from cloudinit import helpers as c_helpers
-from cloudinit.util import (b64e, subp)
+from cloudinit.util import (
+ b64e, subp, ProcessExecutionError, which, write_file)
-from cloudinit.tests.helpers import mock, FilesystemMockingTestCase, TestCase
+from cloudinit.tests.helpers import (
+ CiTestCase, mock, FilesystemMockingTestCase, skipIf)
+
+try:
+ import serial as _pyserial
+ assert _pyserial # avoid pyflakes error F401: import unused
+ HAS_PYSERIAL = True
+except ImportError:
+ HAS_PYSERIAL = False
+
+DSMOS = 'cloudinit.sources.DataSourceSmartOS'
SDC_NICS = json.loads("""
[
{
@@ -366,37 +376,33 @@ class PsuedoJoyentClient(object):
class TestSmartOSDataSource(FilesystemMockingTestCase):
+ jmc_cfact = None
+ get_smartos_environ = None
+
def setUp(self):
super(TestSmartOSDataSource, self).setUp()
- dsmos = 'cloudinit.sources.DataSourceSmartOS'
- patcher = mock.patch(dsmos + ".jmc_client_factory")
- self.jmc_cfact = patcher.start()
- self.addCleanup(patcher.stop)
- patcher = mock.patch(dsmos + ".get_smartos_environ")
- self.get_smartos_environ = patcher.start()
- self.addCleanup(patcher.stop)
-
- self.tmp = tempfile.mkdtemp()
- self.addCleanup(shutil.rmtree, self.tmp)
- self.paths = c_helpers.Paths(
- {'cloud_dir': self.tmp, 'run_dir': self.tmp})
-
- self.legacy_user_d = os.path.join(self.tmp, 'legacy_user_tmp')
+ self.add_patch(DSMOS + ".get_smartos_environ", "get_smartos_environ")
+ self.add_patch(DSMOS + ".jmc_client_factory", "jmc_cfact")
+ self.legacy_user_d = self.tmp_path('legacy_user_tmp')
os.mkdir(self.legacy_user_d)
-
- self.orig_lud = DataSourceSmartOS.LEGACY_USER_D
- DataSourceSmartOS.LEGACY_USER_D = self.legacy_user_d
-
- def tearDown(self):
- DataSourceSmartOS.LEGACY_USER_D = self.orig_lud
- super(TestSmartOSDataSource, self).tearDown()
+ self.add_patch(DSMOS + ".LEGACY_USER_D", "m_legacy_user_d",
+ autospec=False, new=self.legacy_user_d)
+ self.add_patch(DSMOS + ".identify_file", "m_identify_file",
+ return_value="text/plain")
def _get_ds(self, mockdata=None, mode=DataSourceSmartOS.SMARTOS_ENV_KVM,
sys_cfg=None, ds_cfg=None):
self.jmc_cfact.return_value = PsuedoJoyentClient(mockdata)
self.get_smartos_environ.return_value = mode
+ tmpd = self.tmp_dir()
+ dirs = {'cloud_dir': self.tmp_path('cloud_dir', tmpd),
+ 'run_dir': self.tmp_path('run_dir')}
+ for d in dirs.values():
+ os.mkdir(d)
+ paths = c_helpers.Paths(dirs)
+
if sys_cfg is None:
sys_cfg = {}
@@ -405,7 +411,7 @@ class TestSmartOSDataSource(FilesystemMockingTestCase):
sys_cfg['datasource']['SmartOS'] = ds_cfg
return DataSourceSmartOS.DataSourceSmartOS(
- sys_cfg, distro=None, paths=self.paths)
+ sys_cfg, distro=None, paths=paths)
def test_no_base64(self):
ds_cfg = {'no_base64_decode': ['test_var1'], 'all_base': True}
@@ -493,6 +499,7 @@ class TestSmartOSDataSource(FilesystemMockingTestCase):
dsrc.metadata['user-script'])
legacy_script_f = "%s/user-script" % self.legacy_user_d
+ print("legacy_script_f=%s" % legacy_script_f)
self.assertTrue(os.path.exists(legacy_script_f))
self.assertTrue(os.path.islink(legacy_script_f))
user_script_perm = oct(os.stat(legacy_script_f)[stat.ST_MODE])[-3:]
@@ -640,6 +647,28 @@ class TestSmartOSDataSource(FilesystemMockingTestCase):
mydscfg['disk_aliases']['FOO'])
+class TestIdentifyFile(CiTestCase):
+ """Test the 'identify_file' utility."""
+ @skipIf(not which("file"), "command 'file' not available.")
+ def test_file_happy_path(self):
+ """Test file is available and functional on plain text."""
+ fname = self.tmp_path("myfile")
+ write_file(fname, "plain text content here\n")
+ with self.allow_subp(["file"]):
+ self.assertEqual("text/plain", identify_file(fname))
+
+ @mock.patch(DSMOS + ".util.subp")
+ def test_returns_none_on_error(self, m_subp):
+ """On 'file' execution error, None should be returned."""
+ m_subp.side_effect = ProcessExecutionError("FILE_FAILED", exit_code=99)
+ fname = self.tmp_path("myfile")
+ write_file(fname, "plain text content here\n")
+ self.assertEqual(None, identify_file(fname))
+ self.assertEqual(
+ [mock.call(["file", "--brief", "--mime-type", fname])],
+ m_subp.call_args_list)
+
+
class ShortReader(object):
"""Implements a 'read' interface for bytes provided.
much like io.BytesIO but the 'endbyte' acts as if EOF.
@@ -893,7 +922,7 @@ class TestJoyentMetadataClient(FilesystemMockingTestCase):
self.assertEqual(client.list(), [])
-class TestNetworkConversion(TestCase):
+class TestNetworkConversion(CiTestCase):
def test_convert_simple(self):
expected = {
'version': 1,
@@ -1058,7 +1087,8 @@ class TestNetworkConversion(TestCase):
"Only supported on KVM and bhyve guests under SmartOS")
@unittest2.skipUnless(os.access(SERIAL_DEVICE, os.W_OK),
"Requires write access to " + SERIAL_DEVICE)
-class TestSerialConcurrency(TestCase):
+@unittest2.skipUnless(HAS_PYSERIAL is True, "pyserial not available")
+class TestSerialConcurrency(CiTestCase):
"""
This class tests locking on an actual serial port, and as such can only
be run in a kvm or bhyve guest running on a SmartOS host. A test run on
@@ -1066,7 +1096,11 @@ class TestSerialConcurrency(TestCase):
there is only one session over a connection. In contrast, in the
absence of proper locking multiple processes opening the same serial
port can corrupt each others' exchanges with the metadata server.
+
+ This takes on the order of 2 to 3 minutes to run.
"""
+ allowed_subp = ['mdata-get']
+
def setUp(self):
self.mdata_proc = multiprocessing.Process(target=self.start_mdata_loop)
self.mdata_proc.start()
@@ -1097,7 +1131,7 @@ class TestSerialConcurrency(TestCase):
keys = [tup[0] for tup in ds.SMARTOS_ATTRIB_MAP.values()]
keys.extend(ds.SMARTOS_ATTRIB_JSON.values())
- client = ds.jmc_client_factory()
+ client = ds.jmc_client_factory(smartos_type=SMARTOS_ENV_KVM)
self.assertIsNotNone(client)
# The behavior that we are testing for was observed mdata-get running
diff --git a/tests/unittests/test_distros/test_netconfig.py b/tests/unittests/test_distros/test_netconfig.py
index 7765e40..740fb76 100644
--- a/tests/unittests/test_distros/test_netconfig.py
+++ b/tests/unittests/test_distros/test_netconfig.py
@@ -2,24 +2,19 @@
import os
from six import StringIO
-import stat
from textwrap import dedent
try:
from unittest import mock
except ImportError:
import mock
-try:
- from contextlib import ExitStack
-except ImportError:
- from contextlib2 import ExitStack
from cloudinit import distros
from cloudinit.distros.parsers.sys_conf import SysConf
from cloudinit import helpers
-from cloudinit.net import eni
from cloudinit import settings
-from cloudinit.tests.helpers import FilesystemMockingTestCase
+from cloudinit.tests.helpers import (
+ FilesystemMockingTestCase, dir2dict, populate_dir)
from cloudinit import util
@@ -82,7 +77,7 @@ V1_NET_CFG = {'config': [{'name': 'eth0',
'type': 'physical'}],
'version': 1}
-V1_NET_CFG_OUTPUT = """
+V1_NET_CFG_OUTPUT = """\
# This file is generated from information provided by
# the datasource. Changes to it will not persist across an instance.
# To disable cloud-init's network configuration capabilities, write a file
@@ -116,7 +111,7 @@ V1_NET_CFG_IPV6 = {'config': [{'name': 'eth0',
'version': 1}
-V1_TO_V2_NET_CFG_OUTPUT = """
+V1_TO_V2_NET_CFG_OUTPUT = """\
# This file is generated from information provided by
# the datasource. Changes to it will not persist across an instance.
# To disable cloud-init's network configuration capabilities, write a file
@@ -145,7 +140,7 @@ V2_NET_CFG = {
}
-V2_TO_V2_NET_CFG_OUTPUT = """
+V2_TO_V2_NET_CFG_OUTPUT = """\
# This file is generated from information provided by
# the datasource. Changes to it will not persist across an instance.
# To disable cloud-init's network configuration capabilities, write a file
@@ -176,21 +171,10 @@ class WriteBuffer(object):
return self.buffer.getvalue()
-class TestNetCfgDistro(FilesystemMockingTestCase):
-
- frbsd_ifout = """\
-hn0: flags=8843<UP,BROADCAST,RUNNING,SIMPLEX,MULTICAST> metric 0 mtu 1500
- options=51b<RXCSUM,TXCSUM,VLAN_MTU,VLAN_HWTAGGING,TSO4,LRO>
- ether 00:15:5d:4c:73:00
- inet6 fe80::215:5dff:fe4c:7300%hn0 prefixlen 64 scopeid 0x2
- inet 10.156.76.127 netmask 0xfffffc00 broadcast 10.156.79.255
- nd6 options=23<PERFORMNUD,ACCEPT_RTADV,AUTO_LINKLOCAL>
- media: Ethernet autoselect (10Gbase-T <full-duplex>)
- status: active
-"""
+class TestNetCfgDistroBase(FilesystemMockingTestCase):
def setUp(self):
- super(TestNetCfgDistro, self).setUp()
+ super(TestNetCfgDistroBase, self).setUp()
self.add_patch('cloudinit.util.system_is_snappy', 'm_snappy')
self.add_patch('cloudinit.util.system_info', 'm_sysinfo')
self.m_sysinfo.return_value = {'dist': ('Distro', '99.1', 'Codename')}
@@ -204,144 +188,6 @@ hn0: flags=8843<UP,BROADCAST,RUNNING,SIMPLEX,MULTICAST> metric 0 mtu 1500
paths = helpers.Paths({})
return cls(dname, cfg.get('system_info'), paths)
- def test_simple_write_ub(self):
- ub_distro = self._get_distro('ubuntu')
- with ExitStack() as mocks:
- write_bufs = {}
-
- def replace_write(filename, content, mode=0o644, omode="wb"):
- buf = WriteBuffer()
- buf.mode = mode
- buf.omode = omode
- buf.write(content)
- write_bufs[filename] = buf
-
- mocks.enter_context(
- mock.patch.object(util, 'write_file', replace_write))
- mocks.enter_context(
- mock.patch.object(os.path, 'isfile', return_value=False))
-
- ub_distro.apply_network(BASE_NET_CFG, False)
-
- self.assertEqual(len(write_bufs), 1)
- eni_name = '/etc/network/interfaces.d/50-cloud-init.cfg'
- self.assertIn(eni_name, write_bufs)
- write_buf = write_bufs[eni_name]
- self.assertEqual(str(write_buf).strip(), BASE_NET_CFG.strip())
- self.assertEqual(write_buf.mode, 0o644)
-
- def test_apply_network_config_eni_ub(self):
- ub_distro = self._get_distro('ubuntu')
- with ExitStack() as mocks:
- write_bufs = {}
-
- def replace_write(filename, content, mode=0o644, omode="wb"):
- buf = WriteBuffer()
- buf.mode = mode
- buf.omode = omode
- buf.write(content)
- write_bufs[filename] = buf
-
- # eni availability checks
- mocks.enter_context(
- mock.patch.object(util, 'which', return_value=True))
- mocks.enter_context(
- mock.patch.object(eni, 'available', return_value=True))
- mocks.enter_context(
- mock.patch.object(util, 'ensure_dir'))
- mocks.enter_context(
- mock.patch.object(util, 'write_file', replace_write))
- mocks.enter_context(
- mock.patch.object(os.path, 'isfile', return_value=False))
- mocks.enter_context(
- mock.patch("cloudinit.net.eni.glob.glob",
- return_value=[]))
-
- ub_distro.apply_network_config(V1_NET_CFG, False)
-
- self.assertEqual(len(write_bufs), 2)
- eni_name = '/etc/network/interfaces.d/50-cloud-init.cfg'
- self.assertIn(eni_name, write_bufs)
- write_buf = write_bufs[eni_name]
- self.assertEqual(str(write_buf).strip(), V1_NET_CFG_OUTPUT.strip())
- self.assertEqual(write_buf.mode, 0o644)
-
- def test_apply_network_config_v1_to_netplan_ub(self):
- renderers = ['netplan']
- devlist = ['eth0', 'lo']
- ub_distro = self._get_distro('ubuntu', renderers=renderers)
- with ExitStack() as mocks:
- write_bufs = {}
-
- def replace_write(filename, content, mode=0o644, omode="wb"):
- buf = WriteBuffer()
- buf.mode = mode
- buf.omode = omode
- buf.write(content)
- write_bufs[filename] = buf
-
- mocks.enter_context(
- mock.patch.object(util, 'which', return_value=True))
- mocks.enter_context(
- mock.patch.object(util, 'write_file', replace_write))
- mocks.enter_context(
- mock.patch.object(util, 'ensure_dir'))
- mocks.enter_context(
- mock.patch.object(util, 'subp', return_value=(0, 0)))
- mocks.enter_context(
- mock.patch.object(os.path, 'isfile', return_value=False))
- mocks.enter_context(
- mock.patch("cloudinit.net.netplan.get_devicelist",
- return_value=devlist))
-
- ub_distro.apply_network_config(V1_NET_CFG, False)
-
- self.assertEqual(len(write_bufs), 1)
- netplan_name = '/etc/netplan/50-cloud-init.yaml'
- self.assertIn(netplan_name, write_bufs)
- write_buf = write_bufs[netplan_name]
- self.assertEqual(str(write_buf).strip(),
- V1_TO_V2_NET_CFG_OUTPUT.strip())
- self.assertEqual(write_buf.mode, 0o644)
-
- def test_apply_network_config_v2_passthrough_ub(self):
- renderers = ['netplan']
- devlist = ['eth0', 'lo']
- ub_distro = self._get_distro('ubuntu', renderers=renderers)
- with ExitStack() as mocks:
- write_bufs = {}
-
- def replace_write(filename, content, mode=0o644, omode="wb"):
- buf = WriteBuffer()
- buf.mode = mode
- buf.omode = omode
- buf.write(content)
- write_bufs[filename] = buf
-
- mocks.enter_context(
- mock.patch.object(util, 'which', return_value=True))
- mocks.enter_context(
- mock.patch.object(util, 'write_file', replace_write))
- mocks.enter_context(
- mock.patch.object(util, 'ensure_dir'))
- mocks.enter_context(
- mock.patch.object(util, 'subp', return_value=(0, 0)))
- mocks.enter_context(
- mock.patch.object(os.path, 'isfile', return_value=False))
- # FreeBSD does not have '/sys/class/net' file,
- # so we need mock here.
- mocks.enter_context(
- mock.patch.object(os, 'listdir', return_value=devlist))
- ub_distro.apply_network_config(V2_NET_CFG, False)
-
- self.assertEqual(len(write_bufs), 1)
- netplan_name = '/etc/netplan/50-cloud-init.yaml'
- self.assertIn(netplan_name, write_bufs)
- write_buf = write_bufs[netplan_name]
- self.assertEqual(str(write_buf).strip(),
- V2_TO_V2_NET_CFG_OUTPUT.strip())
- self.assertEqual(write_buf.mode, 0o644)
-
def assertCfgEquals(self, blob1, blob2):
b1 = dict(SysConf(blob1.strip().splitlines()))
b2 = dict(SysConf(blob2.strip().splitlines()))
@@ -353,6 +199,20 @@ hn0: flags=8843<UP,BROADCAST,RUNNING,SIMPLEX,MULTICAST> metric 0 mtu 1500
for (k, v) in b1.items():
self.assertEqual(v, b2[k])
+
+class TestNetCfgDistroFreebsd(TestNetCfgDistroBase):
+
+ frbsd_ifout = """\
+hn0: flags=8843<UP,BROADCAST,RUNNING,SIMPLEX,MULTICAST> metric 0 mtu 1500
+ options=51b<RXCSUM,TXCSUM,VLAN_MTU,VLAN_HWTAGGING,TSO4,LRO>
+ ether 00:15:5d:4c:73:00
+ inet6 fe80::215:5dff:fe4c:7300%hn0 prefixlen 64 scopeid 0x2
+ inet 10.156.76.127 netmask 0xfffffc00 broadcast 10.156.79.255
+ nd6 options=23<PERFORMNUD,ACCEPT_RTADV,AUTO_LINKLOCAL>
+ media: Ethernet autoselect (10Gbase-T <full-duplex>)
+ status: active
+"""
+
@mock.patch('cloudinit.distros.freebsd.Distro.get_ifconfig_list')
@mock.patch('cloudinit.distros.freebsd.Distro.get_ifconfig_ifname_out')
def test_get_ip_nic_freebsd(self, ifname_out, iflist):
@@ -376,349 +236,33 @@ hn0: flags=8843<UP,BROADCAST,RUNNING,SIMPLEX,MULTICAST> metric 0 mtu 1500
res = frbsd_distro.generate_fallback_config()
self.assertIsNotNone(res)
- def test_simple_write_rh(self):
- rh_distro = self._get_distro('rhel')
-
- write_bufs = {}
-
- def replace_write(filename, content, mode=0o644, omode="wb"):
- buf = WriteBuffer()
- buf.mode = mode
- buf.omode = omode
- buf.write(content)
- write_bufs[filename] = buf
-
- with ExitStack() as mocks:
- mocks.enter_context(
- mock.patch.object(util, 'write_file', replace_write))
- mocks.enter_context(
- mock.patch.object(util, 'load_file', return_value=''))
- mocks.enter_context(
- mock.patch.object(os.path, 'isfile', return_value=False))
-
- rh_distro.apply_network(BASE_NET_CFG, False)
-
- self.assertEqual(len(write_bufs), 4)
- self.assertIn('/etc/sysconfig/network-scripts/ifcfg-lo',
- write_bufs)
- write_buf = write_bufs['/etc/sysconfig/network-scripts/ifcfg-lo']
- expected_buf = '''
-DEVICE="lo"
-ONBOOT=yes
-'''
- self.assertCfgEquals(expected_buf, str(write_buf))
- self.assertEqual(write_buf.mode, 0o644)
-
- self.assertIn('/etc/sysconfig/network-scripts/ifcfg-eth0',
- write_bufs)
- write_buf = write_bufs['/etc/sysconfig/network-scripts/ifcfg-eth0']
- expected_buf = '''
-DEVICE="eth0"
-BOOTPROTO="static"
-NETMASK="255.255.255.0"
-IPADDR="192.168.1.5"
-ONBOOT=yes
-GATEWAY="192.168.1.254"
-BROADCAST="192.168.1.0"
-'''
- self.assertCfgEquals(expected_buf, str(write_buf))
- self.assertEqual(write_buf.mode, 0o644)
-
- self.assertIn('/etc/sysconfig/network-scripts/ifcfg-eth1',
- write_bufs)
- write_buf = write_bufs['/etc/sysconfig/network-scripts/ifcfg-eth1']
- expected_buf = '''
-DEVICE="eth1"
-BOOTPROTO="dhcp"
-ONBOOT=yes
-'''
- self.assertCfgEquals(expected_buf, str(write_buf))
- self.assertEqual(write_buf.mode, 0o644)
-
- self.assertIn('/etc/sysconfig/network', write_bufs)
- write_buf = write_bufs['/etc/sysconfig/network']
- expected_buf = '''
-# Created by cloud-init v. 0.7
-NETWORKING=yes
-'''
- self.assertCfgEquals(expected_buf, str(write_buf))
- self.assertEqual(write_buf.mode, 0o644)
-
- def test_apply_network_config_rh(self):
- renderers = ['sysconfig']
- rh_distro = self._get_distro('rhel', renderers=renderers)
-
- write_bufs = {}
-
- def replace_write(filename, content, mode=0o644, omode="wb"):
- buf = WriteBuffer()
- buf.mode = mode
- buf.omode = omode
- buf.write(content)
- write_bufs[filename] = buf
-
- with ExitStack() as mocks:
- # sysconfig availability checks
- mocks.enter_context(
- mock.patch.object(util, 'which', return_value=True))
- mocks.enter_context(
- mock.patch.object(util, 'write_file', replace_write))
- mocks.enter_context(
- mock.patch.object(util, 'load_file', return_value=''))
- mocks.enter_context(
- mock.patch.object(os.path, 'isfile', return_value=True))
-
- rh_distro.apply_network_config(V1_NET_CFG, False)
-
- self.assertEqual(len(write_bufs), 5)
-
- # eth0
- self.assertIn('/etc/sysconfig/network-scripts/ifcfg-eth0',
- write_bufs)
- write_buf = write_bufs['/etc/sysconfig/network-scripts/ifcfg-eth0']
- expected_buf = '''
-# Created by cloud-init on instance boot automatically, do not edit.
-#
-BOOTPROTO=none
-DEFROUTE=yes
-DEVICE=eth0
-GATEWAY=192.168.1.254
-IPADDR=192.168.1.5
-NETMASK=255.255.255.0
-NM_CONTROLLED=no
-ONBOOT=yes
-TYPE=Ethernet
-USERCTL=no
-'''
- self.assertCfgEquals(expected_buf, str(write_buf))
- self.assertEqual(write_buf.mode, 0o644)
-
- # eth1
- self.assertIn('/etc/sysconfig/network-scripts/ifcfg-eth1',
- write_bufs)
- write_buf = write_bufs['/etc/sysconfig/network-scripts/ifcfg-eth1']
- expected_buf = '''
-# Created by cloud-init on instance boot automatically, do not edit.
-#
-BOOTPROTO=dhcp
-DEVICE=eth1
-NM_CONTROLLED=no
-ONBOOT=yes
-TYPE=Ethernet
-USERCTL=no
-'''
- self.assertCfgEquals(expected_buf, str(write_buf))
- self.assertEqual(write_buf.mode, 0o644)
-
- self.assertIn('/etc/sysconfig/network', write_bufs)
- write_buf = write_bufs['/etc/sysconfig/network']
- expected_buf = '''
-# Created by cloud-init v. 0.7
-NETWORKING=yes
-'''
- self.assertCfgEquals(expected_buf, str(write_buf))
- self.assertEqual(write_buf.mode, 0o644)
-
- def test_write_ipv6_rhel(self):
- rh_distro = self._get_distro('rhel')
-
- write_bufs = {}
-
- def replace_write(filename, content, mode=0o644, omode="wb"):
- buf = WriteBuffer()
- buf.mode = mode
- buf.omode = omode
- buf.write(content)
- write_bufs[filename] = buf
-
- with ExitStack() as mocks:
- mocks.enter_context(
- mock.patch.object(util, 'write_file', replace_write))
- mocks.enter_context(
- mock.patch.object(util, 'load_file', return_value=''))
- mocks.enter_context(
- mock.patch.object(os.path, 'isfile', return_value=False))
- rh_distro.apply_network(BASE_NET_CFG_IPV6, False)
-
- self.assertEqual(len(write_bufs), 4)
- self.assertIn('/etc/sysconfig/network-scripts/ifcfg-lo',
- write_bufs)
- write_buf = write_bufs['/etc/sysconfig/network-scripts/ifcfg-lo']
- expected_buf = '''
-DEVICE="lo"
-ONBOOT=yes
-'''
- self.assertCfgEquals(expected_buf, str(write_buf))
- self.assertEqual(write_buf.mode, 0o644)
-
- self.assertIn('/etc/sysconfig/network-scripts/ifcfg-eth0',
- write_bufs)
- write_buf = write_bufs['/etc/sysconfig/network-scripts/ifcfg-eth0']
- expected_buf = '''
-DEVICE="eth0"
-BOOTPROTO="static"
-NETMASK="255.255.255.0"
-IPADDR="192.168.1.5"
-ONBOOT=yes
-GATEWAY="192.168.1.254"
-BROADCAST="192.168.1.0"
-IPV6INIT=yes
-IPV6ADDR="2607:f0d0:1002:0011::2"
-IPV6_DEFAULTGW="2607:f0d0:1002:0011::1"
-'''
- self.assertCfgEquals(expected_buf, str(write_buf))
- self.assertEqual(write_buf.mode, 0o644)
- self.assertIn('/etc/sysconfig/network-scripts/ifcfg-eth1',
- write_bufs)
- write_buf = write_bufs['/etc/sysconfig/network-scripts/ifcfg-eth1']
- expected_buf = '''
-DEVICE="eth1"
-BOOTPROTO="static"
-NETMASK="255.255.255.0"
-IPADDR="192.168.1.6"
-ONBOOT=no
-GATEWAY="192.168.1.254"
-BROADCAST="192.168.1.0"
-IPV6INIT=yes
-IPV6ADDR="2607:f0d0:1002:0011::3"
-IPV6_DEFAULTGW="2607:f0d0:1002:0011::1"
-'''
- self.assertCfgEquals(expected_buf, str(write_buf))
- self.assertEqual(write_buf.mode, 0o644)
-
- self.assertIn('/etc/sysconfig/network', write_bufs)
- write_buf = write_bufs['/etc/sysconfig/network']
- expected_buf = '''
-# Created by cloud-init v. 0.7
-NETWORKING=yes
-NETWORKING_IPV6=yes
-IPV6_AUTOCONF=no
-'''
- self.assertCfgEquals(expected_buf, str(write_buf))
- self.assertEqual(write_buf.mode, 0o644)
-
- def test_apply_network_config_ipv6_rh(self):
- renderers = ['sysconfig']
- rh_distro = self._get_distro('rhel', renderers=renderers)
-
- write_bufs = {}
-
- def replace_write(filename, content, mode=0o644, omode="wb"):
- buf = WriteBuffer()
- buf.mode = mode
- buf.omode = omode
- buf.write(content)
- write_bufs[filename] = buf
-
- with ExitStack() as mocks:
- mocks.enter_context(
- mock.patch.object(util, 'which', return_value=True))
- mocks.enter_context(
- mock.patch.object(util, 'write_file', replace_write))
- mocks.enter_context(
- mock.patch.object(util, 'load_file', return_value=''))
- mocks.enter_context(
- mock.patch.object(os.path, 'isfile', return_value=True))
-
- rh_distro.apply_network_config(V1_NET_CFG_IPV6, False)
-
- self.assertEqual(len(write_bufs), 5)
-
- self.assertIn('/etc/sysconfig/network-scripts/ifcfg-eth0',
- write_bufs)
- write_buf = write_bufs['/etc/sysconfig/network-scripts/ifcfg-eth0']
- expected_buf = '''
-# Created by cloud-init on instance boot automatically, do not edit.
-#
-BOOTPROTO=none
-DEFROUTE=yes
-DEVICE=eth0
-IPV6ADDR=2607:f0d0:1002:0011::2/64
-IPV6INIT=yes
-IPV6_DEFAULTGW=2607:f0d0:1002:0011::1
-NM_CONTROLLED=no
-ONBOOT=yes
-TYPE=Ethernet
-USERCTL=no
-'''
- self.assertCfgEquals(expected_buf, str(write_buf))
- self.assertEqual(write_buf.mode, 0o644)
- self.assertIn('/etc/sysconfig/network-scripts/ifcfg-eth1',
- write_bufs)
- write_buf = write_bufs['/etc/sysconfig/network-scripts/ifcfg-eth1']
- expected_buf = '''
-# Created by cloud-init on instance boot automatically, do not edit.
-#
-BOOTPROTO=dhcp
-DEVICE=eth1
-NM_CONTROLLED=no
-ONBOOT=yes
-TYPE=Ethernet
-USERCTL=no
-'''
- self.assertCfgEquals(expected_buf, str(write_buf))
- self.assertEqual(write_buf.mode, 0o644)
-
- self.assertIn('/etc/sysconfig/network', write_bufs)
- write_buf = write_bufs['/etc/sysconfig/network']
- expected_buf = '''
-# Created by cloud-init v. 0.7
-NETWORKING=yes
-NETWORKING_IPV6=yes
-IPV6_AUTOCONF=no
-'''
- self.assertCfgEquals(expected_buf, str(write_buf))
- self.assertEqual(write_buf.mode, 0o644)
-
def test_simple_write_freebsd(self):
fbsd_distro = self._get_distro('freebsd')
- write_bufs = {}
+ rc_conf = '/etc/rc.conf'
read_bufs = {
- '/etc/rc.conf': '',
- '/etc/resolv.conf': '',
+ rc_conf: 'initial-rc-conf-not-validated',
+ '/etc/resolv.conf': 'initial-resolv-conf-not-validated',
}
- def replace_write(filename, content, mode=0o644, omode="wb"):
- buf = WriteBuffer()
- buf.mode = mode
- buf.omode = omode
- buf.write(content)
- write_bufs[filename] = buf
-
- def replace_read(fname, read_cb=None, quiet=False):
- if fname not in read_bufs:
- if fname in write_bufs:
- return str(write_bufs[fname])
- raise IOError("%s not found" % fname)
- else:
- if fname in write_bufs:
- return str(write_bufs[fname])
- return read_bufs[fname]
-
- with ExitStack() as mocks:
- mocks.enter_context(
- mock.patch.object(util, 'subp', return_value=('vtnet0', '')))
- mocks.enter_context(
- mock.patch.object(os.path, 'exists', return_value=False))
- mocks.enter_context(
- mock.patch.object(util, 'write_file', replace_write))
- mocks.enter_context(
- mock.patch.object(util, 'load_file', replace_read))
-
- fbsd_distro.apply_network(BASE_NET_CFG, False)
-
- self.assertIn('/etc/rc.conf', write_bufs)
- write_buf = write_bufs['/etc/rc.conf']
- expected_buf = '''
-ifconfig_vtnet0="192.168.1.5 netmask 255.255.255.0"
-ifconfig_vtnet1="DHCP"
-defaultrouter="192.168.1.254"
-'''
- self.assertCfgEquals(expected_buf, str(write_buf))
- self.assertEqual(write_buf.mode, 0o644)
-
- def test_apply_network_config_fallback(self):
+ tmpd = self.tmp_dir()
+ populate_dir(tmpd, read_bufs)
+ with self.reRooted(tmpd):
+ with mock.patch("cloudinit.distros.freebsd.util.subp",
+ return_value=('vtnet0', '')):
+ fbsd_distro.apply_network(BASE_NET_CFG, False)
+ results = dir2dict(tmpd)
+
+ self.assertIn(rc_conf, results)
+ self.assertCfgEquals(
+ dedent('''\
+ ifconfig_vtnet0="192.168.1.5 netmask 255.255.255.0"
+ ifconfig_vtnet1="DHCP"
+ defaultrouter="192.168.1.254"
+ '''), results[rc_conf])
+ self.assertEqual(0o644, get_mode(rc_conf, tmpd))
+
+ def test_apply_network_config_fallback_freebsd(self):
fbsd_distro = self._get_distro('freebsd')
# a weak attempt to verify that we don't have an implementation
@@ -735,68 +279,324 @@ defaultrouter="192.168.1.254"
"subnets": [{"type": "dhcp"}]}],
'version': 1}
- write_bufs = {}
+ rc_conf = '/etc/rc.conf'
read_bufs = {
- '/etc/rc.conf': '',
- '/etc/resolv.conf': '',
+ rc_conf: 'initial-rc-conf-not-validated',
+ '/etc/resolv.conf': 'initial-resolv-conf-not-validated',
}
- def replace_write(filename, content, mode=0o644, omode="wb"):
- buf = WriteBuffer()
- buf.mode = mode
- buf.omode = omode
- buf.write(content)
- write_bufs[filename] = buf
-
- def replace_read(fname, read_cb=None, quiet=False):
- if fname not in read_bufs:
- if fname in write_bufs:
- return str(write_bufs[fname])
- raise IOError("%s not found" % fname)
- else:
- if fname in write_bufs:
- return str(write_bufs[fname])
- return read_bufs[fname]
-
- with ExitStack() as mocks:
- mocks.enter_context(
- mock.patch.object(util, 'subp', return_value=('vtnet0', '')))
- mocks.enter_context(
- mock.patch.object(os.path, 'exists', return_value=False))
- mocks.enter_context(
- mock.patch.object(util, 'write_file', replace_write))
- mocks.enter_context(
- mock.patch.object(util, 'load_file', replace_read))
-
- fbsd_distro.apply_network_config(mynetcfg, bring_up=False)
-
- self.assertIn('/etc/rc.conf', write_bufs)
- write_buf = write_bufs['/etc/rc.conf']
- expected_buf = '''
-ifconfig_vtnet0="DHCP"
-'''
- self.assertCfgEquals(expected_buf, str(write_buf))
- self.assertEqual(write_buf.mode, 0o644)
+ tmpd = self.tmp_dir()
+ populate_dir(tmpd, read_bufs)
+ with self.reRooted(tmpd):
+ with mock.patch("cloudinit.distros.freebsd.util.subp",
+ return_value=('vtnet0', '')):
+ fbsd_distro.apply_network_config(mynetcfg, bring_up=False)
+ results = dir2dict(tmpd)
- def test_simple_write_opensuse(self):
- """Opensuse network rendering writes appropriate sysconfg files."""
- tmpdir = self.tmp_dir()
- self.patchOS(tmpdir)
- self.patchUtils(tmpdir)
- distro = self._get_distro('opensuse')
+ self.assertIn(rc_conf, results)
+ self.assertCfgEquals('ifconfig_vtnet0="DHCP"', results[rc_conf])
+ self.assertEqual(0o644, get_mode(rc_conf, tmpd))
- distro.apply_network(BASE_NET_CFG, False)
- lo_path = os.path.join(tmpdir, 'etc/sysconfig/network/ifcfg-lo')
- eth0_path = os.path.join(tmpdir, 'etc/sysconfig/network/ifcfg-eth0')
- eth1_path = os.path.join(tmpdir, 'etc/sysconfig/network/ifcfg-eth1')
+class TestNetCfgDistroUbuntuEni(TestNetCfgDistroBase):
+
+ def setUp(self):
+ super(TestNetCfgDistroUbuntuEni, self).setUp()
+ self.distro = self._get_distro('ubuntu', renderers=['eni'])
+
+ def eni_path(self):
+ return '/etc/network/interfaces.d/50-cloud-init.cfg'
+
+ def _apply_and_verify_eni(self, apply_fn, config, expected_cfgs=None,
+ bringup=False):
+ if not expected_cfgs:
+ raise ValueError('expected_cfg must not be None')
+
+ tmpd = None
+ with mock.patch('cloudinit.net.eni.available') as m_avail:
+ m_avail.return_value = True
+ with self.reRooted(tmpd) as tmpd:
+ apply_fn(config, bringup)
+
+ results = dir2dict(tmpd)
+ for cfgpath, expected in expected_cfgs.items():
+ print("----------")
+ print(expected)
+ print("^^^^ expected | rendered VVVVVVV")
+ print(results[cfgpath])
+ print("----------")
+ self.assertEqual(expected, results[cfgpath])
+ self.assertEqual(0o644, get_mode(cfgpath, tmpd))
+
+ def test_simple_write_ub(self):
+ expected_cfgs = {
+ self.eni_path(): BASE_NET_CFG,
+ }
+
+ # ub_distro.apply_network(BASE_NET_CFG, False)
+ self._apply_and_verify_eni(self.distro.apply_network,
+ BASE_NET_CFG,
+ expected_cfgs=expected_cfgs.copy())
+
+ def test_apply_network_config_eni_ub(self):
expected_cfgs = {
- lo_path: dedent('''
+ self.eni_path(): V1_NET_CFG_OUTPUT,
+ }
+ # ub_distro.apply_network_config(V1_NET_CFG, False)
+ self._apply_and_verify_eni(self.distro.apply_network_config,
+ V1_NET_CFG,
+ expected_cfgs=expected_cfgs.copy())
+
+
+class TestNetCfgDistroUbuntuNetplan(TestNetCfgDistroBase):
+ def setUp(self):
+ super(TestNetCfgDistroUbuntuNetplan, self).setUp()
+ self.distro = self._get_distro('ubuntu', renderers=['netplan'])
+ self.devlist = ['eth0', 'lo']
+
+ def _apply_and_verify_netplan(self, apply_fn, config, expected_cfgs=None,
+ bringup=False):
+ if not expected_cfgs:
+ raise ValueError('expected_cfg must not be None')
+
+ tmpd = None
+ with mock.patch('cloudinit.net.netplan.available',
+ return_value=True):
+ with mock.patch("cloudinit.net.netplan.get_devicelist",
+ return_value=self.devlist):
+ with self.reRooted(tmpd) as tmpd:
+ apply_fn(config, bringup)
+
+ results = dir2dict(tmpd)
+ for cfgpath, expected in expected_cfgs.items():
+ print("----------")
+ print(expected)
+ print("^^^^ expected | rendered VVVVVVV")
+ print(results[cfgpath])
+ print("----------")
+ self.assertEqual(expected, results[cfgpath])
+ self.assertEqual(0o644, get_mode(cfgpath, tmpd))
+
+ def netplan_path(self):
+ return '/etc/netplan/50-cloud-init.yaml'
+
+ def test_apply_network_config_v1_to_netplan_ub(self):
+ expected_cfgs = {
+ self.netplan_path(): V1_TO_V2_NET_CFG_OUTPUT,
+ }
+
+ # ub_distro.apply_network_config(V1_NET_CFG, False)
+ self._apply_and_verify_netplan(self.distro.apply_network_config,
+ V1_NET_CFG,
+ expected_cfgs=expected_cfgs.copy())
+
+ def test_apply_network_config_v2_passthrough_ub(self):
+ expected_cfgs = {
+ self.netplan_path(): V2_TO_V2_NET_CFG_OUTPUT,
+ }
+ # ub_distro.apply_network_config(V2_NET_CFG, False)
+ self._apply_and_verify_netplan(self.distro.apply_network_config,
+ V2_NET_CFG,
+ expected_cfgs=expected_cfgs.copy())
+
+
+class TestNetCfgDistroRedhat(TestNetCfgDistroBase):
+
+ def setUp(self):
+ super(TestNetCfgDistroRedhat, self).setUp()
+ self.distro = self._get_distro('rhel', renderers=['sysconfig'])
+
+ def ifcfg_path(self, ifname):
+ return '/etc/sysconfig/network-scripts/ifcfg-%s' % ifname
+
+ def control_path(self):
+ return '/etc/sysconfig/network'
+
+ def _apply_and_verify(self, apply_fn, config, expected_cfgs=None,
+ bringup=False):
+ if not expected_cfgs:
+ raise ValueError('expected_cfg must not be None')
+
+ tmpd = None
+ with mock.patch('cloudinit.net.sysconfig.available') as m_avail:
+ m_avail.return_value = True
+ with self.reRooted(tmpd) as tmpd:
+ apply_fn(config, bringup)
+
+ results = dir2dict(tmpd)
+ for cfgpath, expected in expected_cfgs.items():
+ self.assertCfgEquals(expected, results[cfgpath])
+ self.assertEqual(0o644, get_mode(cfgpath, tmpd))
+
+ def test_simple_write_rh(self):
+ expected_cfgs = {
+ self.ifcfg_path('lo'): dedent("""\
+ DEVICE="lo"
+ ONBOOT=yes
+ """),
+ self.ifcfg_path('eth0'): dedent("""\
+ DEVICE="eth0"
+ BOOTPROTO="static"
+ NETMASK="255.255.255.0"
+ IPADDR="192.168.1.5"
+ ONBOOT=yes
+ GATEWAY="192.168.1.254"
+ BROADCAST="192.168.1.0"
+ """),
+ self.ifcfg_path('eth1'): dedent("""\
+ DEVICE="eth1"
+ BOOTPROTO="dhcp"
+ ONBOOT=yes
+ """),
+ self.control_path(): dedent("""\
+ NETWORKING=yes
+ """),
+ }
+ # rh_distro.apply_network(BASE_NET_CFG, False)
+ self._apply_and_verify(self.distro.apply_network,
+ BASE_NET_CFG,
+ expected_cfgs=expected_cfgs.copy())
+
+ def test_apply_network_config_rh(self):
+ expected_cfgs = {
+ self.ifcfg_path('eth0'): dedent("""\
+ BOOTPROTO=none
+ DEFROUTE=yes
+ DEVICE=eth0
+ GATEWAY=192.168.1.254
+ IPADDR=192.168.1.5
+ NETMASK=255.255.255.0
+ NM_CONTROLLED=no
+ ONBOOT=yes
+ TYPE=Ethernet
+ USERCTL=no
+ """),
+ self.ifcfg_path('eth1'): dedent("""\
+ BOOTPROTO=dhcp
+ DEVICE=eth1
+ NM_CONTROLLED=no
+ ONBOOT=yes
+ TYPE=Ethernet
+ USERCTL=no
+ """),
+ self.control_path(): dedent("""\
+ NETWORKING=yes
+ """),
+ }
+ # rh_distro.apply_network_config(V1_NET_CFG, False)
+ self._apply_and_verify(self.distro.apply_network_config,
+ V1_NET_CFG,
+ expected_cfgs=expected_cfgs.copy())
+
+ def test_write_ipv6_rhel(self):
+ expected_cfgs = {
+ self.ifcfg_path('lo'): dedent("""\
+ DEVICE="lo"
+ ONBOOT=yes
+ """),
+ self.ifcfg_path('eth0'): dedent("""\
+ DEVICE="eth0"
+ BOOTPROTO="static"
+ NETMASK="255.255.255.0"
+ IPADDR="192.168.1.5"
+ ONBOOT=yes
+ GATEWAY="192.168.1.254"
+ BROADCAST="192.168.1.0"
+ IPV6INIT=yes
+ IPV6ADDR="2607:f0d0:1002:0011::2"
+ IPV6_DEFAULTGW="2607:f0d0:1002:0011::1"
+ """),
+ self.ifcfg_path('eth1'): dedent("""\
+ DEVICE="eth1"
+ BOOTPROTO="static"
+ NETMASK="255.255.255.0"
+ IPADDR="192.168.1.6"
+ ONBOOT=no
+ GATEWAY="192.168.1.254"
+ BROADCAST="192.168.1.0"
+ IPV6INIT=yes
+ IPV6ADDR="2607:f0d0:1002:0011::3"
+ IPV6_DEFAULTGW="2607:f0d0:1002:0011::1"
+ """),
+ self.control_path(): dedent("""\
+ NETWORKING=yes
+ NETWORKING_IPV6=yes
+ IPV6_AUTOCONF=no
+ """),
+ }
+ # rh_distro.apply_network(BASE_NET_CFG_IPV6, False)
+ self._apply_and_verify(self.distro.apply_network,
+ BASE_NET_CFG_IPV6,
+ expected_cfgs=expected_cfgs.copy())
+
+ def test_apply_network_config_ipv6_rh(self):
+ expected_cfgs = {
+ self.ifcfg_path('eth0'): dedent("""\
+ BOOTPROTO=none
+ DEFROUTE=yes
+ DEVICE=eth0
+ IPV6ADDR=2607:f0d0:1002:0011::2/64
+ IPV6INIT=yes
+ IPV6_DEFAULTGW=2607:f0d0:1002:0011::1
+ NM_CONTROLLED=no
+ ONBOOT=yes
+ TYPE=Ethernet
+ USERCTL=no
+ """),
+ self.ifcfg_path('eth1'): dedent("""\
+ BOOTPROTO=dhcp
+ DEVICE=eth1
+ NM_CONTROLLED=no
+ ONBOOT=yes
+ TYPE=Ethernet
+ USERCTL=no
+ """),
+ self.control_path(): dedent("""\
+ NETWORKING=yes
+ NETWORKING_IPV6=yes
+ IPV6_AUTOCONF=no
+ """),
+ }
+ # rh_distro.apply_network_config(V1_NET_CFG_IPV6, False)
+ self._apply_and_verify(self.distro.apply_network_config,
+ V1_NET_CFG_IPV6,
+ expected_cfgs=expected_cfgs.copy())
+
+
+class TestNetCfgDistroOpensuse(TestNetCfgDistroBase):
+
+ def setUp(self):
+ super(TestNetCfgDistroOpensuse, self).setUp()
+ self.distro = self._get_distro('opensuse', renderers=['sysconfig'])
+
+ def ifcfg_path(self, ifname):
+ return '/etc/sysconfig/network/ifcfg-%s' % ifname
+
+ def _apply_and_verify(self, apply_fn, config, expected_cfgs=None,
+ bringup=False):
+ if not expected_cfgs:
+ raise ValueError('expected_cfg must not be None')
+
+ tmpd = None
+ with mock.patch('cloudinit.net.sysconfig.available') as m_avail:
+ m_avail.return_value = True
+ with self.reRooted(tmpd) as tmpd:
+ apply_fn(config, bringup)
+
+ results = dir2dict(tmpd)
+ for cfgpath, expected in expected_cfgs.items():
+ self.assertCfgEquals(expected, results[cfgpath])
+ self.assertEqual(0o644, get_mode(cfgpath, tmpd))
+
+ def test_simple_write_opensuse(self):
+ """Opensuse network rendering writes appropriate sysconfig files."""
+ expected_cfgs = {
+ self.ifcfg_path('lo'): dedent('''
STARTMODE="auto"
USERCONTROL="no"
FIREWALL="no"
'''),
- eth0_path: dedent('''
+ self.ifcfg_path('eth0'): dedent('''
BOOTPROTO="static"
BROADCAST="192.168.1.0"
GATEWAY="192.168.1.254"
@@ -806,18 +606,77 @@ ifconfig_vtnet0="DHCP"
USERCONTROL="no"
ETHTOOL_OPTIONS=""
'''),
- eth1_path: dedent('''
+ self.ifcfg_path('eth1'): dedent('''
BOOTPROTO="dhcp"
STARTMODE="auto"
USERCONTROL="no"
ETHTOOL_OPTIONS=""
''')
}
- for cfgpath in (lo_path, eth0_path, eth1_path):
- self.assertCfgEquals(
- expected_cfgs[cfgpath],
- util.load_file(cfgpath))
- file_stat = os.stat(cfgpath)
- self.assertEqual(0o644, stat.S_IMODE(file_stat.st_mode))
+
+ # distro.apply_network(BASE_NET_CFG, False)
+ self._apply_and_verify(self.distro.apply_network,
+ BASE_NET_CFG,
+ expected_cfgs=expected_cfgs.copy())
+
+ def test_apply_network_config_opensuse(self):
+ """Opensuse uses apply_network_config and renders sysconfig"""
+ expected_cfgs = {
+ self.ifcfg_path('eth0'): dedent("""\
+ BOOTPROTO=none
+ DEFROUTE=yes
+ DEVICE=eth0
+ GATEWAY=192.168.1.254
+ IPADDR=192.168.1.5
+ NETMASK=255.255.255.0
+ NM_CONTROLLED=no
+ ONBOOT=yes
+ TYPE=Ethernet
+ USERCTL=no
+ """),
+ self.ifcfg_path('eth1'): dedent("""\
+ BOOTPROTO=dhcp
+ DEVICE=eth1
+ NM_CONTROLLED=no
+ ONBOOT=yes
+ TYPE=Ethernet
+ USERCTL=no
+ """),
+ }
+ self._apply_and_verify(self.distro.apply_network_config,
+ V1_NET_CFG,
+ expected_cfgs=expected_cfgs.copy())
+
+ def test_apply_network_config_ipv6_opensuse(self):
+ """Opensuse uses apply_network_config and renders sysconfig w/ipv6"""
+ expected_cfgs = {
+ self.ifcfg_path('eth0'): dedent("""\
+ BOOTPROTO=none
+ DEFROUTE=yes
+ DEVICE=eth0
+ IPV6ADDR=2607:f0d0:1002:0011::2/64
+ IPV6INIT=yes
+ IPV6_DEFAULTGW=2607:f0d0:1002:0011::1
+ NM_CONTROLLED=no
+ ONBOOT=yes
+ TYPE=Ethernet
+ USERCTL=no
+ """),
+ self.ifcfg_path('eth1'): dedent("""\
+ BOOTPROTO=dhcp
+ DEVICE=eth1
+ NM_CONTROLLED=no
+ ONBOOT=yes
+ TYPE=Ethernet
+ USERCTL=no
+ """),
+ }
+ self._apply_and_verify(self.distro.apply_network_config,
+ V1_NET_CFG_IPV6,
+ expected_cfgs=expected_cfgs.copy())
+
+
+def get_mode(path, target=None):
+ return os.stat(util.target_path(target, path)).st_mode & 0o777
# vi: ts=4 expandtab
diff --git a/tests/unittests/test_ds_identify.py b/tests/unittests/test_ds_identify.py
index e08e790..46778e9 100644
--- a/tests/unittests/test_ds_identify.py
+++ b/tests/unittests/test_ds_identify.py
@@ -89,6 +89,7 @@ CallReturn = namedtuple('CallReturn',
class DsIdentifyBase(CiTestCase):
dsid_path = os.path.realpath('tools/ds-identify')
+ allowed_subp = ['sh']
def call(self, rootd=None, mocks=None, func="main", args=None, files=None,
policy_dmi=DI_DEFAULT_POLICY,
diff --git a/tests/unittests/test_handler/test_handler_apt_source_v3.py b/tests/unittests/test_handler/test_handler_apt_source_v3.py
index 7a64c23..a81c67c 100644
--- a/tests/unittests/test_handler/test_handler_apt_source_v3.py
+++ b/tests/unittests/test_handler/test_handler_apt_source_v3.py
@@ -48,6 +48,10 @@ ADD_APT_REPO_MATCH = r"^[\w-]+:\w"
TARGET = None
+MOCK_LSB_RELEASE_DATA = {
+ 'id': 'Ubuntu', 'description': 'Ubuntu 18.04.1 LTS',
+ 'release': '18.04', 'codename': 'bionic'}
+
class TestAptSourceConfig(t_help.FilesystemMockingTestCase):
"""TestAptSourceConfig
@@ -64,6 +68,9 @@ class TestAptSourceConfig(t_help.FilesystemMockingTestCase):
self.aptlistfile3 = os.path.join(self.tmp, "single-deb3.list")
self.join = os.path.join
self.matcher = re.compile(ADD_APT_REPO_MATCH).search
+ self.add_patch(
+ 'cloudinit.config.cc_apt_configure.util.lsb_release',
+ 'm_lsb_release', return_value=MOCK_LSB_RELEASE_DATA.copy())
@staticmethod
def _add_apt_sources(*args, **kwargs):
@@ -76,7 +83,7 @@ class TestAptSourceConfig(t_help.FilesystemMockingTestCase):
Get the most basic default mrror and release info to be used in tests
"""
params = {}
- params['RELEASE'] = util.lsb_release()['codename']
+ params['RELEASE'] = MOCK_LSB_RELEASE_DATA['release']
arch = 'amd64'
params['MIRROR'] = cc_apt_configure.\
get_default_mirrors(arch)["PRIMARY"]
@@ -464,7 +471,7 @@ class TestAptSourceConfig(t_help.FilesystemMockingTestCase):
'uri':
'http://testsec.ubuntu.com/%s/' % component}]}
post = ("%s_dists_%s-updates_InRelease" %
- (component, util.lsb_release()['codename']))
+ (component, MOCK_LSB_RELEASE_DATA['codename']))
fromfn = ("%s/%s_%s" % (pre, archive, post))
tofn = ("%s/test.ubuntu.com_%s" % (pre, post))
diff --git a/tests/unittests/test_handler/test_handler_bootcmd.py b/tests/unittests/test_handler/test_handler_bootcmd.py
index b137526..a76760f 100644
--- a/tests/unittests/test_handler/test_handler_bootcmd.py
+++ b/tests/unittests/test_handler/test_handler_bootcmd.py
@@ -118,7 +118,8 @@ class TestBootcmd(CiTestCase):
'echo {0} $INSTANCE_ID > {1}'.format(my_id, out_file)]}
with mock.patch(self._etmpfile_path, FakeExtendedTempFile):
- handle('cc_bootcmd', valid_config, cc, LOG, [])
+ with self.allow_subp(['/bin/sh']):
+ handle('cc_bootcmd', valid_config, cc, LOG, [])
self.assertEqual(my_id + ' iid-datasource-none\n',
util.load_file(out_file))
@@ -128,12 +129,13 @@ class TestBootcmd(CiTestCase):
valid_config = {'bootcmd': ['exit 1']} # Script with error
with mock.patch(self._etmpfile_path, FakeExtendedTempFile):
- with self.assertRaises(util.ProcessExecutionError) as ctxt_manager:
- handle('does-not-matter', valid_config, cc, LOG, [])
+ with self.allow_subp(['/bin/sh']):
+ with self.assertRaises(util.ProcessExecutionError) as ctxt:
+ handle('does-not-matter', valid_config, cc, LOG, [])
self.assertIn(
'Unexpected error while running command.\n'
"Command: ['/bin/sh',",
- str(ctxt_manager.exception))
+ str(ctxt.exception))
self.assertIn(
'Failed to run bootcmd module does-not-matter',
self.logs.getvalue())
diff --git a/tests/unittests/test_handler/test_handler_chef.py b/tests/unittests/test_handler/test_handler_chef.py
index f4bbd66..b16532e 100644
--- a/tests/unittests/test_handler/test_handler_chef.py
+++ b/tests/unittests/test_handler/test_handler_chef.py
@@ -36,13 +36,21 @@ class TestInstallChefOmnibus(HttprettyTestCase):
@mock.patch("cloudinit.config.cc_chef.OMNIBUS_URL", OMNIBUS_URL_HTTP)
def test_install_chef_from_omnibus_runs_chef_url_content(self):
- """install_chef_from_omnibus runs downloaded OMNIBUS_URL as script."""
- chef_outfile = self.tmp_path('chef.out', self.new_root)
- response = '#!/bin/bash\necho "Hi Mom" > {0}'.format(chef_outfile)
+ """install_chef_from_omnibus calls subp_blob_in_tempfile."""
+ response = b'#!/bin/bash\necho "Hi Mom"'
httpretty.register_uri(
httpretty.GET, cc_chef.OMNIBUS_URL, body=response, status=200)
- cc_chef.install_chef_from_omnibus()
- self.assertEqual('Hi Mom\n', util.load_file(chef_outfile))
+ ret = (None, None) # stdout, stderr but capture=False
+
+ with mock.patch("cloudinit.config.cc_chef.util.subp_blob_in_tempfile",
+ return_value=ret) as m_subp_blob:
+ cc_chef.install_chef_from_omnibus()
+ # admittedly whitebox, but assuming subp_blob_in_tempfile works
+ # this should be fine.
+ self.assertEqual(
+ [mock.call(blob=response, args=[], basename='chef-omnibus-install',
+ capture=False)],
+ m_subp_blob.call_args_list)
@mock.patch('cloudinit.config.cc_chef.url_helper.readurl')
@mock.patch('cloudinit.config.cc_chef.util.subp_blob_in_tempfile')
diff --git a/tests/unittests/test_handler/test_handler_resizefs.py b/tests/unittests/test_handler/test_handler_resizefs.py
index f92175f..feca56c 100644
--- a/tests/unittests/test_handler/test_handler_resizefs.py
+++ b/tests/unittests/test_handler/test_handler_resizefs.py
@@ -150,10 +150,12 @@ class TestResizefs(CiTestCase):
self.assertEqual(('growfs', '-y', devpth),
_resize_ufs(mount_point, devpth))
+ @mock.patch('cloudinit.util.is_container', return_value=False)
@mock.patch('cloudinit.util.get_mount_info')
@mock.patch('cloudinit.util.get_device_info_from_zpool')
@mock.patch('cloudinit.util.parse_mount')
- def test_handle_zfs_root(self, mount_info, zpool_info, parse_mount):
+ def test_handle_zfs_root(self, mount_info, zpool_info, parse_mount,
+ is_container):
devpth = 'vmzroot/ROOT/freebsd'
disk = 'gpt/system'
fs_type = 'zfs'
@@ -354,8 +356,10 @@ class TestMaybeGetDevicePathAsWritableBlock(CiTestCase):
('btrfs', 'filesystem', 'resize', 'max', '/'),
_resize_btrfs("/", "/dev/sda1"))
+ @mock.patch('cloudinit.util.is_container', return_value=True)
@mock.patch('cloudinit.util.is_FreeBSD')
- def test_maybe_get_writable_device_path_zfs_freebsd(self, freebsd):
+ def test_maybe_get_writable_device_path_zfs_freebsd(self, freebsd,
+ m_is_container):
freebsd.return_value = True
info = 'dev=gpt/system mnt_point=/ path=/'
devpth = maybe_get_writable_device_path('gpt/system', info, LOG)
diff --git a/tests/unittests/test_handler/test_schema.py b/tests/unittests/test_handler/test_schema.py
index fb266fa..1bad07f 100644
--- a/tests/unittests/test_handler/test_schema.py
+++ b/tests/unittests/test_handler/test_schema.py
@@ -4,7 +4,7 @@ from cloudinit.config.schema import (
CLOUD_CONFIG_HEADER, SchemaValidationError, annotated_cloudconfig_file,
get_schema_doc, get_schema, validate_cloudconfig_file,
validate_cloudconfig_schema, main)
-from cloudinit.util import subp, write_file
+from cloudinit.util import write_file
from cloudinit.tests.helpers import CiTestCase, mock, skipUnlessJsonSchema
@@ -406,8 +406,14 @@ class CloudTestsIntegrationTest(CiTestCase):
integration_testdir = os.path.sep.join(
[testsdir, 'cloud_tests', 'testcases'])
errors = []
- out, _ = subp(['find', integration_testdir, '-name', '*yaml'])
- for filename in out.splitlines():
+
+ yaml_files = []
+ for root, _dirnames, filenames in os.walk(integration_testdir):
+ yaml_files.extend([os.path.join(root, f)
+ for f in filenames if f.endswith(".yaml")])
+ self.assertTrue(len(yaml_files) > 0)
+
+ for filename in yaml_files:
test_cfg = safe_load(open(filename))
cloud_config = test_cfg.get('cloud_config')
if cloud_config:
diff --git a/tests/unittests/test_net.py b/tests/unittests/test_net.py
index 58e5ea1..f3165da 100644
--- a/tests/unittests/test_net.py
+++ b/tests/unittests/test_net.py
@@ -1,6 +1,7 @@
# This file is part of cloud-init. See LICENSE file for license information.
from cloudinit import net
+from cloudinit import distros
from cloudinit.net import cmdline
from cloudinit.net import (
eni, interface_has_own_mac, natural_sort_key, netplan, network_state,
@@ -129,7 +130,40 @@ OS_SAMPLES = [
'in_macs': {
'fa:16:3e:ed:9a:59': 'eth0',
},
- 'out_sysconfig': [
+ 'out_sysconfig_opensuse': [
+ ('etc/sysconfig/network/ifcfg-eth0',
+ """
+# Created by cloud-init on instance boot automatically, do not edit.
+#
+BOOTPROTO=none
+DEFROUTE=yes
+DEVICE=eth0
+GATEWAY=172.19.3.254
+HWADDR=fa:16:3e:ed:9a:59
+IPADDR=172.19.1.34
+NETMASK=255.255.252.0
+NM_CONTROLLED=no
+ONBOOT=yes
+TYPE=Ethernet
+USERCTL=no
+""".lstrip()),
+ ('etc/resolv.conf',
+ """
+; Created by cloud-init on instance boot automatically, do not edit.
+;
+nameserver 172.19.0.12
+""".lstrip()),
+ ('etc/NetworkManager/conf.d/99-cloud-init.conf',
+ """
+# Created by cloud-init on instance boot automatically, do not edit.
+#
+[main]
+dns = none
+""".lstrip()),
+ ('etc/udev/rules.d/70-persistent-net.rules',
+ "".join(['SUBSYSTEM=="net", ACTION=="add", DRIVERS=="?*", ',
+ 'ATTR{address}=="fa:16:3e:ed:9a:59", NAME="eth0"\n']))],
+ 'out_sysconfig_rhel': [
('etc/sysconfig/network-scripts/ifcfg-eth0',
"""
# Created by cloud-init on instance boot automatically, do not edit.
@@ -162,6 +196,7 @@ dns = none
('etc/udev/rules.d/70-persistent-net.rules',
"".join(['SUBSYSTEM=="net", ACTION=="add", DRIVERS=="?*", ',
'ATTR{address}=="fa:16:3e:ed:9a:59", NAME="eth0"\n']))]
+
},
{
'in_data': {
@@ -195,7 +230,42 @@ dns = none
'in_macs': {
'fa:16:3e:ed:9a:59': 'eth0',
},
- 'out_sysconfig': [
+ 'out_sysconfig_opensuse': [
+ ('etc/sysconfig/network/ifcfg-eth0',
+ """
+# Created by cloud-init on instance boot automatically, do not edit.
+#
+BOOTPROTO=none
+DEFROUTE=yes
+DEVICE=eth0
+GATEWAY=172.19.3.254
+HWADDR=fa:16:3e:ed:9a:59
+IPADDR=172.19.1.34
+IPADDR1=10.0.0.10
+NETMASK=255.255.252.0
+NETMASK1=255.255.255.0
+NM_CONTROLLED=no
+ONBOOT=yes
+TYPE=Ethernet
+USERCTL=no
+""".lstrip()),
+ ('etc/resolv.conf',
+ """
+; Created by cloud-init on instance boot automatically, do not edit.
+;
+nameserver 172.19.0.12
+""".lstrip()),
+ ('etc/NetworkManager/conf.d/99-cloud-init.conf',
+ """
+# Created by cloud-init on instance boot automatically, do not edit.
+#
+[main]
+dns = none
+""".lstrip()),
+ ('etc/udev/rules.d/70-persistent-net.rules',
+ "".join(['SUBSYSTEM=="net", ACTION=="add", DRIVERS=="?*", ',
+ 'ATTR{address}=="fa:16:3e:ed:9a:59", NAME="eth0"\n']))],
+ 'out_sysconfig_rhel': [
('etc/sysconfig/network-scripts/ifcfg-eth0',
"""
# Created by cloud-init on instance boot automatically, do not edit.
@@ -230,6 +300,7 @@ dns = none
('etc/udev/rules.d/70-persistent-net.rules',
"".join(['SUBSYSTEM=="net", ACTION=="add", DRIVERS=="?*", ',
'ATTR{address}=="fa:16:3e:ed:9a:59", NAME="eth0"\n']))]
+
},
{
'in_data': {
@@ -283,7 +354,44 @@ dns = none
'in_macs': {
'fa:16:3e:ed:9a:59': 'eth0',
},
- 'out_sysconfig': [
+ 'out_sysconfig_opensuse': [
+ ('etc/sysconfig/network/ifcfg-eth0',
+ """
+# Created by cloud-init on instance boot automatically, do not edit.
+#
+BOOTPROTO=none
+DEFROUTE=yes
+DEVICE=eth0
+GATEWAY=172.19.3.254
+HWADDR=fa:16:3e:ed:9a:59
+IPADDR=172.19.1.34
+IPV6ADDR=2001:DB8::10/64
+IPV6ADDR_SECONDARIES="2001:DB9::10/64 2001:DB10::10/64"
+IPV6INIT=yes
+IPV6_DEFAULTGW=2001:DB8::1
+NETMASK=255.255.252.0
+NM_CONTROLLED=no
+ONBOOT=yes
+TYPE=Ethernet
+USERCTL=no
+""".lstrip()),
+ ('etc/resolv.conf',
+ """
+; Created by cloud-init on instance boot automatically, do not edit.
+;
+nameserver 172.19.0.12
+""".lstrip()),
+ ('etc/NetworkManager/conf.d/99-cloud-init.conf',
+ """
+# Created by cloud-init on instance boot automatically, do not edit.
+#
+[main]
+dns = none
+""".lstrip()),
+ ('etc/udev/rules.d/70-persistent-net.rules',
+ "".join(['SUBSYSTEM=="net", ACTION=="add", DRIVERS=="?*", ',
+ 'ATTR{address}=="fa:16:3e:ed:9a:59", NAME="eth0"\n']))],
+ 'out_sysconfig_rhel': [
('etc/sysconfig/network-scripts/ifcfg-eth0',
"""
# Created by cloud-init on instance boot automatically, do not edit.
@@ -1154,7 +1262,59 @@ pre-down route del -net 10.0.0.0 netmask 255.0.0.0 gw 11.0.0.1 metric 3 || true
version: 2
"""),
- 'expected_sysconfig': {
+ 'expected_sysconfig_opensuse': {
+ 'ifcfg-bond0': textwrap.dedent("""\
+ BONDING_MASTER=yes
+ BONDING_OPTS="mode=active-backup xmit_hash_policy=layer3+4 miimon=100"
+ BONDING_SLAVE0=bond0s0
+ BONDING_SLAVE1=bond0s1
+ BOOTPROTO=none
+ DEFROUTE=yes
+ DEVICE=bond0
+ GATEWAY=192.168.0.1
+ MACADDR=aa:bb:cc:dd:e8:ff
+ IPADDR=192.168.0.2
+ IPADDR1=192.168.1.2
+ IPV6ADDR=2001:1::1/92
+ IPV6INIT=yes
+ MTU=9000
+ NETMASK=255.255.255.0
+ NETMASK1=255.255.255.0
+ NM_CONTROLLED=no
+ ONBOOT=yes
+ TYPE=Bond
+ USERCTL=no
+ """),
+ 'ifcfg-bond0s0': textwrap.dedent("""\
+ BOOTPROTO=none
+ DEVICE=bond0s0
+ HWADDR=aa:bb:cc:dd:e8:00
+ MASTER=bond0
+ NM_CONTROLLED=no
+ ONBOOT=yes
+ SLAVE=yes
+ TYPE=Ethernet
+ USERCTL=no
+ """),
+ 'ifroute-bond0': textwrap.dedent("""\
+ ADDRESS0=10.1.3.0
+ GATEWAY0=192.168.0.3
+ NETMASK0=255.255.255.0
+ """),
+ 'ifcfg-bond0s1': textwrap.dedent("""\
+ BOOTPROTO=none
+ DEVICE=bond0s1
+ HWADDR=aa:bb:cc:dd:e8:01
+ MASTER=bond0
+ NM_CONTROLLED=no
+ ONBOOT=yes
+ SLAVE=yes
+ TYPE=Ethernet
+ USERCTL=no
+ """),
+ },
+
+ 'expected_sysconfig_rhel': {
'ifcfg-bond0': textwrap.dedent("""\
BONDING_MASTER=yes
BONDING_OPTS="mode=active-backup xmit_hash_policy=layer3+4 miimon=100"
@@ -1493,6 +1653,12 @@ def _setup_test(tmp_dir, mock_get_devicelist, mock_read_sys_net,
class TestGenerateFallbackConfig(CiTestCase):
+ def setUp(self):
+ super(TestGenerateFallbackConfig, self).setUp()
+ self.add_patch(
+ "cloudinit.util.get_cmdline", "m_get_cmdline",
+ return_value="root=/dev/sda1")
+
@mock.patch("cloudinit.net.sys_dev_path")
@mock.patch("cloudinit.net.read_sys_net")
@mock.patch("cloudinit.net.get_devicelist")
@@ -1527,7 +1693,7 @@ class TestGenerateFallbackConfig(CiTestCase):
# don't set rulepath so eni writes them
renderer = eni.Renderer(
{'eni_path': 'interfaces', 'netrules_path': 'netrules'})
- renderer.render_network_state(ns, render_dir)
+ renderer.render_network_state(ns, target=render_dir)
self.assertTrue(os.path.exists(os.path.join(render_dir,
'interfaces')))
@@ -1591,7 +1757,7 @@ iface eth0 inet dhcp
# don't set rulepath so eni writes them
renderer = eni.Renderer(
{'eni_path': 'interfaces', 'netrules_path': 'netrules'})
- renderer.render_network_state(ns, render_dir)
+ renderer.render_network_state(ns, target=render_dir)
self.assertTrue(os.path.exists(os.path.join(render_dir,
'interfaces')))
@@ -1682,7 +1848,7 @@ iface eth1 inet dhcp
self.assertEqual(0, mock_settle.call_count)
-class TestSysConfigRendering(CiTestCase):
+class TestRhelSysConfigRendering(CiTestCase):
with_logs = True
@@ -1690,6 +1856,13 @@ class TestSysConfigRendering(CiTestCase):
header = ('# Created by cloud-init on instance boot automatically, '
'do not edit.\n#\n')
+ expected_name = 'expected_sysconfig'
+
+ def _get_renderer(self):
+ distro_cls = distros.fetch('rhel')
+ return sysconfig.Renderer(
+ config=distro_cls.renderer_configs.get('sysconfig'))
+
def _render_and_read(self, network_config=None, state=None, dir=None):
if dir is None:
dir = self.tmp_dir()
@@ -1701,8 +1874,8 @@ class TestSysConfigRendering(CiTestCase):
else:
raise ValueError("Expected data or state, got neither")
- renderer = sysconfig.Renderer()
- renderer.render_network_state(ns, dir)
+ renderer = self._get_renderer()
+ renderer.render_network_state(ns, target=dir)
return dir2dict(dir)
def _compare_files_to_expected(self, expected, found):
@@ -1728,12 +1901,13 @@ class TestSysConfigRendering(CiTestCase):
if missing:
raise AssertionError("Missing headers in: %s" % missing)
+ @mock.patch("cloudinit.net.util.get_cmdline", return_value="root=myroot")
@mock.patch("cloudinit.net.sys_dev_path")
@mock.patch("cloudinit.net.read_sys_net")
@mock.patch("cloudinit.net.get_devicelist")
def test_default_generation(self, mock_get_devicelist,
mock_read_sys_net,
- mock_sys_dev_path):
+ mock_sys_dev_path, m_get_cmdline):
tmp_dir = self.tmp_dir()
_setup_test(tmp_dir, mock_get_devicelist,
mock_read_sys_net, mock_sys_dev_path)
@@ -1745,8 +1919,8 @@ class TestSysConfigRendering(CiTestCase):
render_dir = os.path.join(tmp_dir, "render")
os.makedirs(render_dir)
- renderer = sysconfig.Renderer()
- renderer.render_network_state(ns, render_dir)
+ renderer = self._get_renderer()
+ renderer.render_network_state(ns, target=render_dir)
render_file = 'etc/sysconfig/network-scripts/ifcfg-eth1000'
with open(os.path.join(render_dir, render_file)) as fh:
@@ -1797,9 +1971,9 @@ USERCTL=no
network_cfg = openstack.convert_net_json(net_json, known_macs=macs)
ns = network_state.parse_net_config_data(network_cfg,
skip_broken=False)
- renderer = sysconfig.Renderer()
+ renderer = self._get_renderer()
with self.assertRaises(ValueError):
- renderer.render_network_state(ns, render_dir)
+ renderer.render_network_state(ns, target=render_dir)
self.assertEqual([], os.listdir(render_dir))
def test_multiple_ipv6_default_gateways(self):
@@ -1835,9 +2009,9 @@ USERCTL=no
network_cfg = openstack.convert_net_json(net_json, known_macs=macs)
ns = network_state.parse_net_config_data(network_cfg,
skip_broken=False)
- renderer = sysconfig.Renderer()
+ renderer = self._get_renderer()
with self.assertRaises(ValueError):
- renderer.render_network_state(ns, render_dir)
+ renderer.render_network_state(ns, target=render_dir)
self.assertEqual([], os.listdir(render_dir))
def test_openstack_rendering_samples(self):
@@ -1849,12 +2023,13 @@ USERCTL=no
ex_input, known_macs=ex_mac_addrs)
ns = network_state.parse_net_config_data(network_cfg,
skip_broken=False)
- renderer = sysconfig.Renderer()
+ renderer = self._get_renderer()
# render a multiple times to simulate reboots
- renderer.render_network_state(ns, render_dir)
- renderer.render_network_state(ns, render_dir)
- renderer.render_network_state(ns, render_dir)
- for fn, expected_content in os_sample.get('out_sysconfig', []):
+ renderer.render_network_state(ns, target=render_dir)
+ renderer.render_network_state(ns, target=render_dir)
+ renderer.render_network_state(ns, target=render_dir)
+ for fn, expected_content in os_sample.get('out_sysconfig_rhel',
+ []):
with open(os.path.join(render_dir, fn)) as fh:
self.assertEqual(expected_content, fh.read())
@@ -1862,8 +2037,8 @@ USERCTL=no
ns = network_state.parse_net_config_data(CONFIG_V1_SIMPLE_SUBNET)
render_dir = self.tmp_path("render")
os.makedirs(render_dir)
- renderer = sysconfig.Renderer()
- renderer.render_network_state(ns, render_dir)
+ renderer = self._get_renderer()
+ renderer.render_network_state(ns, target=render_dir)
found = dir2dict(render_dir)
nspath = '/etc/sysconfig/network-scripts/'
self.assertNotIn(nspath + 'ifcfg-lo', found.keys())
@@ -1888,8 +2063,8 @@ USERCTL=no
ns = network_state.parse_net_config_data(CONFIG_V1_EXPLICIT_LOOPBACK)
render_dir = self.tmp_path("render")
os.makedirs(render_dir)
- renderer = sysconfig.Renderer()
- renderer.render_network_state(ns, render_dir)
+ renderer = self._get_renderer()
+ renderer.render_network_state(ns, target=render_dir)
found = dir2dict(render_dir)
nspath = '/etc/sysconfig/network-scripts/'
self.assertNotIn(nspath + 'ifcfg-lo', found.keys())
@@ -1906,33 +2081,332 @@ USERCTL=no
self.assertEqual(expected, found[nspath + 'ifcfg-eth0'])
def test_bond_config(self):
+ expected_name = 'expected_sysconfig_rhel'
entry = NETWORK_CONFIGS['bond']
found = self._render_and_read(network_config=yaml.load(entry['yaml']))
- self._compare_files_to_expected(entry['expected_sysconfig'], found)
+ self._compare_files_to_expected(entry[expected_name], found)
+ self._assert_headers(found)
+
+ def test_vlan_config(self):
+ entry = NETWORK_CONFIGS['vlan']
+ found = self._render_and_read(network_config=yaml.load(entry['yaml']))
+ self._compare_files_to_expected(entry[self.expected_name], found)
+ self._assert_headers(found)
+
+ def test_bridge_config(self):
+ entry = NETWORK_CONFIGS['bridge']
+ found = self._render_and_read(network_config=yaml.load(entry['yaml']))
+ self._compare_files_to_expected(entry[self.expected_name], found)
+ self._assert_headers(found)
+
+ def test_manual_config(self):
+ entry = NETWORK_CONFIGS['manual']
+ found = self._render_and_read(network_config=yaml.load(entry['yaml']))
+ self._compare_files_to_expected(entry[self.expected_name], found)
+ self._assert_headers(found)
+
+ def test_all_config(self):
+ entry = NETWORK_CONFIGS['all']
+ found = self._render_and_read(network_config=yaml.load(entry['yaml']))
+ self._compare_files_to_expected(entry[self.expected_name], found)
+ self._assert_headers(found)
+ self.assertNotIn(
+ 'WARNING: Network config: ignoring eth0.101 device-level mtu',
+ self.logs.getvalue())
+
+ def test_small_config(self):
+ entry = NETWORK_CONFIGS['small']
+ found = self._render_and_read(network_config=yaml.load(entry['yaml']))
+ self._compare_files_to_expected(entry[self.expected_name], found)
+ self._assert_headers(found)
+
+ def test_v4_and_v6_static_config(self):
+ entry = NETWORK_CONFIGS['v4_and_v6_static']
+ found = self._render_and_read(network_config=yaml.load(entry['yaml']))
+ self._compare_files_to_expected(entry[self.expected_name], found)
+ self._assert_headers(found)
+ expected_msg = (
+ 'WARNING: Network config: ignoring iface0 device-level mtu:8999'
+ ' because ipv4 subnet-level mtu:9000 provided.')
+ self.assertIn(expected_msg, self.logs.getvalue())
+
+ def test_dhcpv6_only_config(self):
+ entry = NETWORK_CONFIGS['dhcpv6_only']
+ found = self._render_and_read(network_config=yaml.load(entry['yaml']))
+ self._compare_files_to_expected(entry[self.expected_name], found)
+ self._assert_headers(found)
+
+
+class TestOpenSuseSysConfigRendering(CiTestCase):
+
+ with_logs = True
+
+ scripts_dir = '/etc/sysconfig/network'
+ header = ('# Created by cloud-init on instance boot automatically, '
+ 'do not edit.\n#\n')
+
+ expected_name = 'expected_sysconfig'
+
+ def _get_renderer(self):
+ distro_cls = distros.fetch('opensuse')
+ return sysconfig.Renderer(
+ config=distro_cls.renderer_configs.get('sysconfig'))
+
+ def _render_and_read(self, network_config=None, state=None, dir=None):
+ if dir is None:
+ dir = self.tmp_dir()
+
+ if network_config:
+ ns = network_state.parse_net_config_data(network_config)
+ elif state:
+ ns = state
+ else:
+ raise ValueError("Expected data or state, got neither")
+
+ renderer = self._get_renderer()
+ renderer.render_network_state(ns, target=dir)
+ return dir2dict(dir)
+
+ def _compare_files_to_expected(self, expected, found):
+ orig_maxdiff = self.maxDiff
+ expected_d = dict(
+ (os.path.join(self.scripts_dir, k), util.load_shell_content(v))
+ for k, v in expected.items())
+
+ # only compare the files in scripts_dir
+ scripts_found = dict(
+ (k, util.load_shell_content(v)) for k, v in found.items()
+ if k.startswith(self.scripts_dir))
+ try:
+ self.maxDiff = None
+ self.assertEqual(expected_d, scripts_found)
+ finally:
+ self.maxDiff = orig_maxdiff
+
+ def _assert_headers(self, found):
+ missing = [f for f in found
+ if (f.startswith(self.scripts_dir) and
+ not found[f].startswith(self.header))]
+ if missing:
+ raise AssertionError("Missing headers in: %s" % missing)
+
+ @mock.patch("cloudinit.net.util.get_cmdline", return_value="root=myroot")
+ @mock.patch("cloudinit.net.sys_dev_path")
+ @mock.patch("cloudinit.net.read_sys_net")
+ @mock.patch("cloudinit.net.get_devicelist")
+ def test_default_generation(self, mock_get_devicelist,
+ mock_read_sys_net,
+ mock_sys_dev_path, m_get_cmdline):
+ tmp_dir = self.tmp_dir()
+ _setup_test(tmp_dir, mock_get_devicelist,
+ mock_read_sys_net, mock_sys_dev_path)
+
+ network_cfg = net.generate_fallback_config()
+ ns = network_state.parse_net_config_data(network_cfg,
+ skip_broken=False)
+
+ render_dir = os.path.join(tmp_dir, "render")
+ os.makedirs(render_dir)
+
+ renderer = self._get_renderer()
+ renderer.render_network_state(ns, target=render_dir)
+
+ render_file = 'etc/sysconfig/network/ifcfg-eth1000'
+ with open(os.path.join(render_dir, render_file)) as fh:
+ content = fh.read()
+ expected_content = """
+# Created by cloud-init on instance boot automatically, do not edit.
+#
+BOOTPROTO=dhcp
+DEVICE=eth1000
+HWADDR=07-1C-C6-75-A4-BE
+NM_CONTROLLED=no
+ONBOOT=yes
+TYPE=Ethernet
+USERCTL=no
+""".lstrip()
+ self.assertEqual(expected_content, content)
+
+ def test_multiple_ipv4_default_gateways(self):
+ """ValueError is raised when duplicate ipv4 gateways exist."""
+ net_json = {
+ "services": [{"type": "dns", "address": "172.19.0.12"}],
+ "networks": [{
+ "network_id": "dacd568d-5be6-4786-91fe-750c374b78b4",
+ "type": "ipv4", "netmask": "255.255.252.0",
+ "link": "tap1a81968a-79",
+ "routes": [{
+ "netmask": "0.0.0.0",
+ "network": "0.0.0.0",
+ "gateway": "172.19.3.254",
+ }, {
+ "netmask": "0.0.0.0", # A second default gateway
+ "network": "0.0.0.0",
+ "gateway": "172.20.3.254",
+ }],
+ "ip_address": "172.19.1.34", "id": "network0"
+ }],
+ "links": [
+ {
+ "ethernet_mac_address": "fa:16:3e:ed:9a:59",
+ "mtu": None, "type": "bridge", "id":
+ "tap1a81968a-79",
+ "vif_id": "1a81968a-797a-400f-8a80-567f997eb93f"
+ },
+ ],
+ }
+ macs = {'fa:16:3e:ed:9a:59': 'eth0'}
+ render_dir = self.tmp_dir()
+ network_cfg = openstack.convert_net_json(net_json, known_macs=macs)
+ ns = network_state.parse_net_config_data(network_cfg,
+ skip_broken=False)
+ renderer = self._get_renderer()
+ with self.assertRaises(ValueError):
+ renderer.render_network_state(ns, target=render_dir)
+ self.assertEqual([], os.listdir(render_dir))
+
+ def test_multiple_ipv6_default_gateways(self):
+ """ValueError is raised when duplicate ipv6 gateways exist."""
+ net_json = {
+ "services": [{"type": "dns", "address": "172.19.0.12"}],
+ "networks": [{
+ "network_id": "public-ipv6",
+ "type": "ipv6", "netmask": "",
+ "link": "tap1a81968a-79",
+ "routes": [{
+ "gateway": "2001:DB8::1",
+ "netmask": "::",
+ "network": "::"
+ }, {
+ "gateway": "2001:DB9::1",
+ "netmask": "::",
+ "network": "::"
+ }],
+ "ip_address": "2001:DB8::10", "id": "network1"
+ }],
+ "links": [
+ {
+ "ethernet_mac_address": "fa:16:3e:ed:9a:59",
+ "mtu": None, "type": "bridge", "id":
+ "tap1a81968a-79",
+ "vif_id": "1a81968a-797a-400f-8a80-567f997eb93f"
+ },
+ ],
+ }
+ macs = {'fa:16:3e:ed:9a:59': 'eth0'}
+ render_dir = self.tmp_dir()
+ network_cfg = openstack.convert_net_json(net_json, known_macs=macs)
+ ns = network_state.parse_net_config_data(network_cfg,
+ skip_broken=False)
+ renderer = self._get_renderer()
+ with self.assertRaises(ValueError):
+ renderer.render_network_state(ns, target=render_dir)
+ self.assertEqual([], os.listdir(render_dir))
+
+ def test_openstack_rendering_samples(self):
+ for os_sample in OS_SAMPLES:
+ render_dir = self.tmp_dir()
+ ex_input = os_sample['in_data']
+ ex_mac_addrs = os_sample['in_macs']
+ network_cfg = openstack.convert_net_json(
+ ex_input, known_macs=ex_mac_addrs)
+ ns = network_state.parse_net_config_data(network_cfg,
+ skip_broken=False)
+ renderer = self._get_renderer()
+ # render a multiple times to simulate reboots
+ renderer.render_network_state(ns, target=render_dir)
+ renderer.render_network_state(ns, target=render_dir)
+ renderer.render_network_state(ns, target=render_dir)
+ for fn, expected_content in os_sample.get('out_sysconfig_opensuse',
+ []):
+ with open(os.path.join(render_dir, fn)) as fh:
+ self.assertEqual(expected_content, fh.read())
+
+ def test_network_config_v1_samples(self):
+ ns = network_state.parse_net_config_data(CONFIG_V1_SIMPLE_SUBNET)
+ render_dir = self.tmp_path("render")
+ os.makedirs(render_dir)
+ renderer = self._get_renderer()
+ renderer.render_network_state(ns, target=render_dir)
+ found = dir2dict(render_dir)
+ nspath = '/etc/sysconfig/network/'
+ self.assertNotIn(nspath + 'ifcfg-lo', found.keys())
+ expected = """\
+# Created by cloud-init on instance boot automatically, do not edit.
+#
+BOOTPROTO=none
+DEFROUTE=yes
+DEVICE=interface0
+GATEWAY=10.0.2.2
+HWADDR=52:54:00:12:34:00
+IPADDR=10.0.2.15
+NETMASK=255.255.255.0
+NM_CONTROLLED=no
+ONBOOT=yes
+TYPE=Ethernet
+USERCTL=no
+"""
+ self.assertEqual(expected, found[nspath + 'ifcfg-interface0'])
+
+ def test_config_with_explicit_loopback(self):
+ ns = network_state.parse_net_config_data(CONFIG_V1_EXPLICIT_LOOPBACK)
+ render_dir = self.tmp_path("render")
+ os.makedirs(render_dir)
+ renderer = self._get_renderer()
+ renderer.render_network_state(ns, target=render_dir)
+ found = dir2dict(render_dir)
+ nspath = '/etc/sysconfig/network/'
+ self.assertNotIn(nspath + 'ifcfg-lo', found.keys())
+ expected = """\
+# Created by cloud-init on instance boot automatically, do not edit.
+#
+BOOTPROTO=dhcp
+DEVICE=eth0
+NM_CONTROLLED=no
+ONBOOT=yes
+TYPE=Ethernet
+USERCTL=no
+"""
+ self.assertEqual(expected, found[nspath + 'ifcfg-eth0'])
+
+ def test_bond_config(self):
+ expected_name = 'expected_sysconfig_opensuse'
+ entry = NETWORK_CONFIGS['bond']
+ found = self._render_and_read(network_config=yaml.load(entry['yaml']))
+ for fname, contents in entry[expected_name].items():
+ print(fname)
+ print(contents)
+ print()
+ print('-- expected ^ | v rendered --')
+ for fname, contents in found.items():
+ print(fname)
+ print(contents)
+ print()
+ self._compare_files_to_expected(entry[expected_name], found)
self._assert_headers(found)
def test_vlan_config(self):
entry = NETWORK_CONFIGS['vlan']
found = self._render_and_read(network_config=yaml.load(entry['yaml']))
- self._compare_files_to_expected(entry['expected_sysconfig'], found)
+ self._compare_files_to_expected(entry[self.expected_name], found)
self._assert_headers(found)
def test_bridge_config(self):
entry = NETWORK_CONFIGS['bridge']
found = self._render_and_read(network_config=yaml.load(entry['yaml']))
- self._compare_files_to_expected(entry['expected_sysconfig'], found)
+ self._compare_files_to_expected(entry[self.expected_name], found)
self._assert_headers(found)
def test_manual_config(self):
entry = NETWORK_CONFIGS['manual']
found = self._render_and_read(network_config=yaml.load(entry['yaml']))
- self._compare_files_to_expected(entry['expected_sysconfig'], found)
+ self._compare_files_to_expected(entry[self.expected_name], found)
self._assert_headers(found)
def test_all_config(self):
entry = NETWORK_CONFIGS['all']
found = self._render_and_read(network_config=yaml.load(entry['yaml']))
- self._compare_files_to_expected(entry['expected_sysconfig'], found)
+ self._compare_files_to_expected(entry[self.expected_name], found)
self._assert_headers(found)
self.assertNotIn(
'WARNING: Network config: ignoring eth0.101 device-level mtu',
@@ -1941,13 +2415,13 @@ USERCTL=no
def test_small_config(self):
entry = NETWORK_CONFIGS['small']
found = self._render_and_read(network_config=yaml.load(entry['yaml']))
- self._compare_files_to_expected(entry['expected_sysconfig'], found)
+ self._compare_files_to_expected(entry[self.expected_name], found)
self._assert_headers(found)
def test_v4_and_v6_static_config(self):
entry = NETWORK_CONFIGS['v4_and_v6_static']
found = self._render_and_read(network_config=yaml.load(entry['yaml']))
- self._compare_files_to_expected(entry['expected_sysconfig'], found)
+ self._compare_files_to_expected(entry[self.expected_name], found)
self._assert_headers(found)
expected_msg = (
'WARNING: Network config: ignoring iface0 device-level mtu:8999'
@@ -1957,18 +2431,19 @@ USERCTL=no
def test_dhcpv6_only_config(self):
entry = NETWORK_CONFIGS['dhcpv6_only']
found = self._render_and_read(network_config=yaml.load(entry['yaml']))
- self._compare_files_to_expected(entry['expected_sysconfig'], found)
+ self._compare_files_to_expected(entry[self.expected_name], found)
self._assert_headers(found)
class TestEniNetRendering(CiTestCase):
+ @mock.patch("cloudinit.net.util.get_cmdline", return_value="root=myroot")
@mock.patch("cloudinit.net.sys_dev_path")
@mock.patch("cloudinit.net.read_sys_net")
@mock.patch("cloudinit.net.get_devicelist")
def test_default_generation(self, mock_get_devicelist,
mock_read_sys_net,
- mock_sys_dev_path):
+ mock_sys_dev_path, m_get_cmdline):
tmp_dir = self.tmp_dir()
_setup_test(tmp_dir, mock_get_devicelist,
mock_read_sys_net, mock_sys_dev_path)
@@ -1982,7 +2457,7 @@ class TestEniNetRendering(CiTestCase):
renderer = eni.Renderer(
{'eni_path': 'interfaces', 'netrules_path': None})
- renderer.render_network_state(ns, render_dir)
+ renderer.render_network_state(ns, target=render_dir)
self.assertTrue(os.path.exists(os.path.join(render_dir,
'interfaces')))
@@ -2002,7 +2477,7 @@ iface eth1000 inet dhcp
tmp_dir = self.tmp_dir()
ns = network_state.parse_net_config_data(CONFIG_V1_EXPLICIT_LOOPBACK)
renderer = eni.Renderer()
- renderer.render_network_state(ns, tmp_dir)
+ renderer.render_network_state(ns, target=tmp_dir)
expected = """\
auto lo
iface lo inet loopback
@@ -2016,6 +2491,7 @@ iface eth0 inet dhcp
class TestNetplanNetRendering(CiTestCase):
+ @mock.patch("cloudinit.net.util.get_cmdline", return_value="root=myroot")
@mock.patch("cloudinit.net.netplan._clean_default")
@mock.patch("cloudinit.net.sys_dev_path")
@mock.patch("cloudinit.net.read_sys_net")
@@ -2023,7 +2499,7 @@ class TestNetplanNetRendering(CiTestCase):
def test_default_generation(self, mock_get_devicelist,
mock_read_sys_net,
mock_sys_dev_path,
- mock_clean_default):
+ mock_clean_default, m_get_cmdline):
tmp_dir = self.tmp_dir()
_setup_test(tmp_dir, mock_get_devicelist,
mock_read_sys_net, mock_sys_dev_path)
@@ -2038,7 +2514,7 @@ class TestNetplanNetRendering(CiTestCase):
render_target = 'netplan.yaml'
renderer = netplan.Renderer(
{'netplan_path': render_target, 'postcmds': False})
- renderer.render_network_state(ns, render_dir)
+ renderer.render_network_state(ns, target=render_dir)
self.assertTrue(os.path.exists(os.path.join(render_dir,
render_target)))
@@ -2143,7 +2619,7 @@ class TestNetplanPostcommands(CiTestCase):
render_target = 'netplan.yaml'
renderer = netplan.Renderer(
{'netplan_path': render_target, 'postcmds': True})
- renderer.render_network_state(ns, render_dir)
+ renderer.render_network_state(ns, target=render_dir)
mock_netplan_generate.assert_called_with(run=True)
mock_net_setup_link.assert_called_with(run=True)
@@ -2168,7 +2644,7 @@ class TestNetplanPostcommands(CiTestCase):
'/sys/class/net/lo'], capture=True),
]
with mock.patch.object(os.path, 'islink', return_value=True):
- renderer.render_network_state(ns, render_dir)
+ renderer.render_network_state(ns, target=render_dir)
mock_subp.assert_has_calls(expected)
@@ -2363,7 +2839,7 @@ class TestNetplanRoundTrip(CiTestCase):
renderer = netplan.Renderer(
config={'netplan_path': netplan_path})
- renderer.render_network_state(ns, target)
+ renderer.render_network_state(ns, target=target)
return dir2dict(target)
def testsimple_render_bond_netplan(self):
@@ -2453,7 +2929,7 @@ class TestEniRoundTrip(CiTestCase):
renderer = eni.Renderer(
config={'eni_path': eni_path, 'netrules_path': netrules_path})
- renderer.render_network_state(ns, dir)
+ renderer.render_network_state(ns, target=dir)
return dir2dict(dir)
def testsimple_convert_and_render(self):
diff --git a/tests/unittests/test_util.py b/tests/unittests/test_util.py
index 7a203ce..5a14479 100644
--- a/tests/unittests/test_util.py
+++ b/tests/unittests/test_util.py
@@ -24,6 +24,7 @@ except ImportError:
BASH = util.which('bash')
+BOGUS_COMMAND = 'this-is-not-expected-to-be-a-program-name'
class FakeSelinux(object):
@@ -742,6 +743,8 @@ class TestReadSeeded(helpers.TestCase):
class TestSubp(helpers.CiTestCase):
with_logs = True
+ allowed_subp = [BASH, 'cat', helpers.CiTestCase.SUBP_SHELL_TRUE,
+ BOGUS_COMMAND, sys.executable]
stdin2err = [BASH, '-c', 'cat >&2']
stdin2out = ['cat']
@@ -749,7 +752,6 @@ class TestSubp(helpers.CiTestCase):
utf8_valid = b'start \xc3\xa9 end'
utf8_valid_2 = b'd\xc3\xa9j\xc8\xa7'
printenv = [BASH, '-c', 'for n in "$@"; do echo "$n=${!n}"; done', '--']
- bogus_command = 'this-is-not-expected-to-be-a-program-name'
def printf_cmd(self, *args):
# bash's printf supports \xaa. So does /usr/bin/printf
@@ -848,9 +850,10 @@ class TestSubp(helpers.CiTestCase):
util.write_file(noshebang, 'true\n')
os.chmod(noshebang, os.stat(noshebang).st_mode | stat.S_IEXEC)
- self.assertRaisesRegex(util.ProcessExecutionError,
- r'Missing #! in script\?',
- util.subp, (noshebang,))
+ with self.allow_subp([noshebang]):
+ self.assertRaisesRegex(util.ProcessExecutionError,
+ r'Missing #! in script\?',
+ util.subp, (noshebang,))
def test_subp_combined_stderr_stdout(self):
"""Providing combine_capture as True redirects stderr to stdout."""
@@ -868,14 +871,14 @@ class TestSubp(helpers.CiTestCase):
def test_exception_has_out_err_are_bytes_if_decode_false(self):
"""Raised exc should have stderr, stdout as bytes if no decode."""
with self.assertRaises(util.ProcessExecutionError) as cm:
- util.subp([self.bogus_command], decode=False)
+ util.subp([BOGUS_COMMAND], decode=False)
self.assertTrue(isinstance(cm.exception.stdout, bytes))
self.assertTrue(isinstance(cm.exception.stderr, bytes))
def test_exception_has_out_err_are_bytes_if_decode_true(self):
"""Raised exc should have stderr, stdout as string if no decode."""
with self.assertRaises(util.ProcessExecutionError) as cm:
- util.subp([self.bogus_command], decode=True)
+ util.subp([BOGUS_COMMAND], decode=True)
self.assertTrue(isinstance(cm.exception.stdout, six.string_types))
self.assertTrue(isinstance(cm.exception.stderr, six.string_types))
@@ -925,10 +928,10 @@ class TestSubp(helpers.CiTestCase):
logs.append(log)
with self.assertRaises(util.ProcessExecutionError):
- util.subp([self.bogus_command], status_cb=status_cb)
+ util.subp([BOGUS_COMMAND], status_cb=status_cb)
expected = [
- 'Begin run command: {cmd}\n'.format(cmd=self.bogus_command),
+ 'Begin run command: {cmd}\n'.format(cmd=BOGUS_COMMAND),
'ERROR: End run command: invalid command provided\n']
self.assertEqual(expected, logs)
@@ -940,13 +943,13 @@ class TestSubp(helpers.CiTestCase):
logs.append(log)
with self.assertRaises(util.ProcessExecutionError):
- util.subp(['ls', '/I/dont/exist'], status_cb=status_cb)
- util.subp(['ls'], status_cb=status_cb)
+ util.subp([BASH, '-c', 'exit 2'], status_cb=status_cb)
+ util.subp([BASH, '-c', 'exit 0'], status_cb=status_cb)
expected = [
- 'Begin run command: ls /I/dont/exist\n',
+ 'Begin run command: %s -c exit 2\n' % BASH,
'ERROR: End run command: exit(2)\n',
- 'Begin run command: ls\n',
+ 'Begin run command: %s -c exit 0\n' % BASH,
'End run command: exit(0)\n']
self.assertEqual(expected, logs)
Follow ups
