summaryrefslogtreecommitdiff
path: root/tests
diff options
context:
space:
mode:
Diffstat (limited to 'tests')
-rw-r--r--tests/unittests/test_datasource/test_altcloud.py44
-rw-r--r--tests/unittests/test_datasource/test_cloudsigma.py3
-rw-r--r--tests/unittests/test_datasource/test_configdrive.py9
-rw-r--r--tests/unittests/test_datasource/test_nocloud.py2
-rw-r--r--tests/unittests/test_datasource/test_opennebula.py1
-rw-r--r--tests/unittests/test_datasource/test_ovf.py8
-rw-r--r--tests/unittests/test_datasource/test_smartos.py94
-rw-r--r--tests/unittests/test_distros/test_netconfig.py971
-rw-r--r--tests/unittests/test_ds_identify.py1
-rw-r--r--tests/unittests/test_handler/test_handler_apt_source_v3.py11
-rw-r--r--tests/unittests/test_handler/test_handler_bootcmd.py10
-rw-r--r--tests/unittests/test_handler/test_handler_chef.py18
-rw-r--r--tests/unittests/test_handler/test_handler_resizefs.py8
-rw-r--r--tests/unittests/test_handler/test_schema.py12
-rw-r--r--tests/unittests/test_net.py560
-rw-r--r--tests/unittests/test_util.py27
16 files changed, 1092 insertions, 687 deletions
diff --git a/tests/unittests/test_datasource/test_altcloud.py b/tests/unittests/test_datasource/test_altcloud.py
index 3253f3ad..ff35904e 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 f6a59b6b..380ad1b5 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 68400f22..b0abfc5f 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 cdbd1e1a..21931eb7 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 36b4d771..61591017 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 fc4eb36e..9d52eb99 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 dca0b3d4..46d67b94 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 7765e408..740fb76c 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 e08e7908..46778e95 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 7a64c230..a81c67c7 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 b1375269..a76760fa 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 f4bbd66d..b16532ea 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 f92175fd..feca56c2 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 fb266faf..1bad07f6 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 58e5ea14..f3165da9 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 7a203ce2..5a14479a 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)
View Lorry Controller Status