testing: add get_cloud function (SC-461)

cloudinit/config/tests/test_resolv_conf.py

@@ -2,7 +2,7 @@
 
 from unittest import mock
 from cloudinit.config.cc_resolv_conf import generate_resolv_conf
-from tests.unittests.test_distros.test_create_users import MyBaseDistro
+from tests.unittests.util import TestingDistro
 
 EXPECTED_HEADER = """\
 # Your system has been configured with 'manage-resolv-conf' set to true.
@@ -14,7 +14,7 @@
 
 class TestGenerateResolvConf:
 
-    dist = MyBaseDistro()
+    dist = TestingDistro()
     tmpl_fn = "templates/resolv.conf.tmpl"
 
     @mock.patch("cloudinit.config.cc_resolv_conf.templater.render_to_file")

tests/unittests/test_distros/test_create_users.py

@@ -5,44 +5,7 @@
 from cloudinit import distros
 from cloudinit import ssh_util
 from cloudinit.tests.helpers import (CiTestCase, mock)
-
-
-class MyBaseDistro(distros.Distro):
-    # MyBaseDistro is here to test base Distro class implementations
-
-    def __init__(self, name="basedistro", cfg=None, paths=None):
-        if not cfg:
-            cfg = {}
-        if not paths:
-            paths = {}
-        super(MyBaseDistro, self).__init__(name, cfg, paths)
-
-    def install_packages(self, pkglist):
-        raise NotImplementedError()
-
-    def _write_network(self, settings):
-        raise NotImplementedError()
-
-    def package_command(self, command, args=None, pkgs=None):
-        raise NotImplementedError()
-
-    def update_package_sources(self):
-        raise NotImplementedError()
-
-    def apply_locale(self, locale, out_fn=None):
-        raise NotImplementedError()
-
-    def set_timezone(self, tz):
-        raise NotImplementedError()
-
-    def _read_hostname(self, filename, default=None):
-        raise NotImplementedError()
-
-    def _write_hostname(self, hostname, filename):
-        raise NotImplementedError()
-
-    def _read_system_hostname(self):
-        raise NotImplementedError()
+from tests.unittests.util import abstract_to_concrete
 
 
 @mock.patch("cloudinit.distros.util.system_is_snappy", return_value=False)
@@ -53,7 +16,9 @@ class TestCreateUser(CiTestCase):
 
     def setUp(self):
         super(TestCreateUser, self).setUp()
-        self.dist = MyBaseDistro()
+        self.dist = abstract_to_concrete(distros.Distro)(
+            name='test', cfg=None, paths=None
+        )
 
     def _useradd2call(self, args):
         # return a mock call for the useradd command in args

tests/unittests/test_handler/test_handler_apt_configure_sources_list_v1.py

@@ -9,19 +9,16 @@
 import tempfile
 from unittest import mock
 
-from cloudinit import cloud
-from cloudinit import distros
-from cloudinit import helpers
 from cloudinit import templater
 from cloudinit import subp
 from cloudinit import util
 
 from cloudinit.config import cc_apt_configure
-from cloudinit.sources import DataSourceNone
 
 from cloudinit.distros.debian import Distro
 
 from cloudinit.tests import helpers as t_help
+from tests.unittests.util import get_cloud
 
 LOG = logging.getLogger(__name__)
 
@@ -80,16 +77,6 @@ def setUp(self):
         get_arch.return_value = 'amd64'
         self.addCleanup(apatcher.stop)
 
-    def _get_cloud(self, distro, metadata=None):
-        self.patchUtils(self.new_root)
-        paths = helpers.Paths({})
-        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)
-
     def apt_source_list(self, distro, mirror, mirrorcheck=None):
         """apt_source_list
         Test rendering of a source.list from template for a given distro
@@ -102,7 +89,7 @@ def apt_source_list(self, distro, mirror, mirrorcheck=None):
         else:
             cfg = {'apt_mirror': mirror}
 
-        mycloud = self._get_cloud(distro)
+        mycloud = get_cloud(distro)
 
         with mock.patch.object(util, 'write_file') as mockwf:
             with mock.patch.object(util, 'load_file',
@@ -175,7 +162,7 @@ def test_apt_v1_srcl_ubuntu_mirrorfail(self):
     def test_apt_v1_srcl_custom(self):
         """Test rendering from a custom source.list template"""
         cfg = util.load_yaml(YAML_TEXT_CUSTOM_SL)
-        mycloud = self._get_cloud('ubuntu')
+        mycloud = get_cloud()
 
         # the second mock restores the original subp
         with mock.patch.object(util, 'write_file') as mockwrite:

tests/unittests/test_handler/test_handler_apt_configure_sources_list_v3.py

@@ -3,26 +3,22 @@
 """ test_apt_custom_sources_list
 Test templating of custom sources list
 """
+from contextlib import ExitStack
 import logging
 import os
 import shutil
 import tempfile
 from unittest import mock
 from unittest.mock import call
 
-from cloudinit import cloud
-from cloudinit import distros
-from cloudinit import helpers
 from cloudinit import subp
 from cloudinit import util
-
 from cloudinit.config import cc_apt_configure
-from cloudinit.sources import DataSourceNone
-
 from cloudinit.distros.debian import Distro
-
 from cloudinit.tests import helpers as t_help
 
+from tests.unittests.util import get_cloud
+
 LOG = logging.getLogger(__name__)
 
 TARGET = "/"
@@ -108,37 +104,29 @@ def setUp(self):
         get_arch.return_value = 'amd64'
         self.addCleanup(apatcher.stop)
 
-    def _get_cloud(self, distro, metadata=None):
-        self.patchUtils(self.new_root)
-        paths = helpers.Paths({})
-        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)
-
     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 mock_writefile:
-            with mock.patch.object(util, 'load_file',
-                                   return_value=MOCKED_APT_SRC_LIST
-                                   ) as mock_loadfile:
-                with mock.patch.object(os.path, 'isfile',
-                                       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
+        mycloud = get_cloud(distro)
+
+        with ExitStack() as stack:
+            mock_writefile = stack.enter_context(mock.patch.object(
+                util, 'write_file'))
+            mock_loadfile = stack.enter_context(mock.patch.object(
+                util, 'load_file', return_value=MOCKED_APT_SRC_LIST))
+            mock_isfile = stack.enter_context(mock.patch.object(
+                os.path, 'isfile', return_value=True))
+            stack.enter_context(mock.patch.object(
+                util, 'del_file'))
+            cfg_func = ('cloudinit.config.cc_apt_configure.'
+                        '_should_configure_on_empty_apt')
+            mock_shouldcfg = stack.enter_context(mock.patch(
+                cfg_func, return_value=(cfg_on_empty, 'test')
+            ))
+            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"""
@@ -176,7 +164,7 @@ 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')
+        mycloud = get_cloud()
 
         with mock.patch.object(util, 'write_file') as mock_writefile:
             with mock.patch.object(util, 'system_is_snappy',
@@ -219,7 +207,7 @@ def test_apt_v3_source_list_psm(self):
     def test_apt_v3_srcl_custom(self):
         """test_apt_v3_srcl_custom - Test rendering a custom source template"""
         cfg = util.load_yaml(YAML_TEXT_CUSTOM_SL)
-        mycloud = self._get_cloud('ubuntu')
+        mycloud = get_cloud()
 
         # the second mock restores the original subp
         with mock.patch.object(util, 'write_file') as mockwrite:

tests/unittests/test_handler/test_handler_apt_source_v3.py

@@ -14,18 +14,14 @@
 from unittest import TestCase, mock
 from unittest.mock import call
 
-from cloudinit import cloud
-from cloudinit import distros
 from cloudinit import gpg
-from cloudinit import helpers
 from cloudinit import subp
 from cloudinit import util
-
 from cloudinit.config import cc_apt_configure
-from cloudinit.sources import DataSourceNone
-
 from cloudinit.tests import helpers as t_help
 
+from tests.unittests.util import get_cloud
+
 EXPECTEDKEY = """-----BEGIN PGP PUBLIC KEY BLOCK-----
 Version: GnuPG v1
 
@@ -106,16 +102,6 @@ def _myjoin(self, *args, **kwargs):
         else:
             return self.join(*args, **kwargs)
 
-    def _get_cloud(self, distro, metadata=None):
-        self.patchUtils(self.new_root)
-        paths = helpers.Paths({})
-        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)
-
     def _apt_src_basic(self, filename, cfg):
         """_apt_src_basic
         Test Fix deb source string, has to overwrite mirror conf in params
@@ -587,7 +573,7 @@ def test_apt_v3_mirror_default(self):
         default_mirrors = cc_apt_configure.get_default_mirrors(arch)
         pmir = default_mirrors["PRIMARY"]
         smir = default_mirrors["SECURITY"]
-        mycloud = self._get_cloud('ubuntu')
+        mycloud = get_cloud()
         mirrors = cc_apt_configure.find_apt_mirror_info({}, mycloud, arch)
 
         self.assertEqual(mirrors['MIRROR'],
@@ -659,7 +645,7 @@ def test_apt_v3_mirror_arches_sysdefault(self):
         default_mirrors = cc_apt_configure.get_default_mirrors(arch)
         pmir = default_mirrors["PRIMARY"]
         smir = default_mirrors["SECURITY"]
-        mycloud = self._get_cloud('ubuntu')
+        mycloud = get_cloud()
         cfg = {"primary": [{'arches': ["thisarchdoesntexist_64"],
                             "uri": "notthis"},
                            {'arches': ["thisarchdoesntexist"],
@@ -969,7 +955,7 @@ def test_apt_v3_mirror_search_dns(self, m_get_hostname):
         pmir = "phit"
         smir = "shit"
         arch = 'amd64'
-        mycloud = self._get_cloud('ubuntu')
+        mycloud = get_cloud('ubuntu')
         cfg = {"primary": [{'arches': ["default"],
                             "search_dns": True}],
                "security": [{'arches': ["default"],

tests/unittests/test_handler/test_handler_bootcmd.py

@@ -1,14 +1,13 @@
 # This file is part of cloud-init. See LICENSE file for license information.
+import logging
+import tempfile
 
 from cloudinit.config.cc_bootcmd import handle, schema
-from cloudinit.sources import DataSourceNone
-from cloudinit import (distros, helpers, cloud, subp, util)
+from cloudinit import (subp, util)
 from cloudinit.tests.helpers import (
     CiTestCase, mock, SchemaTestCaseMixin, skipUnlessJsonSchema)
 
-import logging
-import tempfile
-
+from tests.unittests.util import get_cloud
 
 LOG = logging.getLogger(__name__)
 
@@ -39,18 +38,10 @@ def setUp(self):
         self.subp = subp.subp
         self.new_root = self.tmp_dir()
 
-    def _get_cloud(self, distro):
-        paths = helpers.Paths({})
-        cls = distros.fetch(distro)
-        mydist = cls(distro, {}, paths)
-        myds = DataSourceNone.DataSourceNone({}, mydist, paths)
-        paths.datasource = myds
-        return cloud.Cloud(myds, paths, {}, mydist, None)
-
     def test_handler_skip_if_no_bootcmd(self):
         """When the provided config doesn't contain bootcmd, skip it."""
         cfg = {}
-        mycloud = self._get_cloud('ubuntu')
+        mycloud = get_cloud()
         handle('notimportant', cfg, mycloud, LOG, None)
         self.assertIn(
             "Skipping module named notimportant, no 'bootcmd' key",
@@ -59,7 +50,7 @@ def test_handler_skip_if_no_bootcmd(self):
     def test_handler_invalid_command_set(self):
         """Commands which can't be converted to shell will raise errors."""
         invalid_config = {'bootcmd': 1}
-        cc = self._get_cloud('ubuntu')
+        cc = get_cloud()
         with self.assertRaises(TypeError) as context_manager:
             handle('cc_bootcmd', invalid_config, cc, LOG, [])
         self.assertIn('Failed to shellify bootcmd', self.logs.getvalue())
@@ -75,7 +66,7 @@ def test_handler_schema_validation_warns_non_array_type(self):
         invalid content.
         """
         invalid_config = {'bootcmd': 1}
-        cc = self._get_cloud('ubuntu')
+        cc = get_cloud()
         with self.assertRaises(TypeError):
             handle('cc_bootcmd', invalid_config, cc, LOG, [])
         self.assertIn(
@@ -92,7 +83,7 @@ def test_handler_schema_validation_warns_non_array_item_type(self):
         """
         invalid_config = {
             'bootcmd': ['ls /', 20, ['wget', 'http://stuff/blah'], {'a': 'n'}]}
-        cc = self._get_cloud('ubuntu')
+        cc = get_cloud()
         with self.assertRaises(TypeError) as context_manager:
             handle('cc_bootcmd', invalid_config, cc, LOG, [])
         expected_warnings = [
@@ -111,7 +102,7 @@ def test_handler_schema_validation_warns_non_array_item_type(self):
 
     def test_handler_creates_and_runs_bootcmd_script_with_instance_id(self):
         """Valid schema runs a bootcmd script with INSTANCE_ID in the env."""
-        cc = self._get_cloud('ubuntu')
+        cc = get_cloud()
         out_file = self.tmp_path('bootcmd.out', self.new_root)
         my_id = "b6ea0f59-e27d-49c6-9f87-79f19765a425"
         valid_config = {'bootcmd': [
@@ -125,7 +116,7 @@ def test_handler_creates_and_runs_bootcmd_script_with_instance_id(self):
 
     def test_handler_runs_bootcmd_script_with_error(self):
         """When a valid script generates an error, that error is raised."""
-        cc = self._get_cloud('ubuntu')
+        cc = get_cloud()
         valid_config = {'bootcmd': ['exit 1']}  # Script with error
 
         with mock.patch(self._etmpfile_path, FakeExtendedTempFile):