diff options
Diffstat (limited to 'tests/unittests/test_handler')
5 files changed, 516 insertions, 1 deletions
diff --git a/tests/unittests/test_handler/test_handler_apt_configure.py b/tests/unittests/test_handler/test_handler_apt_configure.py new file mode 100644 index 00000000..203dd2aa --- /dev/null +++ b/tests/unittests/test_handler/test_handler_apt_configure.py @@ -0,0 +1,106 @@ +from mocker import MockerTestCase + +from cloudinit import util + +from cloudinit.config import cc_apt_configure + +import os +import re + + +class TestAptProxyConfig(MockerTestCase): + def setUp(self): + super(TestAptProxyConfig, self).setUp() + self.tmp = self.makeDir() + self.pfile = os.path.join(self.tmp, "proxy.cfg") + self.cfile = os.path.join(self.tmp, "config.cfg") + + def _search_apt_config(self, contents, ptype, value): + print( + r"acquire::%s::proxy\s+[\"']%s[\"'];\n" % (ptype, value), + contents, "flags=re.IGNORECASE") + return(re.search( + r"acquire::%s::proxy\s+[\"']%s[\"'];\n" % (ptype, value), + contents, flags=re.IGNORECASE)) + + def test_apt_proxy_written(self): + cfg = {'apt_proxy': 'myproxy'} + cc_apt_configure.apply_apt_config(cfg, self.pfile, self.cfile) + + self.assertTrue(os.path.isfile(self.pfile)) + self.assertFalse(os.path.isfile(self.cfile)) + + contents = str(util.read_file_or_url(self.pfile)) + self.assertTrue(self._search_apt_config(contents, "http", "myproxy")) + + def test_apt_http_proxy_written(self): + cfg = {'apt_http_proxy': 'myproxy'} + cc_apt_configure.apply_apt_config(cfg, self.pfile, self.cfile) + + self.assertTrue(os.path.isfile(self.pfile)) + self.assertFalse(os.path.isfile(self.cfile)) + + contents = str(util.read_file_or_url(self.pfile)) + self.assertTrue(self._search_apt_config(contents, "http", "myproxy")) + + def test_apt_all_proxy_written(self): + cfg = {'apt_http_proxy': 'myproxy_http_proxy', + 'apt_https_proxy': 'myproxy_https_proxy', + 'apt_ftp_proxy': 'myproxy_ftp_proxy'} + + values = {'http': cfg['apt_http_proxy'], + 'https': cfg['apt_https_proxy'], + 'ftp': cfg['apt_ftp_proxy'], + } + + cc_apt_configure.apply_apt_config(cfg, self.pfile, self.cfile) + + self.assertTrue(os.path.isfile(self.pfile)) + self.assertFalse(os.path.isfile(self.cfile)) + + contents = str(util.read_file_or_url(self.pfile)) + + for ptype, pval in values.iteritems(): + self.assertTrue(self._search_apt_config(contents, ptype, pval)) + + def test_proxy_deleted(self): + util.write_file(self.cfile, "content doesnt matter") + cc_apt_configure.apply_apt_config({}, self.pfile, self.cfile) + self.assertFalse(os.path.isfile(self.pfile)) + self.assertFalse(os.path.isfile(self.cfile)) + + def test_proxy_replaced(self): + util.write_file(self.cfile, "content doesnt matter") + cc_apt_configure.apply_apt_config({'apt_proxy': "foo"}, + self.pfile, self.cfile) + self.assertTrue(os.path.isfile(self.pfile)) + contents = str(util.read_file_or_url(self.pfile)) + self.assertTrue(self._search_apt_config(contents, "http", "foo")) + + def test_config_written(self): + payload = 'this is my apt config' + cfg = {'apt_config': payload} + + cc_apt_configure.apply_apt_config(cfg, self.pfile, self.cfile) + + self.assertTrue(os.path.isfile(self.cfile)) + self.assertFalse(os.path.isfile(self.pfile)) + + self.assertEqual(str(util.read_file_or_url(self.cfile)), payload) + + def test_config_replaced(self): + util.write_file(self.pfile, "content doesnt matter") + cc_apt_configure.apply_apt_config({'apt_config': "foo"}, + self.pfile, self.cfile) + self.assertTrue(os.path.isfile(self.cfile)) + self.assertEqual(str(util.read_file_or_url(self.cfile)), "foo") + + def test_config_deleted(self): + # if no 'apt_config' is provided, delete any previously written file + util.write_file(self.pfile, "content doesnt matter") + cc_apt_configure.apply_apt_config({}, self.pfile, self.cfile) + self.assertFalse(os.path.isfile(self.pfile)) + self.assertFalse(os.path.isfile(self.cfile)) + + +# vi: ts=4 expandtab diff --git a/tests/unittests/test_handler/test_handler_growpart.py b/tests/unittests/test_handler/test_handler_growpart.py new file mode 100644 index 00000000..c0497e08 --- /dev/null +++ b/tests/unittests/test_handler/test_handler_growpart.py @@ -0,0 +1,258 @@ +from mocker import MockerTestCase + +from cloudinit import cloud +from cloudinit import util + +from cloudinit.config import cc_growpart + +import errno +import logging +import os +import re +import unittest + +# growpart: +# mode: auto # off, on, auto, 'growpart', 'parted' +# devices: ['root'] + +HELP_PARTED_NO_RESIZE = """ +Usage: parted [OPTION]... [DEVICE [COMMAND [PARAMETERS]...]...] +Apply COMMANDs with PARAMETERS to DEVICE. If no COMMAND(s) are given, run in +interactive mode. + +OPTIONs: +<SNIP> + +COMMANDs: +<SNIP> + quit exit program + rescue START END rescue a lost partition near START + and END + resize NUMBER START END resize partition NUMBER and its file + system + rm NUMBER delete partition NUMBER +<SNIP> +Report bugs to bug-parted@gnu.org +""" + +HELP_PARTED_RESIZE = """ +Usage: parted [OPTION]... [DEVICE [COMMAND [PARAMETERS]...]...] +Apply COMMANDs with PARAMETERS to DEVICE. If no COMMAND(s) are given, run in +interactive mode. + +OPTIONs: +<SNIP> + +COMMANDs: +<SNIP> + quit exit program + rescue START END rescue a lost partition near START + and END + resize NUMBER START END resize partition NUMBER and its file + system + resizepart NUMBER END resize partition NUMBER + rm NUMBER delete partition NUMBER +<SNIP> +Report bugs to bug-parted@gnu.org +""" + +HELP_GROWPART_RESIZE = """ +growpart disk partition + rewrite partition table so that partition takes up all the space it can + options: + -h | --help print Usage and exit +<SNIP> + -u | --update R update the the kernel partition table info after growing + this requires kernel support and 'partx --update' + R is one of: + - 'auto' : [default] update partition if possible +<SNIP> + Example: + - growpart /dev/sda 1 + Resize partition 1 on /dev/sda +""" + +HELP_GROWPART_NO_RESIZE = """ +growpart disk partition + rewrite partition table so that partition takes up all the space it can + options: + -h | --help print Usage and exit +<SNIP> + Example: + - growpart /dev/sda 1 + Resize partition 1 on /dev/sda +""" + + +class TestDisabled(MockerTestCase): + def setUp(self): + super(TestDisabled, self).setUp() + self.name = "growpart" + self.cloud_init = None + self.log = logging.getLogger("TestDisabled") + self.args = [] + + self.handle = cc_growpart.handle + + def test_mode_off(self): + #Test that nothing is done if mode is off. + + # this really only verifies that resizer_factory isn't called + config = {'growpart': {'mode': 'off'}} + self.mocker.replace(cc_growpart.resizer_factory, + passthrough=False) + self.mocker.replay() + + self.handle(self.name, config, self.cloud_init, self.log, self.args) + + +class TestConfig(MockerTestCase): + def setUp(self): + super(TestConfig, self).setUp() + self.name = "growpart" + self.paths = None + self.cloud = cloud.Cloud(None, self.paths, None, None, None) + self.log = logging.getLogger("TestConfig") + self.args = [] + os.environ = {} + + self.cloud_init = None + self.handle = cc_growpart.handle + + # Order must be correct + self.mocker.order() + + @unittest.skip("until LP: #1212444 fixed") + def test_no_resizers_auto_is_fine(self): + subp = self.mocker.replace(util.subp, passthrough=False) + subp(['parted', '--help'], env={'LANG': 'C'}) + self.mocker.result((HELP_PARTED_NO_RESIZE, "")) + subp(['growpart', '--help'], env={'LANG': 'C'}) + self.mocker.result((HELP_GROWPART_NO_RESIZE, "")) + self.mocker.replay() + + config = {'growpart': {'mode': 'auto'}} + self.handle(self.name, config, self.cloud_init, self.log, self.args) + + def test_no_resizers_mode_growpart_is_exception(self): + subp = self.mocker.replace(util.subp, passthrough=False) + subp(['growpart', '--help'], env={'LANG': 'C'}) + self.mocker.result((HELP_GROWPART_NO_RESIZE, "")) + self.mocker.replay() + + config = {'growpart': {'mode': "growpart"}} + self.assertRaises(ValueError, self.handle, self.name, config, + self.cloud_init, self.log, self.args) + + @unittest.skip("until LP: #1212444 fixed") + def test_mode_auto_prefers_parted(self): + subp = self.mocker.replace(util.subp, passthrough=False) + subp(['parted', '--help'], env={'LANG': 'C'}) + self.mocker.result((HELP_PARTED_RESIZE, "")) + self.mocker.replay() + + ret = cc_growpart.resizer_factory(mode="auto") + self.assertTrue(isinstance(ret, cc_growpart.ResizeParted)) + + def test_handle_with_no_growpart_entry(self): + #if no 'growpart' entry in config, then mode=auto should be used + + myresizer = object() + + factory = self.mocker.replace(cc_growpart.resizer_factory, + passthrough=False) + rsdevs = self.mocker.replace(cc_growpart.resize_devices, + passthrough=False) + factory("auto") + self.mocker.result(myresizer) + rsdevs(myresizer, ["/"]) + self.mocker.result((("/", cc_growpart.RESIZE.CHANGED, "my-message",),)) + self.mocker.replay() + + try: + orig_resizers = cc_growpart.RESIZERS + cc_growpart.RESIZERS = (('mysizer', object),) + self.handle(self.name, {}, self.cloud_init, self.log, self.args) + finally: + cc_growpart.RESIZERS = orig_resizers + + +class TestResize(MockerTestCase): + def setUp(self): + super(TestResize, self).setUp() + self.name = "growpart" + self.log = logging.getLogger("TestResize") + + # Order must be correct + self.mocker.order() + + def test_simple_devices(self): + #test simple device list + # this patches out devent2dev, os.stat, and device_part_info + # so in the end, doesn't test a lot + devs = ["/dev/XXda1", "/dev/YYda2"] + devstat_ret = Bunch(st_mode=25008, st_ino=6078, st_dev=5L, + st_nlink=1, st_uid=0, st_gid=6, st_size=0, + st_atime=0, st_mtime=0, st_ctime=0) + enoent = ["/dev/NOENT"] + real_stat = os.stat + resize_calls = [] + + class myresizer(object): + def resize(self, diskdev, partnum, partdev): + resize_calls.append((diskdev, partnum, partdev)) + if partdev == "/dev/YYda2": + return (1024, 2048) + return (1024, 1024) # old size, new size + + def mystat(path): + if path in devs: + return devstat_ret + if path in enoent: + e = OSError("%s: does not exist" % path) + e.errno = errno.ENOENT + raise e + return real_stat(path) + + try: + opinfo = cc_growpart.device_part_info + cc_growpart.device_part_info = simple_device_part_info + os.stat = mystat + + resized = cc_growpart.resize_devices(myresizer(), devs + enoent) + + def find(name, res): + for f in res: + if f[0] == name: + return f + return None + + self.assertEqual(cc_growpart.RESIZE.NOCHANGE, + find("/dev/XXda1", resized)[1]) + self.assertEqual(cc_growpart.RESIZE.CHANGED, + find("/dev/YYda2", resized)[1]) + self.assertEqual(cc_growpart.RESIZE.SKIPPED, + find(enoent[0], resized)[1]) + #self.assertEqual(resize_calls, + #[("/dev/XXda", "1", "/dev/XXda1"), + #("/dev/YYda", "2", "/dev/YYda2")]) + finally: + cc_growpart.device_part_info = opinfo + os.stat = real_stat + + +def simple_device_part_info(devpath): + # simple stupid return (/dev/vda, 1) for /dev/vda + ret = re.search("([^0-9]*)([0-9]*)$", devpath) + x = (ret.group(1), ret.group(2)) + return x + + +class Bunch: + st_mode = None # fix pylint complaint + + def __init__(self, **kwds): + self.__dict__.update(kwds) + + +# vi: ts=4 expandtab diff --git a/tests/unittests/test_handler/test_handler_locale.py b/tests/unittests/test_handler/test_handler_locale.py new file mode 100644 index 00000000..72ad00fd --- /dev/null +++ b/tests/unittests/test_handler/test_handler_locale.py @@ -0,0 +1,64 @@ +# Copyright (C) 2013 Hewlett-Packard Development Company, L.P. +# +# Author: Juerg Haefliger <juerg.haefliger@hp.com> +# +# Based on test_handler_set_hostname.py +# +# This program is free software: you can redistribute it and/or modify +# it under the terms of the GNU General Public License version 3, as +# published by the Free Software Foundation. +# +# This program is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU General Public License for more details. +# +# You should have received a copy of the GNU General Public License +# along with this program. If not, see <http://www.gnu.org/licenses/>. + +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 tests.unittests import helpers as t_help + +from configobj import ConfigObj + +from StringIO import StringIO + +import logging + +LOG = logging.getLogger(__name__) + + +class TestLocale(t_help.FilesystemMockingTestCase): + def setUp(self): + super(TestLocale, self).setUp() + self.new_root = self.makeDir(prefix="unittest_") + + 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_sles(self): + + cfg = { + 'locale': 'My.Locale', + } + cc = self._get_cloud('sles') + cc_locale.handle('cc_locale', cfg, cc, LOG, []) + + contents = util.load_file('/etc/sysconfig/language') + n_cfg = ConfigObj(StringIO(contents)) + self.assertEquals({'RC_LANG': cfg['locale']}, dict(n_cfg)) diff --git a/tests/unittests/test_handler/test_handler_set_hostname.py b/tests/unittests/test_handler/test_handler_set_hostname.py index a1aba62f..6344ec0c 100644 --- a/tests/unittests/test_handler/test_handler_set_hostname.py +++ b/tests/unittests/test_handler/test_handler_set_hostname.py @@ -35,7 +35,6 @@ class TestHostname(t_help.FilesystemMockingTestCase): 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") @@ -56,3 +55,16 @@ class TestHostname(t_help.FilesystemMockingTestCase): cfg, cc, LOG, []) contents = util.load_file("/etc/hostname") self.assertEquals('blah', contents.strip()) + + def test_write_hostname_sles(self): + cfg = { + 'hostname': 'blah.blah.blah.suse.com', + } + distro = self._fetch_distro('sles') + 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_timezone.py b/tests/unittests/test_handler/test_handler_timezone.py new file mode 100644 index 00000000..40b69773 --- /dev/null +++ b/tests/unittests/test_handler/test_handler_timezone.py @@ -0,0 +1,75 @@ +# Copyright (C) 2013 Hewlett-Packard Development Company, L.P. +# +# Author: Juerg Haefliger <juerg.haefliger@hp.com> +# +# Based on test_handler_set_hostname.py +# +# This program is free software: you can redistribute it and/or modify +# it under the terms of the GNU General Public License version 3, as +# published by the Free Software Foundation. +# +# This program is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU General Public License for more details. +# +# You should have received a copy of the GNU General Public License +# along with this program. If not, see <http://www.gnu.org/licenses/>. + +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 tests.unittests import helpers as t_help + +from configobj import ConfigObj + +from StringIO import StringIO + +import logging + +LOG = logging.getLogger(__name__) + + +class TestTimezone(t_help.FilesystemMockingTestCase): + def setUp(self): + super(TestTimezone, self).setUp() + self.new_root = self.makeDir(prefix="unittest_") + + 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') + + # Create a dummy timezone file + dummy_contents = '0123456789abcdefgh' + util.write_file('/usr/share/zoneinfo/%s' % cfg['timezone'], + dummy_contents) + + cc_timezone.handle('cc_timezone', cfg, cc, LOG, []) + + contents = util.load_file('/etc/sysconfig/clock') + n_cfg = ConfigObj(StringIO(contents)) + self.assertEquals({'TIMEZONE': cfg['timezone']}, dict(n_cfg)) + + contents = util.load_file('/etc/localtime') + self.assertEquals(dummy_contents, contents.strip()) |