diff options
Diffstat (limited to 'tests/unittests/sources/test_maas.py')
| -rw-r--r-- | tests/unittests/sources/test_maas.py | 147 |
1 files changed, 87 insertions, 60 deletions
diff --git a/tests/unittests/sources/test_maas.py b/tests/unittests/sources/test_maas.py index 34b79587..e95ba374 100644 --- a/tests/unittests/sources/test_maas.py +++ b/tests/unittests/sources/test_maas.py @@ -1,19 +1,19 @@ # This file is part of cloud-init. See LICENSE file for license information. -from copy import copy import os import shutil import tempfile -import yaml +from copy import copy from unittest import mock -from cloudinit.sources import DataSourceMAAS +import yaml + from cloudinit import url_helper +from cloudinit.sources import DataSourceMAAS from tests.unittests.helpers import CiTestCase, populate_dir class TestMAASDataSource(CiTestCase): - def setUp(self): super(TestMAASDataSource, self).setUp() # Make a temp directoy for tests to use. @@ -23,11 +23,13 @@ class TestMAASDataSource(CiTestCase): def test_seed_dir_valid(self): """Verify a valid seeddir is read as such.""" - userdata = b'valid01-userdata' - data = {'meta-data/instance-id': 'i-valid01', - 'meta-data/local-hostname': 'valid01-hostname', - 'user-data': userdata, - 'public-keys': 'ssh-rsa AAAAB3Nz...aC1yc2E= keyname'} + userdata = b"valid01-userdata" + data = { + "meta-data/instance-id": "i-valid01", + "meta-data/local-hostname": "valid01-hostname", + "user-data": userdata, + "public-keys": "ssh-rsa AAAAB3Nz...aC1yc2E= keyname", + } my_d = os.path.join(self.tmp, "valid") populate_dir(my_d, data) @@ -35,20 +37,23 @@ class TestMAASDataSource(CiTestCase): ud, md, vd = DataSourceMAAS.read_maas_seed_dir(my_d) self.assertEqual(userdata, ud) - for key in ('instance-id', 'local-hostname'): + for key in ("instance-id", "local-hostname"): self.assertEqual(data["meta-data/" + key], md[key]) # verify that 'userdata' is not returned as part of the metadata - self.assertFalse(('user-data' in md)) + self.assertFalse(("user-data" in md)) self.assertIsNone(vd) def test_seed_dir_valid_extra(self): """Verify extra files do not affect seed_dir validity.""" - userdata = b'valid-extra-userdata' - data = {'meta-data/instance-id': 'i-valid-extra', - 'meta-data/local-hostname': 'valid-extra-hostname', - 'user-data': userdata, 'foo': 'bar'} + userdata = b"valid-extra-userdata" + data = { + "meta-data/instance-id": "i-valid-extra", + "meta-data/local-hostname": "valid-extra-hostname", + "user-data": userdata, + "foo": "bar", + } my_d = os.path.join(self.tmp, "valid_extra") populate_dir(my_d, data) @@ -56,62 +61,77 @@ class TestMAASDataSource(CiTestCase): ud, md, _vd = DataSourceMAAS.read_maas_seed_dir(my_d) self.assertEqual(userdata, ud) - for key in ('instance-id', 'local-hostname'): - self.assertEqual(data['meta-data/' + key], md[key]) + for key in ("instance-id", "local-hostname"): + self.assertEqual(data["meta-data/" + key], md[key]) # additional files should not just appear as keys in metadata atm - self.assertFalse(('foo' in md)) + self.assertFalse(("foo" in md)) def test_seed_dir_invalid(self): """Verify that invalid seed_dir raises MAASSeedDirMalformed.""" - valid = {'instance-id': 'i-instanceid', - 'local-hostname': 'test-hostname', 'user-data': ''} + valid = { + "instance-id": "i-instanceid", + "local-hostname": "test-hostname", + "user-data": "", + } my_based = os.path.join(self.tmp, "valid_extra") # missing 'userdata' file my_d = "%s-01" % my_based invalid_data = copy(valid) - del invalid_data['local-hostname'] + del invalid_data["local-hostname"] populate_dir(my_d, invalid_data) - self.assertRaises(DataSourceMAAS.MAASSeedDirMalformed, - DataSourceMAAS.read_maas_seed_dir, my_d) + self.assertRaises( + DataSourceMAAS.MAASSeedDirMalformed, + DataSourceMAAS.read_maas_seed_dir, + my_d, + ) # missing 'instance-id' my_d = "%s-02" % my_based invalid_data = copy(valid) - del invalid_data['instance-id'] + del invalid_data["instance-id"] populate_dir(my_d, invalid_data) - self.assertRaises(DataSourceMAAS.MAASSeedDirMalformed, - DataSourceMAAS.read_maas_seed_dir, my_d) + self.assertRaises( + DataSourceMAAS.MAASSeedDirMalformed, + DataSourceMAAS.read_maas_seed_dir, + my_d, + ) def test_seed_dir_none(self): """Verify that empty seed_dir raises MAASSeedDirNone.""" my_d = os.path.join(self.tmp, "valid_empty") - self.assertRaises(DataSourceMAAS.MAASSeedDirNone, - DataSourceMAAS.read_maas_seed_dir, my_d) + self.assertRaises( + DataSourceMAAS.MAASSeedDirNone, + DataSourceMAAS.read_maas_seed_dir, + my_d, + ) def test_seed_dir_missing(self): """Verify that missing seed_dir raises MAASSeedDirNone.""" - self.assertRaises(DataSourceMAAS.MAASSeedDirNone, - DataSourceMAAS.read_maas_seed_dir, - os.path.join(self.tmp, "nonexistantdirectory")) + self.assertRaises( + DataSourceMAAS.MAASSeedDirNone, + DataSourceMAAS.read_maas_seed_dir, + os.path.join(self.tmp, "nonexistantdirectory"), + ) def mock_read_maas_seed_url(self, data, seed, version="19991231"): """mock up readurl to appear as a web server at seed has provided data. return what read_maas_seed_url returns.""" + def my_readurl(*args, **kwargs): if len(args): url = args[0] else: - url = kwargs['url'] + url = kwargs["url"] prefix = "%s/%s/" % (seed, version) if not url.startswith(prefix): raise ValueError("unexpected call %s" % url) - short = url[len(prefix):] + short = url[len(prefix) :] if short not in data: raise url_helper.UrlError("not found", code=404, url=url) return url_helper.StringResponse(data[short]) @@ -124,44 +144,48 @@ class TestMAASDataSource(CiTestCase): def test_seed_url_valid(self): """Verify that valid seed_url is read as such.""" valid = { - 'meta-data/instance-id': 'i-instanceid', - 'meta-data/local-hostname': 'test-hostname', - 'meta-data/public-keys': 'test-hostname', - 'meta-data/vendor-data': b'my-vendordata', - 'user-data': b'foodata', + "meta-data/instance-id": "i-instanceid", + "meta-data/local-hostname": "test-hostname", + "meta-data/public-keys": "test-hostname", + "meta-data/vendor-data": b"my-vendordata", + "user-data": b"foodata", } my_seed = "http://example.com/xmeta" my_ver = "1999-99-99" ud, md, vd = self.mock_read_maas_seed_url(valid, my_seed, my_ver) - self.assertEqual(valid['meta-data/instance-id'], md['instance-id']) + self.assertEqual(valid["meta-data/instance-id"], md["instance-id"]) self.assertEqual( - valid['meta-data/local-hostname'], md['local-hostname']) - self.assertEqual(valid['meta-data/public-keys'], md['public-keys']) - self.assertEqual(valid['user-data'], ud) + valid["meta-data/local-hostname"], md["local-hostname"] + ) + self.assertEqual(valid["meta-data/public-keys"], md["public-keys"]) + self.assertEqual(valid["user-data"], ud) # vendor-data is yaml, which decodes a string - self.assertEqual(valid['meta-data/vendor-data'].decode(), vd) + self.assertEqual(valid["meta-data/vendor-data"].decode(), vd) def test_seed_url_vendor_data_dict(self): - expected_vd = {'key1': 'value1'} + expected_vd = {"key1": "value1"} valid = { - 'meta-data/instance-id': 'i-instanceid', - 'meta-data/local-hostname': 'test-hostname', - 'meta-data/vendor-data': yaml.safe_dump(expected_vd).encode(), + "meta-data/instance-id": "i-instanceid", + "meta-data/local-hostname": "test-hostname", + "meta-data/vendor-data": yaml.safe_dump(expected_vd).encode(), } _ud, md, vd = self.mock_read_maas_seed_url( - valid, "http://example.com/foo") + valid, "http://example.com/foo" + ) - self.assertEqual(valid['meta-data/instance-id'], md['instance-id']) + self.assertEqual(valid["meta-data/instance-id"], md["instance-id"]) self.assertEqual(expected_vd, vd) @mock.patch("cloudinit.sources.DataSourceMAAS.url_helper.OauthUrlHelper") class TestGetOauthHelper(CiTestCase): - base_cfg = {'consumer_key': 'FAKE_CONSUMER_KEY', - 'token_key': 'FAKE_TOKEN_KEY', - 'token_secret': 'FAKE_TOKEN_SECRET', - 'consumer_secret': None} + base_cfg = { + "consumer_key": "FAKE_CONSUMER_KEY", + "token_key": "FAKE_TOKEN_KEY", + "token_secret": "FAKE_TOKEN_SECRET", + "consumer_secret": None, + } def test_all_required(self, m_helper): """Valid config as expected.""" @@ -171,17 +195,20 @@ class TestGetOauthHelper(CiTestCase): def test_other_fields_not_passed_through(self, m_helper): """Only relevant fields are passed through.""" mycfg = self.base_cfg.copy() - mycfg['unrelated_field'] = 'unrelated' + mycfg["unrelated_field"] = "unrelated" DataSourceMAAS.get_oauth_helper(mycfg) m_helper.assert_has_calls([mock.call(**self.base_cfg)]) class TestGetIdHash(CiTestCase): - v1_cfg = {'consumer_key': 'CKEY', 'token_key': 'TKEY', - 'token_secret': 'TSEC'} + v1_cfg = { + "consumer_key": "CKEY", + "token_key": "TKEY", + "token_secret": "TSEC", + } v1_id = ( - 'v1:' - '403ee5f19c956507f1d0e50814119c405902137ea4f8838bde167c5da8110392') + "v1:403ee5f19c956507f1d0e50814119c405902137ea4f8838bde167c5da8110392" + ) def test_v1_expected(self): """Test v1 id generated as expected working behavior from config.""" @@ -191,8 +218,8 @@ class TestGetIdHash(CiTestCase): def test_v1_extra_fields_are_ignored(self): """Test v1 id ignores unused entries in config.""" cfg = self.v1_cfg.copy() - cfg['consumer_secret'] = "BOO" - cfg['unrelated'] = "HI MOM" + cfg["consumer_secret"] = "BOO" + cfg["unrelated"] = "HI MOM" result = DataSourceMAAS.get_id_from_ds_cfg(cfg) self.assertEqual(self.v1_id, result) |