summaryrefslogtreecommitdiff
path: root/tests/unittests
diff options
context:
space:
mode:
Diffstat (limited to 'tests/unittests')
-rw-r--r--tests/unittests/__init__.py0
-rw-r--r--tests/unittests/helpers.py105
-rw-r--r--tests/unittests/test_datasource/__init__.py0
-rw-r--r--tests/unittests/test_datasource/test_configdrive.py31
-rw-r--r--tests/unittests/test_distros/__init__.py0
-rw-r--r--tests/unittests/test_distros/test_generic.py32
-rw-r--r--tests/unittests/test_distros/test_user_data_normalize.py52
-rw-r--r--tests/unittests/test_filters/__init__.py0
-rw-r--r--tests/unittests/test_filters/test_launch_index.py4
-rw-r--r--tests/unittests/test_handler/__init__.py0
-rw-r--r--tests/unittests/test_handler/test_handler_ca_certs.py18
-rw-r--r--tests/unittests/test_handler/test_handler_set_hostname.py58
-rw-r--r--tests/unittests/test_handler/test_handler_yum_add_repo.py68
-rw-r--r--tests/unittests/test_merging.py62
-rw-r--r--tests/unittests/test_runs/__init__.py0
-rw-r--r--tests/unittests/test_runs/test_merge_run.py52
-rw-r--r--tests/unittests/test_runs/test_simple_run.py80
-rw-r--r--tests/unittests/test_util.py59
18 files changed, 534 insertions, 87 deletions
diff --git a/tests/unittests/__init__.py b/tests/unittests/__init__.py
new file mode 100644
index 00000000..e69de29b
--- /dev/null
+++ b/tests/unittests/__init__.py
diff --git a/tests/unittests/helpers.py b/tests/unittests/helpers.py
index d0f09e70..e8080668 100644
--- a/tests/unittests/helpers.py
+++ b/tests/unittests/helpers.py
@@ -3,6 +3,38 @@ import os
from mocker import MockerTestCase
from cloudinit import helpers as ch
+from cloudinit import util
+
+import shutil
+
+
+# Makes the old path start
+# with new base instead of whatever
+# it previously had
+def rebase_path(old_path, new_base):
+ if old_path.startswith(new_base):
+ # Already handled...
+ return old_path
+ # Retarget the base of that path
+ # to the new base instead of the
+ # old one...
+ path = os.path.join(new_base, old_path.lstrip("/"))
+ path = os.path.abspath(path)
+ return path
+
+
+# Can work on anything that takes a path as arguments
+def retarget_many_wrapper(new_base, am, old_func):
+ def wrapper(*args, **kwds):
+ n_args = list(args)
+ nam = am
+ if am == -1:
+ nam = len(n_args)
+ for i in range(0, nam):
+ path = args[i]
+ n_args[i] = rebase_path(path, new_base)
+ return old_func(*n_args, **kwds)
+ return wrapper
class ResourceUsingTestCase(MockerTestCase):
@@ -40,3 +72,76 @@ class ResourceUsingTestCase(MockerTestCase):
'templates_dir': self.resourceLocation(),
})
return cp
+
+
+class FilesystemMockingTestCase(ResourceUsingTestCase):
+ def __init__(self, methodName="runTest"):
+ ResourceUsingTestCase.__init__(self, methodName)
+ self.patched_funcs = []
+
+ def replicateTestRoot(self, example_root, target_root):
+ real_root = self.resourceLocation()
+ real_root = os.path.join(real_root, 'roots', example_root)
+ for (dir_path, _dirnames, filenames) in os.walk(real_root):
+ real_path = dir_path
+ make_path = rebase_path(real_path[len(real_root):], target_root)
+ util.ensure_dir(make_path)
+ for f in filenames:
+ real_path = util.abs_join(real_path, f)
+ make_path = util.abs_join(make_path, f)
+ shutil.copy(real_path, make_path)
+
+ def tearDown(self):
+ self.restore()
+ ResourceUsingTestCase.tearDown(self)
+
+ def restore(self):
+ for (mod, f, func) in self.patched_funcs:
+ setattr(mod, f, func)
+ self.patched_funcs = []
+
+ def patchUtils(self, new_root):
+ patch_funcs = {
+ util: [('write_file', 1),
+ ('append_file', 1),
+ ('load_file', 1),
+ ('ensure_dir', 1),
+ ('chmod', 1),
+ ('delete_dir_contents', 1),
+ ('del_file', 1),
+ ('sym_link', -1)],
+ }
+ for (mod, funcs) in patch_funcs.items():
+ for (f, am) in funcs:
+ func = getattr(mod, f)
+ trap_func = retarget_many_wrapper(new_root, am, func)
+ setattr(mod, f, trap_func)
+ self.patched_funcs.append((mod, f, func))
+
+ # Handle subprocess calls
+ func = getattr(util, 'subp')
+
+ def nsubp(*_args, **_kwargs):
+ return ('', '')
+
+ setattr(util, 'subp', nsubp)
+ self.patched_funcs.append((util, 'subp', func))
+
+ def null_func(*_args, **_kwargs):
+ return None
+
+ for f in ['chownbyid', 'chownbyname']:
+ func = getattr(util, f)
+ setattr(util, f, null_func)
+ self.patched_funcs.append((util, f, func))
+
+ def patchOS(self, new_root):
+ patch_funcs = {
+ os.path: ['isfile', 'exists', 'islink', 'isdir'],
+ }
+ for (mod, funcs) in patch_funcs.items():
+ for f in funcs:
+ func = getattr(mod, f)
+ trap_func = retarget_many_wrapper(new_root, 1, func)
+ setattr(mod, f, trap_func)
+ self.patched_funcs.append((mod, f, func))
diff --git a/tests/unittests/test_datasource/__init__.py b/tests/unittests/test_datasource/__init__.py
new file mode 100644
index 00000000..e69de29b
--- /dev/null
+++ b/tests/unittests/test_datasource/__init__.py
diff --git a/tests/unittests/test_datasource/test_configdrive.py b/tests/unittests/test_datasource/test_configdrive.py
index 4fa13db8..aa5b98ed 100644
--- a/tests/unittests/test_datasource/test_configdrive.py
+++ b/tests/unittests/test_datasource/test_configdrive.py
@@ -6,11 +6,10 @@ import os.path
import mocker
from mocker import MockerTestCase
-from cloudinit.sources import DataSourceConfigDrive as ds
+from cloudinit import helpers
from cloudinit import settings
+from cloudinit.sources import DataSourceConfigDrive as ds
from cloudinit import util
-from cloudinit import helpers
-
PUBKEY = u'ssh-rsa AAAAB3NzaC1....sIkJhq8wdX+4I3A4cYbYP ubuntu@server-460\n'
@@ -289,6 +288,32 @@ class TestConfigDriveDataSource(MockerTestCase):
finally:
util.find_devs_with = orig_find_devs_with
+ def test_pubkeys_v2(self):
+ """Verify that public-keys work in config-drive-v2."""
+ populate_dir(self.tmp, CFG_DRIVE_FILES_V2)
+ myds = cfg_ds_from_dir(self.tmp)
+ self.assertEqual(myds.get_public_ssh_keys(),
+ [OSTACK_META['public_keys']['mykey']])
+
+
+def cfg_ds_from_dir(seed_d):
+ found = ds.read_config_drive_dir(seed_d)
+ cfg_ds = ds.DataSourceConfigDrive(settings.CFG_BUILTIN, None,
+ helpers.Paths({}))
+ populate_ds_from_read_config(cfg_ds, seed_d, found)
+ return cfg_ds
+
+
+def populate_ds_from_read_config(cfg_ds, source, results):
+ """Patch the DataSourceConfigDrive from the results of
+ read_config_drive_dir hopefully in line with what it would have
+ if cfg_ds.get_data had been successfully called"""
+ cfg_ds.source = source
+ cfg_ds.metadata = results.get('metadata')
+ cfg_ds.ec2_metadata = results.get('ec2-metadata')
+ cfg_ds.userdata_raw = results.get('userdata')
+ cfg_ds.version = results.get('cfgdrive_ver')
+
def populate_dir(seed_dir, files):
for (name, content) in files.iteritems():
diff --git a/tests/unittests/test_distros/__init__.py b/tests/unittests/test_distros/__init__.py
new file mode 100644
index 00000000..e69de29b
--- /dev/null
+++ b/tests/unittests/test_distros/__init__.py
diff --git a/tests/unittests/test_distros/test_generic.py b/tests/unittests/test_distros/test_generic.py
index 2df4c2f0..704699b5 100644
--- a/tests/unittests/test_distros/test_generic.py
+++ b/tests/unittests/test_distros/test_generic.py
@@ -1,6 +1,9 @@
-from mocker import MockerTestCase
-
from cloudinit import distros
+from cloudinit import util
+
+from tests.unittests import helpers
+
+import os
unknown_arch_info = {
'arches': ['default'],
@@ -27,7 +30,7 @@ gpmi = distros._get_package_mirror_info # pylint: disable=W0212
gapmi = distros._get_arch_package_mirror_info # pylint: disable=W0212
-class TestGenericDistro(MockerTestCase):
+class TestGenericDistro(helpers.FilesystemMockingTestCase):
def return_first(self, mlist):
if not mlist:
@@ -52,6 +55,29 @@ class TestGenericDistro(MockerTestCase):
# Make a temp directoy for tests to use.
self.tmp = self.makeDir()
+ def test_sudoers_ensure_new(self):
+ cls = distros.fetch("ubuntu")
+ d = cls("ubuntu", {}, None)
+ self.patchOS(self.tmp)
+ self.patchUtils(self.tmp)
+ d.ensure_sudo_dir("/b")
+ contents = util.load_file("/etc/sudoers")
+ self.assertIn("includedir /b", contents)
+ self.assertTrue(os.path.isdir("/b"))
+
+ def test_sudoers_ensure_append(self):
+ cls = distros.fetch("ubuntu")
+ d = cls("ubuntu", {}, None)
+ self.patchOS(self.tmp)
+ self.patchUtils(self.tmp)
+ util.write_file("/etc/sudoers", "josh, josh\n")
+ d.ensure_sudo_dir("/b")
+ contents = util.load_file("/etc/sudoers")
+ self.assertIn("includedir /b", contents)
+ self.assertTrue(os.path.isdir("/b"))
+ self.assertIn("josh", contents)
+ self.assertEquals(2, contents.count("josh"))
+
def test_arch_package_mirror_info_unknown(self):
"""for an unknown arch, we should get back that with arch 'default'."""
arch_mirrors = gapmi(package_mirrors, arch="unknown")
diff --git a/tests/unittests/test_distros/test_user_data_normalize.py b/tests/unittests/test_distros/test_user_data_normalize.py
index 9d6fb996..5d9d4311 100644
--- a/tests/unittests/test_distros/test_user_data_normalize.py
+++ b/tests/unittests/test_distros/test_user_data_normalize.py
@@ -4,24 +4,54 @@ from cloudinit import distros
from cloudinit import helpers
from cloudinit import settings
+bcfg = {
+ 'name': 'bob',
+ 'plain_text_passwd': 'ubuntu',
+ 'home': "/home/ubuntu",
+ 'shell': "/bin/bash",
+ 'lock_passwd': True,
+ 'gecos': "Ubuntu",
+ 'groups': ["foo"]
+}
+
class TestUGNormalize(MockerTestCase):
- def _make_distro(self, dtype, def_user=None, def_groups=None):
+ def _make_distro(self, dtype, def_user=None):
cfg = dict(settings.CFG_BUILTIN)
cfg['system_info']['distro'] = dtype
paths = helpers.Paths(cfg['system_info']['paths'])
distro_cls = distros.fetch(dtype)
- distro = distro_cls(dtype, cfg['system_info'], paths)
if def_user:
- distro.default_user = def_user
- if def_groups:
- distro.default_user_groups = def_groups
+ cfg['system_info']['default_user'] = def_user.copy()
+ distro = distro_cls(dtype, cfg['system_info'], paths)
return distro
def _norm(self, cfg, distro):
return distros.normalize_users_groups(cfg, distro)
+ def test_group_dict(self):
+ distro = self._make_distro('ubuntu')
+ g = {'groups': [
+ {
+ 'ubuntu': ['foo', 'bar'],
+ 'bob': 'users',
+ },
+ 'cloud-users',
+ {
+ 'bob': 'users2',
+ },
+ ]
+ }
+ (_users, groups) = self._norm(g, distro)
+ self.assertIn('ubuntu', groups)
+ ub_members = groups['ubuntu']
+ self.assertEquals(sorted(['foo', 'bar']), sorted(ub_members))
+ self.assertIn('bob', groups)
+ b_members = groups['bob']
+ self.assertEquals(sorted(['users', 'users2']),
+ sorted(b_members))
+
def test_basic_groups(self):
distro = self._make_distro('ubuntu')
ug_cfg = {
@@ -71,7 +101,7 @@ class TestUGNormalize(MockerTestCase):
self.assertEquals({}, users)
def test_users_simple_dict(self):
- distro = self._make_distro('ubuntu', 'bob')
+ distro = self._make_distro('ubuntu', bcfg)
ug_cfg = {
'users': {
'default': True,
@@ -95,7 +125,7 @@ class TestUGNormalize(MockerTestCase):
self.assertIn('bob', users)
def test_users_simple_dict_no(self):
- distro = self._make_distro('ubuntu', 'bob')
+ distro = self._make_distro('ubuntu', bcfg)
ug_cfg = {
'users': {
'default': False,
@@ -137,7 +167,7 @@ class TestUGNormalize(MockerTestCase):
self.assertEquals({'default': False}, users['bob'])
def test_users_old_user(self):
- distro = self._make_distro('ubuntu', 'bob')
+ distro = self._make_distro('ubuntu', bcfg)
ug_cfg = {
'user': 'zetta',
'users': 'default'
@@ -185,7 +215,7 @@ class TestUGNormalize(MockerTestCase):
self.assertEquals({}, groups)
def test_users_dict_default_additional(self):
- distro = self._make_distro('ubuntu', 'bob')
+ distro = self._make_distro('ubuntu', bcfg)
ug_cfg = {
'users': [
{'name': 'default', 'blah': True}
@@ -201,7 +231,7 @@ class TestUGNormalize(MockerTestCase):
users['bob']['default'])
def test_users_dict_extract(self):
- distro = self._make_distro('ubuntu', 'bob')
+ distro = self._make_distro('ubuntu', bcfg)
ug_cfg = {
'users': [
'default',
@@ -228,7 +258,7 @@ class TestUGNormalize(MockerTestCase):
self.assertEquals(config, expected_config)
def test_users_dict_default(self):
- distro = self._make_distro('ubuntu', 'bob')
+ distro = self._make_distro('ubuntu', bcfg)
ug_cfg = {
'users': [
'default',
diff --git a/tests/unittests/test_filters/__init__.py b/tests/unittests/test_filters/__init__.py
new file mode 100644
index 00000000..e69de29b
--- /dev/null
+++ b/tests/unittests/test_filters/__init__.py
diff --git a/tests/unittests/test_filters/test_launch_index.py b/tests/unittests/test_filters/test_launch_index.py
index 7ca7cbb6..773bb312 100644
--- a/tests/unittests/test_filters/test_launch_index.py
+++ b/tests/unittests/test_filters/test_launch_index.py
@@ -1,6 +1,6 @@
import copy
-import helpers as th
+from tests.unittests import helpers
import itertools
@@ -18,7 +18,7 @@ def count_messages(root):
return am
-class TestLaunchFilter(th.ResourceUsingTestCase):
+class TestLaunchFilter(helpers.ResourceUsingTestCase):
def assertCounts(self, message, expected_counts):
orig_message = copy.deepcopy(message)
diff --git a/tests/unittests/test_handler/__init__.py b/tests/unittests/test_handler/__init__.py
new file mode 100644
index 00000000..e69de29b
--- /dev/null
+++ b/tests/unittests/test_handler/__init__.py
diff --git a/tests/unittests/test_handler/test_handler_ca_certs.py b/tests/unittests/test_handler/test_handler_ca_certs.py
index d3df5c50..d73c9fa9 100644
--- a/tests/unittests/test_handler/test_handler_ca_certs.py
+++ b/tests/unittests/test_handler/test_handler_ca_certs.py
@@ -77,7 +77,7 @@ class TestConfig(MockerTestCase):
"""Test that a single cert gets passed to add_ca_certs."""
config = {"ca-certs": {"trusted": ["CERT1"]}}
- self.mock_add(self.paths, ["CERT1"])
+ self.mock_add(["CERT1"])
self.mock_update()
self.mocker.replay()
@@ -87,7 +87,7 @@ class TestConfig(MockerTestCase):
"""Test that multiple certs get passed to add_ca_certs."""
config = {"ca-certs": {"trusted": ["CERT1", "CERT2"]}}
- self.mock_add(self.paths, ["CERT1", "CERT2"])
+ self.mock_add(["CERT1", "CERT2"])
self.mock_update()
self.mocker.replay()
@@ -97,7 +97,7 @@ class TestConfig(MockerTestCase):
"""Test remove_defaults works as expected."""
config = {"ca-certs": {"remove-defaults": True}}
- self.mock_remove(self.paths)
+ self.mock_remove()
self.mock_update()
self.mocker.replay()
@@ -116,8 +116,8 @@ class TestConfig(MockerTestCase):
"""Test remove_defaults is not called when config value is False."""
config = {"ca-certs": {"remove-defaults": True, "trusted": ["CERT1"]}}
- self.mock_remove(self.paths)
- self.mock_add(self.paths, ["CERT1"])
+ self.mock_remove()
+ self.mock_add(["CERT1"])
self.mock_update()
self.mocker.replay()
@@ -136,7 +136,7 @@ class TestAddCaCerts(MockerTestCase):
"""Test that no certificate are written if not provided."""
self.mocker.replace(util.write_file, passthrough=False)
self.mocker.replay()
- cc_ca_certs.add_ca_certs(self.paths, [])
+ cc_ca_certs.add_ca_certs([])
def test_single_cert(self):
"""Test adding a single certificate to the trusted CAs."""
@@ -149,7 +149,7 @@ class TestAddCaCerts(MockerTestCase):
"\ncloud-init-ca-certs.crt", omode="ab")
self.mocker.replay()
- cc_ca_certs.add_ca_certs(self.paths, [cert])
+ cc_ca_certs.add_ca_certs([cert])
def test_multiple_certs(self):
"""Test adding multiple certificates to the trusted CAs."""
@@ -163,7 +163,7 @@ class TestAddCaCerts(MockerTestCase):
"\ncloud-init-ca-certs.crt", omode="ab")
self.mocker.replay()
- cc_ca_certs.add_ca_certs(self.paths, certs)
+ cc_ca_certs.add_ca_certs(certs)
class TestUpdateCaCerts(MockerTestCase):
@@ -198,4 +198,4 @@ class TestRemoveDefaultCaCerts(MockerTestCase):
"ca-certificates ca-certificates/trust_new_crts select no")
self.mocker.replay()
- cc_ca_certs.remove_default_ca_certs(self.paths)
+ cc_ca_certs.remove_default_ca_certs()
diff --git a/tests/unittests/test_handler/test_handler_set_hostname.py b/tests/unittests/test_handler/test_handler_set_hostname.py
new file mode 100644
index 00000000..a1aba62f
--- /dev/null
+++ b/tests/unittests/test_handler/test_handler_set_hostname.py
@@ -0,0 +1,58 @@
+from cloudinit.config import cc_set_hostname
+
+from cloudinit import cloud
+from cloudinit import distros
+from cloudinit import helpers
+from cloudinit import util
+
+from tests.unittests import helpers as t_help
+
+import logging
+
+from StringIO import StringIO
+
+from configobj import ConfigObj
+
+LOG = logging.getLogger(__name__)
+
+
+class TestHostname(t_help.FilesystemMockingTestCase):
+ def setUp(self):
+ super(TestHostname, self).setUp()
+ self.tmp = self.makeDir(prefix="unittest_")
+
+ def _fetch_distro(self, kind):
+ cls = distros.fetch(kind)
+ paths = helpers.Paths({})
+ return cls(kind, {}, paths)
+
+ def test_write_hostname_rhel(self):
+ cfg = {
+ 'hostname': 'blah.blah.blah.yahoo.com',
+ }
+ distro = self._fetch_distro('rhel')
+ paths = helpers.Paths({})
+ ds = None
+ cc = cloud.Cloud(ds, paths, {}, distro, None)
+ self.patchUtils(self.tmp)
+ self.patchOS(self.tmp)
+ cc_set_hostname.handle('cc_set_hostname',
+ cfg, cc, LOG, [])
+ contents = util.load_file("/etc/sysconfig/network")
+ n_cfg = ConfigObj(StringIO(contents))
+ self.assertEquals({'HOSTNAME': 'blah.blah.blah.yahoo.com'},
+ dict(n_cfg))
+
+ def test_write_hostname_debian(self):
+ cfg = {
+ 'hostname': 'blah.blah.blah.yahoo.com',
+ }
+ distro = self._fetch_distro('debian')
+ paths = helpers.Paths({})
+ ds = None
+ cc = cloud.Cloud(ds, paths, {}, distro, None)
+ self.patchUtils(self.tmp)
+ cc_set_hostname.handle('cc_set_hostname',
+ cfg, cc, LOG, [])
+ contents = util.load_file("/etc/hostname")
+ self.assertEquals('blah', contents.strip())
diff --git a/tests/unittests/test_handler/test_handler_yum_add_repo.py b/tests/unittests/test_handler/test_handler_yum_add_repo.py
new file mode 100644
index 00000000..8df592f9
--- /dev/null
+++ b/tests/unittests/test_handler/test_handler_yum_add_repo.py
@@ -0,0 +1,68 @@
+from cloudinit import helpers
+from cloudinit import util
+
+from cloudinit.config import cc_yum_add_repo
+
+from tests.unittests import helpers
+
+import logging
+
+from StringIO import StringIO
+
+import configobj
+
+LOG = logging.getLogger(__name__)
+
+
+class TestConfig(helpers.FilesystemMockingTestCase):
+ def setUp(self):
+ super(TestConfig, self).setUp()
+ self.tmp = self.makeDir(prefix="unittest_")
+
+ def test_bad_config(self):
+ cfg = {
+ 'yum_repos': {
+ 'epel-testing': {
+ 'name': 'Extra Packages for Enterprise Linux 5 - Testing',
+ # Missing this should cause the repo not to be written
+ #'baseurl': 'http://blah.org/pub/epel/testing/5/$basearch',
+ 'enabled': False,
+ 'gpgcheck': True,
+ 'gpgkey': 'file:///etc/pki/rpm-gpg/RPM-GPG-KEY-EPEL',
+ 'failovermethod': 'priority',
+ },
+ },
+ }
+ self.patchUtils(self.tmp)
+ cc_yum_add_repo.handle('yum_add_repo', cfg, None, LOG, [])
+ self.assertRaises(IOError, util.load_file,
+ "/etc/yum.repos.d/epel_testing.repo")
+
+ def test_write_config(self):
+ cfg = {
+ 'yum_repos': {
+ 'epel-testing': {
+ 'name': 'Extra Packages for Enterprise Linux 5 - Testing',
+ 'baseurl': 'http://blah.org/pub/epel/testing/5/$basearch',
+ 'enabled': False,
+ 'gpgcheck': True,
+ 'gpgkey': 'file:///etc/pki/rpm-gpg/RPM-GPG-KEY-EPEL',
+ 'failovermethod': 'priority',
+ },
+ },
+ }
+ self.patchUtils(self.tmp)
+ cc_yum_add_repo.handle('yum_add_repo', cfg, None, LOG, [])
+ contents = util.load_file("/etc/yum.repos.d/epel_testing.repo")
+ contents = configobj.ConfigObj(StringIO(contents))
+ expected = {
+ 'epel_testing': {
+ 'name': 'Extra Packages for Enterprise Linux 5 - Testing',
+ 'failovermethod': 'priority',
+ 'gpgkey': 'file:///etc/pki/rpm-gpg/RPM-GPG-KEY-EPEL',
+ 'enabled': '0',
+ 'baseurl': 'http://blah.org/pub/epel/testing/5/$basearch',
+ 'gpgcheck': '1',
+ }
+ }
+ self.assertEquals(expected, dict(contents))
diff --git a/tests/unittests/test_merging.py b/tests/unittests/test_merging.py
new file mode 100644
index 00000000..0037b966
--- /dev/null
+++ b/tests/unittests/test_merging.py
@@ -0,0 +1,62 @@
+from mocker import MockerTestCase
+
+from cloudinit import util
+
+
+class TestMergeDict(MockerTestCase):
+ def test_simple_merge(self):
+ """Test simple non-conflict merge."""
+ source = {"key1": "value1"}
+ candidate = {"key2": "value2"}
+ result = util.mergedict(source, candidate)
+ self.assertEqual({"key1": "value1", "key2": "value2"}, result)
+
+ def test_nested_merge(self):
+ """Test nested merge."""
+ source = {"key1": {"key1.1": "value1.1"}}
+ candidate = {"key1": {"key1.2": "value1.2"}}
+ result = util.mergedict(source, candidate)
+ self.assertEqual(
+ {"key1": {"key1.1": "value1.1", "key1.2": "value1.2"}}, result)
+
+ def test_merge_does_not_override(self):
+ """Test that candidate doesn't override source."""
+ source = {"key1": "value1", "key2": "value2"}
+ candidate = {"key1": "value2", "key2": "NEW VALUE"}
+ result = util.mergedict(source, candidate)
+ self.assertEqual(source, result)
+
+ def test_empty_candidate(self):
+ """Test empty candidate doesn't change source."""
+ source = {"key": "value"}
+ candidate = {}
+ result = util.mergedict(source, candidate)
+ self.assertEqual(source, result)
+
+ def test_empty_source(self):
+ """Test empty source is replaced by candidate."""
+ source = {}
+ candidate = {"key": "value"}
+ result = util.mergedict(source, candidate)
+ self.assertEqual(candidate, result)
+
+ def test_non_dict_candidate(self):
+ """Test non-dict candidate is discarded."""
+ source = {"key": "value"}
+ candidate = "not a dict"
+ result = util.mergedict(source, candidate)
+ self.assertEqual(source, result)
+
+ def test_non_dict_source(self):
+ """Test non-dict source is not modified with a dict candidate."""
+ source = "not a dict"
+ candidate = {"key": "value"}
+ result = util.mergedict(source, candidate)
+ self.assertEqual(source, result)
+
+ def test_neither_dict(self):
+ """Test if neither candidate or source is dict source wins."""
+ source = "source"
+ candidate = "candidate"
+ result = util.mergedict(source, candidate)
+ self.assertEqual(source, result)
diff --git a/tests/unittests/test_runs/__init__.py b/tests/unittests/test_runs/__init__.py
new file mode 100644
index 00000000..e69de29b
--- /dev/null
+++ b/tests/unittests/test_runs/__init__.py
diff --git a/tests/unittests/test_runs/test_merge_run.py b/tests/unittests/test_runs/test_merge_run.py
new file mode 100644
index 00000000..d9c3a455
--- /dev/null
+++ b/tests/unittests/test_runs/test_merge_run.py
@@ -0,0 +1,52 @@
+import os
+
+from tests.unittests import helpers
+
+from cloudinit.settings import (PER_INSTANCE)
+from cloudinit import stages
+from cloudinit import util
+
+
+class TestMergeRun(helpers.FilesystemMockingTestCase):
+ def _patchIn(self, root):
+ self.restore()
+ self.patchOS(root)
+ self.patchUtils(root)
+
+ def test_none_ds(self):
+ new_root = self.makeDir()
+ self.replicateTestRoot('simple_ubuntu', new_root)
+ cfg = {
+ 'datasource_list': ['None'],
+ 'cloud_init_modules': ['write-files'],
+ }
+ ud = self.readResource('user_data.1.txt')
+ cloud_cfg = util.yaml_dumps(cfg)
+ util.ensure_dir(os.path.join(new_root, 'etc', 'cloud'))
+ util.write_file(os.path.join(new_root, 'etc',
+ 'cloud', 'cloud.cfg'), cloud_cfg)
+ self._patchIn(new_root)
+
+ # Now start verifying whats created
+ initer = stages.Init()
+ initer.read_cfg()
+ initer.initialize()
+ initer.fetch()
+ initer.datasource.userdata_raw = ud
+ _iid = initer.instancify()
+ initer.update()
+ initer.cloudify().run('consume_userdata',
+ initer.consume_userdata,
+ args=[PER_INSTANCE],
+ freq=PER_INSTANCE)
+ mirrors = initer.distro.get_option('package_mirrors')
+ self.assertEquals(1, len(mirrors))
+ mirror = mirrors[0]
+ self.assertEquals(mirror['arches'], ['i386', 'amd64', 'blah'])
+ mods = stages.Modules(initer)
+ (which_ran, failures) = mods.run_section('cloud_init_modules')
+ self.assertTrue(len(failures) == 0)
+ self.assertTrue(os.path.exists('/etc/blah.ini'))
+ self.assertIn('write-files', which_ran)
+ contents = util.load_file('/etc/blah.ini')
+ self.assertEquals(contents, 'blah')
diff --git a/tests/unittests/test_runs/test_simple_run.py b/tests/unittests/test_runs/test_simple_run.py
new file mode 100644
index 00000000..60ef812a
--- /dev/null
+++ b/tests/unittests/test_runs/test_simple_run.py
@@ -0,0 +1,80 @@
+import os
+
+from tests.unittests import helpers
+
+from cloudinit.settings import (PER_INSTANCE)
+from cloudinit import stages
+from cloudinit import util
+
+
+class TestSimpleRun(helpers.FilesystemMockingTestCase):
+ def _patchIn(self, root):
+ self.restore()
+ self.patchOS(root)
+ self.patchUtils(root)
+
+ def _pp_root(self, root, repatch=True):
+ self.restore()
+ for (dirpath, dirnames, filenames) in os.walk(root):
+ print(dirpath)
+ for f in filenames:
+ joined = os.path.join(dirpath, f)
+ if os.path.islink(joined):
+ print("f %s - (symlink)" % (f))
+ else:
+ print("f %s" % (f))
+ for d in dirnames:
+ joined = os.path.join(dirpath, d)
+ if os.path.islink(joined):
+ print("d %s - (symlink)" % (d))
+ else:
+ print("d %s" % (d))
+ if repatch:
+ self._patchIn(root)
+
+ def test_none_ds(self):
+ new_root = self.makeDir()
+ self.replicateTestRoot('simple_ubuntu', new_root)
+ cfg = {
+ 'datasource_list': ['None'],
+ 'write_files': [
+ {
+ 'path': '/etc/blah.ini',
+ 'content': 'blah',
+ 'permissions': 0755,
+ },
+ ],
+ 'cloud_init_modules': ['write-files'],
+ }
+ cloud_cfg = util.yaml_dumps(cfg)
+ util.ensure_dir(os.path.join(new_root, 'etc', 'cloud'))
+ util.write_file(os.path.join(new_root, 'etc',
+ 'cloud', 'cloud.cfg'), cloud_cfg)
+ self._patchIn(new_root)
+
+ # Now start verifying whats created
+ initer = stages.Init()
+ initer.read_cfg()
+ initer.initialize()
+ self.assertTrue(os.path.exists("/var/lib/cloud"))
+ for d in ['scripts', 'seed', 'instances', 'handlers', 'sem', 'data']:
+ self.assertTrue(os.path.isdir(os.path.join("/var/lib/cloud", d)))
+
+ initer.fetch()
+ iid = initer.instancify()
+ self.assertEquals(iid, 'iid-datasource-none')
+ initer.update()
+ self.assertTrue(os.path.islink("var/lib/cloud/instance"))
+
+ initer.cloudify().run('consume_userdata',
+ initer.consume_userdata,
+ args=[PER_INSTANCE],
+ freq=PER_INSTANCE)
+
+ mods = stages.Modules(initer)
+ (which_ran, failures) = mods.run_section('cloud_init_modules')
+ self.assertTrue(len(failures) == 0)
+ self.assertTrue(os.path.exists('/etc/blah.ini'))
+ self.assertIn('write-files', which_ran)
+ contents = util.load_file('/etc/blah.ini')
+ self.assertEquals(contents, 'blah')
diff --git a/tests/unittests/test_util.py b/tests/unittests/test_util.py
index 96962b91..02611581 100644
--- a/tests/unittests/test_util.py
+++ b/tests/unittests/test_util.py
@@ -28,65 +28,6 @@ class FakeSelinux(object):
self.restored.append(path)
-class TestMergeDict(MockerTestCase):
- def test_simple_merge(self):
- """Test simple non-conflict merge."""
- source = {"key1": "value1"}
- candidate = {"key2": "value2"}
- result = util.mergedict(source, candidate)
- self.assertEqual({"key1": "value1", "key2": "value2"}, result)
-
- def test_nested_merge(self):
- """Test nested merge."""
- source = {"key1": {"key1.1": "value1.1"}}
- candidate = {"key1": {"key1.2": "value1.2"}}
- result = util.mergedict(source, candidate)
- self.assertEqual(
- {"key1": {"key1.1": "value1.1", "key1.2": "value1.2"}}, result)
-
- def test_merge_does_not_override(self):
- """Test that candidate doesn't override source."""
- source = {"key1": "value1", "key2": "value2"}
- candidate = {"key1": "value2", "key2": "NEW VALUE"}
- result = util.mergedict(source, candidate)
- self.assertEqual(source, result)
-
- def test_empty_candidate(self):
- """Test empty candidate doesn't change source."""
- source = {"key": "value"}
- candidate = {}
- result = util.mergedict(source, candidate)
- self.assertEqual(source, result)
-
- def test_empty_source(self):
- """Test empty source is replaced by candidate."""
- source = {}
- candidate = {"key": "value"}
- result = util.mergedict(source, candidate)
- self.assertEqual(candidate, result)
-
- def test_non_dict_candidate(self):
- """Test non-dict candidate is discarded."""
- source = {"key": "value"}
- candidate = "not a dict"
- result = util.mergedict(source, candidate)
- self.assertEqual(source, result)
-
- def test_non_dict_source(self):
- """Test non-dict source is not modified with a dict candidate."""
- source = "not a dict"
- candidate = {"key": "value"}
- result = util.mergedict(source, candidate)
- self.assertEqual(source, result)
-
- def test_neither_dict(self):
- """Test if neither candidate or source is dict source wins."""
- source = "source"
- candidate = "candidate"
- result = util.mergedict(source, candidate)
- self.assertEqual(source, result)
-
-
class TestGetCfgOptionListOrStr(TestCase):
def test_not_found_no_default(self):
"""None is returned if key is not found and no default given."""
generated by cgit v1.2.3 (git 2.39.1) at 2024-09-23 02:17:19 +0000