diff options
Diffstat (limited to 'tests/unittests')
| -rw-r--r-- | tests/unittests/helpers.py | 3 | ||||
| -rw-r--r-- | tests/unittests/test__init__.py | 6 | ||||
| -rw-r--r-- | tests/unittests/test_datasource/test_configdrive.py | 5 | ||||
| -rw-r--r-- | tests/unittests/test_datasource/test_maas.py | 5 | ||||
| -rw-r--r-- | tests/unittests/test_datasource/test_nocloud.py | 35 | ||||
| -rw-r--r-- | tests/unittests/test_datasource/test_smartos.py | 145 | ||||
| -rw-r--r-- | tests/unittests/test_ec2_util.py | 8 | ||||
| -rw-r--r-- | tests/unittests/test_pathprefix2dict.py | 40 |
8 files changed, 231 insertions, 16 deletions
diff --git a/tests/unittests/helpers.py b/tests/unittests/helpers.py index c0da0983..5b4f4208 100644 --- a/tests/unittests/helpers.py +++ b/tests/unittests/helpers.py @@ -187,7 +187,8 @@ class FilesystemMockingTestCase(ResourceUsingTestCase): def populate_dir(path, files): - os.makedirs(path) + if not os.path.exists(path): + os.makedirs(path) for (name, content) in files.iteritems(): with open(os.path.join(path, name), "w") as fp: fp.write(content) diff --git a/tests/unittests/test__init__.py b/tests/unittests/test__init__.py index b4b20e51..8c41c1ca 100644 --- a/tests/unittests/test__init__.py +++ b/tests/unittests/test__init__.py @@ -196,7 +196,7 @@ class TestCmdlineUrl(MockerTestCase): mock_readurl = self.mocker.replace(url_helper.readurl, passthrough=False) mock_readurl(url, ARGS, KWARGS) - self.mocker.result(util.StringResponse(payload)) + self.mocker.result(url_helper.StringResponse(payload)) self.mocker.replay() self.assertEqual((key, url, None), @@ -212,7 +212,7 @@ class TestCmdlineUrl(MockerTestCase): mock_readurl = self.mocker.replace(url_helper.readurl, passthrough=False) mock_readurl(url, ARGS, KWARGS) - self.mocker.result(util.StringResponse(payload)) + self.mocker.result(url_helper.StringResponse(payload)) self.mocker.replay() self.assertEqual((key, url, payload), @@ -225,7 +225,7 @@ class TestCmdlineUrl(MockerTestCase): cmdline = "ro %s=%s bar=1" % (key, url) self.mocker.replace(url_helper.readurl, passthrough=False) - self.mocker.result(util.StringResponse("")) + self.mocker.result(url_helper.StringResponse("")) self.mocker.replay() self.assertEqual((None, None, None), diff --git a/tests/unittests/test_datasource/test_configdrive.py b/tests/unittests/test_datasource/test_configdrive.py index d5935294..1f4a0a0b 100644 --- a/tests/unittests/test_datasource/test_configdrive.py +++ b/tests/unittests/test_datasource/test_configdrive.py @@ -285,10 +285,11 @@ class TestConfigDriveDataSource(MockerTestCase): self.assertEqual(["/dev/vdb", "/dev/zdd"], ds.find_candidate_devs()) - # verify that partitions are not considered + # verify that partitions are considered, that have correct label. devs_with_answers = {"TYPE=vfat": ["/dev/sda1"], "TYPE=iso9660": [], "LABEL=config-2": ["/dev/vdb3"]} - self.assertEqual([], ds.find_candidate_devs()) + self.assertEqual(["/dev/vdb3"], + ds.find_candidate_devs()) finally: util.find_devs_with = orig_find_devs_with diff --git a/tests/unittests/test_datasource/test_maas.py b/tests/unittests/test_datasource/test_maas.py index 2007a6df..bd5d23fd 100644 --- a/tests/unittests/test_datasource/test_maas.py +++ b/tests/unittests/test_datasource/test_maas.py @@ -119,9 +119,10 @@ class TestMAASDataSource(mocker.MockerTestCase): mock_request(url, headers=None, timeout=mocker.ANY, data=mocker.ANY, sec_between=mocker.ANY, ssl_details=mocker.ANY, retries=mocker.ANY, - headers_cb=my_headers_cb) + headers_cb=my_headers_cb, + exception_cb=mocker.ANY) resp = valid.get(key) - self.mocker.result(util.StringResponse(resp)) + self.mocker.result(url_helper.StringResponse(resp)) self.mocker.replay() (userdata, metadata) = DataSourceMAAS.read_maas_seed_url(my_seed, diff --git a/tests/unittests/test_datasource/test_nocloud.py b/tests/unittests/test_datasource/test_nocloud.py index 7328b240..af575a10 100644 --- a/tests/unittests/test_datasource/test_nocloud.py +++ b/tests/unittests/test_datasource/test_nocloud.py @@ -97,6 +97,41 @@ class TestNoCloudDataSource(MockerTestCase): self.assertEqual(dsrc.metadata.get('instance-id'), 'IID') self.assertTrue(ret) + def test_nocloud_seed_with_vendordata(self): + md = {'instance-id': 'IID', 'dsmode': 'local'} + ud = "USER_DATA_HERE" + vd = "THIS IS MY VENDOR_DATA" + + populate_dir(os.path.join(self.paths.seed_dir, "nocloud"), + {'user-data': ud, 'meta-data': yaml.safe_dump(md), + 'vendor-data': vd}) + + sys_cfg = { + 'datasource': {'NoCloud': {'fs_label': None}} + } + + ds = DataSourceNoCloud.DataSourceNoCloud + + dsrc = ds(sys_cfg=sys_cfg, distro=None, paths=self.paths) + ret = dsrc.get_data() + self.assertEqual(dsrc.userdata_raw, ud) + self.assertEqual(dsrc.metadata, md) + self.assertEqual(dsrc.vendordata, vd) + self.assertTrue(ret) + + def test_nocloud_no_vendordata(self): + populate_dir(os.path.join(self.paths.seed_dir, "nocloud"), + {'user-data': "ud", 'meta-data': "instance-id: IID\n"}) + + sys_cfg = {'datasource': {'NoCloud': {'fs_label': None}}} + + ds = DataSourceNoCloud.DataSourceNoCloud + + dsrc = ds(sys_cfg=sys_cfg, distro=None, paths=self.paths) + ret = dsrc.get_data() + self.assertEqual(dsrc.userdata_raw, "ud") + self.assertFalse(dsrc.vendordata) + self.assertTrue(ret) class TestParseCommandLineData(MockerTestCase): diff --git a/tests/unittests/test_datasource/test_smartos.py b/tests/unittests/test_datasource/test_smartos.py index 956767d8..ae427bb5 100644 --- a/tests/unittests/test_datasource/test_smartos.py +++ b/tests/unittests/test_datasource/test_smartos.py @@ -27,6 +27,10 @@ from cloudinit import helpers from cloudinit.sources import DataSourceSmartOS from mocker import MockerTestCase +import os +import os.path +import re +import stat import uuid MOCK_RETURNS = { @@ -35,7 +39,11 @@ MOCK_RETURNS = { 'disable_iptables_flag': None, 'enable_motd_sys_info': None, 'test-var1': 'some data', - 'user-data': '\n'.join(['#!/bin/sh', '/bin/true', '']), + 'cloud-init:user-data': '\n'.join(['#!/bin/sh', '/bin/true', '']), + 'sdc:datacenter_name': 'somewhere2', + 'sdc:operator-script': '\n'.join(['bin/true', '']), + 'user-data': '\n'.join(['something', '']), + 'user-script': '\n'.join(['/bin/true', '']), } DMI_DATA_RETURN = (str(uuid.uuid4()), 'smartdc') @@ -101,6 +109,7 @@ class TestSmartOSDataSource(MockerTestCase): def setUp(self): # makeDir comes from MockerTestCase self.tmp = self.makeDir() + self.legacy_user_d = self.makeDir() # patch cloud_dir, so our 'seed_dir' is guaranteed empty self.paths = helpers.Paths({'cloud_dir': self.tmp}) @@ -138,6 +147,7 @@ class TestSmartOSDataSource(MockerTestCase): sys_cfg['datasource'] = sys_cfg.get('datasource', {}) sys_cfg['datasource']['SmartOS'] = ds_cfg + self.apply_patches([(mod, 'LEGACY_USER_D', self.legacy_user_d)]) self.apply_patches([(mod, 'get_serial', _get_serial)]) self.apply_patches([(mod, 'dmi_data', _dmi_data)]) dsrc = mod.DataSourceSmartOS(sys_cfg, distro=None, @@ -194,7 +204,7 @@ class TestSmartOSDataSource(MockerTestCase): # metadata provided base64_all of true my_returns = MOCK_RETURNS.copy() my_returns['base64_all'] = "true" - for k in ('hostname', 'user-data'): + for k in ('hostname', 'cloud-init:user-data'): my_returns[k] = base64.b64encode(my_returns[k]) dsrc = self._get_ds(mockdata=my_returns) @@ -202,7 +212,7 @@ class TestSmartOSDataSource(MockerTestCase): self.assertTrue(ret) self.assertEquals(MOCK_RETURNS['hostname'], dsrc.metadata['local-hostname']) - self.assertEquals(MOCK_RETURNS['user-data'], + self.assertEquals(MOCK_RETURNS['cloud-init:user-data'], dsrc.userdata_raw) self.assertEquals(MOCK_RETURNS['root_authorized_keys'], dsrc.metadata['public-keys']) @@ -213,9 +223,9 @@ class TestSmartOSDataSource(MockerTestCase): def test_b64_userdata(self): my_returns = MOCK_RETURNS.copy() - my_returns['b64-user-data'] = "true" + my_returns['b64-cloud-init:user-data'] = "true" my_returns['b64-hostname'] = "true" - for k in ('hostname', 'user-data'): + for k in ('hostname', 'cloud-init:user-data'): my_returns[k] = base64.b64encode(my_returns[k]) dsrc = self._get_ds(mockdata=my_returns) @@ -223,7 +233,8 @@ class TestSmartOSDataSource(MockerTestCase): self.assertTrue(ret) self.assertEquals(MOCK_RETURNS['hostname'], dsrc.metadata['local-hostname']) - self.assertEquals(MOCK_RETURNS['user-data'], dsrc.userdata_raw) + self.assertEquals(MOCK_RETURNS['cloud-init:user-data'], + dsrc.userdata_raw) self.assertEquals(MOCK_RETURNS['root_authorized_keys'], dsrc.metadata['public-keys']) @@ -238,13 +249,131 @@ class TestSmartOSDataSource(MockerTestCase): self.assertTrue(ret) self.assertEquals(MOCK_RETURNS['hostname'], dsrc.metadata['local-hostname']) - self.assertEquals(MOCK_RETURNS['user-data'], dsrc.userdata_raw) + self.assertEquals(MOCK_RETURNS['cloud-init:user-data'], + dsrc.userdata_raw) def test_userdata(self): dsrc = self._get_ds(mockdata=MOCK_RETURNS) ret = dsrc.get_data() self.assertTrue(ret) - self.assertEquals(MOCK_RETURNS['user-data'], dsrc.userdata_raw) + self.assertEquals(MOCK_RETURNS['user-data'], + dsrc.metadata['legacy-user-data']) + self.assertEquals(MOCK_RETURNS['cloud-init:user-data'], + dsrc.userdata_raw) + + def test_sdc_scripts(self): + dsrc = self._get_ds(mockdata=MOCK_RETURNS) + ret = dsrc.get_data() + self.assertTrue(ret) + self.assertEquals(MOCK_RETURNS['user-script'], + dsrc.metadata['user-script']) + + legacy_script_f = "%s/user-script" % self.legacy_user_d + self.assertTrue(os.path.exists(legacy_script_f)) + self.assertTrue(os.path.islink(legacy_script_f)) + user_script_perm = oct(os.stat(legacy_script_f)[stat.ST_MODE])[-3:] + self.assertEquals(user_script_perm, '700') + + def test_scripts_shebanged(self): + dsrc = self._get_ds(mockdata=MOCK_RETURNS) + ret = dsrc.get_data() + self.assertTrue(ret) + self.assertEquals(MOCK_RETURNS['user-script'], + dsrc.metadata['user-script']) + + legacy_script_f = "%s/user-script" % self.legacy_user_d + self.assertTrue(os.path.exists(legacy_script_f)) + self.assertTrue(os.path.islink(legacy_script_f)) + shebang = None + with open(legacy_script_f, 'r') as f: + shebang = f.readlines()[0].strip() + self.assertEquals(shebang, "#!/bin/bash") + user_script_perm = oct(os.stat(legacy_script_f)[stat.ST_MODE])[-3:] + self.assertEquals(user_script_perm, '700') + + def test_scripts_shebang_not_added(self): + """ + Test that the SmartOS requirement that plain text scripts + are executable. This test makes sure that plain texts scripts + with out file magic have it added appropriately by cloud-init. + """ + + my_returns = MOCK_RETURNS.copy() + my_returns['user-script'] = '\n'.join(['#!/usr/bin/perl', + 'print("hi")', '']) + + dsrc = self._get_ds(mockdata=my_returns) + ret = dsrc.get_data() + self.assertTrue(ret) + self.assertEquals(my_returns['user-script'], + dsrc.metadata['user-script']) + + legacy_script_f = "%s/user-script" % self.legacy_user_d + self.assertTrue(os.path.exists(legacy_script_f)) + self.assertTrue(os.path.islink(legacy_script_f)) + shebang = None + with open(legacy_script_f, 'r') as f: + shebang = f.readlines()[0].strip() + self.assertEquals(shebang, "#!/usr/bin/perl") + + def test_scripts_removed(self): + """ + Since SmartOS requires that the user script is fetched + each boot, we want to make sure that the information + is backed-up for user-review later. + + This tests the behavior of when a script is removed. It makes + sure that a) the previous script is backed-up; and 2) that + there is no script remaining. + """ + + script_d = os.path.join(self.tmp, "scripts", "per-boot") + os.makedirs(script_d) + + test_script_f = "%s/99_user_script" % script_d + with open(test_script_f, 'w') as f: + f.write("TEST DATA") + + my_returns = MOCK_RETURNS.copy() + del my_returns['user-script'] + + dsrc = self._get_ds(mockdata=my_returns) + ret = dsrc.get_data() + self.assertTrue(ret) + self.assertFalse(dsrc.metadata['user-script']) + self.assertFalse(os.path.exists(test_script_f)) + + def test_userdata_removed(self): + """ + User-data in the SmartOS world is supposed to be written to a file + each and every boot. This tests to make sure that in the event the + legacy user-data is removed, the existing user-data is backed-up and + there is no /var/db/user-data left. + """ + + user_data_f = "%s/mdata-user-data" % self.legacy_user_d + with open(user_data_f, 'w') as f: + f.write("PREVIOUS") + + my_returns = MOCK_RETURNS.copy() + del my_returns['user-data'] + + dsrc = self._get_ds(mockdata=my_returns) + ret = dsrc.get_data() + self.assertTrue(ret) + self.assertFalse(dsrc.metadata.get('legacy-user-data')) + + found_new = False + for root, _dirs, files in os.walk(self.legacy_user_d): + for name in files: + name_f = os.path.join(root, name) + permissions = oct(os.stat(name_f)[stat.ST_MODE])[-3:] + if re.match(r'.*\/mdata-user-data$', name_f): + found_new = True + print name_f + self.assertEquals(permissions, '400') + + self.assertFalse(found_new) def test_disable_iptables_flag(self): dsrc = self._get_ds(mockdata=MOCK_RETURNS) diff --git a/tests/unittests/test_ec2_util.py b/tests/unittests/test_ec2_util.py index dd588aca..957dc3f2 100644 --- a/tests/unittests/test_ec2_util.py +++ b/tests/unittests/test_ec2_util.py @@ -34,6 +34,14 @@ class TestEc2Util(helpers.TestCase): self.assertEquals('', userdata) @hp.activate + def test_userdata_fetch_fail_server_not_found(self): + hp.register_uri(hp.GET, + 'http://169.254.169.254/%s/user-data' % (self.VERSION), + status=404) + userdata = eu.get_instance_userdata(self.VERSION) + self.assertEquals('', userdata) + + @hp.activate def test_metadata_fetch_no_keys(self): base_url = 'http://169.254.169.254/%s/meta-data' % (self.VERSION) hp.register_uri(hp.GET, base_url, status=200, diff --git a/tests/unittests/test_pathprefix2dict.py b/tests/unittests/test_pathprefix2dict.py new file mode 100644 index 00000000..c68c263c --- /dev/null +++ b/tests/unittests/test_pathprefix2dict.py @@ -0,0 +1,40 @@ +from cloudinit import util + +from mocker import MockerTestCase +from tests.unittests.helpers import populate_dir + + +class TestPathPrefix2Dict(MockerTestCase): + + def setUp(self): + self.tmp = self.makeDir() + + def test_required_only(self): + dirdata = {'f1': 'f1content', 'f2': 'f2content'} + populate_dir(self.tmp, dirdata) + + ret = util.pathprefix2dict(self.tmp, required=['f1', 'f2']) + self.assertEqual(dirdata, ret) + + def test_required_missing(self): + dirdata = {'f1': 'f1content'} + populate_dir(self.tmp, dirdata) + kwargs = {'required': ['f1', 'f2']} + self.assertRaises(ValueError, util.pathprefix2dict, self.tmp, **kwargs) + + def test_no_required_and_optional(self): + dirdata = {'f1': 'f1c', 'f2': 'f2c'} + populate_dir(self.tmp, dirdata) + + ret = util.pathprefix2dict(self.tmp, required=None, + optional=['f1', 'f2']) + self.assertEqual(dirdata, ret) + + def test_required_and_optional(self): + dirdata = {'f1': 'f1c', 'f2': 'f2c'} + populate_dir(self.tmp, dirdata) + + ret = util.pathprefix2dict(self.tmp, required=['f1'], optional=['f2']) + self.assertEqual(dirdata, ret) + +# vi: ts=4 expandtab |