x
/
cloudbase-init
Code Issues Proposed changes

Merge "Improve code coverage"

This commit is contained in:
Jenkins 2017-02-23 01:49:55 +00:00 committed by Gerrit Code Review
parent 48592a8c24 0bbedfb3a1
commit e6739ac6c8
15 changed files with 292 additions and 7 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

15
cloudbaseinit/tests/metadata/services/test_base.py
View File

@ -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):

11
cloudbaseinit/tests/metadata/services/test_cloudstack.py
View File

@ -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')
@ -262,3 +262,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())

15
cloudbaseinit/tests/metadata/services/test_httpservice.py
View File

@ -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')

9
cloudbaseinit/tests/metadata/services/test_maasservice.py
View File

@ -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):

11
cloudbaseinit/tests/metadata/services/test_opennebulaservice.py
View File

@ -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, [])

18
cloudbaseinit/tests/plugins/common/test_userdata.py
View File

@ -206,10 +206,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())
@ -245,6 +242,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')

45
cloudbaseinit/tests/plugins/common/userdataplugins/cloudconfigplugins/test_set_hostname.py Normal file
View File

@ -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)

14
cloudbaseinit/tests/plugins/common/userdataplugins/cloudconfigplugins/test_write_files.py
View File

@ -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",

9
cloudbaseinit/tests/plugins/common/userdataplugins/test_heat.py
View File

@ -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')

42
cloudbaseinit/tests/test_init.py
View File

@ -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.'

47
cloudbaseinit/tests/utils/test_classloader.py Normal file
View File

@ -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)

40
cloudbaseinit/tests/utils/test_crypt.py Normal file
View File

@ -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)

14
cloudbaseinit/tests/utils/test_dhcp.py
View File

@ -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)

4
cloudbaseinit/tests/utils/test_encoding.py
View File

@ -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)

5
cloudbaseinit/tests/utils/test_network.py
View File

@ -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')