x/cloudbase-init
Code Issues Proposed changes

Unit test refactoring

Browse Source 
Refactored unit tests to fix issues due to windows dependecies removal
along with httpservice refactoring and readed EC2Service tests
This commit is contained in:
Robert Tingirica 2014-02-24 11:31:24 +02:00
parent 654334975c
commit 1c7c48469a
7 changed files with 199 additions and 85 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

22
cloudbaseinit/tests/metadata/services/configdrive/test_configdrive.py
View File

@ -19,7 +19,6 @@ import mock
import os
import sys
import unittest
import uuid
from oslo.config import cfg
@ -40,31 +39,12 @@ class ConfigDriveServiceTest(unittest.TestCase):
@mock.patch.dict(sys.modules, _mock_dict)
def setUp(self):
configdrive = importlib.import_module('cloudbaseinit.metadata.services'
'.configdrive.configdrive')
'.configdrive')
self._config_drive = configdrive.ConfigDriveService()
def tearDown(self):
reload(sys)
@mock.patch('cloudbaseinit.metadata.services.configdrive.manager.'
'ConfigDriveManager.get_config_drive_files')
@mock.patch('tempfile.gettempdir')
@mock.patch('os.path.join')
def test_load(self, mock_join, mock_gettempdir,
mock_get_config_drive_files):
uuid.uuid4 = mock.MagicMock()
fake_path = os.path.join('fake', 'path')
fake_path_found = os.path.join(fake_path, 'found')
uuid.uuid4.return_value = 'random'
mock_get_config_drive_files.return_value = fake_path_found
mock_join.return_value = fake_path
response = self._config_drive.load()
mock_join.assert_called_with(mock_gettempdir(), 'random')
mock_get_config_drive_files.assert_called_once_with(
fake_path, CONF.config_drive_raw_hhd, CONF.config_drive_cdrom)
self.assertEqual(self._config_drive._metadata_path, fake_path)
self.assertEqual(response, fake_path_found)
@mock.patch('os.path.normpath')
@mock.patch('os.path.join')
def test_get_data(self, mock_join, mock_normpath):

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

@ -14,27 +14,22 @@
# License for the specific language governing permissions and limitations
# under the License.
import importlib
import mock
import posixpath
import sys
import unittest
from cloudbaseinit.metadata.services import base
from cloudbaseinit.metadata.services import baseopenstackservice
from cloudbaseinit.utils import x509constants
from oslo.config import cfg
CONF = cfg.CONF
_ctypes_mock = mock.MagicMock()
mock_dict = {'ctypes': _ctypes_mock}
class BaseOpenStackServiceTest(unittest.TestCase):
@mock.patch.dict(sys.modules, mock_dict)
def setUp(self):
self.baseopenstackservice = importlib.import_module(
"cloudbaseinit.metadata.services.baseopenstackservice")
CONF.set_override('retry_count_interval', 0)
self._service = self.baseopenstackservice.BaseOpenStackService()
self._service = baseopenstackservice.BaseOpenStackService()
@mock.patch("cloudbaseinit.metadata.services.baseopenstackservice"
".BaseOpenStackService._get_cache_data")
@ -142,7 +137,7 @@ class BaseOpenStackServiceTest(unittest.TestCase):
if 'meta' in meta_data:
self.assertEqual(response, ['fake cert'])
elif type(ret_value) is str and ret_value.startswith(
self.baseopenstackservice.x509.PEM_HEADER):
x509constants.PEM_HEADER):
mock_get_user_data.assert_called_once_with()
self.assertEqual(response, [ret_value])
elif ret_value is base.NotExistingMetadataException:
@ -154,7 +149,7 @@ class BaseOpenStackServiceTest(unittest.TestCase):
def test_get_client_auth_certs_no_cert_data(self):
self._test_get_client_auth_certs(
meta_data={}, ret_value=self.baseopenstackservice.x509.PEM_HEADER)
meta_data={}, ret_value=x509constants.PEM_HEADER)
def test_get_client_auth_certs_no_cert_data_exception(self):
self._test_get_client_auth_certs(

123
cloudbaseinit/tests/metadata/services/test_ec2service.py Normal file
View File

@ -0,0 +1,123 @@
# vim: tabstop=4 shiftwidth=4 softtabstop=4
# Copyright 2014 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 mock
import posixpath
import unittest
import urllib2
from cloudbaseinit.metadata.services import base
from cloudbaseinit.metadata.services import ec2service
from oslo.config import cfg
CONF = cfg.CONF
class EC2ServiceTest(unittest.TestCase):
def setUp(self):
CONF.set_override('retry_count_interval', 0)
self._service = ec2service.EC2Service()
@mock.patch('cloudbaseinit.utils.network.check_metadata_ip_route')
@mock.patch('cloudbaseinit.metadata.services.ec2service.EC2Service'
'._get_data')
def _test_load(self, mock_get_data, mock_check_metadata_ip_route,
side_effect):
mock_get_data.side_effect = [side_effect]
response = self._service.load()
mock_check_metadata_ip_route.assert_called_once_with(
CONF.ec2_metadata_base_url)
mock_get_data.assert_called_once_with('latest/meta-data/')
if side_effect is Exception:
self.assertFalse(response)
else:
self.assertTrue(response)
def test_load(self):
self._test_load(side_effect=None)
def test_load_exception(self):
self._test_load(side_effect=Exception)
@mock.patch('urllib2.urlopen')
def _test_get_response(self, mock_urlopen, ret_value):
req = mock.MagicMock()
mock_urlopen.side_effect = [ret_value]
is_instance = isinstance(ret_value, urllib2.HTTPError)
if is_instance and ret_value.code == 404:
self.assertRaises(base.NotExistingMetadataException,
self._service._get_response, req)
elif is_instance and ret_value.code != 404:
self.assertRaises(urllib2.HTTPError,
self._service._get_response, req)
else:
response = self._service._get_response(req)
self.assertEqual(response, ret_value)
mock_urlopen.assert_called_once_with(req)
def test_get_response(self):
self._test_get_response(ret_value=None)
def test_get_response_error_404(self):
err = urllib2.HTTPError("http://169.254.169.254/", 404,
'test error 404', {}, None)
self._test_get_response(ret_value=err)
def test_get_response_error_other(self):
err = urllib2.HTTPError("http://169.254.169.254/", 409,
'test error 409', {}, None)
self._test_get_response(ret_value=err)
@mock.patch('urllib2.Request')
@mock.patch('cloudbaseinit.metadata.services.ec2service.EC2Service'
'._get_response')
def test_get_data(self, mock_get_response, mock_Request):
response = self._service._get_data('fake')
fake_path = posixpath.join(CONF.ec2_metadata_base_url, 'fake')
mock_Request.assert_called_once_with(fake_path)
mock_get_response.assert_called_once_with(mock_Request())
self.assertEqual(response, mock_get_response().read())
@mock.patch('cloudbaseinit.metadata.services.ec2service.EC2Service'
'._get_cache_data')
def test_get_host_name(self, mock_get_cache_data):
response = self._service.get_host_name()
mock_get_cache_data.assert_called_once_with(
'%s/meta-data/local-hostname' % self._service._metadata_version)
self.assertEqual(response, mock_get_cache_data())
@mock.patch('cloudbaseinit.metadata.services.ec2service.EC2Service'
'._get_cache_data')
def test_get_instance_id(self, mock_get_cache_data):
response = self._service.get_instance_id()
mock_get_cache_data.assert_called_once_with(
'%s/meta-data/instance-id' % self._service._metadata_version)
self.assertEqual(response, mock_get_cache_data())
@mock.patch('cloudbaseinit.metadata.services.ec2service.EC2Service'
'._get_cache_data')
def test_get_public_keys(self, mock_get_cache_data):
mock_get_cache_data.side_effect = ['key=info', 'fake key']
response = self._service.get_public_keys()
expected = [
mock.call('%s/meta-data/public-keys' %
self._service._metadata_version),
mock.call('%(version)s/meta-data/public-keys/%('
'idx)s/openssh-key' %
{'version': self._service._metadata_version,
'idx': 'key'})]
self.assertEqual(mock_get_cache_data.call_args_list, expected)
self.assertEqual(response, ['fake key'])

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

@ -1,6 +1,6 @@
# vim: tabstop=4 shiftwidth=4 softtabstop=4
# Copyright 2013 Cloudbase Solutions Srl
# Copyright 2014 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
@ -14,68 +14,33 @@
# License for the specific language governing permissions and limitations
# under the License.
import importlib
import mock
import os
import sys
import unittest
import urllib2
from oslo.config import cfg
from cloudbaseinit.metadata.services import httpservice
from cloudbaseinit.metadata.services import base
CONF = cfg.CONF
_ctypes_mock = mock.MagicMock()
mock_dict = {'ctypes': _ctypes_mock}
class HttpServiceTest(unittest.TestCase):
@mock.patch.dict(sys.modules, mock_dict)
def setUp(self):
httpservice = importlib.import_module("cloudbaseinit.metadata.services"
".httpservice")
CONF.set_override('retry_count_interval', 0)
self._httpservice = httpservice.HttpService()
@mock.patch('cloudbaseinit.osutils.factory.get_os_utils')
@mock.patch('urlparse.urlparse')
def _test_check_metadata_ip_route(self, mock_urlparse, mock_get_os_utils,
side_effect):
mock_utils = mock.MagicMock()
mock_split = mock.MagicMock()
mock_get_os_utils.return_value = mock_utils
mock_utils.check_os_version.return_value = True
mock_urlparse().netloc.split.return_value = mock_split
mock_split[0].startswith.return_value = True
mock_utils.check_static_route_exists.return_value = False
mock_utils.get_default_gateway.return_value = (1, '0.0.0.0')
mock_utils.add_static_route.side_effect = [side_effect]
self._httpservice._check_metadata_ip_route()
mock_utils.check_os_version.assert_called_once_with(6, 0)
mock_urlparse.assert_called_with(CONF.metadata_base_url)
mock_split[0].startswith.assert_called_once_with("169.254.")
mock_utils.check_static_route_exists.assert_called_once_with(
mock_split[0])
mock_utils.get_default_gateway.assert_called_once_with()
mock_utils.add_static_route.assert_called_once_with(
mock_split[0], "255.255.255.255", '0.0.0.0', 1, 10)
def test_test_check_metadata_ip_route(self):
self._test_check_metadata_ip_route(side_effect=None)
def test_test_check_metadata_ip_route_fail(self):
self._test_check_metadata_ip_route(side_effect=Exception)
@mock.patch('cloudbaseinit.metadata.services.httpservice.HttpService'
'._check_metadata_ip_route')
@mock.patch('cloudbaseinit.utils.network.check_metadata_ip_route')
@mock.patch('cloudbaseinit.metadata.services.httpservice.HttpService'
'._get_meta_data')
def _test_load(self, mock_get_meta_data, mock_check_metadata_ip_route,
side_effect):
mock_get_meta_data.side_effect = [side_effect]
response = self._httpservice.load()
mock_check_metadata_ip_route.assert_called_once_with()
mock_check_metadata_ip_route.assert_called_once_with(
CONF.metadata_base_url)
mock_get_meta_data.assert_called_once_with()
if side_effect:
self.assertEqual(response, False)

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

@ -14,31 +14,24 @@
# License for the specific language governing permissions and limitations
# under the License.
import importlib
import mock
import os
import posixpath
import sys
import unittest
import urllib2
from oslo.config import cfg
from cloudbaseinit.metadata.services import base
from cloudbaseinit.metadata.services import maasservice
from cloudbaseinit.utils import x509constants
CONF = cfg.CONF
_ctypes_mock = mock.MagicMock()
mock_dict = {'ctypes': _ctypes_mock}
class MaaSHttpServiceTest(unittest.TestCase):
@mock.patch.dict(sys.modules, mock_dict)
def setUp(self):
maasservice = importlib.import_module("cloudbaseinit.metadata.services"
".maasservice")
self.mock_oauth = mock.MagicMock()
self.mock_x509 = mock.MagicMock()
maasservice.oauth = self.mock_oauth
maasservice.x509 = self.mock_x509
self._maasservice = maasservice.MaaSHttpService()
@mock.patch("cloudbaseinit.metadata.services.maasservice.MaaSHttpService"
@ -175,5 +168,5 @@ class MaaSHttpServiceTest(unittest.TestCase):
mock_get_cache_data.assert_called_with(
'%s/meta-data/x509' % self._maasservice._metadata_version)
mock_get_list_from_text.assert_called_once_with(
mock_get_cache_data(), "%s\n" % self.mock_x509.PEM_FOOTER)
mock_get_cache_data(), "%s\n" % x509constants.PEM_FOOTER)
self.assertEqual(response, mock_get_list_from_text())

57
cloudbaseinit/tests/utils/test_network.py Normal file
View File

@ -0,0 +1,57 @@
# vim: tabstop=4 shiftwidth=4 softtabstop=4
# Copyright 2013 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 mock
import sys
import unittest
from oslo.config import cfg
from cloudbaseinit.utils import network
CONF = cfg.CONF
class NetworkUtilsTest(unittest.TestCase):
@mock.patch('cloudbaseinit.osutils.factory.get_os_utils')
@mock.patch('urlparse.urlparse')
def _test_check_metadata_ip_route(self, mock_urlparse, mock_get_os_utils,
side_effect):
mock_utils = mock.MagicMock()
mock_split = mock.MagicMock()
sys.platform = 'win32'
mock_get_os_utils.return_value = mock_utils
mock_utils.check_os_version.return_value = True
mock_urlparse().netloc.split.return_value = mock_split
mock_split[0].startswith.return_value = True
mock_utils.check_static_route_exists.return_value = False
mock_utils.get_default_gateway.return_value = (1, '0.0.0.0')
mock_utils.add_static_route.side_effect = [side_effect]
network.check_metadata_ip_route('196.254.196.254')
mock_utils.check_os_version.assert_called_once_with(6, 0)
mock_urlparse.assert_called_with('196.254.196.254')
mock_split[0].startswith.assert_called_once_with("169.254.")
mock_utils.check_static_route_exists.assert_called_once_with(
mock_split[0])
mock_utils.get_default_gateway.assert_called_once_with()
mock_utils.add_static_route.assert_called_once_with(
mock_split[0], "255.255.255.255", '0.0.0.0', 1, 10)
def test_test_check_metadata_ip_route(self):
self._test_check_metadata_ip_route(side_effect=None)
def test_test_check_metadata_ip_route_fail(self):
self._test_check_metadata_ip_route(side_effect=Exception)

9
cloudbaseinit/tests/utils/windows/test_x509.py
View File

@ -20,6 +20,7 @@ import unittest
from oslo.config import cfg
from cloudbaseinit.utils import x509constants
if sys.platform == 'win32':
from cloudbaseinit.utils.windows import cryptoapi
from cloudbaseinit.utils.windows import x509
@ -267,9 +268,9 @@ class CryptoAPICertManagerTests(unittest.TestCase):
def test_get_cert_base64(self):
fake_cert_data = ''
fake_cert_data += x509.PEM_HEADER + '\n'
fake_cert_data += x509constants.PEM_HEADER + '\n'
fake_cert_data += 'fake cert' + '\n'
fake_cert_data += x509.PEM_FOOTER
fake_cert_data += x509constants.PEM_FOOTER
response = self._x509._get_cert_base64(fake_cert_data)
self.assertEqual(response, 'fake cert')
@ -307,9 +308,9 @@ class CryptoAPICertManagerTests(unittest.TestCase):
mock_free, crypttstr, store_handle, add_enc_cert,
upn_len):
fake_cert_data = ''
fake_cert_data += x509.PEM_HEADER + '\n'
fake_cert_data += x509constants.PEM_HEADER + '\n'
fake_cert_data += 'fake cert' + '\n'
fake_cert_data += x509.PEM_FOOTER
fake_cert_data += x509constants.PEM_FOOTER
mock_get_cert_base64.return_value = 'fake cert'
mock_CryptStringToBinaryA.return_value = crypttstr
mock_CertOpenStore.return_value = store_handle