summaryrefslogtreecommitdiff
path: root/tests/unittests
diff options
context:
space:
mode:
authorJoshua Harlow <harlowja@yahoo-inc.com>2014-02-07 15:14:26 -0800
committerJoshua Harlow <harlowja@yahoo-inc.com>2014-02-07 15:14:26 -0800
commit2983a26ecf9716dc957ec4bacf15544072774190 (patch)
treea7e074c21789ebc553713b3fd591fd9bbe4a9684 /tests/unittests
parent810df2c55c108e7e4064263e508d9786d8b1dc8e (diff)
parent3cfe9b3d8958b1a4e450d5ff31d805c424945027 (diff)
downloadvyos-cloud-init-2983a26ecf9716dc957ec4bacf15544072774190.tar.gz
vyos-cloud-init-2983a26ecf9716dc957ec4bacf15544072774190.zip
Remerged with trunk
Diffstat (limited to 'tests/unittests')
-rw-r--r--tests/unittests/helpers.py3
-rw-r--r--tests/unittests/test__init__.py6
-rw-r--r--tests/unittests/test_datasource/test_configdrive.py5
-rw-r--r--tests/unittests/test_datasource/test_maas.py5
-rw-r--r--tests/unittests/test_datasource/test_nocloud.py35
-rw-r--r--tests/unittests/test_datasource/test_smartos.py145
-rw-r--r--tests/unittests/test_ec2_util.py8
-rw-r--r--tests/unittests/test_pathprefix2dict.py40
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 bd6cdd5d..937b88c1 100644
--- a/tests/unittests/test_datasource/test_configdrive.py
+++ b/tests/unittests/test_datasource/test_configdrive.py
@@ -288,10 +288,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 18d36d86..dd87665d 100644
--- a/tests/unittests/test_ec2_util.py
+++ b/tests/unittests/test_ec2_util.py
@@ -35,6 +35,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