# http://www.apache.org/licenses/LICENSE-2.0
##############################################################################
-# Unittest for yardstick.common.utils
-
-from __future__ import absolute_import
-
from copy import deepcopy
import errno
+import importlib
import ipaddress
from itertools import product, chain
import mock
import os
import six
from six.moves import configparser
+import time
import unittest
-import yardstick
-from yardstick.common import utils
+from yardstick import ssh
from yardstick.common import constants
+from yardstick.common import utils
+from yardstick.common import exceptions
+from yardstick.tests.unit import base as ut_base
-class IterSubclassesTestCase(unittest.TestCase):
+class IterSubclassesTestCase(ut_base.BaseUnitTestCase):
# Disclaimer: this class is a modified copy from
# rally/tests/unit/common/plugin/test_discover.py
# Copyright 2015: Mirantis Inc.
self.assertEqual([B, C, D], list(utils.itersubclasses(A)))
-class ImportModulesFromPackageTestCase(unittest.TestCase):
+class ImportModulesFromPackageTestCase(ut_base.BaseUnitTestCase):
@mock.patch('yardstick.common.utils.os.walk')
def test_import_modules_from_package_no_mod(self, mock_walk):
utils.import_modules_from_package('foo.bar')
@mock.patch('yardstick.common.utils.os.walk')
- @mock.patch('yardstick.common.utils.importutils')
- def test_import_modules_from_package(self, mock_importutils, mock_walk):
+ @mock.patch.object(importlib, 'import_module')
+ def test_import_modules_from_package(self, mock_import_module, mock_walk):
yardstick_root = os.path.dirname(os.path.dirname(yardstick.__file__))
mock_walk.return_value = ([
])
utils.import_modules_from_package('foo.bar')
- mock_importutils.import_module.assert_called_with('bar.baz')
+ mock_import_module.assert_called_once_with('bar.baz')
-class GetParaFromYaml(unittest.TestCase):
+class GetParaFromYaml(ut_base.BaseUnitTestCase):
@mock.patch('yardstick.common.utils.os.environ.get')
def test_get_param_para_not_found(self, get_env):
return file_path
-class CommonUtilTestCase(unittest.TestCase):
+class CommonUtilTestCase(ut_base.BaseUnitTestCase):
def setUp(self):
self.data = {
("=".join(item) for item in sorted(flattened_data.items())))
self.assertEqual(result, line)
+ def test_get_key_with_default_negative(self):
+ with self.assertRaises(KeyError):
+ utils.get_key_with_default({}, 'key1')
+
+ @mock.patch('yardstick.common.utils.open', create=True)
+ def test_(self, mock_open):
+ mock_open.side_effect = IOError
+
+ with self.assertRaises(IOError):
+ utils.find_relative_file('my/path', 'task/path')
+
+ self.assertEqual(mock_open.call_count, 2)
+
+ @mock.patch('yardstick.common.utils.open', create=True)
+ def test_open_relative_path(self, mock_open):
+ mock_open_result = mock_open()
+ mock_open_call_count = 1 # initial call to get result
+
+ self.assertEqual(utils.open_relative_file('foo', 'bar'), mock_open_result)
+
+ mock_open_call_count += 1 # one more call expected
+ self.assertEqual(mock_open.call_count, mock_open_call_count)
+ self.assertIn('foo', mock_open.call_args_list[-1][0][0])
+ self.assertNotIn('bar', mock_open.call_args_list[-1][0][0])
+
+ def open_effect(*args, **kwargs):
+ if kwargs.get('name', args[0]) == os.path.join('bar', 'foo'):
+ return mock_open_result
+ raise IOError(errno.ENOENT, 'not found')
+
+ mock_open.side_effect = open_effect
+ self.assertEqual(utils.open_relative_file('foo', 'bar'), mock_open_result)
+
+ mock_open_call_count += 2 # two more calls expected
+ self.assertEqual(mock_open.call_count, mock_open_call_count)
+ self.assertIn('foo', mock_open.call_args_list[-1][0][0])
+ self.assertIn('bar', mock_open.call_args_list[-1][0][0])
+
+ # test an IOError of type ENOENT
+ mock_open.side_effect = IOError(errno.ENOENT, 'not found')
+ with self.assertRaises(IOError):
+ # the second call still raises
+ utils.open_relative_file('foo', 'bar')
-class TestMacAddressToHex(unittest.TestCase):
+ mock_open_call_count += 2 # two more calls expected
+ self.assertEqual(mock_open.call_count, mock_open_call_count)
+ self.assertIn('foo', mock_open.call_args_list[-1][0][0])
+ self.assertIn('bar', mock_open.call_args_list[-1][0][0])
+
+ # test an IOError other than ENOENT
+ mock_open.side_effect = IOError(errno.EBUSY, 'busy')
+ with self.assertRaises(IOError):
+ utils.open_relative_file('foo', 'bar')
+
+ mock_open_call_count += 1 # one more call expected
+ self.assertEqual(mock_open.call_count, mock_open_call_count)
+
+
+class TestMacAddressToHex(ut_base.BaseUnitTestCase):
def test_mac_address_to_hex_list(self):
self.assertEqual(utils.mac_address_to_hex_list("ea:3e:e1:9a:99:e8"),
['0xea', '0x3e', '0xe1', '0x9a', '0x99', '0xe8'])
-class TranslateToStrTestCase(unittest.TestCase):
+class TranslateToStrTestCase(ut_base.BaseUnitTestCase):
def test_translate_to_str_unicode(self):
input_str = u'hello'
self.assertIs(input_value, result)
-class TestParseCpuInfo(unittest.TestCase):
+class TestParseCpuInfo(ut_base.BaseUnitTestCase):
def test_single_socket_no_hyperthread(self):
cpuinfo = """\
"""
socket_map = utils.SocketTopology.parse_cpuinfo(cpuinfo)
- assert sorted(socket_map.keys()) == [0]
- assert sorted(socket_map[0].keys()) == [2, 3, 4]
+ self.assertEqual(sorted(socket_map.keys()), [0])
+ self.assertEqual(sorted(socket_map[0].keys()), [2, 3, 4])
def test_single_socket_hyperthread(self):
cpuinfo = """\
"""
socket_map = utils.SocketTopology.parse_cpuinfo(cpuinfo)
- assert sorted(socket_map.keys()) == [0]
- assert sorted(socket_map[0].keys()) == [1, 2, 3]
- assert sorted(socket_map[0][1]) == [5]
- assert sorted(socket_map[0][2]) == [6]
- assert sorted(socket_map[0][3]) == [7]
+ self.assertEqual(sorted(socket_map.keys()), [0])
+ self.assertEqual(sorted(socket_map[0].keys()), [1, 2, 3])
+ self.assertEqual(sorted(socket_map[0][1]), [5])
+ self.assertEqual(sorted(socket_map[0][2]), [6])
+ self.assertEqual(sorted(socket_map[0][3]), [7])
def test_dual_socket_hyperthread(self):
cpuinfo = """\
"""
socket_map = utils.SocketTopology.parse_cpuinfo(cpuinfo)
- assert sorted(socket_map.keys()) == [0, 1]
- assert sorted(socket_map[0].keys()) == [0, 1, 2]
- assert sorted(socket_map[1].keys()) == [26, 27, 28]
- assert sorted(socket_map[0][0]) == [44]
- assert sorted(socket_map[0][1]) == [1]
- assert sorted(socket_map[0][2]) == [2]
- assert sorted(socket_map[1][26]) == [85]
- assert sorted(socket_map[1][27]) == [86]
- assert sorted(socket_map[1][28]) == [43, 87]
+ self.assertEqual(sorted(socket_map.keys()), [0, 1])
+ self.assertEqual(sorted(socket_map[0].keys()), [0, 1, 2])
+ self.assertEqual(sorted(socket_map[1].keys()), [26, 27, 28])
+ self.assertEqual(sorted(socket_map[0][0]), [44])
+ self.assertEqual(sorted(socket_map[0][1]), [1])
+ self.assertEqual(sorted(socket_map[0][2]), [2])
+ self.assertEqual(sorted(socket_map[1][26]), [85])
+ self.assertEqual(sorted(socket_map[1][27]), [86])
+ self.assertEqual(sorted(socket_map[1][28]), [43, 87])
def test_dual_socket_no_hyperthread(self):
cpuinfo = """\
"""
socket_map = utils.SocketTopology.parse_cpuinfo(cpuinfo)
processors = socket_map.processors()
- assert processors == [1, 2, 43, 44, 85, 86, 87]
+ self.assertEqual(processors, [1, 2, 43, 44, 85, 86, 87])
cores = socket_map.cores()
- assert cores == [0, 1, 2, 26, 27, 28]
+ self.assertEqual(cores, [0, 1, 2, 26, 27, 28])
sockets = socket_map.sockets()
- assert sockets == [0, 1]
+ self.assertEqual(sockets, [0, 1])
-class ChangeObjToDictTestCase(unittest.TestCase):
+class ChangeObjToDictTestCase(ut_base.BaseUnitTestCase):
def test_change_obj_to_dict(self):
class A(object):
self.assertEqual(obj_r, obj_s)
-class SetDictValueTestCase(unittest.TestCase):
+class SetDictValueTestCase(ut_base.BaseUnitTestCase):
def test_set_dict_value(self):
input_dic = {
self.assertEqual(output_dic.get('welcome', {}).get('to'), 'yardstick')
-class RemoveFileTestCase(unittest.TestCase):
+class RemoveFileTestCase(ut_base.BaseUnitTestCase):
def test_remove_file(self):
try:
self.assertTrue(isinstance(e, OSError))
-class TestUtils(unittest.TestCase):
+class ParseIniFileTestCase(ut_base.BaseUnitTestCase):
+
+ def setUp(self):
+ self._mock_config_parser_type = mock.patch.object(configparser,
+ 'ConfigParser')
+ self.mock_config_parser_type = self._mock_config_parser_type.start()
+ self.addCleanup(self._stop_mocks)
+
+ def _stop_mocks(self):
+ self._mock_config_parser_type.stop()
+
+ def test_parse_ini_file(self):
+ defaults = {'default1': 'value1',
+ 'default2': 'value2'}
+ s1 = {'key1': 'value11',
+ 'key2': 'value22'}
+ s2 = {'key1': 'value123',
+ 'key2': 'value234'}
+
+ mock_config_parser = mock.Mock()
+ self.mock_config_parser_type.return_value = mock_config_parser
+ mock_config_parser.read.return_value = True
+ mock_config_parser.sections.return_value = ['s1', 's2']
+ mock_config_parser.items.side_effect = iter([
+ defaults.items(),
+ s1.items(),
+ s2.items(),
+ ])
+
+ expected = {'DEFAULT': defaults,
+ 's1': s1,
+ 's2': s2}
+ result = utils.parse_ini_file('my_path')
+ self.assertDictEqual(expected, result)
+
+ @mock.patch.object(utils, 'logger')
+ def test_parse_ini_file_missing_section_header(self, *args):
+ mock_config_parser = mock.Mock()
+ self.mock_config_parser_type.return_value = mock_config_parser
+ mock_config_parser.read.side_effect = (
+ configparser.MissingSectionHeaderError(
+ mock.Mock(), 321, mock.Mock()))
+
+ with self.assertRaises(configparser.MissingSectionHeaderError):
+ utils.parse_ini_file('my_path')
+
+ def test_parse_ini_file_no_file(self):
+ mock_config_parser = mock.Mock()
+ self.mock_config_parser_type.return_value = mock_config_parser
+ mock_config_parser.read.return_value = False
+ with self.assertRaises(RuntimeError):
+ utils.parse_ini_file('my_path')
+
+ def test_parse_ini_file_no_default_section_header(self):
+ s1 = {'key1': 'value11',
+ 'key2': 'value22'}
+ s2 = {'key1': 'value123',
+ 'key2': 'value234'}
+
+ mock_config_parser = mock.Mock()
+ self.mock_config_parser_type.return_value = mock_config_parser
+ mock_config_parser.read.return_value = True
+ mock_config_parser.sections.return_value = ['s1', 's2']
+ mock_config_parser.items.side_effect = iter([
+ configparser.NoSectionError(mock.Mock()),
+ s1.items(),
+ s2.items(),
+ ])
+
+ expected = {'DEFAULT': {},
+ 's1': s1,
+ 's2': s2}
+ result = utils.parse_ini_file('my_path')
+ self.assertDictEqual(expected, result)
+
+
+class TestUtils(ut_base.BaseUnitTestCase):
@mock.patch('yardstick.common.utils.os.makedirs')
def test_makedirs(self, *_):
os.environ.clear()
os.environ.update(base_env)
- @mock.patch('yardstick.common.utils.configparser.ConfigParser')
+ @mock.patch.object(configparser, 'ConfigParser')
def test_parse_ini_file(self, mock_config_parser_type):
defaults = {
'default1': 'value1',
result = utils.parse_ini_file('my_path')
self.assertDictEqual(result, expected)
- @mock.patch('yardstick.common.utils.configparser.ConfigParser')
- def test_parse_ini_file_missing_section_header(self, mock_config_parser_type):
+ @mock.patch.object(utils, 'logger')
+ @mock.patch.object(configparser, 'ConfigParser')
+ def test_parse_ini_file_missing_section_header(
+ self, mock_config_parser_type, *args):
mock_config_parser = mock_config_parser_type()
- mock_config_parser.read.side_effect = \
- configparser.MissingSectionHeaderError(mock.Mock(), 321, mock.Mock())
+ mock_config_parser.read.side_effect = (
+ configparser.MissingSectionHeaderError(mock.Mock(), 321,
+ mock.Mock()))
with self.assertRaises(configparser.MissingSectionHeaderError):
utils.parse_ini_file('my_path')
- @mock.patch('yardstick.common.utils.configparser.ConfigParser')
+ @mock.patch.object(configparser, 'ConfigParser')
def test_parse_ini_file_no_file(self, mock_config_parser_type):
mock_config_parser = mock_config_parser_type()
mock_config_parser.read.return_value = False
with self.assertRaises(RuntimeError):
utils.parse_ini_file('my_path')
- @mock.patch('yardstick.common.utils.configparser.ConfigParser')
+ @mock.patch.object(configparser, 'ConfigParser')
def test_parse_ini_file_no_default_section_header(self, mock_config_parser_type):
s1 = {
'key1': 'value11',
with self.assertRaises(RuntimeError):
utils.validate_non_string_sequence(1, raise_exc=RuntimeError)
- def test_error_class(self):
- with self.assertRaises(RuntimeError):
- utils.ErrorClass()
-
- error_instance = utils.ErrorClass(test='')
- with self.assertRaises(AttributeError):
- error_instance.get_name()
-
-class TestUtilsIpAddrMethods(unittest.TestCase):
+class TestUtilsIpAddrMethods(ut_base.BaseUnitTestCase):
GOOD_IP_V4_ADDRESS_STR_LIST = [
u'0.0.0.0',
self.assertEqual(utils.ip_to_hex(value), value)
-class SafeDecodeUtf8TestCase(unittest.TestCase):
+class SafeDecodeUtf8TestCase(ut_base.BaseUnitTestCase):
@unittest.skipIf(six.PY2,
'This test should only be launched with Python 3.x')
self.assertEqual('this is a byte array', out)
-def main():
- unittest.main()
-
-if __name__ == '__main__':
- main()
+class ReadMeminfoTestCase(ut_base.BaseUnitTestCase):
+
+ MEMINFO = (b'MemTotal: 65860500 kB\n'
+ b'MemFree: 28690900 kB\n'
+ b'MemAvailable: 52873764 kB\n'
+ b'Active(anon): 3015676 kB\n'
+ b'HugePages_Total: 8\n'
+ b'Hugepagesize: 1048576 kB')
+ MEMINFO_DICT = {'MemTotal': '65860500',
+ 'MemFree': '28690900',
+ 'MemAvailable': '52873764',
+ 'Active(anon)': '3015676',
+ 'HugePages_Total': '8',
+ 'Hugepagesize': '1048576'}
+
+ def test_read_meminfo(self):
+ ssh_client = ssh.SSH('user', 'host')
+ with mock.patch.object(ssh_client, 'get_file_obj') as \
+ mock_get_client, \
+ mock.patch.object(six, 'BytesIO',
+ return_value=six.BytesIO(self.MEMINFO)):
+ output = utils.read_meminfo(ssh_client)
+ mock_get_client.assert_called_once_with('/proc/meminfo', mock.ANY)
+ self.assertEqual(self.MEMINFO_DICT, output)
+
+
+class TimerTestCase(ut_base.BaseUnitTestCase):
+
+ def test__getattr(self):
+ with utils.Timer() as timer:
+ time.sleep(1)
+ self.assertEqual(1, round(timer.total_seconds(), 0))
+ self.assertEqual(1, timer.delta.seconds)
+
+ def test__enter_with_timeout(self):
+ with utils.Timer(timeout=10) as timer:
+ time.sleep(1)
+ self.assertEqual(1, round(timer.total_seconds(), 0))
+
+ def test__enter_with_timeout_exception(self):
+ with self.assertRaises(exceptions.TimerTimeout):
+ with utils.Timer(timeout=1):
+ time.sleep(2)
+
+ def test__enter_with_timeout_no_exception(self):
+ with utils.Timer(timeout=1, raise_exception=False):
+ time.sleep(2)
+
+ def test__iter(self):
+ iterations = []
+ for i in utils.Timer(timeout=2):
+ iterations.append(i)
+ time.sleep(1.1)
+ self.assertEqual(2, len(iterations))
+
+
+class WaitUntilTrueTestCase(ut_base.BaseUnitTestCase):
+
+ def test_no_timeout(self):
+ self.assertIsNone(utils.wait_until_true(lambda: True,
+ timeout=1, sleep=1))
+
+ def test_timeout_generic_exception(self):
+ with self.assertRaises(exceptions.WaitTimeout):
+ self.assertIsNone(utils.wait_until_true(lambda: False,
+ timeout=1, sleep=1))
+
+ def test_timeout_given_exception(self):
+ class MyTimeoutException(exceptions.YardstickException):
+ message = 'My timeout exception'
+
+ with self.assertRaises(MyTimeoutException):
+ self.assertIsNone(
+ utils.wait_until_true(lambda: False, timeout=1, sleep=1,
+ exception=MyTimeoutException))
Code Review / yardstick.git / blobdiff