summaryrefslogtreecommitdiff
path: root/tests
diff options
context:
space:
mode:
Diffstat (limited to 'tests')
-rw-r--r--tests/unittests/test_distros/test_create_users.py43
-rw-r--r--tests/unittests/test_handler/test_handler_apt_configure_sources_list_v1.py19
-rw-r--r--tests/unittests/test_handler/test_handler_apt_configure_sources_list_v3.py60
-rw-r--r--tests/unittests/test_handler/test_handler_apt_source_v3.py24
-rw-r--r--tests/unittests/test_handler/test_handler_bootcmd.py29
-rw-r--r--tests/unittests/test_handler/test_handler_ca_certs.py33
-rw-r--r--tests/unittests/test_handler/test_handler_chef.py25
-rw-r--r--tests/unittests/test_handler/test_handler_debug.py29
-rw-r--r--tests/unittests/test_handler/test_handler_landscape.py28
-rw-r--r--tests/unittests/test_handler/test_handler_locale.py41
-rw-r--r--tests/unittests/test_handler/test_handler_lxd.py21
-rw-r--r--tests/unittests/test_handler/test_handler_mcollective.py23
-rw-r--r--tests/unittests/test_handler/test_handler_ntp.py29
-rw-r--r--tests/unittests/test_handler/test_handler_puppet.py111
-rw-r--r--tests/unittests/test_handler/test_handler_runcmd.py33
-rw-r--r--tests/unittests/test_handler/test_handler_seed_random.py46
-rw-r--r--tests/unittests/test_handler/test_handler_timezone.py24
-rw-r--r--tests/unittests/util.py143
18 files changed, 352 insertions, 409 deletions
diff --git a/tests/unittests/test_distros/test_create_users.py b/tests/unittests/test_distros/test_create_users.py
index 021866b7..685f08ba 100644
--- a/tests/unittests/test_distros/test_create_users.py
+++ b/tests/unittests/test_distros/test_create_users.py
@@ -5,44 +5,7 @@ import re
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
diff --git a/tests/unittests/test_handler/test_handler_apt_configure_sources_list_v1.py b/tests/unittests/test_handler/test_handler_apt_configure_sources_list_v1.py
index 369480be..d69916f9 100644
--- a/tests/unittests/test_handler/test_handler_apt_configure_sources_list_v1.py
+++ b/tests/unittests/test_handler/test_handler_apt_configure_sources_list_v1.py
@@ -9,19 +9,16 @@ import shutil
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 @@ class TestAptSourceConfigSourceList(t_help.FilesystemMockingTestCase):
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 @@ class TestAptSourceConfigSourceList(t_help.FilesystemMockingTestCase):
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 @@ class TestAptSourceConfigSourceList(t_help.FilesystemMockingTestCase):
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:
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 b96fd4d4..cd6f9239 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
@@ -3,6 +3,7 @@
""" test_apt_custom_sources_list
Test templating of custom sources list
"""
+from contextlib import ExitStack
import logging
import os
import shutil
@@ -10,19 +11,14 @@ 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 @@ class TestAptSourceConfigSourceList(t_help.FilesystemMockingTestCase):
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 @@ class TestAptSourceConfigSourceList(t_help.FilesystemMockingTestCase):
"""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 @@ class TestAptSourceConfigSourceList(t_help.FilesystemMockingTestCase):
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:
diff --git a/tests/unittests/test_handler/test_handler_apt_source_v3.py b/tests/unittests/test_handler/test_handler_apt_source_v3.py
index 687cfbf1..d4db610f 100644
--- a/tests/unittests/test_handler/test_handler_apt_source_v3.py
+++ b/tests/unittests/test_handler/test_handler_apt_source_v3.py
@@ -14,18 +14,14 @@ import tempfile
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 @@ class TestAptSourceConfig(t_help.FilesystemMockingTestCase):
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 @@ class TestAptSourceConfig(t_help.FilesystemMockingTestCase):
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 @@ class TestAptSourceConfig(t_help.FilesystemMockingTestCase):
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 @@ deb http://ubuntu.com/ubuntu/ xenial-proposed main""")
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"],
diff --git a/tests/unittests/test_handler/test_handler_bootcmd.py b/tests/unittests/test_handler/test_handler_bootcmd.py
index b53d60d4..8cd3a5e1 100644
--- a/tests/unittests/test_handler/test_handler_bootcmd.py
+++ b/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 @@ class TestBootcmd(CiTestCase):
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 @@ class TestBootcmd(CiTestCase):
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 @@ class TestBootcmd(CiTestCase):
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 @@ class TestBootcmd(CiTestCase):
"""
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 @@ class TestBootcmd(CiTestCase):
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 @@ class TestBootcmd(CiTestCase):
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):
diff --git a/tests/unittests/test_handler/test_handler_ca_certs.py b/tests/unittests/test_handler/test_handler_ca_certs.py
index 6e3831ed..2a4ab49e 100644
--- a/tests/unittests/test_handler/test_handler_ca_certs.py
+++ b/tests/unittests/test_handler/test_handler_ca_certs.py
@@ -1,20 +1,19 @@
# This file is part of cloud-init. See LICENSE file for license information.
+import logging
+import shutil
+import tempfile
+import unittest
+from contextlib import ExitStack
+from unittest import mock
-from cloudinit import cloud
from cloudinit import distros
from cloudinit.config import cc_ca_certs
from cloudinit import helpers
from cloudinit import subp
from cloudinit import util
-
from cloudinit.tests.helpers import TestCase
-import logging
-import shutil
-import tempfile
-import unittest
-from contextlib import ExitStack
-from unittest import mock
+from tests.unittests.util import get_cloud
class TestNoConfig(unittest.TestCase):
@@ -56,10 +55,6 @@ class TestConfig(TestCase):
paths = helpers.Paths({})
return cls(kind, {}, paths)
- def _get_cloud(self, kind):
- distro = self._fetch_distro(kind)
- return cloud.Cloud(None, self.paths, None, distro, None)
-
def _mock_init(self):
self.mocks = ExitStack()
self.addCleanup(self.mocks.close)
@@ -81,7 +76,7 @@ class TestConfig(TestCase):
for distro_name in cc_ca_certs.distros:
self._mock_init()
- cloud = self._get_cloud(distro_name)
+ cloud = get_cloud(distro_name)
cc_ca_certs.handle(self.name, config, cloud, self.log, self.args)
self.assertEqual(self.mock_add.call_count, 0)
@@ -94,7 +89,7 @@ class TestConfig(TestCase):
for distro_name in cc_ca_certs.distros:
self._mock_init()
- cloud = self._get_cloud(distro_name)
+ cloud = get_cloud(distro_name)
cc_ca_certs.handle(self.name, config, cloud, self.log, self.args)
self.assertEqual(self.mock_add.call_count, 0)
@@ -107,7 +102,7 @@ class TestConfig(TestCase):
for distro_name in cc_ca_certs.distros:
self._mock_init()
- cloud = self._get_cloud(distro_name)
+ cloud = get_cloud(distro_name)
conf = cc_ca_certs._distro_ca_certs_configs(distro_name)
cc_ca_certs.handle(self.name, config, cloud, self.log, self.args)
@@ -121,7 +116,7 @@ class TestConfig(TestCase):
for distro_name in cc_ca_certs.distros:
self._mock_init()
- cloud = self._get_cloud(distro_name)
+ cloud = get_cloud(distro_name)
conf = cc_ca_certs._distro_ca_certs_configs(distro_name)
cc_ca_certs.handle(self.name, config, cloud, self.log, self.args)
@@ -135,7 +130,7 @@ class TestConfig(TestCase):
for distro_name in cc_ca_certs.distros:
self._mock_init()
- cloud = self._get_cloud(distro_name)
+ cloud = get_cloud(distro_name)
cc_ca_certs.handle(self.name, config, cloud, self.log, self.args)
self.assertEqual(self.mock_add.call_count, 0)
@@ -148,7 +143,7 @@ class TestConfig(TestCase):
for distro_name in cc_ca_certs.distros:
self._mock_init()
- cloud = self._get_cloud(distro_name)
+ cloud = get_cloud(distro_name)
cc_ca_certs.handle(self.name, config, cloud, self.log, self.args)
self.assertEqual(self.mock_add.call_count, 0)
@@ -161,7 +156,7 @@ class TestConfig(TestCase):
for distro_name in cc_ca_certs.distros:
self._mock_init()
- cloud = self._get_cloud(distro_name)
+ cloud = get_cloud(distro_name)
conf = cc_ca_certs._distro_ca_certs_configs(distro_name)
cc_ca_certs.handle(self.name, config, cloud, self.log, self.args)
diff --git a/tests/unittests/test_handler/test_handler_chef.py b/tests/unittests/test_handler/test_handler_chef.py
index 7918c609..0672cebc 100644
--- a/tests/unittests/test_handler/test_handler_chef.py
+++ b/tests/unittests/test_handler/test_handler_chef.py
@@ -5,16 +5,14 @@ import json
import logging
import os
-from cloudinit import cloud
from cloudinit.config import cc_chef
-from cloudinit import distros
-from cloudinit import helpers
-from cloudinit.sources import DataSourceNone
from cloudinit import util
from cloudinit.tests.helpers import (
HttprettyTestCase, FilesystemMockingTestCase, mock, skipIf)
+from tests.unittests.util import get_cloud
+
LOG = logging.getLogger(__name__)
CLIENT_TEMPL = os.path.sep.join(["templates", "chef_client.rb.tmpl"])
@@ -106,19 +104,12 @@ class TestChef(FilesystemMockingTestCase):
super(TestChef, self).setUp()
self.tmp = self.tmp_dir()
- def fetch_cloud(self, distro_kind):
- cls = distros.fetch(distro_kind)
- paths = helpers.Paths({})
- distro = cls(distro_kind, {}, paths)
- ds = DataSourceNone.DataSourceNone({}, distro, paths, None)
- return cloud.Cloud(ds, paths, {}, distro, None)
-
def test_no_config(self):
self.patchUtils(self.tmp)
self.patchOS(self.tmp)
cfg = {}
- cc_chef.handle('chef', cfg, self.fetch_cloud('ubuntu'), LOG, [])
+ cc_chef.handle('chef', cfg, get_cloud(), LOG, [])
for d in cc_chef.CHEF_DIRS:
self.assertFalse(os.path.isdir(d))
@@ -163,7 +154,7 @@ class TestChef(FilesystemMockingTestCase):
'/etc/chef/encrypted_data_bag_secret'
},
}
- cc_chef.handle('chef', cfg, self.fetch_cloud('ubuntu'), LOG, [])
+ cc_chef.handle('chef', cfg, get_cloud(), LOG, [])
for d in cc_chef.CHEF_DIRS:
self.assertTrue(os.path.isdir(d))
c = util.load_file(cc_chef.CHEF_RB_PATH)
@@ -198,7 +189,7 @@ class TestChef(FilesystemMockingTestCase):
}
},
}
- cc_chef.handle('chef', cfg, self.fetch_cloud('ubuntu'), LOG, [])
+ cc_chef.handle('chef', cfg, get_cloud(), LOG, [])
c = util.load_file(cc_chef.CHEF_FB_PATH)
self.assertEqual(
{
@@ -222,7 +213,7 @@ class TestChef(FilesystemMockingTestCase):
'show_time': None,
},
}
- cc_chef.handle('chef', cfg, self.fetch_cloud('ubuntu'), LOG, [])
+ cc_chef.handle('chef', cfg, get_cloud(), LOG, [])
c = util.load_file(cc_chef.CHEF_RB_PATH)
self.assertNotIn('json_attribs', c)
self.assertNotIn('Formatter.show_time', c)
@@ -246,7 +237,7 @@ class TestChef(FilesystemMockingTestCase):
'validation_cert': v_cert
},
}
- cc_chef.handle('chef', cfg, self.fetch_cloud('ubuntu'), LOG, [])
+ cc_chef.handle('chef', cfg, get_cloud(), LOG, [])
content = util.load_file(cc_chef.CHEF_RB_PATH)
self.assertIn(v_path, content)
util.load_file(v_path)
@@ -271,7 +262,7 @@ class TestChef(FilesystemMockingTestCase):
}
util.write_file('/etc/cloud/templates/chef_client.rb.tmpl', tpl_file)
util.write_file(v_path, expected_cert)
- cc_chef.handle('chef', cfg, self.fetch_cloud('ubuntu'), LOG, [])
+ cc_chef.handle('chef', cfg, get_cloud(), LOG, [])
content = util.load_file(cc_chef.CHEF_RB_PATH)
self.assertIn(v_path, content)
util.load_file(v_path)
diff --git a/tests/unittests/test_handler/test_handler_debug.py b/tests/unittests/test_handler/test_handler_debug.py
index 7d43e020..41e9d9bd 100644
--- a/tests/unittests/test_handler/test_handler_debug.py
+++ b/tests/unittests/test_handler/test_handler_debug.py
@@ -1,21 +1,15 @@
# Copyright (C) 2014 Yahoo! Inc.
#
# This file is part of cloud-init. See LICENSE file for license information.
+import logging
+import shutil
+import tempfile
-from cloudinit.config import cc_debug
-
-from cloudinit import cloud
-from cloudinit import distros
-from cloudinit import helpers
from cloudinit import util
-
-from cloudinit.sources import DataSourceNone
-
+from cloudinit.config import cc_debug
from cloudinit.tests.helpers import (FilesystemMockingTestCase, mock)
-import logging
-import shutil
-import tempfile
+from tests.unittests.util import get_cloud
LOG = logging.getLogger(__name__)
@@ -26,16 +20,7 @@ class TestDebug(FilesystemMockingTestCase):
super(TestDebug, self).setUp()
self.new_root = tempfile.mkdtemp()
self.addCleanup(shutil.rmtree, self.new_root)
-
- def _get_cloud(self, distro, metadata=None):
self.patchUtils(self.new_root)
- paths = helpers.Paths({})
- cls = distros.fetch(distro)
- d = cls(distro, {}, paths)
- ds = DataSourceNone.DataSourceNone({}, d, paths)
- if metadata:
- ds.metadata.update(metadata)
- return cloud.Cloud(ds, paths, {}, d, None)
def test_debug_write(self, m_locale):
m_locale.return_value = 'en_US.UTF-8'
@@ -48,7 +33,7 @@ class TestDebug(FilesystemMockingTestCase):
'output': '/var/log/cloud-init-debug.log',
},
}
- cc = self._get_cloud('ubuntu')
+ cc = get_cloud()
cc_debug.handle('cc_debug', cfg, cc, LOG, [])
contents = util.load_file('/var/log/cloud-init-debug.log')
# Some basic sanity tests...
@@ -66,7 +51,7 @@ class TestDebug(FilesystemMockingTestCase):
'output': '/var/log/cloud-init-debug.log',
},
}
- cc = self._get_cloud('ubuntu')
+ cc = get_cloud()
cc_debug.handle('cc_debug', cfg, cc, LOG, [])
self.assertRaises(IOError,
util.load_file, '/var/log/cloud-init-debug.log')
diff --git a/tests/unittests/test_handler/test_handler_landscape.py b/tests/unittests/test_handler/test_handler_landscape.py
index 7d165687..00333985 100644
--- a/tests/unittests/test_handler/test_handler_landscape.py
+++ b/tests/unittests/test_handler/test_handler_landscape.py
@@ -1,14 +1,13 @@
# This file is part of cloud-init. See LICENSE file for license information.
+import logging
+from configobj import ConfigObj
from cloudinit.config import cc_landscape
-from cloudinit import (distros, helpers, cloud, util)
-from cloudinit.sources import DataSourceNone
+from cloudinit import util
from cloudinit.tests.helpers import (FilesystemMockingTestCase, mock,
wrap_and_call)
-from configobj import ConfigObj
-import logging
-
+from tests.unittests.util import get_cloud
LOG = logging.getLogger(__name__)
@@ -22,18 +21,11 @@ class TestLandscape(FilesystemMockingTestCase):
self.new_root = self.tmp_dir()
self.conf = self.tmp_path('client.conf', self.new_root)
self.default_file = self.tmp_path('default_landscape', self.new_root)
-
- def _get_cloud(self, distro):
self.patchUtils(self.new_root)
- paths = helpers.Paths({'templates_dir': self.new_root})
- cls = distros.fetch(distro)
- mydist = cls(distro, {}, paths)
- myds = DataSourceNone.DataSourceNone({}, mydist, paths)
- return cloud.Cloud(myds, paths, {}, mydist, None)
def test_handler_skips_empty_landscape_cloudconfig(self):
"""Empty landscape cloud-config section does no work."""
- mycloud = self._get_cloud('ubuntu')
+ mycloud = get_cloud('ubuntu')
mycloud.distro = mock.MagicMock()
cfg = {'landscape': {}}
cc_landscape.handle('notimportant', cfg, mycloud, LOG, None)
@@ -41,7 +33,7 @@ class TestLandscape(FilesystemMockingTestCase):
def test_handler_error_on_invalid_landscape_type(self):
"""Raise an error when landscape configuraiton option is invalid."""
- mycloud = self._get_cloud('ubuntu')
+ mycloud = get_cloud('ubuntu')
cfg = {'landscape': 'wrongtype'}
with self.assertRaises(RuntimeError) as context_manager:
cc_landscape.handle('notimportant', cfg, mycloud, LOG, None)
@@ -52,7 +44,7 @@ class TestLandscape(FilesystemMockingTestCase):
@mock.patch('cloudinit.config.cc_landscape.subp')
def test_handler_restarts_landscape_client(self, m_subp):
"""handler restarts lansdscape-client after install."""
- mycloud = self._get_cloud('ubuntu')
+ mycloud = get_cloud('ubuntu')
cfg = {'landscape': {'client': {}}}
wrap_and_call(
'cloudinit.config.cc_landscape',
@@ -64,7 +56,7 @@ class TestLandscape(FilesystemMockingTestCase):
def test_handler_installs_client_and_creates_config_file(self):
"""Write landscape client.conf and install landscape-client."""
- mycloud = self._get_cloud('ubuntu')
+ mycloud = get_cloud('ubuntu')
cfg = {'landscape': {'client': {}}}
expected = {'client': {
'log_level': 'info',
@@ -91,7 +83,7 @@ class TestLandscape(FilesystemMockingTestCase):
"""Merge and write options from LSC_CLIENT_CFG_FILE with defaults."""
# Write existing sparse client.conf file
util.write_file(self.conf, '[client]\ncomputer_title = My PC\n')
- mycloud = self._get_cloud('ubuntu')
+ mycloud = get_cloud('ubuntu')
cfg = {'landscape': {'client': {}}}
expected = {'client': {
'log_level': 'info',
@@ -112,7 +104,7 @@ class TestLandscape(FilesystemMockingTestCase):
"""Merge and write options from cloud-config options with defaults."""
# Write empty sparse client.conf file
util.write_file(self.conf, '')
- mycloud = self._get_cloud('ubuntu')
+ mycloud = get_cloud('ubuntu')
cfg = {'landscape': {'client': {'computer_title': 'My PC'}}}
expected = {'client': {
'log_level': 'info',
diff --git a/tests/unittests/test_handler/test_handler_locale.py b/tests/unittests/test_handler/test_handler_locale.py
index 15fe7b23..3c17927e 100644
--- a/tests/unittests/test_handler/test_handler_locale.py
+++ b/tests/unittests/test_handler/test_handler_locale.py
@@ -3,27 +3,21 @@
# Author: Juerg Haefliger <juerg.haefliger@hp.com>
#
# This file is part of cloud-init. See LICENSE file for license information.
-
-from cloudinit.config import cc_locale
-
-from cloudinit import cloud
-from cloudinit import distros
-from cloudinit import helpers
-from cloudinit import util
-
-from cloudinit.sources import DataSourceNoCloud
-
-from cloudinit.tests import helpers as t_help
-
-from configobj import ConfigObj
-
import logging
import os
import shutil
import tempfile
from io import BytesIO
+from configobj import ConfigObj
from unittest import mock
+from cloudinit import util
+from cloudinit.config import cc_locale
+from cloudinit.tests import helpers as t_help
+
+from tests.unittests.util import get_cloud
+
+
LOG = logging.getLogger(__name__)
@@ -33,16 +27,7 @@ class TestLocale(t_help.FilesystemMockingTestCase):
super(TestLocale, self).setUp()
self.new_root = tempfile.mkdtemp()
self.addCleanup(shutil.rmtree, self.new_root)
-
- def _get_cloud(self, distro):
self.patchUtils(self.new_root)
- paths = helpers.Paths({})
-
- cls = distros.fetch(distro)
- d = cls(distro, {}, paths)
- ds = DataSourceNoCloud.DataSourceNoCloud({}, d, paths)
- cc = cloud.Cloud(ds, paths, {}, d, None)
- return cc
def test_set_locale_arch(self):
locale = 'en_GB.UTF-8'
@@ -51,7 +36,7 @@ class TestLocale(t_help.FilesystemMockingTestCase):
'locale': locale,
'locale_configfile': locale_configfile,
}
- cc = self._get_cloud('arch')
+ cc = get_cloud('arch')
with mock.patch('cloudinit.distros.arch.subp.subp') as m_subp:
with mock.patch('cloudinit.distros.arch.LOG.warning') as m_LOG:
@@ -72,7 +57,7 @@ class TestLocale(t_help.FilesystemMockingTestCase):
cfg = {
'locale': 'My.Locale',
}
- cc = self._get_cloud('sles')
+ cc = get_cloud('sles')
cc_locale.handle('cc_locale', cfg, cc, LOG, [])
if cc.distro.uses_systemd():
locale_conf = cc.distro.systemd_locale_conf_fn
@@ -87,7 +72,7 @@ class TestLocale(t_help.FilesystemMockingTestCase):
def test_set_locale_sles_default(self):
cfg = {}
- cc = self._get_cloud('sles')
+ cc = get_cloud('sles')
cc_locale.handle('cc_locale', cfg, cc, LOG, [])
if cc.distro.uses_systemd():
@@ -106,7 +91,7 @@ class TestLocale(t_help.FilesystemMockingTestCase):
locale_conf = os.path.join(self.new_root, "etc/default/locale")
util.write_file(locale_conf, 'LANG="en_US.UTF-8"\n')
cfg = {'locale': 'C.UTF-8'}
- cc = self._get_cloud('ubuntu')
+ cc = get_cloud('ubuntu')
with mock.patch('cloudinit.distros.debian.subp.subp') as m_subp:
with mock.patch('cloudinit.distros.debian.LOCALE_CONF_FN',
locale_conf):
@@ -118,7 +103,7 @@ class TestLocale(t_help.FilesystemMockingTestCase):
def test_locale_rhel_defaults_en_us_utf8(self):
"""Test cc_locale gets en_US.UTF-8 from distro get_locale fallback"""
cfg = {}
- cc = self._get_cloud('rhel')
+ cc = get_cloud('rhel')
update_sysconfig = 'cloudinit.distros.rhel_util.update_sysconfig_file'
with mock.patch.object(cc.distro, 'uses_systemd') as m_use_sd:
m_use_sd.return_value = True
diff --git a/tests/unittests/test_handler/test_handler_lxd.py b/tests/unittests/test_handler/test_handler_lxd.py
index b2181992..ea8b6e90 100644
--- a/tests/unittests/test_handler/test_handler_lxd.py
+++ b/tests/unittests/test_handler/test_handler_lxd.py
@@ -1,11 +1,10 @@
# This file is part of cloud-init. See LICENSE file for license information.
+from unittest import mock
from cloudinit.config import cc_lxd
-from cloudinit.sources import DataSourceNoCloud
-from cloudinit import (distros, helpers, cloud)
from cloudinit.tests import helpers as t_help
-from unittest import mock
+from tests.unittests.util import get_cloud
class TestLxd(t_help.CiTestCase):
@@ -22,18 +21,10 @@ class TestLxd(t_help.CiTestCase):
}
}
- def _get_cloud(self, distro):
- cls = distros.fetch(distro)
- paths = helpers.Paths({})
- d = cls(distro, {}, paths)
- ds = DataSourceNoCloud.DataSourceNoCloud({}, d, paths)
- cc = cloud.Cloud(ds, paths, {}, d, None)
- return cc
-
@mock.patch("cloudinit.config.cc_lxd.maybe_cleanup_default")
@mock.patch("cloudinit.config.cc_lxd.subp")
def test_lxd_init(self, mock_subp, m_maybe_clean):
- cc = self._get_cloud('ubuntu')
+ cc = get_cloud()
mock_subp.which.return_value = True
m_maybe_clean.return_value = None
cc_lxd.handle('cc_lxd', self.lxd_cfg, cc, self.logger, [])
@@ -50,7 +41,7 @@ class TestLxd(t_help.CiTestCase):
@mock.patch("cloudinit.config.cc_lxd.maybe_cleanup_default")
@mock.patch("cloudinit.config.cc_lxd.subp")
def test_lxd_install(self, mock_subp, m_maybe_clean):
- cc = self._get_cloud('ubuntu')
+ cc = get_cloud()
cc.distro = mock.MagicMock()
mock_subp.which.return_value = None
cc_lxd.handle('cc_lxd', self.lxd_cfg, cc, self.logger, [])
@@ -64,7 +55,7 @@ class TestLxd(t_help.CiTestCase):
@mock.patch("cloudinit.config.cc_lxd.maybe_cleanup_default")
@mock.patch("cloudinit.config.cc_lxd.subp")
def test_no_init_does_nothing(self, mock_subp, m_maybe_clean):
- cc = self._get_cloud('ubuntu')
+ cc = get_cloud()
cc.distro = mock.MagicMock()
cc_lxd.handle('cc_lxd', {'lxd': {}}, cc, self.logger, [])
self.assertFalse(cc.distro.install_packages.called)
@@ -74,7 +65,7 @@ class TestLxd(t_help.CiTestCase):
@mock.patch("cloudinit.config.cc_lxd.maybe_cleanup_default")
@mock.patch("cloudinit.config.cc_lxd.subp")
def test_no_lxd_does_nothing(self, mock_subp, m_maybe_clean):
- cc = self._get_cloud('ubuntu')
+ cc = get_cloud()
cc.distro = mock.MagicMock()
cc_lxd.handle('cc_lxd', {'package_update': True}, cc, self.logger, [])
self.assertFalse(cc.distro.install_packages.called)
diff --git a/tests/unittests/test_handler/test_handler_mcollective.py b/tests/unittests/test_handler/test_handler_mcollective.py
index 6891e15f..9cda6fbe 100644
--- a/tests/unittests/test_handler/test_handler_mcollective.py
+++ b/tests/unittests/test_handler/test_handler_mcollective.py
@@ -1,11 +1,4 @@
# This file is part of cloud-init. See LICENSE file for license information.
-
-from cloudinit import (cloud, distros, helpers, util)
-from cloudinit.config import cc_mcollective
-from cloudinit.sources import DataSourceNoCloud
-
-from cloudinit.tests import helpers as t_help
-
import configobj
import logging
import os
@@ -13,6 +6,12 @@ import shutil
import tempfile
from io import BytesIO
+from cloudinit import (util)
+from cloudinit.config import cc_mcollective
+from cloudinit.tests import helpers as t_help
+
+from tests.unittests.util import get_cloud
+
LOG = logging.getLogger(__name__)
@@ -128,18 +127,10 @@ class TestConfig(t_help.FilesystemMockingTestCase):
class TestHandler(t_help.TestCase):
- def _get_cloud(self, distro):
- cls = distros.fetch(distro)
- paths = helpers.Paths({})
- d = cls(distro, {}, paths)
- ds = DataSourceNoCloud.DataSourceNoCloud({}, d, paths)
- cc = cloud.Cloud(ds, paths, {}, d, None)
- return cc
-
@t_help.mock.patch("cloudinit.config.cc_mcollective.subp")
@t_help.mock.patch("cloudinit.config.cc_mcollective.util")
def test_mcollective_install(self, mock_util, mock_subp):
- cc = self._get_cloud('ubuntu')
+ cc = get_cloud()
cc.distro = t_help.mock.MagicMock()
mock_util.load_file.return_value = b""
mycfg = {'mcollective': {'conf': {'loglevel': 'debug'}}}
diff --git a/tests/unittests/test_handler/test_handler_ntp.py b/tests/unittests/test_handler/test_handler_ntp.py
index 6b9c8377..5dfa564f 100644
--- a/tests/unittests/test_handler/test_handler_ntp.py
+++ b/tests/unittests/test_handler/test_handler_ntp.py
@@ -1,17 +1,17 @@
# This file is part of cloud-init. See LICENSE file for license information.
+import copy
+import os
+import shutil
+from functools import partial
+from os.path import dirname
+from cloudinit import (helpers, util)
from cloudinit.config import cc_ntp
-from cloudinit.sources import DataSourceNone
-from cloudinit import (distros, helpers, cloud, util)
-
from cloudinit.tests.helpers import (
CiTestCase, FilesystemMockingTestCase, mock, skipUnlessJsonSchema)
+from tests.unittests.util import get_cloud
-import copy
-import os
-from os.path import dirname
-import shutil
NTP_TEMPLATE = """\
## template: jinja
@@ -39,16 +39,11 @@ class TestNtp(FilesystemMockingTestCase):
self.m_snappy.return_value = False
self.add_patch('cloudinit.util.system_info', 'm_sysinfo')
self.m_sysinfo.return_value = {'dist': ('Distro', '99.1', 'Codename')}
-
- def _get_cloud(self, distro, sys_cfg=None):
- self.new_root = self.reRoot(root=self.new_root)
- paths = helpers.Paths({'templates_dir': self.new_root})
- cls = distros.fetch(distro)
- if not sys_cfg:
- sys_cfg = {}
- mydist = cls(distro, sys_cfg, paths)
- myds = DataSourceNone.DataSourceNone(sys_cfg, mydist, paths)
- return cloud.Cloud(myds, paths, sys_cfg, mydist, None)
+ self.new_root = self.reRoot()
+ self._get_cloud = partial(
+ get_cloud,
+ paths=helpers.Paths({'templates_dir': self.new_root})
+ )
def _get_template_path(self, template_name, distro, basepath=None):
# ntp.conf.{distro} -> ntp.conf.debian.tmpl
diff --git a/tests/unittests/test_handler/test_handler_puppet.py b/tests/unittests/test_handler/test_handler_puppet.py
index 19f72a0c..8d99f535 100644
--- a/tests/unittests/test_handler/test_handler_puppet.py
+++ b/tests/unittests/test_handler/test_handler_puppet.py
@@ -1,13 +1,12 @@
# This file is part of cloud-init. See LICENSE file for license information.
+import logging
+import textwrap
from cloudinit.config import cc_puppet
-from cloudinit.sources import DataSourceNone
-from cloudinit import (distros, helpers, cloud, util)
+from cloudinit import util
from cloudinit.tests.helpers import CiTestCase, HttprettyTestCase, mock
-import logging
-import textwrap
-
+from tests.unittests.util import get_cloud
LOG = logging.getLogger(__name__)
@@ -65,19 +64,13 @@ class TestPuppetHandle(CiTestCase):
self.conf = self.tmp_path('puppet.conf')
self.csr_attributes_path = self.tmp_path(
'csr_attributes.yaml')
-
- def _get_cloud(self, distro):
- paths = helpers.Paths({'templates_dir': self.new_root})
- cls = distros.fetch(distro)
- mydist = cls(distro, {}, paths)
- myds = DataSourceNone.DataSourceNone({}, mydist, paths)
- return cloud.Cloud(myds, paths, {}, mydist, None)
+ self.cloud = get_cloud()
def test_skips_missing_puppet_key_in_cloudconfig(self, m_auto):
"""Cloud-config containing no 'puppet' key is skipped."""
- mycloud = self._get_cloud('ubuntu')
+
cfg = {}
- cc_puppet.handle('notimportant', cfg, mycloud, LOG, None)
+ cc_puppet.handle('notimportant', cfg, self.cloud, LOG, None)
self.assertIn(
"no 'puppet' configuration found", self.logs.getvalue())
self.assertEqual(0, m_auto.call_count)
@@ -85,9 +78,9 @@ class TestPuppetHandle(CiTestCase):
@mock.patch('cloudinit.config.cc_puppet.subp.subp', return_value=("", ""))
def test_puppet_config_starts_puppet_service(self, m_subp, m_auto):
"""Cloud-config 'puppet' configuration starts puppet."""
- mycloud = self._get_cloud('ubuntu')
+
cfg = {'puppet': {'install': False}}
- cc_puppet.handle('notimportant', cfg, mycloud, LOG, None)
+ cc_puppet.handle('notimportant', cfg, self.cloud, LOG, None)
self.assertEqual(1, m_auto.call_count)
self.assertIn(
[mock.call(['service', 'puppet', 'start'], capture=False)],
@@ -96,34 +89,34 @@ class TestPuppetHandle(CiTestCase):
@mock.patch('cloudinit.config.cc_puppet.subp.subp', return_value=("", ""))
def test_empty_puppet_config_installs_puppet(self, m_subp, m_auto):
"""Cloud-config empty 'puppet' configuration installs latest puppet."""
- mycloud = self._get_cloud('ubuntu')
- mycloud.distro = mock.MagicMock()
+
+ self.cloud.distro = mock.MagicMock()
cfg = {'puppet': {}}
- cc_puppet.handle('notimportant', cfg, mycloud, LOG, None)
+ cc_puppet.handle('notimportant', cfg, self.cloud, LOG, None)
self.assertEqual(
[mock.call(('puppet', None))],
- mycloud.distro.install_packages.call_args_list)
+ self.cloud.distro.install_packages.call_args_list)
@mock.patch('cloudinit.config.cc_puppet.subp.subp', return_value=("", ""))
def test_puppet_config_installs_puppet_on_true(self, m_subp, _):
"""Cloud-config with 'puppet' key installs when 'install' is True."""
- mycloud = self._get_cloud('ubuntu')
- mycloud.distro = mock.MagicMock()
+
+ self.cloud.distro = mock.MagicMock()
cfg = {'puppet': {'install': True}}
- cc_puppet.handle('notimportant', cfg, mycloud, LOG, None)
+ cc_puppet.handle('notimportant', cfg, self.cloud, LOG, None)
self.assertEqual(
[mock.call(('puppet', None))],
- mycloud.distro.install_packages.call_args_list)
+ self.cloud.distro.install_packages.call_args_list)
@mock.patch('cloudinit.config.cc_puppet.install_puppet_aio', autospec=True)
@mock.patch('cloudinit.config.cc_puppet.subp.subp', return_value=("", ""))
def test_puppet_config_installs_puppet_aio(self, m_subp, m_aio, _):
"""Cloud-config with 'puppet' key installs
when 'install_type' is 'aio'."""
- mycloud = self._get_cloud('ubuntu')
- mycloud.distro = mock.MagicMock()
+
+ self.cloud.distro = mock.MagicMock()
cfg = {'puppet': {'install': True, 'install_type': 'aio'}}
- cc_puppet.handle('notimportant', cfg, mycloud, LOG, None)
+ cc_puppet.handle('notimportant', cfg, self.cloud, LOG, None)
m_aio.assert_called_with(
cc_puppet.AIO_INSTALL_URL,
None, None, True)
@@ -134,11 +127,11 @@ class TestPuppetHandle(CiTestCase):
m_subp, m_aio, _):
"""Cloud-config with 'puppet' key installs
when 'install_type' is 'aio' and 'version' is specified."""
- mycloud = self._get_cloud('ubuntu')
- mycloud.distro = mock.MagicMock()
+
+ self.cloud.distro = mock.MagicMock()
cfg = {'puppet': {'install': True,
'version': '6.24.0', 'install_type': 'aio'}}
- cc_puppet.handle('notimportant', cfg, mycloud, LOG, None)
+ cc_puppet.handle('notimportant', cfg, self.cloud, LOG, None)
m_aio.assert_called_with(
cc_puppet.AIO_INSTALL_URL,
'6.24.0', None, True)
@@ -150,11 +143,11 @@ class TestPuppetHandle(CiTestCase):
m_aio, _):
"""Cloud-config with 'puppet' key installs
when 'install_type' is 'aio' and 'collection' is specified."""
- mycloud = self._get_cloud('ubuntu')
- mycloud.distro = mock.MagicMock()
+
+ self.cloud.distro = mock.MagicMock()
cfg = {'puppet': {'install': True,
'collection': 'puppet6', 'install_type': 'aio'}}
- cc_puppet.handle('notimportant', cfg, mycloud, LOG, None)
+ cc_puppet.handle('notimportant', cfg, self.cloud, LOG, None)
m_aio.assert_called_with(
cc_puppet.AIO_INSTALL_URL,
None, 'puppet6', True)
@@ -166,13 +159,13 @@ class TestPuppetHandle(CiTestCase):
m_aio, _):
"""Cloud-config with 'puppet' key installs
when 'install_type' is 'aio' and 'aio_install_url' is specified."""
- mycloud = self._get_cloud('ubuntu')
- mycloud.distro = mock.MagicMock()
+
+ self.cloud.distro = mock.MagicMock()
cfg = {'puppet':
{'install': True,
'aio_install_url': 'http://test.url/path/to/script.sh',
'install_type': 'aio'}}
- cc_puppet.handle('notimportant', cfg, mycloud, LOG, None)
+ cc_puppet.handle('notimportant', cfg, self.cloud, LOG, None)
m_aio.assert_called_with(
'http://test.url/path/to/script.sh', None, None, True)
@@ -183,11 +176,11 @@ class TestPuppetHandle(CiTestCase):
m_aio, _):
"""Cloud-config with 'puppet' key installs
when 'install_type' is 'aio' and no cleanup."""
- mycloud = self._get_cloud('ubuntu')
- mycloud.distro = mock.MagicMock()
+
+ self.cloud.distro = mock.MagicMock()
cfg = {'puppet': {'install': True,
'cleanup': False, 'install_type': 'aio'}}
- cc_puppet.handle('notimportant', cfg, mycloud, LOG, None)
+ cc_puppet.handle('notimportant', cfg, self.cloud, LOG, None)
m_aio.assert_called_with(
cc_puppet.AIO_INSTALL_URL,
None, None, False)
@@ -195,13 +188,13 @@ class TestPuppetHandle(CiTestCase):
@mock.patch('cloudinit.config.cc_puppet.subp.subp', return_value=("", ""))
def test_puppet_config_installs_puppet_version(self, m_subp, _):
"""Cloud-config 'puppet' configuration can specify a version."""
- mycloud = self._get_cloud('ubuntu')
- mycloud.distro = mock.MagicMock()
+
+ self.cloud.distro = mock.MagicMock()
cfg = {'puppet': {'version': '3.8'}}
- cc_puppet.handle('notimportant', cfg, mycloud, LOG, None)
+ cc_puppet.handle('notimportant', cfg, self.cloud, LOG, None)
self.assertEqual(
[mock.call(('puppet', '3.8'))],
- mycloud.distro.install_packages.call_args_list)
+ self.cloud.distro.install_packages.call_args_list)
@mock.patch('cloudinit.config.cc_puppet.get_config_value')
@mock.patch('cloudinit.config.cc_puppet.subp.subp', return_value=("", ""))
@@ -213,14 +206,14 @@ class TestPuppetHandle(CiTestCase):
return self.conf
m_default.side_effect = _fake_get_config_value
- mycloud = self._get_cloud('ubuntu')
+
cfg = {
'puppet': {
'conf': {'agent': {'server': 'puppetserver.example.org'}}}}
util.write_file(
self.conf, '[agent]\nserver = origpuppet\nother = 3')
- mycloud.distro = mock.MagicMock()
- cc_puppet.handle('notimportant', cfg, mycloud, LOG, None)
+ self.cloud.distro = mock.MagicMock()
+ cc_puppet.handle('notimportant', cfg, self.cloud, LOG, None)
content = util.load_file(self.conf)
expected = '[agent]\nserver = puppetserver.example.org\nother = 3\n\n'
self.assertEqual(expected, content)
@@ -236,8 +229,8 @@ class TestPuppetHandle(CiTestCase):
return self.csr_attributes_path
m_default.side_effect = _fake_get_config_value
- mycloud = self._get_cloud('ubuntu')
- mycloud.distro = mock.MagicMock()
+
+ self.cloud.distro = mock.MagicMock()
cfg = {
'puppet': {
'csr_attributes': {
@@ -254,7 +247,7 @@ class TestPuppetHandle(CiTestCase):
}
}
}
- cc_puppet.handle('notimportant', cfg, mycloud, LOG, None)
+ cc_puppet.handle('notimportant', cfg, self.cloud, LOG, None)
content = util.load_file(self.csr_attributes_path)
expected = textwrap.dedent("""\
custom_attributes:
@@ -269,9 +262,9 @@ class TestPuppetHandle(CiTestCase):
@mock.patch('cloudinit.config.cc_puppet.subp.subp', return_value=("", ""))
def test_puppet_runs_puppet_if_requested(self, m_subp, m_auto):
"""Run puppet with default args if 'exec' is set to True."""
- mycloud = self._get_cloud('ubuntu')
+
cfg = {'puppet': {'exec': True}}
- cc_puppet.handle('notimportant', cfg, mycloud, LOG, None)
+ cc_puppet.handle('notimportant', cfg, self.cloud, LOG, None)
self.assertEqual(1, m_auto.call_count)
self.assertIn(
[mock.call(['puppet', 'agent', '--test'], capture=False)],
@@ -280,9 +273,9 @@ class TestPuppetHandle(CiTestCase):
@mock.patch('cloudinit.config.cc_puppet.subp.subp', return_value=("", ""))
def test_puppet_starts_puppetd(self, m_subp, m_auto):
"""Run puppet with default args if 'exec' is set to True."""
- mycloud = self._get_cloud('ubuntu')
+
cfg = {'puppet': {}}
- cc_puppet.handle('notimportant', cfg, mycloud, LOG, None)
+ cc_puppet.handle('notimportant', cfg, self.cloud, LOG, None)
self.assertEqual(1, m_auto.call_count)
self.assertIn(
[mock.call(['service', 'puppet', 'start'], capture=False)],
@@ -291,9 +284,9 @@ class TestPuppetHandle(CiTestCase):
@mock.patch('cloudinit.config.cc_puppet.subp.subp', return_value=("", ""))
def test_puppet_skips_puppetd(self, m_subp, m_auto):
"""Run puppet with default args if 'exec' is set to True."""
- mycloud = self._get_cloud('ubuntu')
+
cfg = {'puppet': {'start_service': False}}
- cc_puppet.handle('notimportant', cfg, mycloud, LOG, None)
+ cc_puppet.handle('notimportant', cfg, self.cloud, LOG, None)
self.assertEqual(0, m_auto.call_count)
self.assertNotIn(
[mock.call(['service', 'puppet', 'start'], capture=False)],
@@ -303,10 +296,10 @@ class TestPuppetHandle(CiTestCase):
def test_puppet_runs_puppet_with_args_list_if_requested(self,
m_subp, m_auto):
"""Run puppet with 'exec_args' list if 'exec' is set to True."""
- mycloud = self._get_cloud('ubuntu')
+
cfg = {'puppet': {'exec': True, 'exec_args': [
'--onetime', '--detailed-exitcodes']}}
- cc_puppet.handle('notimportant', cfg, mycloud, LOG, None)
+ cc_puppet.handle('notimportant', cfg, self.cloud, LOG, None)
self.assertEqual(1, m_auto.call_count)
self.assertIn(
[mock.call(
@@ -318,10 +311,10 @@ class TestPuppetHandle(CiTestCase):
def test_puppet_runs_puppet_with_args_string_if_requested(self,
m_subp, m_auto):
"""Run puppet with 'exec_args' string if 'exec' is set to True."""
- mycloud = self._get_cloud('ubuntu')
+
cfg = {'puppet': {'exec': True,
'exec_args': '--onetime --detailed-exitcodes'}}
- cc_puppet.handle('notimportant', cfg, mycloud, LOG, None)
+ cc_puppet.handle('notimportant', cfg, self.cloud, LOG, None)
self.assertEqual(1, m_auto.call_count)
self.assertIn(
[mock.call(
diff --git a/tests/unittests/test_handler/test_handler_runcmd.py b/tests/unittests/test_handler/test_handler_runcmd.py
index c03efa67..672e8093 100644
--- a/tests/unittests/test_handler/test_handler_runcmd.py
+++ b/tests/unittests/test_handler/test_handler_runcmd.py
@@ -1,16 +1,16 @@
# This file is part of cloud-init. See LICENSE file for license information.
+import logging
+import os
+import stat
+from unittest.mock import patch
from cloudinit.config.cc_runcmd import handle, schema
-from cloudinit.sources import DataSourceNone
-from cloudinit import (distros, helpers, cloud, subp, util)
+from cloudinit import (helpers, subp, util)
from cloudinit.tests.helpers import (
CiTestCase, FilesystemMockingTestCase, SchemaTestCaseMixin,
skipUnlessJsonSchema)
-from unittest.mock import patch
-import logging
-import os
-import stat
+from tests.unittests.util import get_cloud
LOG = logging.getLogger(__name__)
@@ -23,20 +23,13 @@ class TestRuncmd(FilesystemMockingTestCase):
super(TestRuncmd, self).setUp()
self.subp = subp.subp
self.new_root = self.tmp_dir()
-
- def _get_cloud(self, distro):
self.patchUtils(self.new_root)
- paths = helpers.Paths({'scripts': self.new_root})
- cls = distros.fetch(distro)
- mydist = cls(distro, {}, paths)
- myds = DataSourceNone.DataSourceNone({}, mydist, paths)
- paths.datasource = myds
- return cloud.Cloud(myds, paths, {}, mydist, None)
+ self.paths = helpers.Paths({'scripts': self.new_root})
def test_handler_skip_if_no_runcmd(self):
"""When the provided config doesn't contain runcmd, skip it."""
cfg = {}
- mycloud = self._get_cloud('ubuntu')
+ mycloud = get_cloud(paths=self.paths)
handle('notimportant', cfg, mycloud, LOG, None)
self.assertIn(
"Skipping module named notimportant, no 'runcmd' key",
@@ -47,7 +40,7 @@ class TestRuncmd(FilesystemMockingTestCase):
"""When shellify fails throw exception"""
cls.side_effect = TypeError("patched shellify")
valid_config = {'runcmd': ['echo 42']}
- cc = self._get_cloud('ubuntu')
+ cc = get_cloud(paths=self.paths)
with self.assertRaises(TypeError) as cm:
with self.allow_subp(['/bin/sh']):
handle('cc_runcmd', valid_config, cc, LOG, None)
@@ -56,7 +49,7 @@ class TestRuncmd(FilesystemMockingTestCase):
def test_handler_invalid_command_set(self):
"""Commands which can't be converted to shell will raise errors."""
invalid_config = {'runcmd': 1}
- cc = self._get_cloud('ubuntu')
+ cc = get_cloud(paths=self.paths)
with self.assertRaises(TypeError) as cm:
handle('cc_runcmd', invalid_config, cc, LOG, [])
self.assertIn(
@@ -72,7 +65,7 @@ class TestRuncmd(FilesystemMockingTestCase):
invalid content.
"""
invalid_config = {'runcmd': 1}
- cc = self._get_cloud('ubuntu')
+ cc = get_cloud(paths=self.paths)
with self.assertRaises(TypeError) as cm:
handle('cc_runcmd', invalid_config, cc, LOG, [])
self.assertIn(
@@ -89,7 +82,7 @@ class TestRuncmd(FilesystemMockingTestCase):
"""
invalid_config = {
'runcmd': ['ls /', 20, ['wget', 'http://stuff/blah'], {'a': 'n'}]}
- cc = self._get_cloud('ubuntu')
+ cc = get_cloud(paths=self.paths)
with self.assertRaises(TypeError) as cm:
handle('cc_runcmd', invalid_config, cc, LOG, [])
expected_warnings = [
@@ -105,7 +98,7 @@ class TestRuncmd(FilesystemMockingTestCase):
def test_handler_write_valid_runcmd_schema_to_file(self):
"""Valid runcmd schema is written to a runcmd shell script."""
valid_config = {'runcmd': [['ls', '/']]}
- cc = self._get_cloud('ubuntu')
+ cc = get_cloud(paths=self.paths)
handle('cc_runcmd', valid_config, cc, LOG, [])
runcmd_file = os.path.join(
self.new_root,
diff --git a/tests/unittests/test_handler/test_handler_seed_random.py b/tests/unittests/test_handler/test_handler_seed_random.py
index 85167f19..2ab153d2 100644
--- a/tests/unittests/test_handler/test_handler_seed_random.py
+++ b/tests/unittests/test_handler/test_handler_seed_random.py
@@ -7,24 +7,17 @@
# Based on test_handler_set_hostname.py
#
# This file is part of cloud-init. See LICENSE file for license information.
-
-from cloudinit.config import cc_seed_random
-
import gzip
+import logging
import tempfile
from io import BytesIO
-from cloudinit import cloud
-from cloudinit import distros
-from cloudinit import helpers
from cloudinit import subp
from cloudinit import util
-
-from cloudinit.sources import DataSourceNone
-
+from cloudinit.config import cc_seed_random
from cloudinit.tests import helpers as t_help
-import logging
+from tests.unittests.util import get_cloud
LOG = logging.getLogger(__name__)
@@ -66,15 +59,6 @@ class TestRandomSeed(t_help.TestCase):
gz_fh.close()
return contents.getvalue()
- def _get_cloud(self, distro, metadata=None):
- paths = helpers.Paths({})
- cls = distros.fetch(distro)
- ubuntu_distro = cls(distro, {}, paths)
- ds = DataSourceNone.DataSourceNone({}, ubuntu_distro, paths)
- if metadata:
- ds.metadata = metadata
- return cloud.Cloud(ds, paths, {}, ubuntu_distro, None)
-
def test_append_random(self):
cfg = {
'random_seed': {
@@ -82,7 +66,7 @@ class TestRandomSeed(t_help.TestCase):
'data': 'tiny-tim-was-here',
}
}
- cc_seed_random.handle('test', cfg, self._get_cloud('ubuntu'), LOG, [])
+ cc_seed_random.handle('test', cfg, get_cloud('ubuntu'), LOG, [])
contents = util.load_file(self._seed_file)
self.assertEqual("tiny-tim-was-here", contents)
@@ -96,7 +80,7 @@ class TestRandomSeed(t_help.TestCase):
}
}
self.assertRaises(IOError, cc_seed_random.handle, 'test', cfg,
- self._get_cloud('ubuntu'), LOG, [])
+ get_cloud('ubuntu'), LOG, [])
def test_append_random_gzip(self):
data = self._compress(b"tiny-toe")
@@ -107,7 +91,7 @@ class TestRandomSeed(t_help.TestCase):
'encoding': 'gzip',
}
}
- cc_seed_random.handle('test', cfg, self._get_cloud('ubuntu'), LOG, [])
+ cc_seed_random.handle('test', cfg, get_cloud('ubuntu'), LOG, [])
contents = util.load_file(self._seed_file)
self.assertEqual("tiny-toe", contents)
@@ -120,7 +104,7 @@ class TestRandomSeed(t_help.TestCase):
'encoding': 'gz',
}
}
- cc_seed_random.handle('test', cfg, self._get_cloud('ubuntu'), LOG, [])
+ cc_seed_random.handle('test', cfg, get_cloud('ubuntu'), LOG, [])
contents = util.load_file(self._seed_file)
self.assertEqual("big-toe", contents)
@@ -133,7 +117,7 @@ class TestRandomSeed(t_help.TestCase):
'encoding': 'base64',
}
}
- cc_seed_random.handle('test', cfg, self._get_cloud('ubuntu'), LOG, [])
+ cc_seed_random.handle('test', cfg, get_cloud('ubuntu'), LOG, [])
contents = util.load_file(self._seed_file)
self.assertEqual("bubbles", contents)
@@ -146,7 +130,7 @@ class TestRandomSeed(t_help.TestCase):
'encoding': 'b64',
}
}
- cc_seed_random.handle('test', cfg, self._get_cloud('ubuntu'), LOG, [])
+ cc_seed_random.handle('test', cfg, get_cloud('ubuntu'), LOG, [])
contents = util.load_file(self._seed_file)
self.assertEqual("kit-kat", contents)
@@ -157,13 +141,13 @@ class TestRandomSeed(t_help.TestCase):
'data': 'tiny-tim-was-here',
}
}
- c = self._get_cloud('ubuntu', {'random_seed': '-so-was-josh'})
+ c = get_cloud('ubuntu', metadata={'random_seed': '-so-was-josh'})
cc_seed_random.handle('test', cfg, c, LOG, [])
contents = util.load_file(self._seed_file)
self.assertEqual('tiny-tim-was-here-so-was-josh', contents)
def test_seed_command_provided_and_available(self):
- c = self._get_cloud('ubuntu', {})
+ c = get_cloud('ubuntu')
self.whichdata = {'pollinate': '/usr/bin/pollinate'}
cfg = {'random_seed': {'command': ['pollinate', '-q']}}
cc_seed_random.handle('test', cfg, c, LOG, [])
@@ -172,7 +156,7 @@ class TestRandomSeed(t_help.TestCase):
self.assertIn(['pollinate', '-q'], subp_args)
def test_seed_command_not_provided(self):
- c = self._get_cloud('ubuntu', {})
+ c = get_cloud('ubuntu')
self.whichdata = {}
cc_seed_random.handle('test', {}, c, LOG, [])
@@ -180,7 +164,7 @@ class TestRandomSeed(t_help.TestCase):
self.assertFalse(self.subp_called)
def test_unavailable_seed_command_and_required_raises_error(self):
- c = self._get_cloud('ubuntu', {})
+ c = get_cloud('ubuntu')
self.whichdata = {}
cfg = {'random_seed': {'command': ['THIS_NO_COMMAND'],
'command_required': True}}
@@ -188,7 +172,7 @@ class TestRandomSeed(t_help.TestCase):
'test', cfg, c, LOG, [])
def test_seed_command_and_required(self):
- c = self._get_cloud('ubuntu', {})
+ c = get_cloud('ubuntu')
self.whichdata = {'foo': 'foo'}
cfg = {'random_seed': {'command_required': True, 'command': ['foo']}}
cc_seed_random.handle('test', cfg, c, LOG, [])
@@ -196,7 +180,7 @@ class TestRandomSeed(t_help.TestCase):
self.assertIn(['foo'], [f['args'] for f in self.subp_called])
def test_file_in_environment_for_command(self):
- c = self._get_cloud('ubuntu', {})
+ c = get_cloud('ubuntu')
self.whichdata = {'foo': 'foo'}
cfg = {'random_seed': {'command_required': True, 'command': ['foo'],
'file': self._seed_file}}
diff --git a/tests/unittests/test_handler/test_handler_timezone.py b/tests/unittests/test_handler/test_handler_timezone.py
index 50c45363..77cdb0c2 100644
--- a/tests/unittests/test_handler/test_handler_timezone.py
+++ b/tests/unittests/test_handler/test_handler_timezone.py
@@ -6,21 +6,19 @@
from cloudinit.config import cc_timezone
-from cloudinit import cloud
-from cloudinit import distros
-from cloudinit import helpers
from cloudinit import util
-from cloudinit.sources import DataSourceNoCloud
-from cloudinit.tests import helpers as t_help
-
-from configobj import ConfigObj
import logging
import shutil
import tempfile
+from configobj import ConfigObj
from io import BytesIO
+from cloudinit.tests import helpers as t_help
+
+from tests.unittests.util import get_cloud
+
LOG = logging.getLogger(__name__)
@@ -29,25 +27,15 @@ class TestTimezone(t_help.FilesystemMockingTestCase):
super(TestTimezone, self).setUp()
self.new_root = tempfile.mkdtemp()
self.addCleanup(shutil.rmtree, self.new_root)
-
- def _get_cloud(self, distro):
self.patchUtils(self.new_root)
self.patchOS(self.new_root)
- paths = helpers.Paths({})
-
- cls = distros.fetch(distro)
- d = cls(distro, {}, paths)
- ds = DataSourceNoCloud.DataSourceNoCloud({}, d, paths)
- cc = cloud.Cloud(ds, paths, {}, d, None)
- return cc
-
def test_set_timezone_sles(self):
cfg = {
'timezone': 'Tatooine/Bestine',
}
- cc = self._get_cloud('sles')
+ cc = get_cloud('sles')
# Create a dummy timezone file
dummy_contents = '0123456789abcdefgh'
diff --git a/tests/unittests/util.py b/tests/unittests/util.py
new file mode 100644
index 00000000..383f5f5c
--- /dev/null
+++ b/tests/unittests/util.py
@@ -0,0 +1,143 @@
+# This file is part of cloud-init. See LICENSE file for license information.
+from cloudinit import cloud, distros, helpers
+from cloudinit.sources.DataSourceNone import DataSourceNone
+
+
+def get_cloud(distro=None, paths=None, sys_cfg=None, metadata=None):
+ """Obtain a "cloud" that can be used for testing.
+
+ Modules take a 'cloud' parameter to call into things that are
+ datasource/distro specific. In most cases, the specifics of this cloud
+ implementation aren't needed to test the module, so provide a fake
+ datasource/distro with stubbed calls to methods that may attempt to
+ read/write files or shell out. If a specific distro is needed, it can
+ be passed in as the distro parameter.
+ """
+ paths = paths or helpers.Paths({})
+ sys_cfg = sys_cfg or {}
+ cls = distros.fetch(distro) if distro else TestingDistro
+ mydist = cls(distro, sys_cfg, paths)
+ myds = DataSourceTesting(sys_cfg, mydist, paths)
+ if metadata:
+ myds.metadata.update(metadata)
+ if paths:
+ paths.datasource = myds
+ return cloud.Cloud(myds, paths, sys_cfg, mydist, None)
+
+
+def abstract_to_concrete(abclass):
+ """Takes an abstract class and returns a concrete version of it."""
+ class concreteCls(abclass):
+ pass
+ concreteCls.__abstractmethods__ = frozenset()
+ return type('DummyConcrete' + abclass.__name__, (concreteCls,), {})
+
+
+class DataSourceTesting(DataSourceNone):
+ def get_hostname(self, fqdn=False, resolve_ip=False, metadata_only=False):
+ return 'hostname'
+
+ def persist_instance_data(self):
+ return True
+
+ @property
+ def fallback_interface(self):
+ return None
+
+ @property
+ def cloud_name(self):
+ return 'testing'
+
+
+class TestingDistro(distros.Distro):
+ # TestingDistro is here to test base Distro class implementations
+ def __init__(self, name="testingdistro", cfg=None, paths=None):
+ if not cfg:
+ cfg = {}
+ if not paths:
+ paths = {}
+ super(TestingDistro, self).__init__(name, cfg, paths)
+
+ def install_packages(self, pkglist):
+ pass
+
+ def set_hostname(self, hostname, fqdn=None):
+ pass
+
+ def uses_systemd(self):
+ return True
+
+ def get_primary_arch(self):
+ return 'i386'
+
+ def get_package_mirror_info(self, arch=None, data_source=None):
+ pass
+
+ def apply_network(self, settings, bring_up=True):
+ return False
+
+ def generate_fallback_config(self):
+ return {}
+
+ def apply_network_config(self, netconfig, bring_up=False) -> bool:
+ return False
+
+ def apply_network_config_names(self, netconfig):
+ pass
+
+ def apply_locale(self, locale, out_fn=None):
+ pass
+
+ def set_timezone(self, tz):
+ pass
+
+ def _read_hostname(self, filename, default=None):
+ raise NotImplementedError()
+
+ def _write_hostname(self, hostname, filename):
+ raise NotImplementedError()
+
+ def _read_system_hostname(self):
+ raise NotImplementedError()
+
+ def update_hostname(self, hostname, fqdn, prev_hostname_fn):
+ pass
+
+ def update_etc_hosts(self, hostname, fqdn):
+ pass
+
+ def add_user(self, name, **kwargs):
+ pass
+
+ def add_snap_user(self, name, **kwargs):
+ return 'snap_user'
+
+ def create_user(self, name, **kwargs):
+ return True
+
+ def lock_passwd(self, name):
+ pass
+
+ def expire_passwd(self, user):
+ pass
+
+ def set_passwd(self, user, passwd, hashed=False):
+ return True
+
+ def ensure_sudo_dir(self, path, sudo_base='/etc/sudoers'):
+ pass
+
+ def write_sudo_rules(self, user, rules, sudo_file=None):
+ pass
+
+ def create_group(self, name, members=None):
+ pass
+
+ def shutdown_command(self, *, mode, delay, message):
+ pass
+
+ def package_command(self, command, args=None, pkgs=None):
+ pass
+
+ def update_package_sources(self):
+ return (True, "yay")