fix a bunch of the tests

1 files changed, 291 insertions, 21 deletions

diff --git a/tests/unittests/test_datasource/test_smartos.py b/tests/unittests/test_datasource/test_smartos.py
index ea20777a..946286bd 100644
--- a/tests/unittests/test_datasource/test_smartos.py
+++ b/tests/unittests/test_datasource/test_smartos.py
@@ -40,13 +40,9 @@ import six
 from cloudinit import helpers as c_helpers
 from cloudinit.sources import DataSourceSmartOS
 from cloudinit.util import b64e
+from cloudinit import util
 
-from .. import helpers
-
-try:
-    from unittest import mock
-except ImportError:
-    import mock
+from ..helpers import mock, TestCase, FilesystemMockingTestCase
 
 SDC_NICS = json.loads("""
 [
@@ -103,7 +99,7 @@ MOCK_RETURNS = {
     'sdc:vendor-data': '\n'.join(['VENDOR_DATA', '']),
     'user-data': '\n'.join(['something', '']),
     'user-script': '\n'.join(['/bin/true', '']),
-    'sdc:nics': SDC_NICS,
+    'sdc:nics': json.dumps(SDC_NICS),
 }
 
 DMI_DATA_RETURN = 'smartdc'
@@ -115,12 +111,286 @@ def get_mock_client(mockdata):
         def __init__(self, serial):
             pass
 
-        def get_metadata(self, metadata_key):
+        def get(self, metadata_key):
             return mockdata.get(metadata_key)
     return MockMetadataClient
 
 
-class TestSmartOSDataSource(helpers.FilesystemMockingTestCase):
+class PsuedoJoyentClient(object):
+    def __init__(self, data=None):
+        if data is None:
+            data = MOCK_RETURNS.copy()
+        self.data = data
+        return
+        
+    def get(self, key, default=None, strip=False):
+        if key in self.data:
+            r = self.data[key]
+            if strip:
+                r = r.strip()
+        else:
+            r = default
+        return r
+
+    def get_json(self, key, default=None):
+        result = self.get(key, default=default)
+        if result is None:
+            return default
+        return json.loads(result)
+
+    def exists(self):
+        return True
+
+
+class TestSmartOSDataSource(FilesystemMockingTestCase):
+    def setUp(self):
+        super(TestSmartOSDataSource, self).setUp()
+
+        dsmos = 'cloudinit.sources.DataSourceSmartOS'
+        patcher = mock.patch(dsmos + ".jmc_client_factory")
+        self.jmc_cfact = patcher.start()
+        self.addCleanup(patcher.stop)
+        patcher = mock.patch(dsmos + ".get_smartos_environ")
+        self.get_smartos_environ = patcher.start()
+        self.addCleanup(patcher.stop)
+
+        self.tmp = tempfile.mkdtemp()
+        self.addCleanup(shutil.rmtree, self.tmp)
+        self.paths = c_helpers.Paths({'cloud_dir': self.tmp})
+
+        self.legacy_user_d = tempfile.mkdtemp()
+        self.orig_lud = DataSourceSmartOS.LEGACY_USER_D
+        DataSourceSmartOS.LEGACY_USER_D = self.legacy_user_d
+
+    def tearDown(self):
+        DataSourceSmartOS.LEGACY_USER_D = self.orig_lud
+        super(TestSmartOSDataSource, self).tearDown()
+
+    def _get_ds(self, mockdata=None, mode=DataSourceSmartOS.SMARTOS_ENV_KVM,
+                sys_cfg=None, ds_cfg=None):
+        self.jmc_cfact.return_value = PsuedoJoyentClient(mockdata)
+        self.get_smartos_environ.return_value = mode
+
+        if sys_cfg is None:
+            sys_cfg = {}
+
+        if ds_cfg is not None:
+            sys_cfg['datasource'] = sys_cfg.get('datasource', {})
+            sys_cfg['datasource']['SmartOS'] = ds_cfg
+
+        return DataSourceSmartOS.DataSourceSmartOS(
+            sys_cfg, distro=None, paths=self.paths)
+        
+    def test_it_got_here(self):
+        dsrc = self._get_ds()
+        ret = dsrc.get_data()
+
+    def test_no_base64(self):
+        ds_cfg = {'no_base64_decode': ['test_var1'], 'all_base': True}
+        dsrc = self._get_ds(ds_cfg=ds_cfg)
+        ret = dsrc.get_data()
+        self.assertTrue(ret)
+
+    def test_uuid(self):
+        dsrc = self._get_ds(mockdata=MOCK_RETURNS)
+        ret = dsrc.get_data()
+        self.assertTrue(ret)
+        self.assertEqual(MOCK_RETURNS['sdc:uuid'],
+                         dsrc.metadata['instance-id'])
+
+    def test_root_keys(self):
+        dsrc = self._get_ds(mockdata=MOCK_RETURNS)
+        ret = dsrc.get_data()
+        self.assertTrue(ret)
+        self.assertEqual(MOCK_RETURNS['root_authorized_keys'],
+                         dsrc.metadata['public-keys'])
+
+    def test_hostname_b64(self):
+        dsrc = self._get_ds(mockdata=MOCK_RETURNS)
+        ret = dsrc.get_data()
+        self.assertTrue(ret)
+        self.assertEqual(MOCK_RETURNS['hostname'],
+                         dsrc.metadata['local-hostname'])
+
+    def test_hostname(self):
+        dsrc = self._get_ds(mockdata=MOCK_RETURNS)
+        ret = dsrc.get_data()
+        self.assertTrue(ret)
+        self.assertEqual(MOCK_RETURNS['hostname'],
+                         dsrc.metadata['local-hostname'])
+
+    def test_userdata(self):
+        dsrc = self._get_ds(mockdata=MOCK_RETURNS)
+        ret = dsrc.get_data()
+        self.assertTrue(ret)
+        self.assertEqual(MOCK_RETURNS['user-data'],
+                         dsrc.metadata['legacy-user-data'])
+        self.assertEqual(MOCK_RETURNS['cloud-init:user-data'],
+                         dsrc.userdata_raw)
+
+    def test_sdc_nics(self):
+        dsrc = self._get_ds(mockdata=MOCK_RETURNS)
+        ret = dsrc.get_data()
+        self.assertTrue(ret)
+        self.assertEquals(json.loads(MOCK_RETURNS['sdc:nics']),
+                          dsrc.metadata['network-data'])
+
+    def test_sdc_scripts(self):
+        dsrc = self._get_ds(mockdata=MOCK_RETURNS)
+        ret = dsrc.get_data()
+        self.assertTrue(ret)
+        self.assertEqual(MOCK_RETURNS['user-script'],
+                         dsrc.metadata['user-script'])
+
+        legacy_script_f = "%s/user-script" % self.legacy_user_d
+        self.assertTrue(os.path.exists(legacy_script_f))
+        self.assertTrue(os.path.islink(legacy_script_f))
+        user_script_perm = oct(os.stat(legacy_script_f)[stat.ST_MODE])[-3:]
+        self.assertEqual(user_script_perm, '700')
+
+    def test_scripts_shebanged(self):
+        dsrc = self._get_ds(mockdata=MOCK_RETURNS)
+        ret = dsrc.get_data()
+        self.assertTrue(ret)
+        self.assertEqual(MOCK_RETURNS['user-script'],
+                         dsrc.metadata['user-script'])
+
+        legacy_script_f = "%s/user-script" % self.legacy_user_d
+        self.assertTrue(os.path.exists(legacy_script_f))
+        self.assertTrue(os.path.islink(legacy_script_f))
+        shebang = None
+        with open(legacy_script_f, 'r') as f:
+            shebang = f.readlines()[0].strip()
+        self.assertEqual(shebang, "#!/bin/bash")
+        user_script_perm = oct(os.stat(legacy_script_f)[stat.ST_MODE])[-3:]
+        self.assertEqual(user_script_perm, '700')
+
+    def test_scripts_shebang_not_added(self):
+        """
+            Test that the SmartOS requirement that plain text scripts
+            are executable. This test makes sure that plain texts scripts
+            with out file magic have it added appropriately by cloud-init.
+        """
+
+        my_returns = MOCK_RETURNS.copy()
+        my_returns['user-script'] = '\n'.join(['#!/usr/bin/perl',
+                                               'print("hi")', ''])
+
+        dsrc = self._get_ds(mockdata=my_returns)
+        ret = dsrc.get_data()
+        self.assertTrue(ret)
+        self.assertEqual(my_returns['user-script'],
+                         dsrc.metadata['user-script'])
+
+        legacy_script_f = "%s/user-script" % self.legacy_user_d
+        self.assertTrue(os.path.exists(legacy_script_f))
+        self.assertTrue(os.path.islink(legacy_script_f))
+        shebang = None
+        with open(legacy_script_f, 'r') as f:
+            shebang = f.readlines()[0].strip()
+        self.assertEqual(shebang, "#!/usr/bin/perl")
+
+    def test_userdata_removed(self):
+        """
+            User-data in the SmartOS world is supposed to be written to a file
+            each and every boot. This tests to make sure that in the event the
+            legacy user-data is removed, the existing user-data is backed-up
+            and there is no /var/db/user-data left.
+        """
+
+        user_data_f = "%s/mdata-user-data" % self.legacy_user_d
+        with open(user_data_f, 'w') as f:
+            f.write("PREVIOUS")
+
+        my_returns = MOCK_RETURNS.copy()
+        del my_returns['user-data']
+
+        dsrc = self._get_ds(mockdata=my_returns)
+        ret = dsrc.get_data()
+        self.assertTrue(ret)
+        self.assertFalse(dsrc.metadata.get('legacy-user-data'))
+
+        found_new = False
+        for root, _dirs, files in os.walk(self.legacy_user_d):
+            for name in files:
+                name_f = os.path.join(root, name)
+                permissions = oct(os.stat(name_f)[stat.ST_MODE])[-3:]
+                if re.match(r'.*\/mdata-user-data$', name_f):
+                    found_new = True
+                    print(name_f)
+                    self.assertEqual(permissions, '400')
+
+        self.assertFalse(found_new)
+
+    def test_vendor_data_not_default(self):
+        dsrc = self._get_ds(mockdata=MOCK_RETURNS)
+        ret = dsrc.get_data()
+        self.assertTrue(ret)
+        self.assertEqual(MOCK_RETURNS['sdc:vendor-data'],
+                         dsrc.metadata['vendor-data'])
+
+    def test_default_vendor_data(self):
+        my_returns = MOCK_RETURNS.copy()
+        def_op_script = my_returns['sdc:vendor-data']
+        del my_returns['sdc:vendor-data']
+        dsrc = self._get_ds(mockdata=my_returns)
+        ret = dsrc.get_data()
+        self.assertTrue(ret)
+        self.assertNotEqual(def_op_script, dsrc.metadata['vendor-data'])
+
+        # we expect default vendor-data is a boothook
+        self.assertTrue(dsrc.vendordata_raw.startswith("#cloud-boothook"))
+
+    def test_disable_iptables_flag(self):
+        dsrc = self._get_ds(mockdata=MOCK_RETURNS)
+        ret = dsrc.get_data()
+        self.assertTrue(ret)
+        self.assertEqual(MOCK_RETURNS['disable_iptables_flag'],
+                         dsrc.metadata['iptables_disable'])
+
+    def test_motd_sys_info(self):
+        dsrc = self._get_ds(mockdata=MOCK_RETURNS)
+        ret = dsrc.get_data()
+        self.assertTrue(ret)
+        self.assertEqual(MOCK_RETURNS['enable_motd_sys_info'],
+                         dsrc.metadata['motd_sys_info'])
+
+    def test_default_ephemeral(self):
+        # Test to make sure that the builtin config has the ephemeral
+        # configuration.
+        dsrc = self._get_ds()
+        cfg = dsrc.get_config_obj()
+
+        ret = dsrc.get_data()
+        self.assertTrue(ret)
+
+        assert 'disk_setup' in cfg
+        assert 'fs_setup' in cfg
+        self.assertIsInstance(cfg['disk_setup'], dict)
+        self.assertIsInstance(cfg['fs_setup'], list)
+
+    def test_override_disk_aliases(self):
+        # Test to make sure that the built-in DS is overriden
+        builtin = DataSourceSmartOS.BUILTIN_DS_CONFIG
+
+        mydscfg = {'disk_aliases': {'FOO': '/dev/bar'}}
+
+        # expect that these values are in builtin, or this is pointless
+        for k in mydscfg:
+            self.assertIn(k, builtin)
+
+        dsrc = self._get_ds(ds_cfg=mydscfg)
+        ret = dsrc.get_data()
+        self.assertTrue(ret)
+
+        self.assertEqual(mydscfg['disk_aliases']['FOO'],
+                         dsrc.ds_cfg['disk_aliases']['FOO'])
+
+        self.assertEqual(dsrc.device_name_to_device('FOO'),
+                         mydscfg['disk_aliases']['FOO'])
+
+
+class OldTestSmartOSDataSource(FilesystemMockingTestCase):
     def setUp(self):
         super(TestSmartOSDataSource, self).setUp()
 
@@ -141,7 +411,7 @@ class TestSmartOSDataSource(helpers.FilesystemMockingTestCase):
         super(TestSmartOSDataSource, self).setUp()
 
     def tearDown(self):
-        helpers.FilesystemMockingTestCase.tearDown(self)
+        FilesystemMockingTestCase.tearDown(self)
         if self._log_handler and self._log:
             self._log.removeHandler(self._log_handler)
         apply_patches([i for i in reversed(self.unapply)])
@@ -166,7 +436,7 @@ class TestSmartOSDataSource(helpers.FilesystemMockingTestCase):
         if dmi_data is None:
             dmi_data = DMI_DATA_RETURN
 
-        def _dmi_data():
+        def _dmi_data(item):
             return dmi_data
 
         def _os_uname():
@@ -188,12 +458,12 @@ class TestSmartOSDataSource(helpers.FilesystemMockingTestCase):
         self.apply_patches([(mod, 'LEGACY_USER_D', self.legacy_user_d)])
         self.apply_patches([
             (mod, 'JoyentMetadataClient', get_mock_client(mockdata))])
-        self.apply_patches([(mod, 'dmi_data', _dmi_data)])
+        self.apply_patches([(util, 'read_dmi_data', _dmi_data)])
         self.apply_patches([(os, 'uname', _os_uname)])
-        self.apply_patches([(mod, 'device_exists', lambda d: True)])
+        self.apply_patches([(os.path, 'exists', lambda d: True)])
         dsrc = mod.DataSourceSmartOS(sys_cfg, distro=None,
                                      paths=self.paths)
-        self.apply_patches([(dsrc, '_get_seed_file_object', mock.MagicMock())])
+        #self.apply_patches([(dsrc, '_get_seed_file_object', mock.MagicMock())])
         return dsrc
 
     def test_seed(self):
@@ -492,7 +762,7 @@ def apply_patches(patches):
     return ret
 
 
-class TestJoyentMetadataClient(helpers.FilesystemMockingTestCase):
+class TestJoyentMetadataClient(FilesystemMockingTestCase):
 
     def setUp(self):
         super(TestJoyentMetadataClient, self).setUp()
@@ -532,7 +802,7 @@ class TestJoyentMetadataClient(helpers.FilesystemMockingTestCase):
                        mock.Mock(return_value=self.request_id)))
 
     def _get_client(self):
-        return DataSourceSmartOS.JoyentMetadataClient(self.serial)
+        return DataSourceSmartOS.JoyentMetadataSerialClient(self.serial)
 
     def assertEndsWith(self, haystack, prefix):
         self.assertTrue(haystack.endswith(prefix),
@@ -546,7 +816,7 @@ class TestJoyentMetadataClient(helpers.FilesystemMockingTestCase):
 
     def test_get_metadata_writes_a_single_line(self):
         client = self._get_client()
-        client.get_metadata('some_key')
+        client.get('some_key')
         self.assertEqual(1, self.serial.write.call_count)
         written_line = self.serial.write.call_args[0][0]
         print(type(written_line))
@@ -556,7 +826,7 @@ class TestJoyentMetadataClient(helpers.FilesystemMockingTestCase):
 
     def _get_written_line(self, key='some_key'):
         client = self._get_client()
-        client.get_metadata(key)
+        client.get(key)
         return self.serial.write.call_args[0][0]
 
     def test_get_metadata_writes_bytes(self):
@@ -600,12 +870,12 @@ class TestJoyentMetadataClient(helpers.FilesystemMockingTestCase):
 
     def test_get_metadata_reads_a_line(self):
         client = self._get_client()
-        client.get_metadata('some_key')
+        client.get('some_key')
         self.assertEqual(self.metasource_data_len, self.serial.read.call_count)
 
     def test_get_metadata_returns_valid_value(self):
         client = self._get_client()
-        value = client.get_metadata('some_key')
+        value = client.get('some_key')
         self.assertEqual(self.metadata_value, value)
 
     def test_get_metadata_throws_exception_for_incorrect_length(self):
@@ -633,4 +903,4 @@ class TestJoyentMetadataClient(helpers.FilesystemMockingTestCase):
         self.response_parts['length'] = 17
         client = self._get_client()
         client._checksum = lambda _: self.response_parts['crc']
-        self.assertIsNone(client.get_metadata('some_key'))
+        self.assertIsNone(client.get('some_key'))