Merged trunk lp:cloud-init

2 files changed, 161 insertions, 10 deletions

diff --git a/tests/unittests/test_datasource/__init__.py b/tests/unittests/test_datasource/__init__.py
new file mode 100644
index 00000000..e69de29b
--- /dev/null
+++ b/tests/unittests/test_datasource/__init__.py
diff --git a/tests/unittests/test_datasource/test_configdrive.py b/tests/unittests/test_datasource/test_configdrive.py
index 55573114..aa5b98ed 100644
--- a/tests/unittests/test_datasource/test_configdrive.py
+++ b/tests/unittests/test_datasource/test_configdrive.py
@@ -2,10 +2,12 @@ from copy import copy
 import json
 import os
 import os.path
-import shutil
-import tempfile
-from unittest import TestCase
 
+import mocker
+from mocker import MockerTestCase
+
+from cloudinit import helpers
+from cloudinit import settings
 from cloudinit.sources import DataSourceConfigDrive as ds
 from cloudinit import util
 
@@ -60,17 +62,140 @@ CFG_DRIVE_FILES_V2 = {
   'openstack/latest/user_data': USER_DATA}
 
 
-class TestConfigDriveDataSource(TestCase):
+class TestConfigDriveDataSource(MockerTestCase):
 
     def setUp(self):
         super(TestConfigDriveDataSource, self).setUp()
-        self.tmp = tempfile.mkdtemp()
+        self.tmp = self.makeDir()
 
-    def tearDown(self):
-        try:
-            shutil.rmtree(self.tmp)
-        except OSError:
-            pass
+    def test_ec2_metadata(self):
+        populate_dir(self.tmp, CFG_DRIVE_FILES_V2)
+        found = ds.read_config_drive_dir(self.tmp)
+        self.assertTrue('ec2-metadata' in found)
+        ec2_md = found['ec2-metadata']
+        self.assertEqual(EC2_META, ec2_md)
+
+    def test_dev_os_remap(self):
+        populate_dir(self.tmp, CFG_DRIVE_FILES_V2)
+        cfg_ds = ds.DataSourceConfigDrive(settings.CFG_BUILTIN,
+                                          None,
+                                          helpers.Paths({}))
+        found = ds.read_config_drive_dir(self.tmp)
+        cfg_ds.metadata = found['metadata']
+        name_tests = {
+            'ami': '/dev/vda1',
+            'root': '/dev/vda1',
+            'ephemeral0': '/dev/vda2',
+            'swap': '/dev/vda3',
+        }
+        for name, dev_name in name_tests.items():
+            my_mock = mocker.Mocker()
+            find_mock = my_mock.replace(util.find_devs_with,
+                                        spec=False, passthrough=False)
+            provided_name = dev_name[len('/dev/'):]
+            provided_name = "s" + provided_name[1:]
+            find_mock(mocker.ARGS)
+            my_mock.result([provided_name])
+            exists_mock = my_mock.replace(os.path.exists,
+                                          spec=False, passthrough=False)
+            exists_mock(mocker.ARGS)
+            my_mock.result(False)
+            exists_mock(mocker.ARGS)
+            my_mock.result(True)
+            my_mock.replay()
+            device = cfg_ds.device_name_to_device(name)
+            my_mock.restore()
+            self.assertEquals(dev_name, device)
+
+    def test_dev_os_map(self):
+        populate_dir(self.tmp, CFG_DRIVE_FILES_V2)
+        cfg_ds = ds.DataSourceConfigDrive(settings.CFG_BUILTIN,
+                                          None,
+                                          helpers.Paths({}))
+        found = ds.read_config_drive_dir(self.tmp)
+        os_md = found['metadata']
+        cfg_ds.metadata = os_md
+        name_tests = {
+            'ami': '/dev/vda1',
+            'root': '/dev/vda1',
+            'ephemeral0': '/dev/vda2',
+            'swap': '/dev/vda3',
+        }
+        for name, dev_name in name_tests.items():
+            my_mock = mocker.Mocker()
+            find_mock = my_mock.replace(util.find_devs_with,
+                                        spec=False, passthrough=False)
+            find_mock(mocker.ARGS)
+            my_mock.result([dev_name])
+            exists_mock = my_mock.replace(os.path.exists,
+                                          spec=False, passthrough=False)
+            exists_mock(mocker.ARGS)
+            my_mock.result(True)
+            my_mock.replay()
+            device = cfg_ds.device_name_to_device(name)
+            my_mock.restore()
+            self.assertEquals(dev_name, device)
+
+    def test_dev_ec2_remap(self):
+        populate_dir(self.tmp, CFG_DRIVE_FILES_V2)
+        cfg_ds = ds.DataSourceConfigDrive(settings.CFG_BUILTIN,
+                                          None,
+                                          helpers.Paths({}))
+        found = ds.read_config_drive_dir(self.tmp)
+        ec2_md = found['ec2-metadata']
+        os_md = found['metadata']
+        cfg_ds.ec2_metadata = ec2_md
+        cfg_ds.metadata = os_md
+        name_tests = {
+            'ami': '/dev/vda1',
+            'root': '/dev/vda1',
+            'ephemeral0': '/dev/vda2',
+            'swap': '/dev/vda3',
+            None: None,
+            'bob': None,
+            'root2k': None,
+        }
+        for name, dev_name in name_tests.items():
+            my_mock = mocker.Mocker()
+            exists_mock = my_mock.replace(os.path.exists,
+                                          spec=False, passthrough=False)
+            exists_mock(mocker.ARGS)
+            my_mock.result(False)
+            exists_mock(mocker.ARGS)
+            my_mock.result(True)
+            my_mock.replay()
+            device = cfg_ds.device_name_to_device(name)
+            self.assertEquals(dev_name, device)
+            my_mock.restore()
+
+    def test_dev_ec2_map(self):
+        populate_dir(self.tmp, CFG_DRIVE_FILES_V2)
+        cfg_ds = ds.DataSourceConfigDrive(settings.CFG_BUILTIN,
+                                          None,
+                                          helpers.Paths({}))
+        found = ds.read_config_drive_dir(self.tmp)
+        exists_mock = self.mocker.replace(os.path.exists,
+                                          spec=False, passthrough=False)
+        exists_mock(mocker.ARGS)
+        self.mocker.count(0, None)
+        self.mocker.result(True)
+        self.mocker.replay()
+        ec2_md = found['ec2-metadata']
+        os_md = found['metadata']
+        cfg_ds.ec2_metadata = ec2_md
+        cfg_ds.metadata = os_md
+        name_tests = {
+            'ami': '/dev/sda1',
+            'root': '/dev/sda1',
+            'ephemeral0': '/dev/sda2',
+            'swap': '/dev/sda3',
+            None: None,
+            'bob': None,
+            'root2k': None,
+        }
+        for name, dev_name in name_tests.items():
+            device = cfg_ds.device_name_to_device(name)
+            self.assertEquals(dev_name, device)
 
     def test_dir_valid(self):
         """Verify a dir is read as such."""
@@ -163,6 +288,32 @@ class TestConfigDriveDataSource(TestCase):
         finally:
             util.find_devs_with = orig_find_devs_with
 
+    def test_pubkeys_v2(self):
+        """Verify that public-keys work in config-drive-v2."""
+        populate_dir(self.tmp, CFG_DRIVE_FILES_V2)
+        myds = cfg_ds_from_dir(self.tmp)
+        self.assertEqual(myds.get_public_ssh_keys(),
+           [OSTACK_META['public_keys']['mykey']])
+
+
+def cfg_ds_from_dir(seed_d):
+    found = ds.read_config_drive_dir(seed_d)
+    cfg_ds = ds.DataSourceConfigDrive(settings.CFG_BUILTIN, None,
+                                      helpers.Paths({}))
+    populate_ds_from_read_config(cfg_ds, seed_d, found)
+    return cfg_ds
+
+
+def populate_ds_from_read_config(cfg_ds, source, results):
+    """Patch the DataSourceConfigDrive from the results of
+    read_config_drive_dir hopefully in line with what it would have
+    if cfg_ds.get_data had been successfully called"""
+    cfg_ds.source = source
+    cfg_ds.metadata = results.get('metadata')
+    cfg_ds.ec2_metadata = results.get('ec2-metadata')
+    cfg_ds.userdata_raw = results.get('userdata')
+    cfg_ds.version = results.get('cfgdrive_ver')
+
 
 def populate_dir(seed_dir, files):
     for (name, content) in files.iteritems():