diff options
Diffstat (limited to 'tests/unittests/test_datasource/test_nocloud.py')
| -rw-r--r-- | tests/unittests/test_datasource/test_nocloud.py | 139 |
1 files changed, 98 insertions, 41 deletions
diff --git a/tests/unittests/test_datasource/test_nocloud.py b/tests/unittests/test_datasource/test_nocloud.py index 21931eb7..3429272c 100644 --- a/tests/unittests/test_datasource/test_nocloud.py +++ b/tests/unittests/test_datasource/test_nocloud.py @@ -1,7 +1,10 @@ # This file is part of cloud-init. See LICENSE file for license information. from cloudinit import helpers -from cloudinit.sources import DataSourceNoCloud +from cloudinit.sources.DataSourceNoCloud import ( + DataSourceNoCloud as dsNoCloud, + _maybe_remove_top_network, + parse_cmdline_data) from cloudinit import util from cloudinit.tests.helpers import CiTestCase, populate_dir, mock, ExitStack @@ -10,6 +13,7 @@ import textwrap import yaml +@mock.patch('cloudinit.sources.DataSourceNoCloud.util.is_lxd') class TestNoCloudDataSource(CiTestCase): def setUp(self): @@ -28,28 +32,46 @@ class TestNoCloudDataSource(CiTestCase): self.mocks.enter_context( mock.patch.object(util, 'read_dmi_data', return_value=None)) - def test_nocloud_seed_dir(self): + def test_nocloud_seed_dir_on_lxd(self, m_is_lxd): md = {'instance-id': 'IID', 'dsmode': 'local'} ud = b"USER_DATA_HERE" - populate_dir(os.path.join(self.paths.seed_dir, "nocloud"), + seed_dir = os.path.join(self.paths.seed_dir, "nocloud") + populate_dir(seed_dir, {'user-data': ud, 'meta-data': yaml.safe_dump(md)}) sys_cfg = { 'datasource': {'NoCloud': {'fs_label': None}} } - ds = DataSourceNoCloud.DataSourceNoCloud - - dsrc = ds(sys_cfg=sys_cfg, distro=None, paths=self.paths) + dsrc = dsNoCloud(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.platform_type, 'lxd') + self.assertEqual( + dsrc.subplatform, 'seed-dir (%s)' % seed_dir) self.assertTrue(ret) - def test_fs_label(self): - # find_devs_with should not be called ff fs_label is None - ds = DataSourceNoCloud.DataSourceNoCloud + def test_nocloud_seed_dir_non_lxd_platform_is_nocloud(self, m_is_lxd): + """Non-lxd environments will list nocloud as the platform.""" + m_is_lxd.return_value = False + md = {'instance-id': 'IID', 'dsmode': 'local'} + seed_dir = os.path.join(self.paths.seed_dir, "nocloud") + populate_dir(seed_dir, + {'user-data': '', 'meta-data': yaml.safe_dump(md)}) + + sys_cfg = { + 'datasource': {'NoCloud': {'fs_label': None}} + } + + dsrc = dsNoCloud(sys_cfg=sys_cfg, distro=None, paths=self.paths) + self.assertTrue(dsrc.get_data()) + self.assertEqual(dsrc.platform_type, 'nocloud') + self.assertEqual( + dsrc.subplatform, 'seed-dir (%s)' % seed_dir) + def test_fs_label(self, m_is_lxd): + # find_devs_with should not be called ff fs_label is None class PsuedoException(Exception): pass @@ -59,26 +81,23 @@ class TestNoCloudDataSource(CiTestCase): # by default, NoCloud should search for filesystems by label sys_cfg = {'datasource': {'NoCloud': {}}} - dsrc = ds(sys_cfg=sys_cfg, distro=None, paths=self.paths) + dsrc = dsNoCloud(sys_cfg=sys_cfg, distro=None, paths=self.paths) self.assertRaises(PsuedoException, dsrc.get_data) # but disabling searching should just end up with None found sys_cfg = {'datasource': {'NoCloud': {'fs_label': None}}} - dsrc = ds(sys_cfg=sys_cfg, distro=None, paths=self.paths) + dsrc = dsNoCloud(sys_cfg=sys_cfg, distro=None, paths=self.paths) ret = dsrc.get_data() self.assertFalse(ret) - def test_no_datasource_expected(self): + def test_no_datasource_expected(self, m_is_lxd): # no source should be found if no cmdline, config, and fs_label=None sys_cfg = {'datasource': {'NoCloud': {'fs_label': None}}} - ds = DataSourceNoCloud.DataSourceNoCloud - dsrc = ds(sys_cfg=sys_cfg, distro=None, paths=self.paths) + dsrc = dsNoCloud(sys_cfg=sys_cfg, distro=None, paths=self.paths) self.assertFalse(dsrc.get_data()) - def test_seed_in_config(self): - ds = DataSourceNoCloud.DataSourceNoCloud - + def test_seed_in_config(self, m_is_lxd): data = { 'fs_label': None, 'meta-data': yaml.safe_dump({'instance-id': 'IID'}), @@ -86,13 +105,13 @@ class TestNoCloudDataSource(CiTestCase): } sys_cfg = {'datasource': {'NoCloud': data}} - dsrc = ds(sys_cfg=sys_cfg, distro=None, paths=self.paths) + dsrc = dsNoCloud(sys_cfg=sys_cfg, distro=None, paths=self.paths) ret = dsrc.get_data() self.assertEqual(dsrc.userdata_raw, b"USER_DATA_RAW") self.assertEqual(dsrc.metadata.get('instance-id'), 'IID') self.assertTrue(ret) - def test_nocloud_seed_with_vendordata(self): + def test_nocloud_seed_with_vendordata(self, m_is_lxd): md = {'instance-id': 'IID', 'dsmode': 'local'} ud = b"USER_DATA_HERE" vd = b"THIS IS MY VENDOR_DATA" @@ -105,30 +124,26 @@ class TestNoCloudDataSource(CiTestCase): 'datasource': {'NoCloud': {'fs_label': None}} } - ds = DataSourceNoCloud.DataSourceNoCloud - - dsrc = ds(sys_cfg=sys_cfg, distro=None, paths=self.paths) + dsrc = dsNoCloud(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_raw, vd) self.assertTrue(ret) - def test_nocloud_no_vendordata(self): + def test_nocloud_no_vendordata(self, m_is_lxd): populate_dir(os.path.join(self.paths.seed_dir, "nocloud"), {'user-data': b"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) + dsrc = dsNoCloud(sys_cfg=sys_cfg, distro=None, paths=self.paths) ret = dsrc.get_data() self.assertEqual(dsrc.userdata_raw, b"ud") self.assertFalse(dsrc.vendordata) self.assertTrue(ret) - def test_metadata_network_interfaces(self): + def test_metadata_network_interfaces(self, m_is_lxd): gateway = "103.225.10.1" md = { 'instance-id': 'i-abcd', @@ -149,15 +164,13 @@ class TestNoCloudDataSource(CiTestCase): sys_cfg = {'datasource': {'NoCloud': {'fs_label': None}}} - ds = DataSourceNoCloud.DataSourceNoCloud - - dsrc = ds(sys_cfg=sys_cfg, distro=None, paths=self.paths) + dsrc = dsNoCloud(sys_cfg=sys_cfg, distro=None, paths=self.paths) ret = dsrc.get_data() self.assertTrue(ret) # very simple check just for the strings above self.assertIn(gateway, str(dsrc.network_config)) - def test_metadata_network_config(self): + def test_metadata_network_config(self, m_is_lxd): # network-config needs to get into network_config netconf = {'version': 1, 'config': [{'type': 'physical', 'name': 'interface0', @@ -170,14 +183,28 @@ class TestNoCloudDataSource(CiTestCase): sys_cfg = {'datasource': {'NoCloud': {'fs_label': None}}} - ds = DataSourceNoCloud.DataSourceNoCloud + dsrc = dsNoCloud(sys_cfg=sys_cfg, distro=None, paths=self.paths) + ret = dsrc.get_data() + self.assertTrue(ret) + self.assertEqual(netconf, dsrc.network_config) - dsrc = ds(sys_cfg=sys_cfg, distro=None, paths=self.paths) + def test_metadata_network_config_with_toplevel_network(self, m_is_lxd): + """network-config may have 'network' top level key.""" + netconf = {'config': 'disabled'} + populate_dir( + os.path.join(self.paths.seed_dir, "nocloud"), + {'user-data': b"ud", + 'meta-data': "instance-id: IID\n", + 'network-config': yaml.dump({'network': netconf}) + "\n"}) + + sys_cfg = {'datasource': {'NoCloud': {'fs_label': None}}} + + dsrc = dsNoCloud(sys_cfg=sys_cfg, distro=None, paths=self.paths) ret = dsrc.get_data() self.assertTrue(ret) self.assertEqual(netconf, dsrc.network_config) - def test_metadata_network_config_over_interfaces(self): + def test_metadata_network_config_over_interfaces(self, m_is_lxd): # network-config should override meta-data/network-interfaces gateway = "103.225.10.1" md = { @@ -203,9 +230,7 @@ class TestNoCloudDataSource(CiTestCase): sys_cfg = {'datasource': {'NoCloud': {'fs_label': None}}} - ds = DataSourceNoCloud.DataSourceNoCloud - - dsrc = ds(sys_cfg=sys_cfg, distro=None, paths=self.paths) + dsrc = dsNoCloud(sys_cfg=sys_cfg, distro=None, paths=self.paths) ret = dsrc.get_data() self.assertTrue(ret) self.assertEqual(netconf, dsrc.network_config) @@ -233,8 +258,7 @@ class TestParseCommandLineData(CiTestCase): for (fmt, expected) in pairs: fill = {} cmdline = fmt % {'ds_id': ds_id} - ret = DataSourceNoCloud.parse_cmdline_data(ds_id=ds_id, fill=fill, - cmdline=cmdline) + ret = parse_cmdline_data(ds_id=ds_id, fill=fill, cmdline=cmdline) self.assertEqual(expected, fill) self.assertTrue(ret) @@ -251,10 +275,43 @@ class TestParseCommandLineData(CiTestCase): for cmdline in cmdlines: fill = {} - ret = DataSourceNoCloud.parse_cmdline_data(ds_id=ds_id, fill=fill, - cmdline=cmdline) + ret = parse_cmdline_data(ds_id=ds_id, fill=fill, cmdline=cmdline) self.assertEqual(fill, {}) self.assertFalse(ret) +class TestMaybeRemoveToplevelNetwork(CiTestCase): + """test _maybe_remove_top_network function.""" + basecfg = [{'type': 'physical', 'name': 'interface0', + 'subnets': [{'type': 'dhcp'}]}] + + def test_should_remove_safely(self): + mcfg = {'config': self.basecfg, 'version': 1} + self.assertEqual(mcfg, _maybe_remove_top_network({'network': mcfg})) + + def test_no_remove_if_other_keys(self): + """should not shift if other keys at top level.""" + mcfg = {'network': {'config': self.basecfg, 'version': 1}, + 'unknown_keyname': 'keyval'} + self.assertEqual(mcfg, _maybe_remove_top_network(mcfg)) + + def test_no_remove_if_non_dict(self): + """should not shift if not a dict.""" + mcfg = {'network': '"content here'} + self.assertEqual(mcfg, _maybe_remove_top_network(mcfg)) + + def test_no_remove_if_missing_config_or_version(self): + """should not shift unless network entry has config and version.""" + mcfg = {'network': {'config': self.basecfg}} + self.assertEqual(mcfg, _maybe_remove_top_network(mcfg)) + + mcfg = {'network': {'version': 1}} + self.assertEqual(mcfg, _maybe_remove_top_network(mcfg)) + + def test_remove_with_config_disabled(self): + """network/config=disabled should be shifted.""" + mcfg = {'config': 'disabled'} + self.assertEqual(mcfg, _maybe_remove_top_network({'network': mcfg})) + + # vi: ts=4 expandtab |