Improve code coverage

Improves the code coverage for the present tests where possible.

Change-Id: I8971ba2ed775dcd772622acea3fed4ebaa177ad7

cloudbaseinit/tests/metadata/services/test_base.py

@@ -39,6 +39,21 @@ class TestBase(unittest.TestCase):
         userdata = self._service.get_decoded_user_data()
         self.assertEqual(b"of course it works", userdata)
 
+    def test_get_name(self):
+        self.assertEqual(self._service.get_name(), 'FakeService')
+
+    def test_can_post_password(self):
+        self.assertFalse(self._service.can_post_password)
+
+    def test_is_password_set(self):
+        self.assertFalse(self._service.is_password_set)
+
+    def test_can_update_password(self):
+        self.assertFalse(self._service.can_update_password)
+
+    def test_is_password_changed(self):
+        self.assertFalse(self._service.is_password_changed())
+
 
 class TestBaseHTTPMetadataService(unittest.TestCase):
 

cloudbaseinit/tests/metadata/services/test_cloudstack.py

@@ -57,7 +57,7 @@ class CloudStackTest(unittest.TestCase):
         ]
 
         self.assertTrue(self._service._test_api(url))
-        for _ in range(3):
+        for _ in range(4):
             self.assertFalse(self._service._test_api(url))
 
     @mock.patch('cloudbaseinit.osutils.factory.get_os_utils')
@@ -280,3 +280,12 @@ class CloudStackTest(unittest.TestCase):
         self.assertIsNone(self._service.get_admin_password())
         self.assertEqual(1, mock_get_password.call_count)
         self.assertEqual(0, mock_delete_password.call_count)
+
+    def test_can_update_password(self):
+        self.assertTrue(self._service.can_update_password)
+
+    @mock.patch('cloudbaseinit.metadata.services.cloudstack.CloudStack.'
+                '_get_password')
+    def test_is_password_changed(self, mock_get_password):
+        mock_get_password.return_value = True
+        self.assertTrue(self._service.is_password_changed())

cloudbaseinit/tests/metadata/services/test_httpservice.py

@@ -22,6 +22,7 @@ except ImportError:
 from six.moves.urllib import error
 
 from cloudbaseinit import conf as cloudbaseinit_conf
+from cloudbaseinit.metadata.services import base
 from cloudbaseinit.metadata.services import httpservice
 
 CONF = cloudbaseinit_conf.CONF
@@ -70,6 +71,13 @@ class HttpServiceTest(unittest.TestCase):
         self.assertEqual('openstack/%s/password' %
                          self._httpservice._POST_PASSWORD_MD_VER, response)
 
+    @mock.patch('cloudbaseinit.metadata.services.httpservice.HttpService'
+                '._get_meta_data')
+    def test_can_post_password(self, mock_get_meta_data):
+        self.assertTrue(self._httpservice.can_post_password)
+        mock_get_meta_data.side_effect = base.NotExistingMetadataException
+        self.assertFalse(self._httpservice.can_post_password)
+
     @mock.patch('cloudbaseinit.metadata.services.httpservice.HttpService'
                 '._get_password_path')
     @mock.patch('cloudbaseinit.metadata.services.httpservice.HttpService'
@@ -92,6 +100,13 @@ class HttpServiceTest(unittest.TestCase):
             mock_get_password_path.assert_called_once_with()
             self.assertEqual(ret_val, response)
 
+    @mock.patch('cloudbaseinit.metadata.services.base.BaseHTTPMetadataService'
+                '._get_data')
+    def test_is_password_set(self, mock_get_data):
+        mock_get_data.return_value = "fake-data"
+        res = self._httpservice.is_password_set
+        self.assertTrue(res)
+
     def test_post_password(self):
         self._test_post_password(ret_val='fake return')
 

cloudbaseinit/tests/metadata/services/test_maasservice.py

@@ -91,6 +91,15 @@ class MaaSHttpServiceTest(unittest.TestCase):
         self.assertEqual('"consumer_secret%26token_secret"',
                          auth_parts['oauth_signature'])
 
+    @mock.patch('cloudbaseinit.metadata.services.base.'
+                'BaseHTTPMetadataService._http_request')
+    @mock.patch('cloudbaseinit.metadata.services.maasservice.MaaSHttpService'
+                '._get_oauth_headers')
+    def test_http_request(self, mock_ouath_headers, mock_http_request):
+        mock_url = "fake.url"
+        self._maasservice._http_request(mock_url)
+        mock_http_request.assert_called_once_with(mock_url, None, {})
+
     @mock.patch("cloudbaseinit.metadata.services.maasservice.MaaSHttpService"
                 "._get_cache_data")
     def test_get_host_name(self, mock_get_cache_data):

cloudbaseinit/tests/metadata/services/test_opennebulaservice.py

@@ -334,3 +334,14 @@ class TestLoadedOpenNebulaService(_TestOpenNebulaService):
             network_details,
             self._service.get_network_details()
         )
+
+    @mock.patch("cloudbaseinit.metadata.services"
+                ".opennebulaservice.OpenNebulaService._get_cache_data")
+    def test_get_network_details_exception(self, mock_get_cache):
+        mock_mac = mock_address = mock.MagicMock()
+        mock_mac.upper.return_value = None
+        mock_address.side_effect = None
+        exc = base.NotExistingMetadataException
+        mock_get_cache.side_effect = [mock_mac, mock_address, exc, exc]
+        result_details = self._service.get_network_details()
+        self.assertEqual(result_details, [])

cloudbaseinit/tests/plugins/common/test_userdata.py

@@ -169,10 +169,7 @@ class UserDataPluginTest(unittest.TestCase):
         mock_part.get_content_type.assert_called_once_with()
         mock_user_handlers.get.assert_called_once_with(
             _content_type)
-        if handler_func and handler_func is Exception:
-            self.assertEqual(2, mock_part.get_content_type.call_count)
-            self.assertEqual(2, mock_part.get_filename.call_count)
-        elif handler_func:
+        if handler_func:
             handler_func.assert_called_once_with(None, _content_type,
                                                  mock_part.get_filename(),
                                                  mock_part.get_payload())
@@ -208,6 +205,19 @@ class UserDataPluginTest(unittest.TestCase):
         self._test_process_part(handler_func=None,
                                 user_data_plugin=user_data_plugin,
                                 content_type=False)
+        self._test_process_part(handler_func=None,
+                                user_data_plugin=None,
+                                content_type=False)
+
+    def test_process_part_exception_occurs(self):
+        mock_part = mock_handlers = mock.MagicMock()
+        mock_handlers.get.side_effect = Exception
+        mock_part.get_content_type().side_effect = Exception
+        self.assertEqual((1, False),
+                         self._userdata._process_part(
+                         part=mock_part,
+                         user_data_plugins=None,
+                         user_handlers=mock_handlers))
 
     @mock.patch('cloudbaseinit.plugins.common.userdata.UserDataPlugin'
                 '._begin_part_process_event')

cloudbaseinit/tests/plugins/common/userdataplugins/cloudconfigplugins/test_set_hostname.py

@@ -0,0 +1,45 @@
+# Copyright 2016 Cloudbase Solutions Srl
+#
+#    Licensed under the Apache License, Version 2.0 (the "License"); you may
+#    not use this file except in compliance with the License. You may obtain
+#    a copy of the License at
+#
+#         http://www.apache.org/licenses/LICENSE-2.0
+#
+#    Unless required by applicable law or agreed to in writing, software
+#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+#    License for the specific language governing permissions and limitations
+#    under the License.
+
+import unittest
+
+try:
+    import unittest.mock as mock
+except ImportError:
+    import mock
+
+from oslo_config import cfg
+
+from cloudbaseinit.plugins.common.userdataplugins.cloudconfigplugins import (
+    set_hostname
+)
+
+
+CONF = cfg.CONF
+
+
+class Set_HostNamePluginPluginTest(unittest.TestCase):
+
+    def setUp(self):
+        self._sethost_name_plugin = set_hostname.SetHostnamePlugin()
+
+    @mock.patch('cloudbaseinit.utils.hostname')
+    @mock.patch('cloudbaseinit.osutils.factory.get_os_utils')
+    def test_process(self, mock_get_os_utils, mock_hostname):
+        mock_data = "fake_data"
+        mock_os_util = mock.MagicMock()
+        mock_os_util.set_hostname.return_value = (mock_data, True)
+        mock_get_os_utils.return_value = mock_os_util
+        result_process = self._sethost_name_plugin.process(mock_data)
+        self.assertTrue(result_process)

cloudbaseinit/tests/plugins/common/userdataplugins/cloudconfigplugins/test_write_files.py

@@ -87,6 +87,20 @@ class WriteFilesPluginTests(unittest.TestCase):
         self.assertEqual(expected_logging, snatcher.output)
         self.assertEqual(write_files.DEFAULT_PERMISSIONS, response)
 
+    @mock.patch('os.makedirs')
+    def test_write_file(self, mk):
+        path = u'fake_path'
+        content = u'fake_content'
+        result = write_files._write_file(path, content, open_mode="w")
+        os.remove(path)
+        self.assertTrue(result)
+
+    @mock.patch('os.makedirs')
+    def test_write_file_excp(self, mock_makedirs):
+        mock_makedirs.side_effect = OSError
+        result = write_files._write_file(u'fake_path', u'fake_content')
+        self.assertFalse(result)
+
     def test_write_file_list(self):
         expected_logging = [
             "Plugin 'invalid' is currently not supported",

cloudbaseinit/tests/plugins/common/userdataplugins/test_heat.py

@@ -13,6 +13,7 @@
 #    under the License.
 
 import os
+import sys
 import unittest
 
 try:
@@ -53,9 +54,11 @@ class HeatUserDataHandlerTests(unittest.TestCase):
                 '.HeatPlugin._check_dir')
     @mock.patch('cloudbaseinit.utils.encoding.write_file')
     def _test_process(self, mock_write_file, mock_check_dir,
-                      mock_execute_user_data_script, filename):
+                      mock_execute_user_data_script, filename, payloads=False):
         mock_part = mock.MagicMock()
         mock_part.get_filename.return_value = filename
+        if payloads is True and sys.version_info < (3, 0):
+            mock_part.get_payload.return_value = filename.decode()
         response = self._heat.process(mock_part)
 
         path = os.path.join(CONF.heat_config_dir, filename)
@@ -75,5 +78,9 @@ class HeatUserDataHandlerTests(unittest.TestCase):
     def test_process(self):
         self._test_process(filename=self._heat._heat_user_data_filename)
 
+    def test_process_payload(self):
+        self._test_process(filename=self._heat._heat_user_data_filename,
+                           payloads=True)
+
     def test_process_content_other_data(self):
         self._test_process(filename='other data')

cloudbaseinit/tests/test_init.py

@@ -64,6 +64,13 @@ class TestInitManager(unittest.TestCase):
                 instance_id + "/" + self._init._PLUGINS_CONFIG_SECTION,
                 response)
 
+    def test_get_plugin_section_id(self):
+        fake_id = "100"
+        self._test_get_plugin_section(instance_id=fake_id)
+
+    def test_get_plugin_section_no_id(self):
+        self._test_get_plugin_section(instance_id=None)
+
     @mock.patch('cloudbaseinit.init.InitManager._get_plugins_section')
     def test_get_plugin_status(self, mock_get_plugins_section):
         self.osutils.get_config_value.return_value = 1
@@ -108,6 +115,21 @@ class TestInitManager(unittest.TestCase):
                                                            fake_name, status)
             self.assertTrue(response)
 
+    def test_exec_plugin_exception_occurs(self):
+        fake_name = 'fake name'
+        mock_plugin = mock.MagicMock()
+        mock_plugin.get_name.return_value = fake_name
+        mock_plugin.execute.side_effect = Exception
+        expected_logging = ["Executing plugin 'fake name'",
+                            "plugin 'fake name' failed with error ''"]
+        with testutils.LogSnatcher('cloudbaseinit.init') as snatcher:
+            self._init._exec_plugin(osutils=self.osutils,
+                                    service='fake service',
+                                    plugin=mock_plugin,
+                                    instance_id='fake id',
+                                    shared_data='shared data')
+        self.assertEqual(expected_logging, snatcher.output[:2])
+
     def test_exec_plugin_execution_done(self):
         self._test_exec_plugin(base.PLUGIN_EXECUTION_DONE)
 
@@ -136,6 +158,26 @@ class TestInitManager(unittest.TestCase):
     def test_check_plugin_os_requirements_other_requirenments(self):
         self._test_check_plugin_os_requirements(('linux', (5, 2)))
 
+    def test_check_plugins_os_not_min_os_version(self):
+        sys.platform = 'win32'
+        fake_name = 'fake name'
+        self.plugin.get_name.return_value = fake_name
+        self.plugin.get_os_requirements.return_value = ('win32', 0)
+        response = self._init._check_plugin_os_requirements(self.osutils,
+                                                            self.plugin)
+        self.plugin.get_name.assert_called_once_with()
+        self.assertTrue(response)
+
+    def test_check_plugins_os_not_supported(self):
+        fake_name = 'fake name'
+        self.plugin.get_name.return_value = fake_name
+        mock_osutils = mock.MagicMock()
+        mock_osutils.check_os_version.return_value = None
+        self.plugin.get_os_requirements.return_value = ('win32', (5, 2))
+        response = self._init._check_plugin_os_requirements(mock_osutils,
+                                                            self.plugin)
+        self.assertFalse(response)
+
     @mock.patch('cloudbaseinit.init.InitManager.'
                 '_exec_plugin')
     @mock.patch('cloudbaseinit.init.InitManager.'

cloudbaseinit/tests/utils/test_classloader.py

@@ -0,0 +1,47 @@
+# Copyright 2016 Cloudbase Solutions Srl
+#
+#    Licensed under the Apache License, Version 2.0 (the "License"); you may
+#    not use this file except in compliance with the License. You may obtain
+#    a copy of the License at
+#
+#         http://www.apache.org/licenses/LICENSE-2.0
+#
+#    Unless required by applicable law or agreed to in writing, software
+#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+#    License for the specific language governing permissions and limitations
+#    under the License.
+
+import os
+import tempfile
+import unittest
+
+try:
+    import unittest.mock as mock
+except ImportError:
+    import mock
+
+from cloudbaseinit.utils import classloader
+
+
+def _create_tempfile():
+    fd, tmp = tempfile.mkstemp()
+    os.close(fd)
+    return tmp
+
+
+class ClassLoaderTest(unittest.TestCase):
+
+    def setUp(self):
+        self._loader = classloader.ClassLoader()
+
+    @mock.patch('imp.load_compiled')
+    @mock.patch('imp.load_source')
+    def test_load_module_py(self, mock_source, mock_compiled):
+        mock_py = os.path.join(_create_tempfile(), "mock.py")
+        mock_pyc = os.path.join(_create_tempfile(), "mock.pyc")
+        mock_source.return_value = mock_compiled.return_value = None
+        result_module_py = self._loader.load_module(mock_py)
+        result_module_pyc = self._loader.load_module(mock_pyc)
+        self.assertIsNone(result_module_py)
+        self.assertIsNone(result_module_pyc)

cloudbaseinit/tests/utils/test_crypt.py

@@ -0,0 +1,40 @@
+# Copyright 2016 Cloudbase Solutions Srl
+#
+#    Licensed under the Apache License, Version 2.0 (the "License"); you may
+#    not use this file except in compliance with the License. You may obtain
+#    a copy of the License at
+#
+#         http://www.apache.org/licenses/LICENSE-2.0
+#
+#    Unless required by applicable law or agreed to in writing, software
+#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+#    License for the specific language governing permissions and limitations
+#    under the License.
+
+import unittest
+
+from cloudbaseinit.utils import crypt
+
+
+class TestOpenSSLException(unittest.TestCase):
+
+    def setUp(self):
+        self._openssl = crypt.OpenSSLException()
+
+    def test_get_openssl_error_msg(self):
+        expected_error_msg = u'error:00000000:lib(0):func(0):reason(0)'
+        error_msg = self._openssl._get_openssl_error_msg()
+        self.assertEqual(expected_error_msg, error_msg)
+
+
+class TestCryptManager(unittest.TestCase):
+
+    def setUp(self):
+        self._crypt_manager = crypt.CryptManager()
+
+    def test_load_ssh_rsa_public_key_invalid(self):
+        ssh_pub_key = "ssh"
+        exc = Exception
+        self.assertRaises(exc, self._crypt_manager.load_ssh_rsa_public_key,
+                          ssh_pub_key)

cloudbaseinit/tests/utils/test_dhcp.py

@@ -92,6 +92,14 @@ class DHCPUtilsTests(unittest.TestCase):
         self._test_parse_dhcp_reply(message_type=2, id_reply=9999,
                                     equals_cookie=True)
 
+    def test_parse_dhcp_reply_false(self):
+        self._test_parse_dhcp_reply(message_type=2, id_reply=111,
+                                    equals_cookie=True)
+
+    def test_parse_dhcp_reply_cookie_false(self):
+        self._test_parse_dhcp_reply(message_type=2, id_reply=9999,
+                                    equals_cookie=False)
+
     def test_parse_dhcp_reply_other_message_type(self):
         self._test_parse_dhcp_reply(message_type=3, id_reply=9999,
                                     equals_cookie=True)
@@ -161,6 +169,12 @@ class DHCPUtilsTests(unittest.TestCase):
         mock_socket().close.assert_called_once_with()
         self.assertEqual('fake replied options', response)
 
+    @mock.patch('cloudbaseinit.utils.dhcp._bind_dhcp_client_socket')
+    def test_get_dhcp_options_timeout(self, mock_client_socket):
+        mock_client_socket.side_effect = [socket.timeout]
+        dhcp.get_dhcp_options(dhcp_host='fake host',
+                              requested_options=['fake option'])
+
     def test__bind_dhcp_client_socket_bind_succeeds(self):
         mock_socket = mock.Mock()
         dhcp._bind_dhcp_client_socket(mock_socket, 0, 0)

cloudbaseinit/tests/utils/test_encoding.py

@@ -24,6 +24,7 @@ from cloudbaseinit.utils import encoding
 class TestEncoding(unittest.TestCase):
 
     def test_get_as_string(self):
+        self.assertIsNone(encoding.get_as_string(None))
         content_map = [
             ("data", "data"),
             (b"data", "data"),
@@ -40,7 +41,8 @@ class TestEncoding(unittest.TestCase):
             (("w", "r"), "my test\ndata\n\n", False),
             (("wb", "rb"), "\r\n".join((chr(x) for x in
                                         (32, 125, 0))).encode(), False),
-            (("wb", "rb"), "my test\ndata\n\n", True)
+            (("wb", "rb"), "my test\ndata\n\n", True),
+            (("wb", "rb"), u"my test\n data", True)
         ]
         with testutils.create_tempdir() as temp:
             fd, path = tempfile.mkstemp(dir=temp)

cloudbaseinit/tests/utils/test_network.py

@@ -29,6 +29,11 @@ CONF = cloudbaseinit_conf.CONF
 
 class NetworkUtilsTest(unittest.TestCase):
 
+    @mock.patch('six.moves.urllib.request.urlopen')
+    def test_check_url(self, mock_url_open):
+        mock_url_open.return_value = None
+        self.assertTrue(network.check_url("fake_url"))
+
     @mock.patch('sys.platform', new='win32')
     @mock.patch('cloudbaseinit.osutils.factory.get_os_utils')
     @mock.patch('six.moves.urllib.parse.urlparse')