From e11d3899d47ec5fcb545e0c7820af9d3995cb574 Mon Sep 17 00:00:00 2001 From: Ryan Harper Date: Fri, 19 May 2017 16:49:11 -0400 Subject: cc_ntp: write template before installing and add service restart On systems which installed ntp and specified servers or pools in the config ntpd didn't notice the updated configuration file and didn't use the correct configuration. Resolve this by rendering the template first which allows the package install to use the existing configuration. Additionally add a service restart to handle the case where ntp does not need to be installed but it may not have started. Add an integration test to confirm that cc_ntp enables ntp to use the specific servers and pools in the cloud-config. LP: #1645644 --- tests/unittests/test_handler/test_handler_ntp.py | 78 +++++++++++++++++++++++- 1 file changed, 76 insertions(+), 2 deletions(-) (limited to 'tests/unittests/test_handler/test_handler_ntp.py') diff --git a/tests/unittests/test_handler/test_handler_ntp.py b/tests/unittests/test_handler/test_handler_ntp.py index ec600077..02bd3046 100644 --- a/tests/unittests/test_handler/test_handler_ntp.py +++ b/tests/unittests/test_handler/test_handler_ntp.py @@ -249,6 +249,8 @@ class TestNtp(FilesystemMockingTestCase): } } mycloud = self._get_cloud('ubuntu') + mycloud.distro = mock.MagicMock() + mycloud.distro.uses_systemd.return_value = True side_effect = [NTP_TEMPLATE.lstrip()] with mock.patch.object(util, 'which', return_value=None): @@ -259,13 +261,70 @@ class TestNtp(FilesystemMockingTestCase): with mock.patch.object(os.path, 'isfile', return_value=True): with mock.patch.object(util, 'rename'): - cc_ntp.handle("notimportant", cfg, - mycloud, LOG, None) + with mock.patch.object(util, 'subp') as msubp: + cc_ntp.handle("notimportant", cfg, + mycloud, LOG, None) mockwrite.assert_called_once_with( '/etc/ntp.conf', NTP_EXPECTED_UBUNTU, mode=420) + msubp.assert_any_call(['systemctl', 'reload-or-restart', 'ntp'], + capture=True) + + @mock.patch("cloudinit.config.cc_ntp.install_ntp") + @mock.patch("cloudinit.config.cc_ntp.write_ntp_config_template") + @mock.patch("cloudinit.config.cc_ntp.rename_ntp_conf") + def test_write_config_before_install(self, mock_ntp_rename, + mock_ntp_write_config, + mock_install_ntp): + pools = ['0.mycompany.pool.ntp.org'] + servers = ['192.168.23.3'] + cfg = { + 'ntp': { + 'pools': pools, + 'servers': servers, + } + } + cc = self._get_cloud('ubuntu') + cc.distro = mock.MagicMock() + mock_parent = mock.MagicMock() + mock_parent.attach_mock(mock_ntp_rename, 'mock_ntp_rename') + mock_parent.attach_mock(mock_ntp_write_config, 'mock_ntp_write_config') + mock_parent.attach_mock(mock_install_ntp, 'mock_install_ntp') + + cc_ntp.handle('cc_ntp', cfg, cc, LOG, None) + + """Check call order""" + mock_parent.assert_has_calls([ + mock.call.mock_ntp_rename(), + mock.call.mock_ntp_write_config(cfg.get('ntp'), cc), + mock.call.mock_install_ntp(cc.distro.install_packages, + packages=['ntp'], check_exe="ntpd")]) + + @mock.patch("cloudinit.config.cc_ntp.reload_ntp") + @mock.patch("cloudinit.config.cc_ntp.install_ntp") + @mock.patch("cloudinit.config.cc_ntp.write_ntp_config_template") + @mock.patch("cloudinit.config.cc_ntp.rename_ntp_conf") + def test_reload_ntp_fail_raises_exception(self, mock_rename, + mock_write_conf, + mock_install, + mock_reload): + pools = ['0.mycompany.pool.ntp.org'] + servers = ['192.168.23.3'] + cfg = { + 'ntp': { + 'pools': pools, + 'servers': servers, + } + } + cc = self._get_cloud('ubuntu') + cc.distro = mock.MagicMock() + + mock_reload.side_effect = [util.ProcessExecutionError] + self.assertRaises(util.ProcessExecutionError, + cc_ntp.handle, 'cc_ntp', + cfg, cc, LOG, None) @mock.patch("cloudinit.config.cc_ntp.util") def test_no_ntpcfg_does_nothing(self, mock_util): @@ -275,4 +334,19 @@ class TestNtp(FilesystemMockingTestCase): self.assertFalse(cc.distro.install_packages.called) self.assertFalse(mock_util.subp.called) + @mock.patch("cloudinit.config.cc_ntp.util") + def test_reload_ntp_systemd(self, mock_util): + cc_ntp.reload_ntp(systemd=True) + self.assertTrue(mock_util.subp.called) + mock_util.subp.assert_called_with( + ['systemctl', 'reload-or-restart', 'ntp'], capture=True) + + @mock.patch("cloudinit.config.cc_ntp.util") + def test_reload_ntp_service(self, mock_util): + cc_ntp.reload_ntp(systemd=False) + self.assertTrue(mock_util.subp.called) + mock_util.subp.assert_called_with( + ['service', 'ntp', 'restart'], capture=True) + + # vi: ts=4 expandtab -- cgit v1.2.3 From fd0c88cf8e8aa015eadb5ab842e872cb627197ec Mon Sep 17 00:00:00 2001 From: Chad Smith Date: Mon, 22 May 2017 23:11:42 -0600 Subject: cc_ntp: Restructure cc_ntp unit tests. Any CiTestCase subclass can now set a class attribute with_logs = True and tests can now make assertions on self.logs.getvalue(). This branch restructures a bit of cc_ntp module to get better test coverage of the module. It also restructures the handler_cc_ntp unit tests to avoid nested mocks where possible. Deeply nested mocks cause a couple of issues: - greater risk: mocks are permanent within the scope, so multiple call-sites could be affected by package mocks - less legible tests: each mock doesn't advertise the actual call-site - tight coupling: the unit test logic to tightly bound to the actual implementation in remote (unrelated) modules which makes it more costly to maintain code - false success: we should be testing the expected behavior not specific remote method names as we want to know if that underlying behavior changes and breaks us. LP: #1692794 --- cloudinit/config/cc_ntp.py | 25 +- tests/unittests/helpers.py | 26 ++ tests/unittests/test_handler/test_handler_ntp.py | 441 ++++++++--------------- 3 files changed, 188 insertions(+), 304 deletions(-) (limited to 'tests/unittests/test_handler/test_handler_ntp.py') diff --git a/cloudinit/config/cc_ntp.py b/cloudinit/config/cc_ntp.py index 225f898a..5cc54536 100644 --- a/cloudinit/config/cc_ntp.py +++ b/cloudinit/config/cc_ntp.py @@ -53,14 +53,12 @@ distros = ['centos', 'debian', 'fedora', 'opensuse', 'ubuntu'] def handle(name, cfg, cloud, log, _args): - """ - Enable and configure ntp + """Enable and configure ntp.""" - ntp: - pools: ['0.{{distro}}.pool.ntp.org', '1.{{distro}}.pool.ntp.org'] - servers: ['192.168.2.1'] - - """ + if 'ntp' not in cfg: + LOG.debug( + "Skipping module named %s, not present or disabled by cfg", name) + return ntp_cfg = cfg.get('ntp', {}) @@ -69,18 +67,12 @@ def handle(name, cfg, cloud, log, _args): " but not a dictionary type," " is a %s %instead"), type_utils.obj_name(ntp_cfg)) - if 'ntp' not in cfg: - LOG.debug("Skipping module named %s," - "not present or disabled by cfg", name) - return True - rename_ntp_conf() # ensure when ntp is installed it has a configuration file # to use instead of starting up with packaged defaults write_ntp_config_template(ntp_cfg, cloud) install_ntp(cloud.distro.install_packages, packages=['ntp'], check_exe="ntpd") - # if ntp was already installed, it may not have started try: reload_ntp(systemd=cloud.distro.uses_systemd()) @@ -98,8 +90,10 @@ def install_ntp(install_func, packages=None, check_exe="ntpd"): install_func(packages) -def rename_ntp_conf(config=NTP_CONF): +def rename_ntp_conf(config=None): """Rename any existing ntp.conf file and render from template""" + if config is None: # For testing + config = NTP_CONF if os.path.exists(config): util.rename(config, config + ".dist") @@ -117,8 +111,9 @@ def write_ntp_config_template(cfg, cloud): pools = cfg.get('pools', []) if len(servers) == 0 and len(pools) == 0: - LOG.debug('Adding distro default ntp pool servers') pools = generate_server_names(cloud.distro.name) + LOG.debug( + 'Adding distro default ntp pool servers: %s', ','.join(pools)) params = { 'servers': servers, diff --git a/tests/unittests/helpers.py b/tests/unittests/helpers.py index d24f817d..9ff15993 100644 --- a/tests/unittests/helpers.py +++ b/tests/unittests/helpers.py @@ -4,6 +4,7 @@ from __future__ import print_function import functools import json +import logging import os import shutil import sys @@ -18,6 +19,10 @@ try: from contextlib import ExitStack except ImportError: from contextlib2 import ExitStack +try: + from cStringIO import StringIO +except ImportError: + from io import StringIO from cloudinit import helpers as ch from cloudinit import util @@ -87,6 +92,27 @@ class TestCase(unittest2.TestCase): class CiTestCase(TestCase): """This is the preferred test case base class unless user needs other test case classes below.""" + + # Subclass overrides for specific test behavior + # Whether or not a unit test needs logfile setup + with_logs = False + + def setUp(self): + super(CiTestCase, self).setUp() + if self.with_logs: + # Create a log handler so unit tests can search expected logs. + logger = logging.getLogger() + self.logs = StringIO() + handler = logging.StreamHandler(self.logs) + self.old_handlers = logger.handlers + logger.handlers = [handler] + + def tearDown(self): + if self.with_logs: + # Remove the handler we setup + logging.getLogger().handlers = self.old_handlers + super(CiTestCase, self).tearDown() + def tmp_dir(self, dir=None, cleanup=True): # return a full path to a temporary directory that will be cleaned up. if dir is None: diff --git a/tests/unittests/test_handler/test_handler_ntp.py b/tests/unittests/test_handler/test_handler_ntp.py index 02bd3046..21f2ab19 100644 --- a/tests/unittests/test_handler/test_handler_ntp.py +++ b/tests/unittests/test_handler/test_handler_ntp.py @@ -2,351 +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("{}.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("{}.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('{}.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..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('{}.tmpl'.format(ntp_conf), 'wb') as stream: + stream.write(b'NOT READ: ntp.conf..tmpl is primary') + # Create ntp.conf.tmpl. + with open('{}.{}.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('{}.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 = [ + "{}.{}.pool.ntp.org".format(x, distro) + for x in range(0, cc_ntp.NR_POOL_SERVERS)] + self.assertEqual( + "servers []\npools {}\n".format(default_pools), + content.decode()) + self.assertIn( + "Adding distro default ntp pool servers: {}".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('{}.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 {}\npools {}\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') - mycloud.distro = mock.MagicMock() - mycloud.distro.uses_systemd.return_value = True - 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'): - with mock.patch.object(util, 'subp') as msubp: - cc_ntp.handle("notimportant", cfg, - mycloud, LOG, None) - - mockwrite.assert_called_once_with( - '/etc/ntp.conf', - NTP_EXPECTED_UBUNTU, - mode=420) - msubp.assert_any_call(['systemctl', 'reload-or-restart', 'ntp'], - capture=True) - - @mock.patch("cloudinit.config.cc_ntp.install_ntp") - @mock.patch("cloudinit.config.cc_ntp.write_ntp_config_template") - @mock.patch("cloudinit.config.cc_ntp.rename_ntp_conf") - def test_write_config_before_install(self, mock_ntp_rename, - mock_ntp_write_config, - mock_install_ntp): - pools = ['0.mycompany.pool.ntp.org'] - servers = ['192.168.23.3'] - cfg = { - 'ntp': { - 'pools': pools, - 'servers': servers, - } - } - cc = self._get_cloud('ubuntu') - cc.distro = mock.MagicMock() - mock_parent = mock.MagicMock() - mock_parent.attach_mock(mock_ntp_rename, 'mock_ntp_rename') - mock_parent.attach_mock(mock_ntp_write_config, 'mock_ntp_write_config') - mock_parent.attach_mock(mock_install_ntp, 'mock_install_ntp') - - cc_ntp.handle('cc_ntp', cfg, cc, LOG, None) - - """Check call order""" - mock_parent.assert_has_calls([ - mock.call.mock_ntp_rename(), - mock.call.mock_ntp_write_config(cfg.get('ntp'), cc), - mock.call.mock_install_ntp(cc.distro.install_packages, - packages=['ntp'], check_exe="ntpd")]) - - @mock.patch("cloudinit.config.cc_ntp.reload_ntp") - @mock.patch("cloudinit.config.cc_ntp.install_ntp") - @mock.patch("cloudinit.config.cc_ntp.write_ntp_config_template") - @mock.patch("cloudinit.config.cc_ntp.rename_ntp_conf") - def test_reload_ntp_fail_raises_exception(self, mock_rename, - mock_write_conf, - mock_install, - mock_reload): - pools = ['0.mycompany.pool.ntp.org'] - servers = ['192.168.23.3'] - cfg = { - 'ntp': { - 'pools': pools, - 'servers': servers, - } - } - cc = self._get_cloud('ubuntu') - cc.distro = mock.MagicMock() - - mock_reload.side_effect = [util.ProcessExecutionError] - self.assertRaises(util.ProcessExecutionError, - cc_ntp.handle, 'cc_ntp', - cfg, cc, LOG, None) - - @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) - - @mock.patch("cloudinit.config.cc_ntp.util") - def test_reload_ntp_systemd(self, mock_util): - cc_ntp.reload_ntp(systemd=True) - self.assertTrue(mock_util.subp.called) - mock_util.subp.assert_called_with( - ['systemctl', 'reload-or-restart', 'ntp'], capture=True) - - @mock.patch("cloudinit.config.cc_ntp.util") - def test_reload_ntp_service(self, mock_util): - cc_ntp.reload_ntp(systemd=False) - self.assertTrue(mock_util.subp.called) - mock_util.subp.assert_called_with( - ['service', 'ntp', 'restart'], capture=True) - + 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( + '{}/templates/ntp.conf.{}.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 {} iburst'.format(server) for server in servers]) + self.assertIn( + expected_servers, content.decode(), + 'failed to render ntp.conf for distro:{}'.format(distro)) + expected_pools = '\n'.join([ + 'pool {} iburst'.format(pool) for pool in pools]) + self.assertIn( + expected_pools, content.decode(), + 'failed to render ntp.conf for distro:{}'.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 -- cgit v1.2.3 From e5b2c011440aefe036c71a8c5e8ec547cc80f270 Mon Sep 17 00:00:00 2001 From: Scott Moser Date: Wed, 24 May 2017 21:33:30 -0400 Subject: python2.6: fix unit tests usage of assertNone and format. python2.6 unittest.TestCase does not have the assertIsNone or assertIsNotNone. We just have to explicitly use the unittest2 version, which we get from helpers. The desire to use assertIsNone comes from flake8 (through hacking, I believe). Also, fix "{}.format('foo')" which is not valid in python2.6. --- tests/unittests/test_datasource/test_altcloud.py | 3 ++- tests/unittests/test_handler/test_handler_ntp.py | 32 ++++++++++++------------ 2 files changed, 18 insertions(+), 17 deletions(-) (limited to 'tests/unittests/test_handler/test_handler_ntp.py') diff --git a/tests/unittests/test_datasource/test_altcloud.py b/tests/unittests/test_datasource/test_altcloud.py index b6d4a453..9c46abc1 100644 --- a/tests/unittests/test_datasource/test_altcloud.py +++ b/tests/unittests/test_datasource/test_altcloud.py @@ -17,7 +17,8 @@ import tempfile from cloudinit import helpers from cloudinit import util -from unittest import TestCase + +from ..helpers import TestCase import cloudinit.sources.DataSourceAltCloud as dsac diff --git a/tests/unittests/test_handler/test_handler_ntp.py b/tests/unittests/test_handler/test_handler_ntp.py index 21f2ab19..bc4277b7 100644 --- a/tests/unittests/test_handler/test_handler_ntp.py +++ b/tests/unittests/test_handler/test_handler_ntp.py @@ -59,7 +59,7 @@ class TestNtp(FilesystemMockingTestCase): 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("{}.dist".format(ntpconf))) + self.assertTrue(os.path.exists("{0}.dist".format(ntpconf))) def test_ntp_rename_ntp_conf_skip_missing(self): """When NTP_CONF doesn't exist rename_ntp doesn't create a file.""" @@ -67,7 +67,7 @@ class TestNtp(FilesystemMockingTestCase): 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("{}.dist".format(ntpconf))) + 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): @@ -84,7 +84,7 @@ class TestNtp(FilesystemMockingTestCase): mycloud = self._get_cloud(distro) ntp_conf = self.tmp_path("ntp.conf", self.new_root) # Doesn't exist # Create ntp.conf.tmpl - with open('{}.tmpl'.format(ntp_conf), 'wb') as stream: + 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) @@ -107,10 +107,10 @@ class TestNtp(FilesystemMockingTestCase): mycloud = self._get_cloud(distro) ntp_conf = self.tmp_path('ntp.conf', self.new_root) # Doesn't exist # Create ntp.conf.tmpl which isn't read - with open('{}.tmpl'.format(ntp_conf), 'wb') as stream: + with open('{0}.tmpl'.format(ntp_conf), 'wb') as stream: stream.write(b'NOT READ: ntp.conf..tmpl is primary') # Create ntp.conf.tmpl. - with open('{}.{}.tmpl'.format(ntp_conf, distro), 'wb') as stream: + 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) @@ -129,19 +129,19 @@ class TestNtp(FilesystemMockingTestCase): mycloud = self._get_cloud(distro) ntp_conf = self.tmp_path('ntp.conf', self.new_root) # Doesn't exist # Create ntp.conf.tmpl - with open('{}.tmpl'.format(ntp_conf), 'wb') as stream: + 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 = [ - "{}.{}.pool.ntp.org".format(x, distro) + "{0}.{1}.pool.ntp.org".format(x, distro) for x in range(0, cc_ntp.NR_POOL_SERVERS)] self.assertEqual( - "servers []\npools {}\n".format(default_pools), + "servers []\npools {0}\n".format(default_pools), content.decode()) self.assertIn( - "Adding distro default ntp pool servers: {}".format( + "Adding distro default ntp pool servers: {0}".format( ",".join(default_pools)), self.logs.getvalue()) @@ -158,7 +158,7 @@ class TestNtp(FilesystemMockingTestCase): mycloud = self._get_cloud('ubuntu') ntp_conf = self.tmp_path('ntp.conf', self.new_root) # Doesn't exist # Create ntp.conf.tmpl - with open('{}.tmpl'.format(ntp_conf), 'wb') as stream: + 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): @@ -166,7 +166,7 @@ class TestNtp(FilesystemMockingTestCase): content = util.read_file_or_url('file://' + ntp_conf).contents self.assertEqual( - 'servers {}\npools {}\n'.format(servers, pools), + 'servers {0}\npools {1}\n'.format(servers, pools), content.decode()) def test_ntp_handler_real_distro_templates(self): @@ -184,7 +184,7 @@ class TestNtp(FilesystemMockingTestCase): mycloud = self._get_cloud(distro) root_dir = dirname(dirname(os.path.realpath(util.__file__))) tmpl_file = os.path.join( - '{}/templates/ntp.conf.{}.tmpl'.format(root_dir, distro)) + '{0}/templates/ntp.conf.{1}.tmpl'.format(root_dir, distro)) # Create a copy in our tmp_dir shutil.copy( tmpl_file, @@ -195,15 +195,15 @@ class TestNtp(FilesystemMockingTestCase): content = util.read_file_or_url('file://' + ntp_conf).contents expected_servers = '\n'.join([ - 'server {} iburst'.format(server) for server in servers]) + 'server {0} iburst'.format(server) for server in servers]) self.assertIn( expected_servers, content.decode(), - 'failed to render ntp.conf for distro:{}'.format(distro)) + 'failed to render ntp.conf for distro:{0}'.format(distro)) expected_pools = '\n'.join([ - 'pool {} iburst'.format(pool) for pool in pools]) + 'pool {0} iburst'.format(pool) for pool in pools]) self.assertIn( expected_pools, content.decode(), - 'failed to render ntp.conf for distro:{}'.format(distro)) + '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.""" -- cgit v1.2.3