diff options
Diffstat (limited to 'tests')
| -rw-r--r-- | tests/unittests/helpers.py | 12 | ||||
| -rw-r--r-- | tests/unittests/test_builtin_handlers.py | 1 | ||||
| -rw-r--r-- | tests/unittests/test_datasource/test_azure.py | 31 | ||||
| -rw-r--r-- | tests/unittests/test_datasource/test_gce.py | 2 | ||||
| -rw-r--r-- | tests/unittests/test_datasource/test_opennebula.py | 10 | ||||
| -rw-r--r-- | tests/unittests/test_datasource/test_smartos.py | 12 | ||||
| -rw-r--r-- | tests/unittests/test_filters/test_launch_index.py | 8 | ||||
| -rw-r--r-- | tests/unittests/test_handler/test_handler_chef.py | 3 | ||||
| -rw-r--r-- | tests/unittests/test_handler/test_handler_seed_random.py | 11 | ||||
| -rw-r--r-- | tests/unittests/test_util.py | 6 |
10 files changed, 63 insertions, 33 deletions
diff --git a/tests/unittests/helpers.py b/tests/unittests/helpers.py index 70b8116f..4b8dcc5c 100644 --- a/tests/unittests/helpers.py +++ b/tests/unittests/helpers.py @@ -1,8 +1,11 @@ import os import sys +import shutil import tempfile import unittest +import six + try: from unittest import mock except ImportError: @@ -15,8 +18,6 @@ except ImportError: from cloudinit import helpers as ch from cloudinit import util -import shutil - # Used for detecting different python versions PY2 = False PY26 = False @@ -115,7 +116,12 @@ def retarget_many_wrapper(new_base, am, old_func): nam = len(n_args) for i in range(0, nam): path = args[i] - n_args[i] = rebase_path(path, new_base) + # patchOS() wraps various os and os.path functions, however in + # Python 3 some of these now accept file-descriptors (integers). + # That breaks rebase_path() so in lieu of a better solution, just + # don't rebase if we get a fd. + if isinstance(path, six.string_types): + n_args[i] = rebase_path(path, new_base) return old_func(*n_args, **kwds) return wrapper diff --git a/tests/unittests/test_builtin_handlers.py b/tests/unittests/test_builtin_handlers.py index 47ff6318..ad32d0b2 100644 --- a/tests/unittests/test_builtin_handlers.py +++ b/tests/unittests/test_builtin_handlers.py @@ -21,7 +21,6 @@ from cloudinit.settings import (PER_ALWAYS, PER_INSTANCE) class TestBuiltins(test_helpers.FilesystemMockingTestCase): - def test_upstart_frequency_no_out(self): c_root = tempfile.mkdtemp() self.addCleanup(shutil.rmtree, c_root) diff --git a/tests/unittests/test_datasource/test_azure.py b/tests/unittests/test_datasource/test_azure.py index 2dbcd389..1f0330b3 100644 --- a/tests/unittests/test_datasource/test_azure.py +++ b/tests/unittests/test_datasource/test_azure.py @@ -22,6 +22,13 @@ import tempfile import unittest +def b64(source): + # In Python 3, b64encode only accepts bytes and returns bytes. + if not isinstance(source, bytes): + source = source.encode('utf-8') + return base64.b64encode(source).decode('us-ascii') + + def construct_valid_ovf_env(data=None, pubkeys=None, userdata=None): if data is None: data = {'HostName': 'FOOHOST'} @@ -51,7 +58,7 @@ def construct_valid_ovf_env(data=None, pubkeys=None, userdata=None): content += "<%s%s>%s</%s>\n" % (key, attrs, val, key) if userdata: - content += "<UserData>%s</UserData>\n" % (base64.b64encode(userdata)) + content += "<UserData>%s</UserData>\n" % (b64(userdata)) if pubkeys: content += "<SSH><PublicKeys>\n" @@ -181,7 +188,7 @@ class TestAzureDataSource(unittest.TestCase): # set dscfg in via base64 encoded yaml cfg = {'agent_command': "my_command"} odata = {'HostName': "myhost", 'UserName': "myuser", - 'dscfg': {'text': base64.b64encode(yaml.dump(cfg)), + 'dscfg': {'text': b64(yaml.dump(cfg)), 'encoding': 'base64'}} data = {'ovfcontent': construct_valid_ovf_env(data=odata)} @@ -233,13 +240,13 @@ class TestAzureDataSource(unittest.TestCase): def test_userdata_found(self): mydata = "FOOBAR" - odata = {'UserData': base64.b64encode(mydata)} + odata = {'UserData': b64(mydata)} data = {'ovfcontent': construct_valid_ovf_env(data=odata)} dsrc = self._get_ds(data) ret = dsrc.get_data() self.assertTrue(ret) - self.assertEqual(dsrc.userdata_raw, mydata) + self.assertEqual(dsrc.userdata_raw, mydata.encode('utf-8')) def test_no_datasource_expected(self): # no source should be found if no seed_dir and no devs @@ -281,7 +288,7 @@ class TestAzureDataSource(unittest.TestCase): 'command': 'my-bounce-command', 'hostname_command': 'my-hostname-command'}} odata = {'HostName': "xhost", - 'dscfg': {'text': base64.b64encode(yaml.dump(cfg)), + 'dscfg': {'text': b64(yaml.dump(cfg)), 'encoding': 'base64'}} data = {'ovfcontent': construct_valid_ovf_env(data=odata)} self._get_ds(data).get_data() @@ -296,7 +303,7 @@ class TestAzureDataSource(unittest.TestCase): # config specifying set_hostname off should not bounce cfg = {'set_hostname': False} odata = {'HostName': "xhost", - 'dscfg': {'text': base64.b64encode(yaml.dump(cfg)), + 'dscfg': {'text': b64(yaml.dump(cfg)), 'encoding': 'base64'}} data = {'ovfcontent': construct_valid_ovf_env(data=odata)} self._get_ds(data).get_data() @@ -325,7 +332,7 @@ class TestAzureDataSource(unittest.TestCase): # Make sure that user can affect disk aliases dscfg = {'disk_aliases': {'ephemeral0': '/dev/sdc'}} odata = {'HostName': "myhost", 'UserName': "myuser", - 'dscfg': {'text': base64.b64encode(yaml.dump(dscfg)), + 'dscfg': {'text': b64(yaml.dump(dscfg)), 'encoding': 'base64'}} usercfg = {'disk_setup': {'/dev/sdc': {'something': '...'}, 'ephemeral0': False}} @@ -347,7 +354,7 @@ class TestAzureDataSource(unittest.TestCase): dsrc = self._get_ds(data) dsrc.get_data() - self.assertEqual(userdata, dsrc.userdata_raw) + self.assertEqual(userdata.encode('us-ascii'), dsrc.userdata_raw) def test_ovf_env_arrives_in_waagent_dir(self): xml = construct_valid_ovf_env(data={}, userdata="FOODATA") @@ -362,7 +369,7 @@ class TestAzureDataSource(unittest.TestCase): def test_existing_ovf_same(self): # waagent/SharedConfig left alone if found ovf-env.xml same as cached - odata = {'UserData': base64.b64encode("SOMEUSERDATA")} + odata = {'UserData': b64("SOMEUSERDATA")} data = {'ovfcontent': construct_valid_ovf_env(data=odata)} populate_dir(self.waagent_d, @@ -386,9 +393,9 @@ class TestAzureDataSource(unittest.TestCase): # 'get_data' should remove SharedConfig.xml in /var/lib/waagent # if ovf-env.xml differs. cached_ovfenv = construct_valid_ovf_env( - {'userdata': base64.b64encode("FOO_USERDATA")}) + {'userdata': b64("FOO_USERDATA")}) new_ovfenv = construct_valid_ovf_env( - {'userdata': base64.b64encode("NEW_USERDATA")}) + {'userdata': b64("NEW_USERDATA")}) populate_dir(self.waagent_d, {'ovf-env.xml': cached_ovfenv, @@ -398,7 +405,7 @@ class TestAzureDataSource(unittest.TestCase): dsrc = self._get_ds({'ovfcontent': new_ovfenv}) ret = dsrc.get_data() self.assertTrue(ret) - self.assertEqual(dsrc.userdata_raw, "NEW_USERDATA") + self.assertEqual(dsrc.userdata_raw, b"NEW_USERDATA") self.assertTrue(os.path.exists( os.path.join(self.waagent_d, 'otherfile'))) self.assertFalse( diff --git a/tests/unittests/test_datasource/test_gce.py b/tests/unittests/test_datasource/test_gce.py index aa60eb33..6dd4b5ed 100644 --- a/tests/unittests/test_datasource/test_gce.py +++ b/tests/unittests/test_datasource/test_gce.py @@ -45,7 +45,7 @@ GCE_META_ENCODING = { 'instance/id': '12345', 'instance/hostname': 'server.project-baz.local', 'instance/zone': 'baz/bang', - 'instance/attributes/user-data': b64encode('/bin/echo baz\n'), + 'instance/attributes/user-data': b64encode(b'/bin/echo baz\n'), 'instance/attributes/user-data-encoding': 'base64', } diff --git a/tests/unittests/test_datasource/test_opennebula.py b/tests/unittests/test_datasource/test_opennebula.py index b79237f0..1a8d2122 100644 --- a/tests/unittests/test_datasource/test_opennebula.py +++ b/tests/unittests/test_datasource/test_opennebula.py @@ -10,6 +10,12 @@ import shutil import tempfile import unittest +def b64(source): + # In Python 3, b64encode only accepts bytes and returns bytes. + if not isinstance(source, bytes): + source = source.encode('utf-8') + return b64encode(source).decode('us-ascii') + TEST_VARS = { 'VAR1': 'single', @@ -180,7 +186,7 @@ class TestOpenNebulaDataSource(unittest.TestCase): self.assertEqual(USER_DATA, results['userdata']) def test_user_data_encoding_required_for_decode(self): - b64userdata = b64encode(USER_DATA) + b64userdata = b64(USER_DATA) for k in ('USER_DATA', 'USERDATA'): my_d = os.path.join(self.tmp, k) populate_context_dir(my_d, {k: b64userdata}) @@ -192,7 +198,7 @@ class TestOpenNebulaDataSource(unittest.TestCase): def test_user_data_base64_encoding(self): for k in ('USER_DATA', 'USERDATA'): my_d = os.path.join(self.tmp, k) - populate_context_dir(my_d, {k: b64encode(USER_DATA), + populate_context_dir(my_d, {k: b64(USER_DATA), 'USERDATA_ENCODING': 'base64'}) results = ds.read_context_disk_dir(my_d) diff --git a/tests/unittests/test_datasource/test_smartos.py b/tests/unittests/test_datasource/test_smartos.py index 01b9b73e..2fb9e1b6 100644 --- a/tests/unittests/test_datasource/test_smartos.py +++ b/tests/unittests/test_datasource/test_smartos.py @@ -36,6 +36,12 @@ import tempfile import stat import uuid +def b64(source): + # In Python 3, b64encode only accepts bytes and returns bytes. + if not isinstance(source, bytes): + source = source.encode('utf-8') + return base64.b64encode(source).decode('us-ascii') + MOCK_RETURNS = { 'hostname': 'test-host', @@ -233,7 +239,7 @@ class TestSmartOSDataSource(helpers.FilesystemMockingTestCase): my_returns = MOCK_RETURNS.copy() my_returns['base64_all'] = "true" for k in ('hostname', 'cloud-init:user-data'): - my_returns[k] = base64.b64encode(my_returns[k]) + my_returns[k] = b64(my_returns[k]) dsrc = self._get_ds(mockdata=my_returns) ret = dsrc.get_data() @@ -254,7 +260,7 @@ class TestSmartOSDataSource(helpers.FilesystemMockingTestCase): my_returns['b64-cloud-init:user-data'] = "true" my_returns['b64-hostname'] = "true" for k in ('hostname', 'cloud-init:user-data'): - my_returns[k] = base64.b64encode(my_returns[k]) + my_returns[k] = b64(my_returns[k]) dsrc = self._get_ds(mockdata=my_returns) ret = dsrc.get_data() @@ -270,7 +276,7 @@ class TestSmartOSDataSource(helpers.FilesystemMockingTestCase): my_returns = MOCK_RETURNS.copy() my_returns['base64_keys'] = 'hostname,ignored' for k in ('hostname',): - my_returns[k] = base64.b64encode(my_returns[k]) + my_returns[k] = b64(my_returns[k]) dsrc = self._get_ds(mockdata=my_returns) ret = dsrc.get_data() diff --git a/tests/unittests/test_filters/test_launch_index.py b/tests/unittests/test_filters/test_launch_index.py index 2f4c2fda..95d24b9b 100644 --- a/tests/unittests/test_filters/test_launch_index.py +++ b/tests/unittests/test_filters/test_launch_index.py @@ -2,7 +2,7 @@ import copy from .. import helpers -import itertools +from six.moves import filterfalse from cloudinit.filters import launch_index from cloudinit import user_data as ud @@ -36,11 +36,9 @@ class TestLaunchFilter(helpers.ResourceUsingTestCase): return False # Do some basic payload checking msg1_msgs = [m for m in msg1.walk()] - msg1_msgs = [m for m in - itertools.ifilterfalse(ud.is_skippable, msg1_msgs)] + msg1_msgs = [m for m in filterfalse(ud.is_skippable, msg1_msgs)] msg2_msgs = [m for m in msg2.walk()] - msg2_msgs = [m for m in - itertools.ifilterfalse(ud.is_skippable, msg2_msgs)] + msg2_msgs = [m for m in filterfalse(ud.is_skippable, msg2_msgs)] for i in range(0, len(msg2_msgs)): m1_msg = msg1_msgs[i] m2_msg = msg2_msgs[i] diff --git a/tests/unittests/test_handler/test_handler_chef.py b/tests/unittests/test_handler/test_handler_chef.py index b06a160c..8ab27911 100644 --- a/tests/unittests/test_handler/test_handler_chef.py +++ b/tests/unittests/test_handler/test_handler_chef.py @@ -11,6 +11,7 @@ from cloudinit.sources import DataSourceNone from .. import helpers as t_help +import six import logging import shutil import tempfile @@ -77,7 +78,7 @@ class TestChef(t_help.FilesystemMockingTestCase): for k, v in cfg['chef'].items(): self.assertIn(v, c) for k, v in cc_chef.CHEF_RB_TPL_DEFAULTS.items(): - if isinstance(v, basestring): + if isinstance(v, six.string_types): self.assertIn(v, c) c = util.load_file(cc_chef.CHEF_FB_PATH) self.assertEqual({}, json.loads(c)) diff --git a/tests/unittests/test_handler/test_handler_seed_random.py b/tests/unittests/test_handler/test_handler_seed_random.py index 579377fb..c2da5ced 100644 --- a/tests/unittests/test_handler/test_handler_seed_random.py +++ b/tests/unittests/test_handler/test_handler_seed_random.py @@ -38,6 +38,13 @@ import logging LOG = logging.getLogger(__name__) +def b64(source): + # In Python 3, b64encode only accepts bytes and returns bytes. + if not isinstance(source, bytes): + source = source.encode('utf-8') + return base64.b64encode(source).decode('us-ascii') + + class TestRandomSeed(t_help.TestCase): def setUp(self): super(TestRandomSeed, self).setUp() @@ -134,7 +141,7 @@ class TestRandomSeed(t_help.TestCase): self.assertEquals("big-toe", contents) def test_append_random_base64(self): - data = base64.b64encode('bubbles') + data = b64('bubbles') cfg = { 'random_seed': { 'file': self._seed_file, @@ -147,7 +154,7 @@ class TestRandomSeed(t_help.TestCase): self.assertEquals("bubbles", contents) def test_append_random_b64(self): - data = base64.b64encode('kit-kat') + data = b64('kit-kat') cfg = { 'random_seed': { 'file': self._seed_file, diff --git a/tests/unittests/test_util.py b/tests/unittests/test_util.py index b1f5d62c..b0207ace 100644 --- a/tests/unittests/test_util.py +++ b/tests/unittests/test_util.py @@ -119,7 +119,7 @@ class TestWriteFile(unittest.TestCase): # Create file first with basic content with open(path, "wb") as f: - f.write("LINE1\n") + f.write(b"LINE1\n") util.write_file(path, contents, omode="a") self.assertTrue(os.path.exists(path)) @@ -194,7 +194,7 @@ class TestDeleteDirContents(unittest.TestCase): os.mkdir(os.path.join(self.tmp, "new_dir")) f_name = os.path.join(self.tmp, "new_dir", "new_file.txt") with open(f_name, "wb") as f: - f.write("DELETE ME") + f.write(b"DELETE ME") util.delete_dir_contents(self.tmp) @@ -205,7 +205,7 @@ class TestDeleteDirContents(unittest.TestCase): file_name = os.path.join(self.tmp, "new_file.txt") link_name = os.path.join(self.tmp, "new_file_link.txt") with open(file_name, "wb") as f: - f.write("DELETE ME") + f.write(b"DELETE ME") os.symlink(file_name, link_name) util.delete_dir_contents(self.tmp) |