X-Git-Url: https://gerrit.opnfv.org/gerrit/gitweb?a=blobdiff_plain;f=yardstick%2Ftests%2Funit%2Fcommon%2Ftest_utils.py;h=3cf6c4d055009175ca86fb46cac86dc1f3da890e;hb=749a77816a4ecd1adee41f896c6e6fc779127046;hp=31b10e6dacde8fd197c2cf8f9660d3bbef09ca41;hpb=c9b1716fd37284d09b4c327b65fd5d7c8769fc7f;p=yardstick.git diff --git a/yardstick/tests/unit/common/test_utils.py b/yardstick/tests/unit/common/test_utils.py index 31b10e6da..3cf6c4d05 100644 --- a/yardstick/tests/unit/common/test_utils.py +++ b/yardstick/tests/unit/common/test_utils.py @@ -12,11 +12,14 @@ import errno import importlib import ipaddress from itertools import product, chain -import mock import os +import socket +import time +import threading + +import mock import six from six.moves import configparser -import time import unittest import yardstick @@ -24,9 +27,10 @@ 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. @@ -47,7 +51,7 @@ class IterSubclassesTestCase(unittest.TestCase): 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): @@ -72,7 +76,7 @@ class ImportModulesFromPackageTestCase(unittest.TestCase): 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): @@ -96,7 +100,7 @@ class GetParaFromYaml(unittest.TestCase): return file_path -class CommonUtilTestCase(unittest.TestCase): +class CommonUtilTestCase(ut_base.BaseUnitTestCase): def setUp(self): self.data = { @@ -186,14 +190,22 @@ class CommonUtilTestCase(unittest.TestCase): self.assertEqual(mock_open.call_count, mock_open_call_count) -class TestMacAddressToHex(unittest.TestCase): +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']) + def test_mac_address_to_hex_list_too_short_mac(self): + with self.assertRaises(exceptions.InvalidMacAddress): + utils.mac_address_to_hex_list("ea:3e:e1:9a") -class TranslateToStrTestCase(unittest.TestCase): + def test_mac_address_to_hex_list_no_int_mac(self): + with self.assertRaises(exceptions.InvalidMacAddress): + utils.mac_address_to_hex_list("invalid_mac") + + +class TranslateToStrTestCase(ut_base.BaseUnitTestCase): def test_translate_to_str_unicode(self): input_str = u'hello' @@ -219,7 +231,7 @@ class TranslateToStrTestCase(unittest.TestCase): self.assertIs(input_value, result) -class TestParseCpuInfo(unittest.TestCase): +class TestParseCpuInfo(ut_base.BaseUnitTestCase): def test_single_socket_no_hyperthread(self): cpuinfo = """\ @@ -804,7 +816,7 @@ power management: self.assertEqual(sockets, [0, 1]) -class ChangeObjToDictTestCase(unittest.TestCase): +class ChangeObjToDictTestCase(ut_base.BaseUnitTestCase): def test_change_obj_to_dict(self): class A(object): @@ -817,7 +829,7 @@ class ChangeObjToDictTestCase(unittest.TestCase): self.assertEqual(obj_r, obj_s) -class SetDictValueTestCase(unittest.TestCase): +class SetDictValueTestCase(ut_base.BaseUnitTestCase): def test_set_dict_value(self): input_dic = { @@ -827,7 +839,7 @@ class SetDictValueTestCase(unittest.TestCase): self.assertEqual(output_dic.get('welcome', {}).get('to'), 'yardstick') -class RemoveFileTestCase(unittest.TestCase): +class RemoveFileTestCase(ut_base.BaseUnitTestCase): def test_remove_file(self): try: @@ -837,7 +849,83 @@ class RemoveFileTestCase(unittest.TestCase): 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, *_): @@ -992,7 +1080,7 @@ class TestUtils(unittest.TestCase): utils.validate_non_string_sequence(1, raise_exc=RuntimeError) -class TestUtilsIpAddrMethods(unittest.TestCase): +class TestUtilsIpAddrMethods(ut_base.BaseUnitTestCase): GOOD_IP_V4_ADDRESS_STR_LIST = [ u'0.0.0.0', @@ -1035,6 +1123,19 @@ class TestUtilsIpAddrMethods(unittest.TestCase): u'123:4567:89ab:cdef:123:4567:89ab:cdef/129', ] + def test_make_ipv4_address(self): + for addr in self.GOOD_IP_V4_ADDRESS_STR_LIST: + # test with no mask + expected = ipaddress.IPv4Address(addr) + self.assertEqual(utils.make_ipv4_address(addr), expected, addr) + + def test_make_ipv4_address_error(self): + addr_list = self.INVALID_IP_ADDRESS_STR_LIST +\ + self.GOOD_IP_V6_ADDRESS_STR_LIST + for addr in addr_list: + self.assertRaises(Exception, utils.make_ipv4_address, addr) + + def test_safe_ip_address(self): addr_list = self.GOOD_IP_V4_ADDRESS_STR_LIST for addr in addr_list: @@ -1118,8 +1219,22 @@ class TestUtilsIpAddrMethods(unittest.TestCase): for value in chain(value_iter, self.INVALID_IP_ADDRESS_STR_LIST): self.assertEqual(utils.ip_to_hex(value), value) + def test_get_mask_from_ip_range_ipv4(self): + ip_str = '1.1.1.1' + for mask in range(8, 30): + ip = ipaddress.ip_network(ip_str + '/' + str(mask), strict=False) + result = utils.get_mask_from_ip_range(ip[2], ip[-2]) + self.assertEqual(mask, result) + + def test_get_mask_from_ip_range_ipv6(self): + ip_str = '2001::1' + for mask in range(8, 120): + ip = ipaddress.ip_network(ip_str + '/' + str(mask), strict=False) + result = utils.get_mask_from_ip_range(ip[2], ip[-2]) + self.assertEqual(mask, result) -class SafeDecodeUtf8TestCase(unittest.TestCase): + +class SafeDecodeUtf8TestCase(ut_base.BaseUnitTestCase): @unittest.skipIf(six.PY2, 'This test should only be launched with Python 3.x') @@ -1130,7 +1245,7 @@ class SafeDecodeUtf8TestCase(unittest.TestCase): self.assertEqual('this is a byte array', out) -class ReadMeminfoTestCase(unittest.TestCase): +class ReadMeminfoTestCase(ut_base.BaseUnitTestCase): MEMINFO = (b'MemTotal: 65860500 kB\n' b'MemFree: 28690900 kB\n' @@ -1156,7 +1271,7 @@ class ReadMeminfoTestCase(unittest.TestCase): self.assertEqual(self.MEMINFO_DICT, output) -class TimerTestCase(unittest.TestCase): +class TimerTestCase(ut_base.BaseUnitTestCase): def test__getattr(self): with utils.Timer() as timer: @@ -1185,8 +1300,12 @@ class TimerTestCase(unittest.TestCase): time.sleep(1.1) self.assertEqual(2, len(iterations)) + def test_delta_time_sec(self): + with utils.Timer() as timer: + self.assertIsInstance(timer.delta_time_sec(), float) + -class WaitUntilTrueTestCase(unittest.TestCase): +class WaitUntilTrueTestCase(ut_base.BaseUnitTestCase): def test_no_timeout(self): self.assertIsNone(utils.wait_until_true(lambda: True, @@ -1205,3 +1324,86 @@ class WaitUntilTrueTestCase(unittest.TestCase): self.assertIsNone( utils.wait_until_true(lambda: False, timeout=1, sleep=1, exception=MyTimeoutException)) + + def _run_thread(self): + with self.assertRaises(exceptions.WaitTimeout): + utils.wait_until_true(lambda: False, timeout=1, sleep=1) + + def test_timeout_no_main_thread(self): + new_thread = threading.Thread(target=self._run_thread) + new_thread.start() + new_thread.join(timeout=3) + + +class SendSocketCommandTestCase(unittest.TestCase): + + @mock.patch.object(socket, 'socket') + def test_execute_correct(self, mock_socket): + mock_socket_obj = mock.Mock() + mock_socket_obj.connect_ex.return_value = 0 + mock_socket.return_value = mock_socket_obj + self.assertEqual(0, utils.send_socket_command('host', 22, 'command')) + mock_socket.assert_called_once_with(socket.AF_INET, socket.SOCK_STREAM) + mock_socket_obj.connect_ex.assert_called_once_with(('host', 22)) + mock_socket_obj.sendall.assert_called_once_with(six.b('command')) + mock_socket_obj.close.assert_called_once() + + @mock.patch.object(socket, 'socket') + def test_execute_exception(self, mock_socket): + mock_socket_obj = mock.Mock() + mock_socket_obj.connect_ex.return_value = 0 + mock_socket.return_value = mock_socket_obj + mock_socket_obj.sendall.side_effect = socket.error + self.assertEqual(1, utils.send_socket_command('host', 22, 'command')) + mock_socket.assert_called_once_with(socket.AF_INET, socket.SOCK_STREAM) + mock_socket_obj.connect_ex.assert_called_once_with(('host', 22)) + mock_socket_obj.sendall.assert_called_once_with(six.b('command')) + mock_socket_obj.close.assert_called_once() + + +class GetPortMacTestCase(unittest.TestCase): + + def setUp(self): + self.ssh_client = mock.Mock() + self.ssh_client.execute.return_value = (0, 'foo ', '') + + def test_ssh_client_execute_called(self): + utils.get_port_mac(self.ssh_client, 99) + self.ssh_client.execute.assert_called_once_with( + "ifconfig |grep HWaddr |grep 99 |awk '{print $5}' ", + raise_on_error=True) + + def test_return_value(self): + self.assertEqual('foo', utils.get_port_mac(self.ssh_client, 99)) + + +class GetPortIPTestCase(unittest.TestCase): + + def setUp(self): + self.ssh_client = mock.Mock() + self.ssh_client.execute.return_value = (0, 'foo ', '') + + def test_ssh_client_execute_called(self): + utils.get_port_ip(self.ssh_client, 99) + self.ssh_client.execute.assert_called_once_with( + "ifconfig 99 |grep 'inet addr' |awk '{print $2}' |cut -d ':' -f2 ", + raise_on_error=True) + + def test_return_value(self): + self.assertEqual('foo', utils.get_port_ip(self.ssh_client, 99)) + + +class SafeCaseTestCase(unittest.TestCase): + + def test_correct_type_int(self): + self.assertEqual(35, utils.safe_cast('35', int, 0)) + + def test_correct_int_as_string(self): + self.assertEqual(25, utils.safe_cast('25', 'int', 0)) + + def test_incorrect_type_as_string(self): + with self.assertRaises(exceptions.InvalidType): + utils.safe_cast('100', 'intt', 0) + + def test_default_value(self): + self.assertEqual(0, utils.safe_cast('', 'int', 0))