diff options
Diffstat (limited to 'tests/unittests/sources/test_maas.py')
| -rw-r--r-- | tests/unittests/sources/test_maas.py | 227 |
1 files changed, 227 insertions, 0 deletions
diff --git a/tests/unittests/sources/test_maas.py b/tests/unittests/sources/test_maas.py new file mode 100644 index 00000000..e95ba374 --- /dev/null +++ b/tests/unittests/sources/test_maas.py @@ -0,0 +1,227 @@ +# This file is part of cloud-init. See LICENSE file for license information. + +import os +import shutil +import tempfile +from copy import copy +from unittest import mock + +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. + self.tmp = tempfile.mkdtemp() + self.addCleanup(shutil.rmtree, self.tmp) + + 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", + } + + my_d = os.path.join(self.tmp, "valid") + populate_dir(my_d, data) + + 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]) + + # verify that 'userdata' is not returned as part of the metadata + 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", + } + + my_d = os.path.join(self.tmp, "valid_extra") + populate_dir(my_d, data) + + 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]) + + # additional files should not just appear as keys in metadata atm + 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": "", + } + + 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"] + populate_dir(my_d, invalid_data) + 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"] + populate_dir(my_d, invalid_data) + 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, + ) + + 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"), + ) + + 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"] + prefix = "%s/%s/" % (seed, version) + if not url.startswith(prefix): + raise ValueError("unexpected call %s" % url) + + 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]) + + # Now do the actual call of the code under test. + with mock.patch("cloudinit.url_helper.readurl") as mock_readurl: + mock_readurl.side_effect = my_readurl + return DataSourceMAAS.read_maas_seed_url(seed, version=version) + + 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", + } + 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/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) + + def test_seed_url_vendor_data_dict(self): + 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(), + } + _ud, md, vd = self.mock_read_maas_seed_url( + valid, "http://example.com/foo" + ) + + 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, + } + + def test_all_required(self, m_helper): + """Valid config as expected.""" + DataSourceMAAS.get_oauth_helper(self.base_cfg.copy()) + m_helper.assert_has_calls([mock.call(**self.base_cfg)]) + + 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" + 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_id = ( + "v1:403ee5f19c956507f1d0e50814119c405902137ea4f8838bde167c5da8110392" + ) + + def test_v1_expected(self): + """Test v1 id generated as expected working behavior from config.""" + result = DataSourceMAAS.get_id_from_ds_cfg(self.v1_cfg.copy()) + self.assertEqual(self.v1_id, result) + + 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" + result = DataSourceMAAS.get_id_from_ds_cfg(cfg) + self.assertEqual(self.v1_id, result) + + +# vi: ts=4 expandtab |