diff options
author | Scott Moser <smoser@brickies.net> | 2012-10-23 18:36:41 +0200 |
---|---|---|
committer | Scott Moser <smoser@brickies.net> | 2012-10-23 18:36:41 +0200 |
commit | cb5893c3e3f635de4fef86f0a19be0ada0054930 (patch) | |
tree | 6fc84345114c4d121c7d3c64ae42503d33ba6b36 /tests | |
parent | 2953d9d448fde3af19fc96ae00f41066f510d6fd (diff) | |
parent | b541f738616349a028c5e54754ea83e439d82734 (diff) | |
download | vyos-cloud-init-cb5893c3e3f635de4fef86f0a19be0ada0054930.tar.gz vyos-cloud-init-cb5893c3e3f635de4fef86f0a19be0ada0054930.zip |
Add helpers for wrapping file operations
Add a new example test that will patch utils and os
functions so that they can be 'retargeted' to a temporary
directory, which allows us the ability to run a full set
of cloud-init stages.
Neat things:
1. All cloud-init code is unchanged (as long as it goes
through the utils functions for most functionality)
2. Allows for a natural way to setup a temporary directory
then patch the new directory as the new 'root' and then
run cloud-init stages and then check the contents of
what was placed as desired.
Note:
This is now exposing what I think was just an issue in
./tools/run-pylint.
./tools/run-pylint ./tests/unittests/test_filters/test_launch_index.py
and
./tools/run-pylint ./tests/unittests/test_simple_run
was always failing, but now I'm seeing errors.
need to fix that. any tests that 'import helper' have issues.
Diffstat (limited to 'tests')
-rw-r--r-- | tests/data/roots/simple_ubuntu/etc/networks/interfaces | 3 | ||||
-rw-r--r-- | tests/unittests/helpers.py | 103 | ||||
-rw-r--r-- | tests/unittests/test_filters/test_launch_index.py | 12 | ||||
-rw-r--r-- | tests/unittests/test_runs/test_simple_run.py | 87 |
4 files changed, 203 insertions, 2 deletions
diff --git a/tests/data/roots/simple_ubuntu/etc/networks/interfaces b/tests/data/roots/simple_ubuntu/etc/networks/interfaces new file mode 100644 index 00000000..77efa67d --- /dev/null +++ b/tests/data/roots/simple_ubuntu/etc/networks/interfaces @@ -0,0 +1,3 @@ +auto lo +iface lo inet loopback + diff --git a/tests/unittests/helpers.py b/tests/unittests/helpers.py index d0f09e70..d5df580b 100644 --- a/tests/unittests/helpers.py +++ b/tests/unittests/helpers.py @@ -3,6 +3,37 @@ 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 +71,75 @@ 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), + ('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_filters/test_launch_index.py b/tests/unittests/test_filters/test_launch_index.py index 7ca7cbb6..1e9b9053 100644 --- a/tests/unittests/test_filters/test_launch_index.py +++ b/tests/unittests/test_filters/test_launch_index.py @@ -1,6 +1,14 @@ import copy +import os +import sys -import helpers as th +top_dir = os.path.join(os.path.dirname(__file__), os.pardir, "helpers.py") +top_dir = os.path.abspath(top_dir) +if os.path.exists(top_dir): + sys.path.insert(0, os.path.dirname(top_dir)) + + +import helpers import itertools @@ -18,7 +26,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_runs/test_simple_run.py b/tests/unittests/test_runs/test_simple_run.py new file mode 100644 index 00000000..7f646b54 --- /dev/null +++ b/tests/unittests/test_runs/test_simple_run.py @@ -0,0 +1,87 @@ +import sys +import os + +# Allow running this test individually +top_dir = os.path.join(os.path.dirname(__file__), os.pardir, "helpers.py") +top_dir = os.path.abspath(top_dir) +if os.path.exists(top_dir): + sys.path.insert(0, os.path.dirname(top_dir)) + + +import helpers + +from cloudinit import util +from cloudinit import stages + +from cloudinit.settings import (PER_INSTANCE) + + +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') |