diff options
Diffstat (limited to 'cloudinit/tests')
| -rw-r--r-- | cloudinit/tests/helpers.py | 110 | ||||
| -rw-r--r-- | cloudinit/tests/test_util.py | 78 |
2 files changed, 168 insertions, 20 deletions
diff --git a/cloudinit/tests/helpers.py b/cloudinit/tests/helpers.py index 5bfe7fa4..2eb7b0cd 100644 --- a/cloudinit/tests/helpers.py +++ b/cloudinit/tests/helpers.py @@ -10,16 +10,16 @@ import shutil import sys import tempfile import time -import unittest import mock import six import unittest2 +from unittest2.util import strclass try: - from contextlib import ExitStack + from contextlib import ExitStack, contextmanager except ImportError: - from contextlib2 import ExitStack + from contextlib2 import ExitStack, contextmanager try: from configparser import ConfigParser @@ -28,11 +28,18 @@ except ImportError: from cloudinit.config.schema import ( SchemaValidationError, validate_cloudconfig_schema) +from cloudinit import cloud +from cloudinit import distros from cloudinit import helpers as ch +from cloudinit.sources import DataSourceNone +from cloudinit.templater import JINJA_AVAILABLE from cloudinit import util +_real_subp = util.subp + # Used for skipping tests SkipTest = unittest2.SkipTest +skipIf = unittest2.skipIf # Used for detecting different python versions PY2 = False @@ -112,6 +119,9 @@ class TestCase(unittest2.TestCase): super(TestCase, self).setUp() self.reset_global_state() + def shortDescription(self): + return strclass(self.__class__) + '.' + self._testMethodName + def add_patch(self, target, attr, *args, **kwargs): """Patches specified target object and sets it as attr on test instance also schedules cleanup""" @@ -140,6 +150,17 @@ class CiTestCase(TestCase): # Subclass overrides for specific test behavior # Whether or not a unit test needs logfile setup with_logs = False + allowed_subp = False + SUBP_SHELL_TRUE = "shell=true" + + @contextmanager + def allow_subp(self, allowed_subp): + orig = self.allowed_subp + try: + self.allowed_subp = allowed_subp + yield + finally: + self.allowed_subp = orig def setUp(self): super(CiTestCase, self).setUp() @@ -152,11 +173,41 @@ class CiTestCase(TestCase): handler.setFormatter(formatter) self.old_handlers = self.logger.handlers self.logger.handlers = [handler] + if self.allowed_subp is True: + util.subp = _real_subp + else: + util.subp = self._fake_subp + + def _fake_subp(self, *args, **kwargs): + if 'args' in kwargs: + cmd = kwargs['args'] + else: + cmd = args[0] + + if not isinstance(cmd, six.string_types): + cmd = cmd[0] + pass_through = False + if not isinstance(self.allowed_subp, (list, bool)): + raise TypeError("self.allowed_subp supports list or bool.") + if isinstance(self.allowed_subp, bool): + pass_through = self.allowed_subp + else: + pass_through = ( + (cmd in self.allowed_subp) or + (self.SUBP_SHELL_TRUE in self.allowed_subp and + kwargs.get('shell'))) + if pass_through: + return _real_subp(*args, **kwargs) + raise Exception( + "called subp. set self.allowed_subp=True to allow\n subp(%s)" % + ', '.join([str(repr(a)) for a in args] + + ["%s=%s" % (k, repr(v)) for k, v in kwargs.items()])) def tearDown(self): if self.with_logs: # Remove the handler we setup logging.getLogger().handlers = self.old_handlers + util.subp = _real_subp super(CiTestCase, self).tearDown() def tmp_dir(self, dir=None, cleanup=True): @@ -187,6 +238,29 @@ class CiTestCase(TestCase): """ raise SystemExit(code) + def tmp_cloud(self, distro, sys_cfg=None, metadata=None): + """Create a cloud with tmp working directory paths. + + @param distro: Name of the distro to attach to the cloud. + @param metadata: Optional metadata to set on the datasource. + + @return: The built cloud instance. + """ + self.new_root = self.tmp_dir() + if not sys_cfg: + sys_cfg = {} + tmp_paths = {} + for var in ['templates_dir', 'run_dir', 'cloud_dir']: + tmp_paths[var] = self.tmp_path(var, dir=self.new_root) + util.ensure_dir(tmp_paths[var]) + self.paths = ch.Paths(tmp_paths) + cls = distros.fetch(distro) + mydist = cls(distro, sys_cfg, self.paths) + myds = DataSourceNone.DataSourceNone(sys_cfg, mydist, self.paths) + if metadata: + myds.metadata.update(metadata) + return cloud.Cloud(myds, self.paths, sys_cfg, mydist, None) + class ResourceUsingTestCase(CiTestCase): @@ -300,6 +374,13 @@ class FilesystemMockingTestCase(ResourceUsingTestCase): self.patchOS(root) return root + @contextmanager + def reRooted(self, root=None): + try: + yield self.reRoot(root) + finally: + self.patched_funcs.close() + class HttprettyTestCase(CiTestCase): # necessary as http_proxy gets in the way of httpretty @@ -426,21 +507,6 @@ def readResource(name, mode='r'): try: - skipIf = unittest.skipIf -except AttributeError: - # Python 2.6. Doesn't have to be high fidelity. - def skipIf(condition, reason): - def decorator(func): - def wrapper(*args, **kws): - if condition: - return func(*args, **kws) - else: - print(reason, file=sys.stderr) - return wrapper - return decorator - - -try: import jsonschema assert jsonschema # avoid pyflakes error F401: import unused _missing_jsonschema_dep = False @@ -453,6 +519,14 @@ def skipUnlessJsonSchema(): _missing_jsonschema_dep, "No python-jsonschema dependency present.") +def skipUnlessJinja(): + return skipIf(not JINJA_AVAILABLE, "No jinja dependency present.") + + +def skipIfJinja(): + return skipIf(JINJA_AVAILABLE, "Jinja dependency present.") + + # older versions of mock do not have the useful 'assert_not_called' if not hasattr(mock.Mock, 'assert_not_called'): def __mock_assert_not_called(mmock): diff --git a/cloudinit/tests/test_util.py b/cloudinit/tests/test_util.py index 6a31e505..edb0c18f 100644 --- a/cloudinit/tests/test_util.py +++ b/cloudinit/tests/test_util.py @@ -57,6 +57,34 @@ OS_RELEASE_CENTOS = dedent("""\ REDHAT_SUPPORT_PRODUCT_VERSION="7" """) +OS_RELEASE_REDHAT_7 = dedent("""\ + NAME="Red Hat Enterprise Linux Server" + VERSION="7.5 (Maipo)" + ID="rhel" + ID_LIKE="fedora" + VARIANT="Server" + VARIANT_ID="server" + VERSION_ID="7.5" + PRETTY_NAME="Red Hat" + ANSI_COLOR="0;31" + CPE_NAME="cpe:/o:redhat:enterprise_linux:7.5:GA:server" + HOME_URL="https://www.redhat.com/" + BUG_REPORT_URL="https://bugzilla.redhat.com/" + + REDHAT_BUGZILLA_PRODUCT="Red Hat Enterprise Linux 7" + REDHAT_BUGZILLA_PRODUCT_VERSION=7.5 + REDHAT_SUPPORT_PRODUCT="Red Hat Enterprise Linux" + REDHAT_SUPPORT_PRODUCT_VERSION="7.5" +""") + +REDHAT_RELEASE_CENTOS_6 = "CentOS release 6.10 (Final)" +REDHAT_RELEASE_CENTOS_7 = "CentOS Linux release 7.5.1804 (Core)" +REDHAT_RELEASE_REDHAT_6 = ( + "Red Hat Enterprise Linux Server release 6.10 (Santiago)") +REDHAT_RELEASE_REDHAT_7 = ( + "Red Hat Enterprise Linux Server release 7.5 (Maipo)") + + OS_RELEASE_DEBIAN = dedent("""\ PRETTY_NAME="Debian GNU/Linux 9 (stretch)" NAME="Debian GNU/Linux" @@ -337,6 +365,12 @@ class TestGetLinuxDistro(CiTestCase): if path == '/etc/os-release': return 1 + @classmethod + def redhat_release_exists(self, path): + """Side effect function """ + if path == '/etc/redhat-release': + return 1 + @mock.patch('cloudinit.util.load_file') def test_get_linux_distro_quoted_name(self, m_os_release, m_path_exists): """Verify we get the correct name if the os-release file has @@ -356,8 +390,48 @@ class TestGetLinuxDistro(CiTestCase): self.assertEqual(('ubuntu', '16.04', 'xenial'), dist) @mock.patch('cloudinit.util.load_file') - def test_get_linux_centos(self, m_os_release, m_path_exists): - """Verify we get the correct name and release name on CentOS.""" + def test_get_linux_centos6(self, m_os_release, m_path_exists): + """Verify we get the correct name and release name on CentOS 6.""" + m_os_release.return_value = REDHAT_RELEASE_CENTOS_6 + m_path_exists.side_effect = TestGetLinuxDistro.redhat_release_exists + dist = util.get_linux_distro() + self.assertEqual(('centos', '6.10', 'Final'), dist) + + @mock.patch('cloudinit.util.load_file') + def test_get_linux_centos7_redhat_release(self, m_os_release, m_exists): + """Verify the correct release info on CentOS 7 without os-release.""" + m_os_release.return_value = REDHAT_RELEASE_CENTOS_7 + m_exists.side_effect = TestGetLinuxDistro.redhat_release_exists + dist = util.get_linux_distro() + self.assertEqual(('centos', '7.5.1804', 'Core'), dist) + + @mock.patch('cloudinit.util.load_file') + def test_get_linux_redhat7_osrelease(self, m_os_release, m_path_exists): + """Verify redhat 7 read from os-release.""" + m_os_release.return_value = OS_RELEASE_REDHAT_7 + m_path_exists.side_effect = TestGetLinuxDistro.os_release_exists + dist = util.get_linux_distro() + self.assertEqual(('redhat', '7.5', 'Maipo'), dist) + + @mock.patch('cloudinit.util.load_file') + def test_get_linux_redhat7_rhrelease(self, m_os_release, m_path_exists): + """Verify redhat 7 read from redhat-release.""" + m_os_release.return_value = REDHAT_RELEASE_REDHAT_7 + m_path_exists.side_effect = TestGetLinuxDistro.redhat_release_exists + dist = util.get_linux_distro() + self.assertEqual(('redhat', '7.5', 'Maipo'), dist) + + @mock.patch('cloudinit.util.load_file') + def test_get_linux_redhat6_rhrelease(self, m_os_release, m_path_exists): + """Verify redhat 6 read from redhat-release.""" + m_os_release.return_value = REDHAT_RELEASE_REDHAT_6 + m_path_exists.side_effect = TestGetLinuxDistro.redhat_release_exists + dist = util.get_linux_distro() + self.assertEqual(('redhat', '6.10', 'Santiago'), dist) + + @mock.patch('cloudinit.util.load_file') + def test_get_linux_copr_centos(self, m_os_release, m_path_exists): + """Verify we get the correct name and release name on COPR CentOS.""" m_os_release.return_value = OS_RELEASE_CENTOS m_path_exists.side_effect = TestGetLinuxDistro.os_release_exists dist = util.get_linux_distro() |