diff options
Diffstat (limited to 'tests/unittests/test_handler')
7 files changed, 366 insertions, 243 deletions
diff --git a/tests/unittests/test_handler/test_handler_apt_configure_sources_list_v3.py b/tests/unittests/test_handler/test_handler_apt_configure_sources_list_v3.py index 24e45233..1ca915b4 100644 --- a/tests/unittests/test_handler/test_handler_apt_configure_sources_list_v3.py +++ b/tests/unittests/test_handler/test_handler_apt_configure_sources_list_v3.py @@ -121,39 +121,82 @@ class TestAptSourceConfigSourceList(t_help.FilesystemMockingTestCase): myds.metadata.update(metadata) return cloud.Cloud(myds, paths, {}, mydist, None) - def _apt_source_list(self, cfg, expected, distro): - "_apt_source_list - Test rendering from template (generic)" - + def _apt_source_list(self, distro, cfg, cfg_on_empty=False): + """_apt_source_list - Test rendering from template (generic)""" # entry at top level now, wrap in 'apt' key cfg = {'apt': cfg} mycloud = self._get_cloud(distro) - with mock.patch.object(util, 'write_file') as mockwf: + + with mock.patch.object(util, 'write_file') as mock_writefile: with mock.patch.object(util, 'load_file', - return_value=MOCKED_APT_SRC_LIST) as mocklf: + return_value=MOCKED_APT_SRC_LIST + ) as mock_loadfile: with mock.patch.object(os.path, 'isfile', - return_value=True) as mockisfile: - with mock.patch.object(util, 'rename'): - cc_apt_configure.handle("test", cfg, mycloud, - LOG, None) - - # check if it would have loaded the distro template - mockisfile.assert_any_call( - ('/etc/cloud/templates/sources.list.%s.tmpl' % distro)) - mocklf.assert_any_call( - ('/etc/cloud/templates/sources.list.%s.tmpl' % distro)) - # check expected content in result - mockwf.assert_called_once_with('/etc/apt/sources.list', expected, - mode=0o644) + return_value=True) as mock_isfile: + cfg_func = ('cloudinit.config.cc_apt_configure.' + + '_should_configure_on_empty_apt') + with mock.patch(cfg_func, + return_value=(cfg_on_empty, "test") + ) as mock_shouldcfg: + cc_apt_configure.handle("test", cfg, mycloud, LOG, + None) + + return mock_writefile, mock_loadfile, mock_isfile, mock_shouldcfg def test_apt_v3_source_list_debian(self): """test_apt_v3_source_list_debian - without custom sources or parms""" cfg = {} - self._apt_source_list(cfg, EXPECTED_BASE_CONTENT, 'debian') + distro = 'debian' + expected = EXPECTED_BASE_CONTENT + + mock_writefile, mock_load_file, mock_isfile, mock_shouldcfg = ( + self._apt_source_list(distro, cfg, cfg_on_empty=True)) + + template = '/etc/cloud/templates/sources.list.%s.tmpl' % distro + mock_writefile.assert_called_once_with('/etc/apt/sources.list', + expected, mode=0o644) + mock_load_file.assert_called_with(template) + mock_isfile.assert_any_call(template) + self.assertEqual(1, mock_shouldcfg.call_count) def test_apt_v3_source_list_ubuntu(self): """test_apt_v3_source_list_ubuntu - without custom sources or parms""" cfg = {} - self._apt_source_list(cfg, EXPECTED_BASE_CONTENT, 'ubuntu') + distro = 'ubuntu' + expected = EXPECTED_BASE_CONTENT + + mock_writefile, mock_load_file, mock_isfile, mock_shouldcfg = ( + self._apt_source_list(distro, cfg, cfg_on_empty=True)) + + template = '/etc/cloud/templates/sources.list.%s.tmpl' % distro + mock_writefile.assert_called_once_with('/etc/apt/sources.list', + expected, mode=0o644) + mock_load_file.assert_called_with(template) + mock_isfile.assert_any_call(template) + self.assertEqual(1, mock_shouldcfg.call_count) + + def test_apt_v3_source_list_ubuntu_snappy(self): + """test_apt_v3_source_list_ubuntu_snappy - without custom sources or + parms""" + cfg = {'apt': {}} + mycloud = self._get_cloud('ubuntu') + + with mock.patch.object(util, 'write_file') as mock_writefile: + with mock.patch.object(util, 'system_is_snappy', + return_value=True) as mock_issnappy: + cc_apt_configure.handle("test", cfg, mycloud, LOG, None) + + self.assertEqual(0, mock_writefile.call_count) + self.assertEqual(1, mock_issnappy.call_count) + + def test_apt_v3_source_list_centos(self): + """test_apt_v3_source_list_centos - without custom sources or parms""" + cfg = {} + distro = 'rhel' + + mock_writefile, _, _, _ = self._apt_source_list(distro, cfg) + + self.assertEqual(0, mock_writefile.call_count) def test_apt_v3_source_list_psm(self): """test_apt_v3_source_list_psm - Test specifying prim+sec mirrors""" @@ -164,8 +207,17 @@ class TestAptSourceConfigSourceList(t_help.FilesystemMockingTestCase): 'uri': pm}], 'security': [{'arches': ["default"], 'uri': sm}]} + distro = 'ubuntu' + expected = EXPECTED_PRIMSEC_CONTENT + + mock_writefile, mock_load_file, mock_isfile, _ = ( + self._apt_source_list(distro, cfg, cfg_on_empty=True)) - self._apt_source_list(cfg, EXPECTED_PRIMSEC_CONTENT, 'ubuntu') + template = '/etc/cloud/templates/sources.list.%s.tmpl' % distro + mock_writefile.assert_called_once_with('/etc/apt/sources.list', + expected, mode=0o644) + mock_load_file.assert_called_with(template) + mock_isfile.assert_any_call(template) def test_apt_v3_srcl_custom(self): """test_apt_v3_srcl_custom - Test rendering a custom source template""" diff --git a/tests/unittests/test_handler/test_handler_disk_setup.py b/tests/unittests/test_handler/test_handler_disk_setup.py index 7ff39225..916a0d7a 100644 --- a/tests/unittests/test_handler/test_handler_disk_setup.py +++ b/tests/unittests/test_handler/test_handler_disk_setup.py @@ -17,6 +17,10 @@ class TestIsDiskUsed(TestCase): self.check_fs = self.patches.enter_context( mock.patch('{0}.check_fs'.format(mod_name))) + def tearDown(self): + super(TestIsDiskUsed, self).tearDown() + self.patches.close() + def test_multiple_child_nodes_returns_true(self): self.enumerate_disk.return_value = (mock.MagicMock() for _ in range(2)) self.check_fs.return_value = (mock.MagicMock(), None, mock.MagicMock()) @@ -62,7 +66,7 @@ class TestGetMbrHddSize(TestCase): size_in_sectors = size_in_bytes / sector_size self._configure_subp_mock(size_in_bytes, sector_size) self.assertEqual(size_in_sectors, - cc_disk_setup.get_mbr_hdd_size('/dev/sda1')) + cc_disk_setup.get_hdd_size('/dev/sda1')) def test_size_for_512_byte_sectors(self): self._test_for_sector_size(512) @@ -147,4 +151,75 @@ class TestUpdateFsSetupDevices(TestCase): 'filesystem': 'xfs' }, fs_setup) + def test_dotted_devname_populates_partition(self): + fs_setup = { + 'device': 'ephemeral0.1', + 'label': 'test2', + 'filesystem': 'xfs' + } + cc_disk_setup.update_fs_setup_devices([fs_setup], + lambda device: device) + self.assertEqual({ + '_origname': 'ephemeral0.1', + 'device': 'ephemeral0', + 'partition': '1', + 'label': 'test2', + 'filesystem': 'xfs' + }, fs_setup) + + +@mock.patch('cloudinit.config.cc_disk_setup.assert_and_settle_device', + return_value=None) +@mock.patch('cloudinit.config.cc_disk_setup.find_device_node', + return_value=('/dev/xdb1', False)) +@mock.patch('cloudinit.config.cc_disk_setup.device_type', return_value=None) +@mock.patch('cloudinit.config.cc_disk_setup.util.subp', return_value=('', '')) +class TestMkfsCommandHandling(TestCase): + + def test_with_cmd(self, subp, *args): + """mkfs honors cmd and logs warnings when extra_opts or overwrite are + provided.""" + with self.assertLogs( + 'cloudinit.config.cc_disk_setup') as logs: + cc_disk_setup.mkfs({ + 'cmd': 'mkfs -t %(filesystem)s -L %(label)s %(device)s', + 'filesystem': 'ext4', + 'device': '/dev/xdb1', + 'label': 'with_cmd', + 'extra_opts': ['should', 'generate', 'warning'], + 'overwrite': 'should generate warning too' + }) + + self.assertIn( + 'WARNING:cloudinit.config.cc_disk_setup:fs_setup:extra_opts ' + + 'ignored because cmd was specified: mkfs -t ext4 -L with_cmd ' + + '/dev/xdb1', + logs.output) + self.assertIn( + 'WARNING:cloudinit.config.cc_disk_setup:fs_setup:overwrite ' + + 'ignored because cmd was specified: mkfs -t ext4 -L with_cmd ' + + '/dev/xdb1', + logs.output) + + subp.assert_called_once_with( + 'mkfs -t ext4 -L with_cmd /dev/xdb1', shell=True) + + @mock.patch('cloudinit.config.cc_disk_setup.util.which') + def test_overwrite_and_extra_opts_without_cmd(self, m_which, subp, *args): + """mkfs observes extra_opts and overwrite settings when cmd is not + present.""" + m_which.side_effect = lambda p: {'mkfs.ext4': '/sbin/mkfs.ext4'}[p] + cc_disk_setup.mkfs({ + 'filesystem': 'ext4', + 'device': '/dev/xdb1', + 'label': 'without_cmd', + 'extra_opts': ['are', 'added'], + 'overwrite': True + }) + + subp.assert_called_once_with( + ['/sbin/mkfs.ext4', '/dev/xdb1', + '-L', 'without_cmd', '-F', 'are', 'added'], + shell=False) + # vi: ts=4 expandtab diff --git a/tests/unittests/test_handler/test_handler_ntp.py b/tests/unittests/test_handler/test_handler_ntp.py index ec600077..bc4277b7 100644 --- a/tests/unittests/test_handler/test_handler_ntp.py +++ b/tests/unittests/test_handler/test_handler_ntp.py @@ -2,277 +2,214 @@ from cloudinit.config import cc_ntp from cloudinit.sources import DataSourceNone -from cloudinit import templater from cloudinit import (distros, helpers, cloud, util) from ..helpers import FilesystemMockingTestCase, mock -import logging + import os +from os.path import dirname import shutil -import tempfile - -LOG = logging.getLogger(__name__) -NTP_TEMPLATE = """ +NTP_TEMPLATE = b"""\ ## template: jinja - -{% if pools %}# pools -{% endif %} -{% for pool in pools -%} -pool {{pool}} iburst -{% endfor %} -{%- if servers %}# servers -{% endif %} -{% for server in servers -%} -server {{server}} iburst -{% endfor %} - -""" - - -NTP_EXPECTED_UBUNTU = """ -# pools -pool 0.mycompany.pool.ntp.org iburst -# servers -server 192.168.23.3 iburst - +servers {{servers}} +pools {{pools}} """ class TestNtp(FilesystemMockingTestCase): + with_logs = True + def setUp(self): super(TestNtp, self).setUp() self.subp = util.subp - self.new_root = tempfile.mkdtemp() - self.addCleanup(shutil.rmtree, self.new_root) + self.new_root = self.tmp_dir() - def _get_cloud(self, distro, metadata=None): + def _get_cloud(self, distro): self.patchUtils(self.new_root) - paths = helpers.Paths({}) + paths = helpers.Paths({'templates_dir': self.new_root}) cls = distros.fetch(distro) mydist = cls(distro, {}, paths) myds = DataSourceNone.DataSourceNone({}, mydist, paths) - if metadata: - myds.metadata.update(metadata) return cloud.Cloud(myds, paths, {}, mydist, None) @mock.patch("cloudinit.config.cc_ntp.util") def test_ntp_install(self, mock_util): - cc = self._get_cloud('ubuntu') - cc.distro = mock.MagicMock() - cc.distro.name = 'ubuntu' - mock_util.which.return_value = None + """ntp_install installs via install_func when check_exe is absent.""" + mock_util.which.return_value = None # check_exe not found. install_func = mock.MagicMock() - cc_ntp.install_ntp(install_func, packages=['ntpx'], check_exe='ntpdx') - self.assertTrue(install_func.called) mock_util.which.assert_called_with('ntpdx') - install_pkg = install_func.call_args_list[0][0][0] - self.assertEqual(sorted(install_pkg), ['ntpx']) + install_func.assert_called_once_with(['ntpx']) @mock.patch("cloudinit.config.cc_ntp.util") def test_ntp_install_not_needed(self, mock_util): - cc = self._get_cloud('ubuntu') - cc.distro = mock.MagicMock() - cc.distro.name = 'ubuntu' - mock_util.which.return_value = ["/usr/sbin/ntpd"] - cc_ntp.install_ntp(cc) - self.assertFalse(cc.distro.install_packages.called) + """ntp_install doesn't attempt install when check_exe is found.""" + mock_util.which.return_value = ["/usr/sbin/ntpd"] # check_exe found. + install_func = mock.MagicMock() + cc_ntp.install_ntp(install_func, packages=['ntp'], check_exe='ntpd') + install_func.assert_not_called() def test_ntp_rename_ntp_conf(self): - with mock.patch.object(os.path, 'exists', - return_value=True) as mockpath: - with mock.patch.object(util, 'rename') as mockrename: - cc_ntp.rename_ntp_conf() - - mockpath.assert_called_with('/etc/ntp.conf') - mockrename.assert_called_with('/etc/ntp.conf', '/etc/ntp.conf.dist') + """When NTP_CONF exists, rename_ntp moves it.""" + ntpconf = self.tmp_path("ntp.conf", self.new_root) + os.mknod(ntpconf) + with mock.patch("cloudinit.config.cc_ntp.NTP_CONF", ntpconf): + cc_ntp.rename_ntp_conf() + self.assertFalse(os.path.exists(ntpconf)) + self.assertTrue(os.path.exists("{0}.dist".format(ntpconf))) def test_ntp_rename_ntp_conf_skip_missing(self): - with mock.patch.object(os.path, 'exists', - return_value=False) as mockpath: - with mock.patch.object(util, 'rename') as mockrename: - cc_ntp.rename_ntp_conf() - - mockpath.assert_called_with('/etc/ntp.conf') - mockrename.assert_not_called() - - def ntp_conf_render(self, distro): - """ntp_conf_render - Test rendering of a ntp.conf from template for a given distro + """When NTP_CONF doesn't exist rename_ntp doesn't create a file.""" + ntpconf = self.tmp_path("ntp.conf", self.new_root) + self.assertFalse(os.path.exists(ntpconf)) + with mock.patch("cloudinit.config.cc_ntp.NTP_CONF", ntpconf): + cc_ntp.rename_ntp_conf() + self.assertFalse(os.path.exists("{0}.dist".format(ntpconf))) + self.assertFalse(os.path.exists(ntpconf)) + + def test_write_ntp_config_template_from_ntp_conf_tmpl_with_servers(self): + """write_ntp_config_template reads content from ntp.conf.tmpl. + + It reads ntp.conf.tmpl if present and renders the value from servers + key. When no pools key is defined, template is rendered using an empty + list for pools. """ - - cfg = {'ntp': {}} - mycloud = self._get_cloud(distro) - distro_names = cc_ntp.generate_server_names(distro) - - with mock.patch.object(templater, 'render_to_file') as mocktmpl: - with mock.patch.object(os.path, 'isfile', return_value=True): - with mock.patch.object(util, 'rename'): - cc_ntp.write_ntp_config_template(cfg, mycloud) - - mocktmpl.assert_called_once_with( - ('/etc/cloud/templates/ntp.conf.%s.tmpl' % distro), - '/etc/ntp.conf', - {'servers': [], 'pools': distro_names}) - - def test_ntp_conf_render_rhel(self): - """Test templater.render_to_file() for rhel""" - self.ntp_conf_render('rhel') - - def test_ntp_conf_render_debian(self): - """Test templater.render_to_file() for debian""" - self.ntp_conf_render('debian') - - def test_ntp_conf_render_fedora(self): - """Test templater.render_to_file() for fedora""" - self.ntp_conf_render('fedora') - - def test_ntp_conf_render_sles(self): - """Test templater.render_to_file() for sles""" - self.ntp_conf_render('sles') - - def test_ntp_conf_render_ubuntu(self): - """Test templater.render_to_file() for ubuntu""" - self.ntp_conf_render('ubuntu') - - def test_ntp_conf_servers_no_pools(self): distro = 'ubuntu' - pools = [] - servers = ['192.168.2.1'] cfg = { - 'ntp': { - 'pools': pools, - 'servers': servers, - } + 'servers': ['192.168.2.1', '192.168.2.2'] } mycloud = self._get_cloud(distro) - - with mock.patch.object(templater, 'render_to_file') as mocktmpl: - with mock.patch.object(os.path, 'isfile', return_value=True): - with mock.patch.object(util, 'rename'): - cc_ntp.write_ntp_config_template(cfg.get('ntp'), mycloud) - - mocktmpl.assert_called_once_with( - ('/etc/cloud/templates/ntp.conf.%s.tmpl' % distro), - '/etc/ntp.conf', - {'servers': servers, 'pools': pools}) - - def test_ntp_conf_custom_pools_no_server(self): + ntp_conf = self.tmp_path("ntp.conf", self.new_root) # Doesn't exist + # Create ntp.conf.tmpl + with open('{0}.tmpl'.format(ntp_conf), 'wb') as stream: + stream.write(NTP_TEMPLATE) + with mock.patch('cloudinit.config.cc_ntp.NTP_CONF', ntp_conf): + cc_ntp.write_ntp_config_template(cfg, mycloud) + content = util.read_file_or_url('file://' + ntp_conf).contents + self.assertEqual( + "servers ['192.168.2.1', '192.168.2.2']\npools []\n", + content.decode()) + + def test_write_ntp_config_template_uses_ntp_conf_distro_no_servers(self): + """write_ntp_config_template reads content from ntp.conf.distro.tmpl. + + It reads ntp.conf.<distro>.tmpl before attempting ntp.conf.tmpl. It + renders the value from the keys servers and pools. When no + servers value is present, template is rendered using an empty list. + """ distro = 'ubuntu' - pools = ['0.mycompany.pool.ntp.org'] - servers = [] cfg = { - 'ntp': { - 'pools': pools, - 'servers': servers, - } + 'pools': ['10.0.0.1', '10.0.0.2'] } mycloud = self._get_cloud(distro) - - with mock.patch.object(templater, 'render_to_file') as mocktmpl: - with mock.patch.object(os.path, 'isfile', return_value=True): - with mock.patch.object(util, 'rename'): - cc_ntp.write_ntp_config_template(cfg.get('ntp'), mycloud) - - mocktmpl.assert_called_once_with( - ('/etc/cloud/templates/ntp.conf.%s.tmpl' % distro), - '/etc/ntp.conf', - {'servers': servers, 'pools': pools}) - - def test_ntp_conf_custom_pools_and_server(self): + ntp_conf = self.tmp_path('ntp.conf', self.new_root) # Doesn't exist + # Create ntp.conf.tmpl which isn't read + with open('{0}.tmpl'.format(ntp_conf), 'wb') as stream: + stream.write(b'NOT READ: ntp.conf.<distro>.tmpl is primary') + # Create ntp.conf.tmpl.<distro> + with open('{0}.{1}.tmpl'.format(ntp_conf, distro), 'wb') as stream: + stream.write(NTP_TEMPLATE) + with mock.patch('cloudinit.config.cc_ntp.NTP_CONF', ntp_conf): + cc_ntp.write_ntp_config_template(cfg, mycloud) + content = util.read_file_or_url('file://' + ntp_conf).contents + self.assertEqual( + "servers []\npools ['10.0.0.1', '10.0.0.2']\n", + content.decode()) + + def test_write_ntp_config_template_defaults_pools_when_empty_lists(self): + """write_ntp_config_template defaults pools servers upon empty config. + + When both pools and servers are empty, default NR_POOL_SERVERS get + configured. + """ distro = 'ubuntu' - pools = ['0.mycompany.pool.ntp.org'] - servers = ['192.168.23.3'] - cfg = { - 'ntp': { - 'pools': pools, - 'servers': servers, - } - } mycloud = self._get_cloud(distro) - - with mock.patch.object(templater, 'render_to_file') as mocktmpl: - with mock.patch.object(os.path, 'isfile', return_value=True): - with mock.patch.object(util, 'rename'): - cc_ntp.write_ntp_config_template(cfg.get('ntp'), mycloud) - - mocktmpl.assert_called_once_with( - ('/etc/cloud/templates/ntp.conf.%s.tmpl' % distro), - '/etc/ntp.conf', - {'servers': servers, 'pools': pools}) - - def test_ntp_conf_contents_match(self): - """Test rendered contents of /etc/ntp.conf for ubuntu""" - pools = ['0.mycompany.pool.ntp.org'] - servers = ['192.168.23.3'] + ntp_conf = self.tmp_path('ntp.conf', self.new_root) # Doesn't exist + # Create ntp.conf.tmpl + with open('{0}.tmpl'.format(ntp_conf), 'wb') as stream: + stream.write(NTP_TEMPLATE) + with mock.patch('cloudinit.config.cc_ntp.NTP_CONF', ntp_conf): + cc_ntp.write_ntp_config_template({}, mycloud) + content = util.read_file_or_url('file://' + ntp_conf).contents + default_pools = [ + "{0}.{1}.pool.ntp.org".format(x, distro) + for x in range(0, cc_ntp.NR_POOL_SERVERS)] + self.assertEqual( + "servers []\npools {0}\n".format(default_pools), + content.decode()) + self.assertIn( + "Adding distro default ntp pool servers: {0}".format( + ",".join(default_pools)), + self.logs.getvalue()) + + def test_ntp_handler_mocked_template(self): + """Test ntp handler renders ubuntu ntp.conf template.""" + pools = ['0.mycompany.pool.ntp.org', '3.mycompany.pool.ntp.org'] + servers = ['192.168.23.3', '192.168.23.4'] cfg = { 'ntp': { 'pools': pools, - 'servers': servers, + 'servers': servers } } mycloud = self._get_cloud('ubuntu') - side_effect = [NTP_TEMPLATE.lstrip()] - - # work backwards from util.write_file and mock out call path - # write_ntp_config_template() - # cloud.get_template_filename() - # os.path.isfile() - # templater.render_to_file() - # templater.render_from_file() - # util.load_file() - # util.write_file() - # - with mock.patch.object(util, 'write_file') as mockwrite: - with mock.patch.object(util, 'load_file', side_effect=side_effect): - with mock.patch.object(os.path, 'isfile', return_value=True): - with mock.patch.object(util, 'rename'): - cc_ntp.write_ntp_config_template(cfg.get('ntp'), - mycloud) - - mockwrite.assert_called_once_with( - '/etc/ntp.conf', - NTP_EXPECTED_UBUNTU, - mode=420) - - def test_ntp_handler(self): - """Test ntp handler renders ubuntu ntp.conf template""" - pools = ['0.mycompany.pool.ntp.org'] - servers = ['192.168.23.3'] + ntp_conf = self.tmp_path('ntp.conf', self.new_root) # Doesn't exist + # Create ntp.conf.tmpl + with open('{0}.tmpl'.format(ntp_conf), 'wb') as stream: + stream.write(NTP_TEMPLATE) + with mock.patch('cloudinit.config.cc_ntp.NTP_CONF', ntp_conf): + with mock.patch.object(util, 'which', return_value=None): + cc_ntp.handle('notimportant', cfg, mycloud, None, None) + + content = util.read_file_or_url('file://' + ntp_conf).contents + self.assertEqual( + 'servers {0}\npools {1}\n'.format(servers, pools), + content.decode()) + + def test_ntp_handler_real_distro_templates(self): + """Test ntp handler renders the shipped distro ntp.conf templates.""" + pools = ['0.mycompany.pool.ntp.org', '3.mycompany.pool.ntp.org'] + servers = ['192.168.23.3', '192.168.23.4'] cfg = { 'ntp': { 'pools': pools, - 'servers': servers, + 'servers': servers } } - mycloud = self._get_cloud('ubuntu') - side_effect = [NTP_TEMPLATE.lstrip()] - - with mock.patch.object(util, 'which', return_value=None): - with mock.patch.object(os.path, 'exists'): - with mock.patch.object(util, 'write_file') as mockwrite: - with mock.patch.object(util, 'load_file', - side_effect=side_effect): - with mock.patch.object(os.path, 'isfile', - return_value=True): - with mock.patch.object(util, 'rename'): - cc_ntp.handle("notimportant", cfg, - mycloud, LOG, None) - - mockwrite.assert_called_once_with( - '/etc/ntp.conf', - NTP_EXPECTED_UBUNTU, - mode=420) - - @mock.patch("cloudinit.config.cc_ntp.util") - def test_no_ntpcfg_does_nothing(self, mock_util): - cc = self._get_cloud('ubuntu') - cc.distro = mock.MagicMock() - cc_ntp.handle('cc_ntp', {}, cc, LOG, []) - self.assertFalse(cc.distro.install_packages.called) - self.assertFalse(mock_util.subp.called) + ntp_conf = self.tmp_path('ntp.conf', self.new_root) # Doesn't exist + for distro in ('debian', 'ubuntu', 'fedora', 'rhel', 'sles'): + mycloud = self._get_cloud(distro) + root_dir = dirname(dirname(os.path.realpath(util.__file__))) + tmpl_file = os.path.join( + '{0}/templates/ntp.conf.{1}.tmpl'.format(root_dir, distro)) + # Create a copy in our tmp_dir + shutil.copy( + tmpl_file, + os.path.join(self.new_root, 'ntp.conf.%s.tmpl' % distro)) + with mock.patch('cloudinit.config.cc_ntp.NTP_CONF', ntp_conf): + with mock.patch.object(util, 'which', return_value=[True]): + cc_ntp.handle('notimportant', cfg, mycloud, None, None) + + content = util.read_file_or_url('file://' + ntp_conf).contents + expected_servers = '\n'.join([ + 'server {0} iburst'.format(server) for server in servers]) + self.assertIn( + expected_servers, content.decode(), + 'failed to render ntp.conf for distro:{0}'.format(distro)) + expected_pools = '\n'.join([ + 'pool {0} iburst'.format(pool) for pool in pools]) + self.assertIn( + expected_pools, content.decode(), + 'failed to render ntp.conf for distro:{0}'.format(distro)) + + def test_no_ntpcfg_does_nothing(self): + """When no ntp section is defined handler logs a warning and noops.""" + cc_ntp.handle('cc_ntp', {}, None, None, []) + self.assertEqual( + 'Skipping module named cc_ntp, not present or disabled by cfg\n', + self.logs.getvalue()) # vi: ts=4 expandtab diff --git a/tests/unittests/test_handler/test_handler_power_state.py b/tests/unittests/test_handler/test_handler_power_state.py index 3fd0069d..e382210d 100644 --- a/tests/unittests/test_handler/test_handler_power_state.py +++ b/tests/unittests/test_handler/test_handler_power_state.py @@ -15,12 +15,12 @@ class TestLoadPowerState(t_help.TestCase): def test_no_config(self): # completely empty config should mean do nothing (cmd, _timeout, _condition) = psc.load_power_state({}) - self.assertEqual(cmd, None) + self.assertIsNone(cmd) def test_irrelevant_config(self): # no power_state field in config should return None for cmd (cmd, _timeout, _condition) = psc.load_power_state({'foo': 'bar'}) - self.assertEqual(cmd, None) + self.assertIsNone(cmd) def test_invalid_mode(self): cfg = {'power_state': {'mode': 'gibberish'}} diff --git a/tests/unittests/test_handler/test_handler_resizefs.py b/tests/unittests/test_handler/test_handler_resizefs.py new file mode 100644 index 00000000..52591b8b --- /dev/null +++ b/tests/unittests/test_handler/test_handler_resizefs.py @@ -0,0 +1,59 @@ +# This file is part of cloud-init. See LICENSE file for license information. + +from cloudinit.config import cc_resizefs + +import textwrap +import unittest + +try: + from unittest import mock +except ImportError: + import mock + + +class TestResizefs(unittest.TestCase): + def setUp(self): + super(TestResizefs, self).setUp() + self.name = "resizefs" + + @mock.patch('cloudinit.config.cc_resizefs._get_dumpfs_output') + @mock.patch('cloudinit.config.cc_resizefs._get_gpart_output') + def test_skip_ufs_resize(self, gpart_out, dumpfs_out): + fs_type = "ufs" + resize_what = "/" + devpth = "/dev/da0p2" + dumpfs_out.return_value = ( + "# newfs command for / (/dev/label/rootfs)\n" + "newfs -O 2 -U -a 4 -b 32768 -d 32768 -e 4096 " + "-f 4096 -g 16384 -h 64 -i 8192 -j -k 6408 -m 8 " + "-o time -s 58719232 /dev/label/rootfs\n") + gpart_out.return_value = textwrap.dedent("""\ + => 40 62914480 da0 GPT (30G) + 40 1024 1 freebsd-boot (512K) + 1064 58719232 2 freebsd-ufs (28G) + 58720296 3145728 3 freebsd-swap (1.5G) + 61866024 1048496 - free - (512M) + """) + res = cc_resizefs.can_skip_resize(fs_type, resize_what, devpth) + self.assertTrue(res) + + @mock.patch('cloudinit.config.cc_resizefs._get_dumpfs_output') + @mock.patch('cloudinit.config.cc_resizefs._get_gpart_output') + def test_skip_ufs_resize_roundup(self, gpart_out, dumpfs_out): + fs_type = "ufs" + resize_what = "/" + devpth = "/dev/da0p2" + dumpfs_out.return_value = ( + "# newfs command for / (/dev/label/rootfs)\n" + "newfs -O 2 -U -a 4 -b 32768 -d 32768 -e 4096 " + "-f 4096 -g 16384 -h 64 -i 8192 -j -k 368 -m 8 " + "-o time -s 297080 /dev/label/rootfs\n") + gpart_out.return_value = textwrap.dedent("""\ + => 34 297086 da0 GPT (145M) + 34 297086 1 freebsd-ufs (145M) + """) + res = cc_resizefs.can_skip_resize(fs_type, resize_what, devpth) + self.assertTrue(res) + + +# vi: ts=4 expandtab diff --git a/tests/unittests/test_handler/test_handler_snappy.py b/tests/unittests/test_handler/test_handler_snappy.py index edb73d6d..e4d07622 100644 --- a/tests/unittests/test_handler/test_handler_snappy.py +++ b/tests/unittests/test_handler/test_handler_snappy.py @@ -419,7 +419,7 @@ class TestSnapConfig(FilesystemMockingTestCase): def test_snap_config_add_snap_user_no_config(self): usercfg = add_snap_user(cfg=None) - self.assertEqual(usercfg, None) + self.assertIsNone(usercfg) def test_snap_config_add_snap_user_not_dict(self): cfg = ['foobar'] @@ -428,7 +428,7 @@ class TestSnapConfig(FilesystemMockingTestCase): def test_snap_config_add_snap_user_no_email(self): cfg = {'assertions': [], 'known': True} usercfg = add_snap_user(cfg=cfg) - self.assertEqual(usercfg, None) + self.assertIsNone(usercfg) @mock.patch('cloudinit.config.cc_snap_config.util') def test_snap_config_add_snap_user_email_only(self, mock_util): diff --git a/tests/unittests/test_handler/test_handler_yum_add_repo.py b/tests/unittests/test_handler/test_handler_yum_add_repo.py index 4815bdb6..c4396df5 100644 --- a/tests/unittests/test_handler/test_handler_yum_add_repo.py +++ b/tests/unittests/test_handler/test_handler_yum_add_repo.py @@ -72,7 +72,7 @@ class TestConfig(helpers.FilesystemMockingTestCase): } for section in expected: self.assertTrue(parser.has_section(section), - "Contains section {}".format(section)) + "Contains section {0}".format(section)) for k, v in expected[section].items(): self.assertEqual(parser.get(section, k), v) @@ -109,7 +109,7 @@ class TestConfig(helpers.FilesystemMockingTestCase): } for section in expected: self.assertTrue(parser.has_section(section), - "Contains section {}".format(section)) + "Contains section {0}".format(section)) for k, v in expected[section].items(): self.assertEqual(parser.get(section, k), v) |