diff options
| author | James Falcon <james.falcon@canonical.com> | 2021-12-15 20:16:38 -0600 |
|---|---|---|
| committer | GitHub <noreply@github.com> | 2021-12-15 19:16:38 -0700 |
| commit | bae9b11da9ed7dd0b16fe5adeaf4774b7cc628cf (patch) | |
| tree | 1fbb3269fc87e39832e3286ef42eefd2b23fcd44 /tests/unittests/sources/test_gce.py | |
| parent | 2bcf4fa972fde686c2e3141c58e640640b44dd00 (diff) | |
| download | vyos-cloud-init-bae9b11da9ed7dd0b16fe5adeaf4774b7cc628cf.tar.gz vyos-cloud-init-bae9b11da9ed7dd0b16fe5adeaf4774b7cc628cf.zip | |
Adopt Black and isort (SC-700) (#1157)
Applied Black and isort, fixed any linting issues, updated tox.ini
and CI.
Diffstat (limited to 'tests/unittests/sources/test_gce.py')
| -rw-r--r-- | tests/unittests/sources/test_gce.py | 304 |
1 files changed, 166 insertions, 138 deletions
diff --git a/tests/unittests/sources/test_gce.py b/tests/unittests/sources/test_gce.py index dc768e99..e030931b 100644 --- a/tests/unittests/sources/test_gce.py +++ b/tests/unittests/sources/test_gce.py @@ -5,58 +5,57 @@ # This file is part of cloud-init. See LICENSE file for license information. import datetime -import httpretty import json import re +from base64 import b64decode, b64encode from unittest import mock from urllib.parse import urlparse -from base64 import b64encode, b64decode +import httpretty -from cloudinit import distros -from cloudinit import helpers -from cloudinit import settings +from cloudinit import distros, helpers, settings from cloudinit.sources import DataSourceGCE - from tests.unittests import helpers as test_helpers - GCE_META = { - 'instance/id': '123', - 'instance/zone': 'foo/bar', - 'instance/hostname': 'server.project-foo.local', + "instance/id": "123", + "instance/zone": "foo/bar", + "instance/hostname": "server.project-foo.local", } GCE_META_PARTIAL = { - 'instance/id': '1234', - 'instance/hostname': 'server.project-bar.local', - 'instance/zone': 'bar/baz', + "instance/id": "1234", + "instance/hostname": "server.project-bar.local", + "instance/zone": "bar/baz", } GCE_META_ENCODING = { - 'instance/id': '12345', - 'instance/hostname': 'server.project-baz.local', - 'instance/zone': 'baz/bang', - 'instance/attributes': { - 'user-data': b64encode(b'#!/bin/echo baz\n').decode('utf-8'), - 'user-data-encoding': 'base64', - } + "instance/id": "12345", + "instance/hostname": "server.project-baz.local", + "instance/zone": "baz/bang", + "instance/attributes": { + "user-data": b64encode(b"#!/bin/echo baz\n").decode("utf-8"), + "user-data-encoding": "base64", + }, } GCE_USER_DATA_TEXT = { - 'instance/id': '12345', - 'instance/hostname': 'server.project-baz.local', - 'instance/zone': 'baz/bang', - 'instance/attributes': { - 'user-data': '#!/bin/sh\necho hi mom\ntouch /run/up-now\n', - } + "instance/id": "12345", + "instance/hostname": "server.project-baz.local", + "instance/zone": "baz/bang", + "instance/attributes": { + "user-data": "#!/bin/sh\necho hi mom\ntouch /run/up-now\n", + }, } -HEADERS = {'Metadata-Flavor': 'Google'} +HEADERS = {"Metadata-Flavor": "Google"} MD_URL_RE = re.compile( - r'http://metadata.google.internal/computeMetadata/v1/.*') -GUEST_ATTRIBUTES_URL = ('http://metadata.google.internal/computeMetadata/' - 'v1/instance/guest-attributes/hostkeys/') + r"http://metadata.google.internal/computeMetadata/v1/.*" +) +GUEST_ATTRIBUTES_URL = ( + "http://metadata.google.internal/computeMetadata/" + "v1/instance/guest-attributes/hostkeys/" +) def _set_mock_metadata(gce_meta=None): @@ -65,10 +64,10 @@ def _set_mock_metadata(gce_meta=None): def _request_callback(method, uri, headers): url_path = urlparse(uri).path - if url_path.startswith('/computeMetadata/v1/'): - path = url_path.split('/computeMetadata/v1/')[1:][0] - recursive = path.endswith('/') - path = path.rstrip('/') + if url_path.startswith("/computeMetadata/v1/"): + path = url_path.split("/computeMetadata/v1/")[1:][0] + recursive = path.endswith("/") + path = path.rstrip("/") else: path = None if path in gce_meta: @@ -77,7 +76,7 @@ def _set_mock_metadata(gce_meta=None): response = json.dumps(response) return (200, headers, response) else: - return (404, headers, '') + return (404, headers, "") # reset is needed. https://github.com/gabrielfalcao/HTTPretty/issues/316 httpretty.register_uri(httpretty.GET, MD_URL_RE, body=_request_callback) @@ -85,28 +84,28 @@ def _set_mock_metadata(gce_meta=None): @httpretty.activate class TestDataSourceGCE(test_helpers.HttprettyTestCase): - def _make_distro(self, dtype, def_user=None): cfg = dict(settings.CFG_BUILTIN) - cfg['system_info']['distro'] = dtype - paths = helpers.Paths(cfg['system_info']['paths']) + cfg["system_info"]["distro"] = dtype + paths = helpers.Paths(cfg["system_info"]["paths"]) distro_cls = distros.fetch(dtype) if def_user: - cfg['system_info']['default_user'] = def_user.copy() - distro = distro_cls(dtype, cfg['system_info'], paths) + cfg["system_info"]["default_user"] = def_user.copy() + distro = distro_cls(dtype, cfg["system_info"], paths) return distro def setUp(self): tmp = self.tmp_dir() self.ds = DataSourceGCE.DataSourceGCE( - settings.CFG_BUILTIN, None, - helpers.Paths({'run_dir': tmp})) + settings.CFG_BUILTIN, None, helpers.Paths({"run_dir": tmp}) + ) ppatch = self.m_platform_reports_gce = mock.patch( - 'cloudinit.sources.DataSourceGCE.platform_reports_gce') + "cloudinit.sources.DataSourceGCE.platform_reports_gce" + ) self.m_platform_reports_gce = ppatch.start() self.m_platform_reports_gce.return_value = True self.addCleanup(ppatch.stop) - self.add_patch('time.sleep', 'm_sleep') # just to speed up tests + self.add_patch("time.sleep", "m_sleep") # just to speed up tests super(TestDataSourceGCE, self).setUp() def test_connection(self): @@ -121,30 +120,33 @@ class TestDataSourceGCE(test_helpers.HttprettyTestCase): def test_metadata(self): # UnicodeDecodeError if set to ds.userdata instead of userdata_raw meta = GCE_META.copy() - meta['instance/attributes/user-data'] = b'/bin/echo \xff\n' + meta["instance/attributes/user-data"] = b"/bin/echo \xff\n" _set_mock_metadata() self.ds.get_data() - shostname = GCE_META.get('instance/hostname').split('.')[0] - self.assertEqual(shostname, - self.ds.get_hostname()) + shostname = GCE_META.get("instance/hostname").split(".")[0] + self.assertEqual(shostname, self.ds.get_hostname()) - self.assertEqual(GCE_META.get('instance/id'), - self.ds.get_instance_id()) + self.assertEqual( + GCE_META.get("instance/id"), self.ds.get_instance_id() + ) - self.assertEqual(GCE_META.get('instance/attributes/user-data'), - self.ds.get_userdata_raw()) + self.assertEqual( + GCE_META.get("instance/attributes/user-data"), + self.ds.get_userdata_raw(), + ) # test partial metadata (missing user-data in particular) def test_metadata_partial(self): _set_mock_metadata(GCE_META_PARTIAL) self.ds.get_data() - self.assertEqual(GCE_META_PARTIAL.get('instance/id'), - self.ds.get_instance_id()) + self.assertEqual( + GCE_META_PARTIAL.get("instance/id"), self.ds.get_instance_id() + ) - shostname = GCE_META_PARTIAL.get('instance/hostname').split('.')[0] + shostname = GCE_META_PARTIAL.get("instance/hostname").split(".")[0] self.assertEqual(shostname, self.ds.get_hostname()) def test_userdata_no_encoding(self): @@ -152,21 +154,25 @@ class TestDataSourceGCE(test_helpers.HttprettyTestCase): _set_mock_metadata(GCE_USER_DATA_TEXT) self.ds.get_data() self.assertEqual( - GCE_USER_DATA_TEXT['instance/attributes']['user-data'].encode(), - self.ds.get_userdata_raw()) + GCE_USER_DATA_TEXT["instance/attributes"]["user-data"].encode(), + self.ds.get_userdata_raw(), + ) def test_metadata_encoding(self): """user-data is base64 encoded if user-data-encoding is 'base64'.""" _set_mock_metadata(GCE_META_ENCODING) self.ds.get_data() - instance_data = GCE_META_ENCODING.get('instance/attributes') - decoded = b64decode(instance_data.get('user-data')) + instance_data = GCE_META_ENCODING.get("instance/attributes") + decoded = b64decode(instance_data.get("user-data")) self.assertEqual(decoded, self.ds.get_userdata_raw()) def test_missing_required_keys_return_false(self): - for required_key in ['instance/id', 'instance/zone', - 'instance/hostname']: + for required_key in [ + "instance/id", + "instance/zone", + "instance/hostname", + ]: meta = GCE_META_PARTIAL.copy() del meta[required_key] _set_mock_metadata(meta) @@ -179,29 +185,35 @@ class TestDataSourceGCE(test_helpers.HttprettyTestCase): self.assertEqual([], self.ds.get_public_ssh_keys()) def test_cloudinit_ssh_keys(self): - valid_key = 'ssh-rsa VALID {0}' - invalid_key = 'ssh-rsa INVALID {0}' + valid_key = "ssh-rsa VALID {0}" + invalid_key = "ssh-rsa INVALID {0}" project_attributes = { - 'sshKeys': '\n'.join([ - 'cloudinit:{0}'.format(valid_key.format(0)), - 'user:{0}'.format(invalid_key.format(0)), - ]), - 'ssh-keys': '\n'.join([ - 'cloudinit:{0}'.format(valid_key.format(1)), - 'user:{0}'.format(invalid_key.format(1)), - ]), + "sshKeys": "\n".join( + [ + "cloudinit:{0}".format(valid_key.format(0)), + "user:{0}".format(invalid_key.format(0)), + ] + ), + "ssh-keys": "\n".join( + [ + "cloudinit:{0}".format(valid_key.format(1)), + "user:{0}".format(invalid_key.format(1)), + ] + ), } instance_attributes = { - 'ssh-keys': '\n'.join([ - 'cloudinit:{0}'.format(valid_key.format(2)), - 'user:{0}'.format(invalid_key.format(2)), - ]), - 'block-project-ssh-keys': 'False', + "ssh-keys": "\n".join( + [ + "cloudinit:{0}".format(valid_key.format(2)), + "user:{0}".format(invalid_key.format(2)), + ] + ), + "block-project-ssh-keys": "False", } meta = GCE_META.copy() - meta['project/attributes'] = project_attributes - meta['instance/attributes'] = instance_attributes + meta["project/attributes"] = project_attributes + meta["instance/attributes"] = instance_attributes _set_mock_metadata(meta) self.ds.get_data() @@ -212,34 +224,42 @@ class TestDataSourceGCE(test_helpers.HttprettyTestCase): @mock.patch("cloudinit.sources.DataSourceGCE.ug_util") def test_default_user_ssh_keys(self, mock_ug_util): mock_ug_util.normalize_users_groups.return_value = None, None - mock_ug_util.extract_default.return_value = 'ubuntu', None + mock_ug_util.extract_default.return_value = "ubuntu", None ubuntu_ds = DataSourceGCE.DataSourceGCE( - settings.CFG_BUILTIN, self._make_distro('ubuntu'), - helpers.Paths({'run_dir': self.tmp_dir()})) + settings.CFG_BUILTIN, + self._make_distro("ubuntu"), + helpers.Paths({"run_dir": self.tmp_dir()}), + ) - valid_key = 'ssh-rsa VALID {0}' - invalid_key = 'ssh-rsa INVALID {0}' + valid_key = "ssh-rsa VALID {0}" + invalid_key = "ssh-rsa INVALID {0}" project_attributes = { - 'sshKeys': '\n'.join([ - 'ubuntu:{0}'.format(valid_key.format(0)), - 'user:{0}'.format(invalid_key.format(0)), - ]), - 'ssh-keys': '\n'.join([ - 'ubuntu:{0}'.format(valid_key.format(1)), - 'user:{0}'.format(invalid_key.format(1)), - ]), + "sshKeys": "\n".join( + [ + "ubuntu:{0}".format(valid_key.format(0)), + "user:{0}".format(invalid_key.format(0)), + ] + ), + "ssh-keys": "\n".join( + [ + "ubuntu:{0}".format(valid_key.format(1)), + "user:{0}".format(invalid_key.format(1)), + ] + ), } instance_attributes = { - 'ssh-keys': '\n'.join([ - 'ubuntu:{0}'.format(valid_key.format(2)), - 'user:{0}'.format(invalid_key.format(2)), - ]), - 'block-project-ssh-keys': 'False', + "ssh-keys": "\n".join( + [ + "ubuntu:{0}".format(valid_key.format(2)), + "user:{0}".format(invalid_key.format(2)), + ] + ), + "block-project-ssh-keys": "False", } meta = GCE_META.copy() - meta['project/attributes'] = project_attributes - meta['instance/attributes'] = instance_attributes + meta["project/attributes"] = project_attributes + meta["instance/attributes"] = instance_attributes _set_mock_metadata(meta) ubuntu_ds.get_data() @@ -248,21 +268,21 @@ class TestDataSourceGCE(test_helpers.HttprettyTestCase): self.assertEqual(set(expected), set(ubuntu_ds.get_public_ssh_keys())) def test_instance_ssh_keys_override(self): - valid_key = 'ssh-rsa VALID {0}' - invalid_key = 'ssh-rsa INVALID {0}' + valid_key = "ssh-rsa VALID {0}" + invalid_key = "ssh-rsa INVALID {0}" project_attributes = { - 'sshKeys': 'cloudinit:{0}'.format(invalid_key.format(0)), - 'ssh-keys': 'cloudinit:{0}'.format(invalid_key.format(1)), + "sshKeys": "cloudinit:{0}".format(invalid_key.format(0)), + "ssh-keys": "cloudinit:{0}".format(invalid_key.format(1)), } instance_attributes = { - 'sshKeys': 'cloudinit:{0}'.format(valid_key.format(0)), - 'ssh-keys': 'cloudinit:{0}'.format(valid_key.format(1)), - 'block-project-ssh-keys': 'False', + "sshKeys": "cloudinit:{0}".format(valid_key.format(0)), + "ssh-keys": "cloudinit:{0}".format(valid_key.format(1)), + "block-project-ssh-keys": "False", } meta = GCE_META.copy() - meta['project/attributes'] = project_attributes - meta['instance/attributes'] = instance_attributes + meta["project/attributes"] = project_attributes + meta["instance/attributes"] = instance_attributes _set_mock_metadata(meta) self.ds.get_data() @@ -271,20 +291,20 @@ class TestDataSourceGCE(test_helpers.HttprettyTestCase): self.assertEqual(set(expected), set(self.ds.get_public_ssh_keys())) def test_block_project_ssh_keys_override(self): - valid_key = 'ssh-rsa VALID {0}' - invalid_key = 'ssh-rsa INVALID {0}' + valid_key = "ssh-rsa VALID {0}" + invalid_key = "ssh-rsa INVALID {0}" project_attributes = { - 'sshKeys': 'cloudinit:{0}'.format(invalid_key.format(0)), - 'ssh-keys': 'cloudinit:{0}'.format(invalid_key.format(1)), + "sshKeys": "cloudinit:{0}".format(invalid_key.format(0)), + "ssh-keys": "cloudinit:{0}".format(invalid_key.format(1)), } instance_attributes = { - 'ssh-keys': 'cloudinit:{0}'.format(valid_key.format(0)), - 'block-project-ssh-keys': 'True', + "ssh-keys": "cloudinit:{0}".format(valid_key.format(0)), + "block-project-ssh-keys": "True", } meta = GCE_META.copy() - meta['project/attributes'] = project_attributes - meta['instance/attributes'] = instance_attributes + meta["project/attributes"] = project_attributes + meta["instance/attributes"] = instance_attributes _set_mock_metadata(meta) self.ds.get_data() @@ -296,7 +316,7 @@ class TestDataSourceGCE(test_helpers.HttprettyTestCase): _set_mock_metadata() r = self.ds.get_data() self.assertEqual(True, r) - self.assertEqual('bar', self.ds.availability_zone) + self.assertEqual("bar", self.ds.availability_zone) @mock.patch("cloudinit.sources.DataSourceGCE.GoogleMetadataFetcher") def test_get_data_returns_false_if_not_on_gce(self, m_fetcher): @@ -306,9 +326,8 @@ class TestDataSourceGCE(test_helpers.HttprettyTestCase): m_fetcher.assert_not_called() def test_has_expired(self): - def _get_timestamp(days): - format_str = '%Y-%m-%dT%H:%M:%S+0000' + format_str = "%Y-%m-%dT%H:%M:%S+0000" today = datetime.datetime.now() timestamp = today + datetime.timedelta(days=days) return timestamp.strftime(format_str) @@ -317,12 +336,12 @@ class TestDataSourceGCE(test_helpers.HttprettyTestCase): future = _get_timestamp(1) ssh_keys = { None: False, - '': False, - 'Invalid': False, - 'user:ssh-rsa key user@domain.com': False, + "": False, + "Invalid": False, + "user:ssh-rsa key user@domain.com": False, 'user:ssh-rsa key google {"expireOn":"%s"}' % past: False, - 'user:ssh-rsa key google-ssh': False, - 'user:ssh-rsa key google-ssh {invalid:json}': False, + "user:ssh-rsa key google-ssh": False, + "user:ssh-rsa key google-ssh {invalid:json}": False, 'user:ssh-rsa key google-ssh {"userName":"user"}': False, 'user:ssh-rsa key google-ssh {"expireOn":"invalid"}': False, 'user:xyz key google-ssh {"expireOn":"%s"}' % future: False, @@ -334,28 +353,36 @@ class TestDataSourceGCE(test_helpers.HttprettyTestCase): def test_parse_public_keys_non_ascii(self): public_key_data = [ - 'cloudinit:rsa ssh-ke%s invalid' % chr(165), - 'use%sname:rsa ssh-key' % chr(174), - 'cloudinit:test 1', - 'default:test 2', - 'user:test 3', + "cloudinit:rsa ssh-ke%s invalid" % chr(165), + "use%sname:rsa ssh-key" % chr(174), + "cloudinit:test 1", + "default:test 2", + "user:test 3", ] - expected = ['test 1', 'test 2'] + expected = ["test 1", "test 2"] found = DataSourceGCE._parse_public_keys( - public_key_data, default_user='default') + public_key_data, default_user="default" + ) self.assertEqual(sorted(found), sorted(expected)) @mock.patch("cloudinit.url_helper.readurl") def test_publish_host_keys(self, m_readurl): - hostkeys = [('ssh-rsa', 'asdfasdf'), - ('ssh-ed25519', 'qwerqwer')] + hostkeys = [("ssh-rsa", "asdfasdf"), ("ssh-ed25519", "qwerqwer")] readurl_expected_calls = [ - mock.call(check_status=False, data=b'asdfasdf', headers=HEADERS, - request_method='PUT', - url='%s%s' % (GUEST_ATTRIBUTES_URL, 'ssh-rsa')), - mock.call(check_status=False, data=b'qwerqwer', headers=HEADERS, - request_method='PUT', - url='%s%s' % (GUEST_ATTRIBUTES_URL, 'ssh-ed25519')), + mock.call( + check_status=False, + data=b"asdfasdf", + headers=HEADERS, + request_method="PUT", + url="%s%s" % (GUEST_ATTRIBUTES_URL, "ssh-rsa"), + ), + mock.call( + check_status=False, + data=b"qwerqwer", + headers=HEADERS, + request_method="PUT", + url="%s%s" % (GUEST_ATTRIBUTES_URL, "ssh-ed25519"), + ), ] self.ds.publish_host_keys(hostkeys) m_readurl.assert_has_calls(readurl_expected_calls, any_order=True) @@ -385,4 +412,5 @@ class TestDataSourceGCE(test_helpers.HttprettyTestCase): ds._get_data() assert m_dhcp.call_count == 0 + # vi: ts=4 expandtab |