diff options
Diffstat (limited to 'cloudinit/tests/test_stages.py')
| -rw-r--r-- | cloudinit/tests/test_stages.py | 406 |
1 files changed, 0 insertions, 406 deletions
diff --git a/cloudinit/tests/test_stages.py b/cloudinit/tests/test_stages.py deleted file mode 100644 index d2d1b37f..00000000 --- a/cloudinit/tests/test_stages.py +++ /dev/null @@ -1,406 +0,0 @@ -# This file is part of cloud-init. See LICENSE file for license information. - -"""Tests related to cloudinit.stages module.""" - -import os -import stat - -import pytest - -from cloudinit import stages -from cloudinit import sources -from cloudinit.sources import NetworkConfigSource - -from cloudinit.event import EventType -from cloudinit.util import write_file - -from cloudinit.tests.helpers import CiTestCase, mock - -TEST_INSTANCE_ID = 'i-testing' - - -class FakeDataSource(sources.DataSource): - - def __init__(self, paths=None, userdata=None, vendordata=None, - network_config=''): - super(FakeDataSource, self).__init__({}, None, paths=paths) - self.metadata = {'instance-id': TEST_INSTANCE_ID} - self.userdata_raw = userdata - self.vendordata_raw = vendordata - self._network_config = None - if network_config: # Permit for None value to setup attribute - self._network_config = network_config - - @property - def network_config(self): - return self._network_config - - def _get_data(self): - return True - - -class TestInit(CiTestCase): - with_logs = True - allowed_subp = False - - def setUp(self): - super(TestInit, self).setUp() - self.tmpdir = self.tmp_dir() - self.init = stages.Init() - # Setup fake Paths for Init to reference - self.init._cfg = {'system_info': { - 'distro': 'ubuntu', 'paths': {'cloud_dir': self.tmpdir, - 'run_dir': self.tmpdir}}} - self.init.datasource = FakeDataSource(paths=self.init.paths) - - def test_wb__find_networking_config_disabled(self): - """find_networking_config returns no config when disabled.""" - disable_file = os.path.join( - self.init.paths.get_cpath('data'), 'upgraded-network') - write_file(disable_file, '') - self.assertEqual( - (None, disable_file), - self.init._find_networking_config()) - - @mock.patch('cloudinit.stages.cmdline.read_initramfs_config') - @mock.patch('cloudinit.stages.cmdline.read_kernel_cmdline_config') - def test_wb__find_networking_config_disabled_by_kernel( - self, m_cmdline, m_initramfs): - """find_networking_config returns when disabled by kernel cmdline.""" - m_cmdline.return_value = {'config': 'disabled'} - m_initramfs.return_value = {'config': ['fake_initrd']} - self.assertEqual( - (None, NetworkConfigSource.cmdline), - self.init._find_networking_config()) - self.assertEqual('DEBUG: network config disabled by cmdline\n', - self.logs.getvalue()) - - @mock.patch('cloudinit.stages.cmdline.read_initramfs_config') - @mock.patch('cloudinit.stages.cmdline.read_kernel_cmdline_config') - def test_wb__find_networking_config_disabled_by_initrd( - self, m_cmdline, m_initramfs): - """find_networking_config returns when disabled by kernel cmdline.""" - m_cmdline.return_value = {} - m_initramfs.return_value = {'config': 'disabled'} - self.assertEqual( - (None, NetworkConfigSource.initramfs), - self.init._find_networking_config()) - self.assertEqual('DEBUG: network config disabled by initramfs\n', - self.logs.getvalue()) - - @mock.patch('cloudinit.stages.cmdline.read_initramfs_config') - @mock.patch('cloudinit.stages.cmdline.read_kernel_cmdline_config') - def test_wb__find_networking_config_disabled_by_datasrc( - self, m_cmdline, m_initramfs): - """find_networking_config returns when disabled by datasource cfg.""" - m_cmdline.return_value = {} # Kernel doesn't disable networking - m_initramfs.return_value = {} # initramfs doesn't disable networking - self.init._cfg = {'system_info': {'paths': {'cloud_dir': self.tmpdir}}, - 'network': {}} # system config doesn't disable - - self.init.datasource = FakeDataSource( - network_config={'config': 'disabled'}) - self.assertEqual( - (None, NetworkConfigSource.ds), - self.init._find_networking_config()) - self.assertEqual('DEBUG: network config disabled by ds\n', - self.logs.getvalue()) - - @mock.patch('cloudinit.stages.cmdline.read_initramfs_config') - @mock.patch('cloudinit.stages.cmdline.read_kernel_cmdline_config') - def test_wb__find_networking_config_disabled_by_sysconfig( - self, m_cmdline, m_initramfs): - """find_networking_config returns when disabled by system config.""" - m_cmdline.return_value = {} # Kernel doesn't disable networking - m_initramfs.return_value = {} # initramfs doesn't disable networking - self.init._cfg = {'system_info': {'paths': {'cloud_dir': self.tmpdir}}, - 'network': {'config': 'disabled'}} - self.assertEqual( - (None, NetworkConfigSource.system_cfg), - self.init._find_networking_config()) - self.assertEqual('DEBUG: network config disabled by system_cfg\n', - self.logs.getvalue()) - - @mock.patch('cloudinit.stages.cmdline.read_initramfs_config') - @mock.patch('cloudinit.stages.cmdline.read_kernel_cmdline_config') - def test__find_networking_config_uses_datasrc_order( - self, m_cmdline, m_initramfs): - """find_networking_config should check sources in DS defined order""" - # cmdline and initramfs, which would normally be preferred over other - # sources, disable networking; in this case, though, the DS moves them - # later so its own config is preferred - m_cmdline.return_value = {'config': 'disabled'} - m_initramfs.return_value = {'config': 'disabled'} - - ds_net_cfg = {'config': {'needle': True}} - self.init.datasource = FakeDataSource(network_config=ds_net_cfg) - self.init.datasource.network_config_sources = [ - NetworkConfigSource.ds, NetworkConfigSource.system_cfg, - NetworkConfigSource.cmdline, NetworkConfigSource.initramfs] - - self.assertEqual( - (ds_net_cfg, NetworkConfigSource.ds), - self.init._find_networking_config()) - - @mock.patch('cloudinit.stages.cmdline.read_initramfs_config') - @mock.patch('cloudinit.stages.cmdline.read_kernel_cmdline_config') - def test__find_networking_config_warns_if_datasrc_uses_invalid_src( - self, m_cmdline, m_initramfs): - """find_networking_config should check sources in DS defined order""" - ds_net_cfg = {'config': {'needle': True}} - self.init.datasource = FakeDataSource(network_config=ds_net_cfg) - self.init.datasource.network_config_sources = [ - 'invalid_src', NetworkConfigSource.ds] - - self.assertEqual( - (ds_net_cfg, NetworkConfigSource.ds), - self.init._find_networking_config()) - self.assertIn('WARNING: data source specifies an invalid network' - ' cfg_source: invalid_src', - self.logs.getvalue()) - - @mock.patch('cloudinit.stages.cmdline.read_initramfs_config') - @mock.patch('cloudinit.stages.cmdline.read_kernel_cmdline_config') - def test__find_networking_config_warns_if_datasrc_uses_unavailable_src( - self, m_cmdline, m_initramfs): - """find_networking_config should check sources in DS defined order""" - ds_net_cfg = {'config': {'needle': True}} - self.init.datasource = FakeDataSource(network_config=ds_net_cfg) - self.init.datasource.network_config_sources = [ - NetworkConfigSource.fallback, NetworkConfigSource.ds] - - self.assertEqual( - (ds_net_cfg, NetworkConfigSource.ds), - self.init._find_networking_config()) - self.assertIn('WARNING: data source specifies an unavailable network' - ' cfg_source: fallback', - self.logs.getvalue()) - - @mock.patch('cloudinit.stages.cmdline.read_initramfs_config') - @mock.patch('cloudinit.stages.cmdline.read_kernel_cmdline_config') - def test_wb__find_networking_config_returns_kernel( - self, m_cmdline, m_initramfs): - """find_networking_config returns kernel cmdline config if present.""" - expected_cfg = {'config': ['fakekernel']} - m_cmdline.return_value = expected_cfg - m_initramfs.return_value = {'config': ['fake_initrd']} - self.init._cfg = {'system_info': {'paths': {'cloud_dir': self.tmpdir}}, - 'network': {'config': ['fakesys_config']}} - self.init.datasource = FakeDataSource( - network_config={'config': ['fakedatasource']}) - self.assertEqual( - (expected_cfg, NetworkConfigSource.cmdline), - self.init._find_networking_config()) - - @mock.patch('cloudinit.stages.cmdline.read_initramfs_config') - @mock.patch('cloudinit.stages.cmdline.read_kernel_cmdline_config') - def test_wb__find_networking_config_returns_initramfs( - self, m_cmdline, m_initramfs): - """find_networking_config returns kernel cmdline config if present.""" - expected_cfg = {'config': ['fake_initrd']} - m_cmdline.return_value = {} - m_initramfs.return_value = expected_cfg - self.init._cfg = {'system_info': {'paths': {'cloud_dir': self.tmpdir}}, - 'network': {'config': ['fakesys_config']}} - self.init.datasource = FakeDataSource( - network_config={'config': ['fakedatasource']}) - self.assertEqual( - (expected_cfg, NetworkConfigSource.initramfs), - self.init._find_networking_config()) - - @mock.patch('cloudinit.stages.cmdline.read_initramfs_config') - @mock.patch('cloudinit.stages.cmdline.read_kernel_cmdline_config') - def test_wb__find_networking_config_returns_system_cfg( - self, m_cmdline, m_initramfs): - """find_networking_config returns system config when present.""" - m_cmdline.return_value = {} # No kernel network config - m_initramfs.return_value = {} # no initramfs network config - expected_cfg = {'config': ['fakesys_config']} - self.init._cfg = {'system_info': {'paths': {'cloud_dir': self.tmpdir}}, - 'network': expected_cfg} - self.init.datasource = FakeDataSource( - network_config={'config': ['fakedatasource']}) - self.assertEqual( - (expected_cfg, NetworkConfigSource.system_cfg), - self.init._find_networking_config()) - - @mock.patch('cloudinit.stages.cmdline.read_initramfs_config') - @mock.patch('cloudinit.stages.cmdline.read_kernel_cmdline_config') - def test_wb__find_networking_config_returns_datasrc_cfg( - self, m_cmdline, m_initramfs): - """find_networking_config returns datasource net config if present.""" - m_cmdline.return_value = {} # No kernel network config - m_initramfs.return_value = {} # no initramfs network config - # No system config for network in setUp - expected_cfg = {'config': ['fakedatasource']} - self.init.datasource = FakeDataSource(network_config=expected_cfg) - self.assertEqual( - (expected_cfg, NetworkConfigSource.ds), - self.init._find_networking_config()) - - @mock.patch('cloudinit.stages.cmdline.read_initramfs_config') - @mock.patch('cloudinit.stages.cmdline.read_kernel_cmdline_config') - def test_wb__find_networking_config_returns_fallback( - self, m_cmdline, m_initramfs): - """find_networking_config returns fallback config if not defined.""" - m_cmdline.return_value = {} # Kernel doesn't disable networking - m_initramfs.return_value = {} # no initramfs network config - # Neither datasource nor system_info disable or provide network - - fake_cfg = {'config': [{'type': 'physical', 'name': 'eth9'}], - 'version': 1} - - def fake_generate_fallback(): - return fake_cfg - - # Monkey patch distro which gets cached on self.init - distro = self.init.distro - distro.generate_fallback_config = fake_generate_fallback - self.assertEqual( - (fake_cfg, NetworkConfigSource.fallback), - self.init._find_networking_config()) - self.assertNotIn('network config disabled', self.logs.getvalue()) - - def test_apply_network_config_disabled(self): - """Log when network is disabled by upgraded-network.""" - disable_file = os.path.join( - self.init.paths.get_cpath('data'), 'upgraded-network') - - def fake_network_config(): - return (None, disable_file) - - self.init._find_networking_config = fake_network_config - - self.init.apply_network_config(True) - self.assertIn( - 'INFO: network config is disabled by %s' % disable_file, - self.logs.getvalue()) - - @mock.patch('cloudinit.net.get_interfaces_by_mac') - @mock.patch('cloudinit.distros.ubuntu.Distro') - def test_apply_network_on_new_instance(self, m_ubuntu, m_macs): - """Call distro apply_network_config methods on is_new_instance.""" - net_cfg = { - 'version': 1, 'config': [ - {'subnets': [{'type': 'dhcp'}], 'type': 'physical', - 'name': 'eth9', 'mac_address': '42:42:42:42:42:42'}]} - - def fake_network_config(): - return net_cfg, NetworkConfigSource.fallback - - m_macs.return_value = {'42:42:42:42:42:42': 'eth9'} - - self.init._find_networking_config = fake_network_config - self.init.apply_network_config(True) - self.init.distro.apply_network_config_names.assert_called_with(net_cfg) - self.init.distro.apply_network_config.assert_called_with( - net_cfg, bring_up=True) - - @mock.patch('cloudinit.distros.ubuntu.Distro') - def test_apply_network_on_same_instance_id(self, m_ubuntu): - """Only call distro.apply_network_config_names on same instance id.""" - old_instance_id = os.path.join( - self.init.paths.get_cpath('data'), 'instance-id') - write_file(old_instance_id, TEST_INSTANCE_ID) - net_cfg = { - 'version': 1, 'config': [ - {'subnets': [{'type': 'dhcp'}], 'type': 'physical', - 'name': 'eth9', 'mac_address': '42:42:42:42:42:42'}]} - - def fake_network_config(): - return net_cfg, NetworkConfigSource.fallback - - self.init._find_networking_config = fake_network_config - self.init.apply_network_config(True) - self.init.distro.apply_network_config_names.assert_called_with(net_cfg) - self.init.distro.apply_network_config.assert_not_called() - self.assertIn( - 'No network config applied. Neither a new instance' - " nor datasource network update on '%s' event" % EventType.BOOT, - self.logs.getvalue()) - - @mock.patch('cloudinit.net.get_interfaces_by_mac') - @mock.patch('cloudinit.distros.ubuntu.Distro') - def test_apply_network_on_datasource_allowed_event(self, m_ubuntu, m_macs): - """Apply network if datasource.update_metadata permits BOOT event.""" - old_instance_id = os.path.join( - self.init.paths.get_cpath('data'), 'instance-id') - write_file(old_instance_id, TEST_INSTANCE_ID) - net_cfg = { - 'version': 1, 'config': [ - {'subnets': [{'type': 'dhcp'}], 'type': 'physical', - 'name': 'eth9', 'mac_address': '42:42:42:42:42:42'}]} - - def fake_network_config(): - return net_cfg, NetworkConfigSource.fallback - - m_macs.return_value = {'42:42:42:42:42:42': 'eth9'} - - self.init._find_networking_config = fake_network_config - self.init.datasource = FakeDataSource(paths=self.init.paths) - self.init.datasource.update_events = {'network': [EventType.BOOT]} - self.init.apply_network_config(True) - self.init.distro.apply_network_config_names.assert_called_with(net_cfg) - self.init.distro.apply_network_config.assert_called_with( - net_cfg, bring_up=True) - - -class TestInit_InitializeFilesystem: - """Tests for cloudinit.stages.Init._initialize_filesystem. - - TODO: Expand these tests to cover all of _initialize_filesystem's behavior. - """ - - @pytest.yield_fixture - def init(self, paths): - """A fixture which yields a stages.Init instance with paths and cfg set - - As it is replaced with a mock, consumers of this fixture can set - `init.cfg` if the default empty dict configuration is not appropriate. - """ - with mock.patch( - "cloudinit.stages.Init.cfg", mock.PropertyMock(return_value={}) - ): - with mock.patch("cloudinit.stages.util.ensure_dirs"): - init = stages.Init() - init._paths = paths - yield init - - @mock.patch("cloudinit.stages.util.ensure_file") - def test_ensure_file_not_called_if_no_log_file_configured( - self, m_ensure_file, init - ): - """If no log file is configured, we should not ensure its existence.""" - init.cfg = {} - - init._initialize_filesystem() - - assert 0 == m_ensure_file.call_count - - def test_log_files_existence_is_ensured_if_configured(self, init, tmpdir): - """If a log file is configured, we should ensure its existence.""" - log_file = tmpdir.join("cloud-init.log") - init.cfg = {"def_log_file": str(log_file)} - - init._initialize_filesystem() - - assert log_file.exists - - def test_existing_file_permissions_are_not_modified(self, init, tmpdir): - """If the log file already exists, we should not modify its permissions - - See https://bugs.launchpad.net/cloud-init/+bug/1900837. - """ - # Use a mode that will never be made the default so this test will - # always be valid - mode = 0o606 - log_file = tmpdir.join("cloud-init.log") - log_file.ensure() - log_file.chmod(mode) - init.cfg = {"def_log_file": str(log_file)} - - init._initialize_filesystem() - - assert mode == stat.S_IMODE(log_file.stat().mode) - -# vi: ts=4 expandtab |