diff options
Diffstat (limited to 'yardstick/tests/unit/common/test_utils.py')
| -rw-r--r-- | yardstick/tests/unit/common/test_utils.py | 158 |
1 files changed, 155 insertions, 3 deletions
diff --git a/yardstick/tests/unit/common/test_utils.py b/yardstick/tests/unit/common/test_utils.py index 446afdd38..8fed5ecf1 100644 --- a/yardstick/tests/unit/common/test_utils.py +++ b/yardstick/tests/unit/common/test_utils.py @@ -12,12 +12,14 @@ import errno import importlib import ipaddress from itertools import product, chain -import mock import os -import six -from six.moves import configparser import socket import time +import threading + +import mock +import six +from six.moves import configparser import unittest import yardstick @@ -194,6 +196,14 @@ class TestMacAddressToHex(ut_base.BaseUnitTestCase): 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") + + 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): @@ -1113,6 +1123,28 @@ class TestUtilsIpAddrMethods(ut_base.BaseUnitTestCase): 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_get_ip_range_count(self): + iprange = "192.168.0.1-192.168.0.25" + count = utils.get_ip_range_count(iprange) + self.assertEqual(count, 24) + + def test_get_ip_range_start(self): + iprange = "192.168.0.1-192.168.0.25" + start = utils.get_ip_range_start(iprange) + self.assertEqual(start, "192.168.0.1") + def test_safe_ip_address(self): addr_list = self.GOOD_IP_V4_ADDRESS_STR_LIST for addr in addr_list: @@ -1196,6 +1228,20 @@ class TestUtilsIpAddrMethods(ut_base.BaseUnitTestCase): 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(ut_base.BaseUnitTestCase): @@ -1263,6 +1309,10 @@ class TimerTestCase(ut_base.BaseUnitTestCase): 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(ut_base.BaseUnitTestCase): @@ -1284,6 +1334,15 @@ class WaitUntilTrueTestCase(ut_base.BaseUnitTestCase): 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): @@ -1309,3 +1368,96 @@ class SendSocketCommandTestCase(unittest.TestCase): 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)) + + +class SetupHugepagesTestCase(unittest.TestCase): + + @mock.patch.object(six, 'BytesIO', return_value=six.BytesIO(b'5\n')) + @mock.patch.object(utils, 'read_meminfo', + return_value={'Hugepagesize': '1024'}) + def test_setup_hugepages(self, mock_meminfo, *args): + ssh = mock.Mock() + ssh.execute = mock.Mock() + hp_size_kb, hp_number, hp_number_set = utils.setup_hugepages(ssh, 10 * 1024) + mock_meminfo.assert_called_once_with(ssh) + ssh.execute.assert_called_once_with( + 'echo 10 | sudo tee /proc/sys/vm/nr_hugepages') + self.assertEqual(hp_size_kb, 1024) + self.assertEqual(hp_number, 10) + self.assertEqual(hp_number_set, 5) + + +class GetOSSampleInfoTestCase(unittest.TestCase): + + def test_get_os_version(self, *args): + ssh = mock.Mock() + ssh.execute.return_value = (0, "18.04", "") + utils.get_os_version(ssh) + ssh.execute.assert_called_once_with("cat /etc/lsb-release") + + def test_get_kernel_version(self, *args): + ssh = mock.Mock() + ssh.execute.return_value = (0, "Linux", "") + utils.get_kernel_version(ssh) + ssh.execute.assert_called_once_with("uname -a") + + def test_get_sample_vnf_info(self, *args): + json_out = """ + {"UDP_Replay": { + "branch_commit": "47123bfc1b3c0d0b01884aebbce1a3e09ad7ddb0", + "md5": "4577702f6d6848380bd912232a1b9ca5", + "path_vnf": "/opt/nsb_bin/UDP_Replay" + } + }""" + json_file = '/opt/nsb_bin/yardstick_sample_vnf.json' + ssh = mock.Mock() + ssh.execute.return_value = (0, json_out, "") + utils.get_sample_vnf_info(ssh, json_file) |