# limitations under the License.
#
-from __future__ import absolute_import
-
-import copy
+from itertools import repeat, chain
import os
import socket
-import unittest
-from itertools import repeat, chain
-from contextlib import contextmanager
+import time
+
import mock
+import unittest
from tests.unit import STL_MOCKS
from yardstick.network_services.vnf_generic.vnf.base import VnfdHelper
+
STLClient = mock.MagicMock()
stl_patch = mock.patch.dict("sys.modules", STL_MOCKS)
stl_patch.start()
from yardstick.network_services.vnf_generic.vnf.prox_helpers import ProxTestDataTuple
from yardstick.network_services.vnf_generic.vnf.prox_helpers import ProxDpdkVnfSetupEnvHelper
from yardstick.network_services.vnf_generic.vnf.prox_helpers import TotStatsTuple
+ from yardstick.network_services.vnf_generic.vnf.prox_helpers import ProxDataHelper
from yardstick.network_services.vnf_generic.vnf.prox_helpers import ProxResourceHelper
+ from yardstick.network_services.vnf_generic.vnf.prox_helpers import ProxProfileHelper
+ from yardstick.network_services.vnf_generic.vnf.prox_helpers import ProxMplsProfileHelper
+ from yardstick.network_services.vnf_generic.vnf.prox_helpers import ProxBngProfileHelper
+ from yardstick.network_services.vnf_generic.vnf.prox_helpers import ProxVpeProfileHelper
+ from yardstick.network_services.vnf_generic.vnf.prox_helpers import ProxlwAFTRProfileHelper
class TestCoreTuple(unittest.TestCase):
"""
-@mock.patch('yardstick.network_services.vnf_generic.vnf.prox_helpers.time')
class TestProxSocketHelper(unittest.TestCase):
+
+ def setUp(self):
+ self.mock_time_sleep = mock.patch.object(time, 'sleep').start()
+
@mock.patch('yardstick.network_services.vnf_generic.vnf.prox_helpers.socket')
- def test___init__(self, mock_socket, mock_time):
+ def test___init__(self, mock_socket):
expected = mock_socket.socket()
prox = ProxSocketHelper()
result = prox._sock
self.assertEqual(result, expected)
- def test_connect(self, mock_time):
+ def test_connect(self):
mock_sock = mock.MagicMock()
prox = ProxSocketHelper(mock_sock)
prox.connect('10.20.30.40', 23456)
self.assertEqual(mock_sock.connect.call_count, 1)
- def test_get_sock(self, mock_time):
+ def test_get_sock(self):
mock_sock = mock.MagicMock()
prox = ProxSocketHelper(mock_sock)
result = prox.get_socket()
self.assertIs(result, mock_sock)
@mock.patch('yardstick.network_services.vnf_generic.vnf.prox_helpers.select')
- def test_get_data(self, mock_select, mock_time):
+ def test_get_data(self, mock_select):
mock_select.select.side_effect = [[1], [0]]
mock_socket = mock.MagicMock()
mock_recv = mock_socket.recv()
self.assertEqual(ret, 'jumped over')
self.assertEqual(len(prox._pkt_dumps), 3)
- def test__parse_socket_data_mixed_data(self, mock_time):
+ def test__parse_socket_data_mixed_data(self):
prox = ProxSocketHelper(mock.MagicMock())
ret = prox._parse_socket_data(PACKET_DUMP_NON_1, False)
self.assertEqual(ret, 'not_a_dump,1,2')
self.assertEqual(ret, 'not_a_dump,1,2')
self.assertEqual(len(prox._pkt_dumps), 1)
- def test__parse_socket_data_bad_data(self, mock_time):
+ def test__parse_socket_data_bad_data(self):
prox = ProxSocketHelper(mock.MagicMock())
with self.assertRaises(ValueError):
prox._parse_socket_data(PACKET_DUMP_BAD_1, False)
ret = prox._parse_socket_data(PACKET_DUMP_BAD_3, False)
self.assertEqual(ret, 'pktdump,3')
- def test__parse_socket_data_pkt_dump_only(self, mock_time):
+ def test__parse_socket_data_pkt_dump_only(self):
prox = ProxSocketHelper(mock.MagicMock())
ret = prox._parse_socket_data('', True)
self.assertFalse(ret)
ret = prox._parse_socket_data(PACKET_DUMP_2, True)
self.assertTrue(ret)
- def test_put_command(self, mock_time):
+ def test_put_command(self):
mock_socket = mock.MagicMock()
prox = ProxSocketHelper(mock_socket)
prox.put_command("data")
mock_socket.sendall.assert_called_once()
- def test_put_command_socket_error(self, mock_time):
+ def test_put_command_socket_error(self):
mock_socket = mock.MagicMock()
mock_socket.sendall.side_effect = OSError
prox = ProxSocketHelper(mock_socket)
prox.put_command("data")
mock_socket.sendall.assert_called_once()
- def test_get_packet_dump(self, mock_time):
+ def test_get_packet_dump(self):
mock_socket = mock.MagicMock()
prox = ProxSocketHelper(mock_socket)
prox._pkt_dumps = []
self.assertEqual(prox.get_packet_dump(), 234)
self.assertEqual(prox._pkt_dumps, [])
- def test_stop_all_reset(self, mock_time):
+ def test_stop_all_reset(self):
mock_socket = mock.MagicMock()
prox = ProxSocketHelper(mock_socket)
prox.stop_all_reset()
mock_socket.sendall.assert_called()
- def test_stop_all(self, mock_time):
+ def test_stop_all(self):
mock_socket = mock.MagicMock()
prox = ProxSocketHelper(mock_socket)
prox.stop_all()
mock_socket.sendall.assert_called()
- def test_stop(self, mock_time):
+ def test_stop(self):
mock_socket = mock.MagicMock()
prox = ProxSocketHelper(mock_socket)
prox.stop([3, 4, 5], 16)
mock_socket.sendall.assert_called()
- def test_start_all(self, mock_time):
+ def test_start_all(self):
mock_socket = mock.MagicMock()
prox = ProxSocketHelper(mock_socket)
prox.start_all()
mock_socket.sendall.assert_called()
- def test_start(self, mock_time):
+ def test_start(self):
mock_socket = mock.MagicMock()
prox = ProxSocketHelper(mock_socket)
prox.start([3, 4, 5])
mock_socket.sendall.assert_called()
- def test_reset_stats(self, mock_time):
+ def test_reset_stats(self):
mock_socket = mock.MagicMock()
prox = ProxSocketHelper(mock_socket)
prox.reset_stats()
mock_socket.sendall.assert_called()
- def test_set_pkt_size(self, mock_time):
+ def test_set_pkt_size(self):
mock_socket = mock.MagicMock()
prox = ProxSocketHelper(mock_socket)
prox.set_pkt_size([3, 4, 5], 1024)
self.assertEqual(mock_socket.sendall.call_count, 3)
- def test_set_value(self, mock_time):
+ def test_set_value(self):
mock_socket = mock.MagicMock()
prox = ProxSocketHelper(mock_socket)
prox.set_value([3, 4, 5], 10, 20, 30)
self.assertEqual(mock_socket.sendall.call_count, 3)
- def test_reset_values(self, mock_time):
+ def test_reset_values(self):
mock_socket = mock.MagicMock()
prox = ProxSocketHelper(mock_socket)
prox.reset_values([3, 4, 5])
self.assertEqual(mock_socket.sendall.call_count, 3)
- def test_set_speed(self, mock_time):
+ def test_set_speed(self):
mock_socket = mock.MagicMock()
prox = ProxSocketHelper(mock_socket)
prox.set_speed([3, 4, 5], 1000)
self.assertEqual(mock_socket.sendall.call_count, 3)
- def test_slope_speed(self, mock_time):
+ def test_slope_speed(self):
core_data = [
{
'cores': [3, 4, 5],
prox.slope_speed(core_data, 5, 5)
self.assertEqual(set_speed.call_count, 10)
- def test_set_pps(self, mock_time):
+ def test_set_pps(self):
mock_socket = mock.MagicMock()
prox = ProxSocketHelper(mock_socket)
prox.set_pps([3, 4, 5], 1000, 512)
self.assertEqual(mock_socket.sendall.call_count, 3)
- def test_lat_stats(self, mock_time):
+ def test_lat_stats(self):
latency_output = [
'1, 2 , 3', # has white space
'4,5', # too short
self.assertEqual(mock_socket.sendall.call_count, 5)
self.assertEqual(result, expected)
- def test_get_all_tot_stats_error(self, mock_time):
+ def test_get_all_tot_stats_error(self):
mock_socket = mock.MagicMock()
prox = ProxSocketHelper(mock_socket)
prox.get_data = mock.MagicMock(return_value='3,4,5')
result = prox.get_all_tot_stats()
self.assertEqual(result, expected)
- def test_get_all_tot_stats(self, mock_time):
+ def test_get_all_tot_stats(self):
mock_socket = mock.MagicMock()
prox = ProxSocketHelper(mock_socket)
prox.get_data = mock.MagicMock(return_value='3,4,5,6')
result = prox.get_all_tot_stats()
self.assertEqual(result, expected)
- def test_hz(self, mock_time):
+ def test_hz(self):
mock_socket = mock.MagicMock()
prox = ProxSocketHelper(mock_socket)
prox.get_data = mock.MagicMock(return_value='3,4,5,6')
result = prox.hz()
self.assertEqual(result, expected)
- def test_rx_stats(self, mock_time):
- core_stats = [
- '3,4,5,6',
- '7,8,9,10,NaN',
- '11,12,13,14,15',
- ]
-
- mock_socket = mock.MagicMock()
- prox = ProxSocketHelper(mock_socket)
- prox.get_data = mock.MagicMock(side_effect=core_stats)
- expected = 21, 24, 27, 14
- result = prox.rx_stats([3, 4, 5], 16)
- self.assertEqual(result, expected)
-
- def test_core_stats(self, mock_time):
+ def test_core_stats(self):
core_stats = [
'3,4,5,6',
'7,8,9,10,NaN',
result = prox.core_stats([3, 4, 5], 16)
self.assertEqual(result, expected)
- def test_port_stats(self, mock_time):
+ def test_port_stats(self):
port_stats = [
','.join(str(n) for n in range(3, 15)),
','.join(str(n) for n in range(8, 32, 2)),
result = prox.port_stats([3, 4, 5])
self.assertEqual(result, expected)
- def test_measure_tot_stats(self, mock_time):
+ def test_measure_tot_stats(self):
start_tot = 3, 4, 5, 6
end_tot = 7, 9, 11, 13
delta_tot = 4, 5, 6, 7
pass
self.assertEqual(result, expected)
- def test_tot_stats(self, mock_time):
+ def test_tot_stats(self):
mock_socket = mock.MagicMock()
prox = ProxSocketHelper(mock_socket)
prox.get_data = mock.MagicMock(return_value='3,4,5,6')
result = prox.tot_stats()
self.assertEqual(result, expected)
- def test_tot_ierrors(self, mock_time):
+ def test_tot_ierrors(self):
mock_socket = mock.MagicMock()
prox = ProxSocketHelper(mock_socket)
prox.get_data = mock.MagicMock(return_value='3,4,5,6')
result = prox.tot_ierrors()
self.assertEqual(result, expected)
- def test_set_count(self, mock_time):
+ def test_set_count(self):
mock_socket = mock.MagicMock()
prox = ProxSocketHelper(mock_socket)
prox.set_count(432, [3, 4, 5])
self.assertEqual(mock_socket.sendall.call_count, 3)
- def test_dump_rx(self, mock_time):
+ def test_dump_rx(self):
mock_socket = mock.MagicMock()
prox = ProxSocketHelper(mock_socket)
prox.dump_rx(3, 5, 8)
self.assertEqual(mock_socket.sendall.call_count, 1)
- def test_quit(self, mock_time):
+ def test_quit(self):
mock_socket = mock.MagicMock()
prox = ProxSocketHelper(mock_socket)
prox.quit()
mock_socket.sendall.assert_called()
- def test_force_quit(self, mock_time):
+ def test_force_quit(self):
mock_socket = mock.MagicMock()
prox = ProxSocketHelper(mock_socket)
prox.force_quit()
},
}
+ def test_global_section(self):
+ setup_helper = ProxDpdkVnfSetupEnvHelper(mock.MagicMock(), mock.MagicMock(),
+ mock.MagicMock())
+
+ setup_helper._prox_config_data = [('a', [])]
+
+ with self.assertRaises(KeyError):
+ _ = setup_helper.global_section
+
+ global_section = (
+ 'global', [
+ ('not_name', 'other data'),
+ ('name_not', 'more data'),
+ ('name', 'prox type'),
+ ],
+ )
+
+ setup_helper._prox_config_data = [
+ ('section1', []),
+ ('section2', [
+ ('a', 'b'),
+ ('c', 'd'),
+ ]),
+ ('core 1', []),
+ ('core 2', [
+ ('index', 8),
+ ('mode', ''),
+ ]),
+ global_section,
+ ('core 3', [
+ ('index', 5),
+ ('mode', 'gen'),
+ ('name', 'tagged'),
+ ]),
+ ('section3', [
+ ('key1', 'value1'),
+ ('key2', 'value2'),
+ ('key3', 'value3'),
+ ]),
+ ('core 4', [
+ ('index', 7),
+ ('mode', 'gen'),
+ ('name', 'udp'),
+ ]),
+ ]
+
+ result = setup_helper.global_section
+ self.assertEqual(result, global_section[1])
+
+ def test_find_in_section(self):
+ setup_helper = ProxDpdkVnfSetupEnvHelper(mock.MagicMock(), mock.MagicMock(),
+ mock.MagicMock())
+
+ setup_helper._prox_config_data = [
+ ('global', [
+ ('not_name', 'other data'),
+ ('name_not', 'more data'),
+ ('name', 'prox type'),
+ ]),
+ ('section1', []),
+ ('section2', [
+ ('a', 'b'),
+ ('c', 'd'),
+ ]),
+ ('core 1', []),
+ ('core 2', [
+ ('index', 8),
+ ('mode', ''),
+ ]),
+ ('core 3', [
+ ('index', 5),
+ ('mode', 'gen'),
+ ('name', 'tagged'),
+ ]),
+ ('section3', [
+ ('key1', 'value1'),
+ ('key2', 'value2'),
+ ('key3', 'value3'),
+ ]),
+ ('core 4', [
+ ('index', 7),
+ ('mode', 'gen'),
+ ('name', 'udp'),
+ ]),
+ ]
+
+ expected = 'value3'
+ result = setup_helper.find_in_section('section3', 'key3')
+ self.assertEqual(result, expected)
+
+ expected = 'default value'
+ result = setup_helper.find_in_section('section3', 'key4', 'default value')
+ self.assertEqual(result, expected)
+
+ with self.assertRaises(KeyError):
+ setup_helper.find_in_section('section4', 'key1')
+
+ with self.assertRaises(KeyError):
+ setup_helper.find_in_section('section1', 'key1')
+
def test__replace_quoted_with_value(self):
# empty string
input_str = ''
result = ProxDpdkVnfSetupEnvHelper.write_prox_config(input_data)
self.assertEqual(result, expected)
+ def test_prox_config_data(self):
+ setup_helper = ProxDpdkVnfSetupEnvHelper(mock.MagicMock(), mock.MagicMock(),
+ mock.MagicMock())
+
+ setup_helper.config_queue = config_queue = mock.MagicMock()
+ config_queue.get.return_value = expected = [('s', [('a', 3), ('b', 45)])]
+
+ result = setup_helper.prox_config_data
+ self.assertEqual(result, expected)
+
@mock.patch('yardstick.network_services.vnf_generic.vnf.prox_helpers.find_relative_file')
def test_build_config_file_no_additional_file(self, mock_find_path):
vnf1 = {
'prox_args': {'-c': ""},
'prox_path': 'd',
'prox_config': 'e/f',
+ 'prox_generate_parameter': False,
}
mock_find_path.side_effect = ['1', '2']
'prox_path': 'd',
'prox_config': 'e/f',
'prox_files': 'g/h.i',
+ 'prox_generate_parameter': True,
}
mock_find_path.side_effect = ['1', '2']
},
}
+ vnfd_helper.port_pairs.all_ports = ['xe0', 'xe1', 'xe2', 'xe3']
helper = ProxDpdkVnfSetupEnvHelper(vnfd_helper, ssh_helper, scenario_helper)
helper.copy_to_target = mock.MagicMock(side_effect=['33', '34', '35'])
helper.generate_prox_config_file = mock.MagicMock(return_value='44')
self.assertEqual(helper._prox_config_data, '44')
self.assertEqual(helper.remote_path, '55')
- @mock.patch('yardstick.network_services.vnf_generic.vnf.prox_helpers.find_relative_file')
- def test_build_config(self, mock_find_path):
+ def test_build_config(self):
vnf1 = {
'prox_args': {'-f': ""},
'prox_path': '/opt/nsb_bin/prox',
],
}
- mock_find_path.side_effect = ['1', '2']
- vnfd_helper = mock.MagicMock()
- ssh_helper = mock.MagicMock()
- ssh_helper.provision_tool.return_value = "/opt/nsb_bin/prox"
+ vnfd_helper = mock.Mock()
+ ssh_helper = mock.Mock()
+ ssh_helper.join_bin_path.return_value = '/opt/nsb_bin/prox'
scenario_helper = ScenarioHelper('vnf1')
scenario_helper.scenario_cfg = {
'task_path': 'a/b',
},
}
- helper = ProxDpdkVnfSetupEnvHelper(vnfd_helper, ssh_helper, scenario_helper)
- helper.remote_path = "/tmp/prox.cfg"
- expected = "sudo bash -c 'cd /opt/nsb_bin; /opt/nsb_bin/prox -o cli -f -f /tmp/prox.cfg '"
- with mock.patch.object(helper, "build_config_file") as mock_build_config:
+ expected = ("sudo bash -c 'cd /opt/nsb_bin; /opt/nsb_bin/prox -o cli "
+ "-f -f /tmp/prox.cfg '")
+
+ helper = ProxDpdkVnfSetupEnvHelper(vnfd_helper, ssh_helper,
+ scenario_helper)
+ with mock.patch.object(helper, 'build_config_file') as mock_cfg_file:
+ helper.remote_path = '/tmp/prox.cfg'
prox_cmd = helper.build_config()
self.assertEqual(prox_cmd, expected)
+ mock_cfg_file.assert_called_once()
def test__insert_additional_file(self):
vnfd_helper = mock.MagicMock()
with self.assertRaises(Exception):
helper.generate_prox_config_file('a/b')
- def test_generate_prox_lua_file(self):
- vnfd_helper = VnfdHelper(self.VNFD0)
- ssh_helper = mock.MagicMock()
- scenario_helper = mock.MagicMock()
-
- helper = ProxDpdkVnfSetupEnvHelper(vnfd_helper, ssh_helper, scenario_helper)
- helper.LUA_PARAMETER_NAME = 'sut'
-
- expected = [
- 'sut_hex_ip_port_0:"98 10 64 13"',
- 'sut_ip_port_0:"152.16.100.19"',
- 'gen_hex_ip_port_0:"98 10 64 13"',
- 'gen_ip_port_0:"152.16.100.19"',
-
- 'sut_hex_ip_port_1:"98 10 28 13"',
- 'sut_ip_port_1:"152.16.40.19"',
- 'gen_hex_ip_port_1:"98 10 28 14"',
- 'gen_ip_port_1:"152.16.40.20"',
- ]
- result = helper.generate_prox_lua_file()
- self.assertListEqual(result.splitlines(), expected)
-
- def test_upload_prox_lua(self):
- def identity(*args):
- return args
-
- vnfd_helper = mock.MagicMock()
- vnfd_helper.interfaces = []
- ssh_helper = mock.MagicMock()
- scenario_helper = mock.MagicMock()
-
- helper = ProxDpdkVnfSetupEnvHelper(vnfd_helper, ssh_helper, scenario_helper)
- helper.generate_prox_lua_file = mock.MagicMock(return_value=234)
- helper.put_string_to_file = identity
-
- expected = ''
- result = helper.upload_prox_lua('my_dir', {})
- self.assertEqual(result, expected)
-
- input_data = {
- 'lua': {
- 'key1': 'value1 ("inside") tail',
- 'key2': 'value2',
- 'key3 ("key_side") head': 'value3',
- },
- }
-
- expected = 234, 'my_dir/key_side'
- result = helper.upload_prox_lua('my_dir', input_data)
- self.assertEqual(result, expected)
-
def test_put_string_to_file(self):
vnfd_helper = mock.MagicMock()
vnfd_helper.interfaces = []
result = helper.put_string_to_file('my long string', 'a/b')
self.assertEqual(result, expected)
- def test__build_pipeline_kwarags(self):
- vnfd_helper = mock.MagicMock()
- ssh_helper = mock.MagicMock()
- ssh_helper.provision_tool.return_value = "/tmp/nosuch"
- scenario_helper = mock.MagicMock()
-
- helper = ProxDpdkVnfSetupEnvHelper(vnfd_helper, ssh_helper, scenario_helper)
- helper._build_pipeline_kwargs()
- self.assertEqual(helper.pipeline_kwargs, {'tool_path': '/tmp/nosuch', 'tool_dir': '/tmp'})
-
def test_copy_to_target(self):
vnfd_helper = mock.MagicMock()
vnfd_helper.interfaces = []
},
}
- def test_line_rate_to_pps(self):
- expected = 0.25 * 1e8
- result = ProxResourceHelper.line_rate_to_pps(180, 4)
- self.assertEqual(result, expected)
-
def test_find_pci(self):
input_str_list = [
'no target here',
@mock.patch('yardstick.network_services.vnf_generic.vnf.prox_helpers.RETRY_INTERVAL', 0)
@mock.patch('yardstick.network_services.vnf_generic.vnf.prox_helpers.ProxSocketHelper')
- def test_sut(self, mock_socket_helper):
+ def test_sut(self, *args):
helper = ProxResourceHelper(mock.MagicMock())
self.assertIsNone(helper.client)
result = helper.sut
self.assertIs(result, helper.client)
self.assertIs(result, helper.sut)
- @mock.patch('yardstick.network_services.vnf_generic.vnf.prox_helpers.SocketTopology')
- def test_cpu_topology(self, mock_socket_topology):
- mock_socket_topology.parse_cpuinfo.return_value = 432
-
+ def test_test_type(self):
setup_helper = mock.MagicMock()
- setup_helper.ssh_helper.execute.return_value = 0, 'output', ''
+ setup_helper.find_in_section.return_value = expected = 'prox type'
helper = ProxResourceHelper(setup_helper)
- self.assertIsNone(helper._cpu_topology)
- result = helper.cpu_topology
- self.assertEqual(result, 432)
- self.assertIs(result, helper._cpu_topology)
- self.assertIs(result, helper.cpu_topology)
- def test_test_cores(self):
+ self.assertIsNone(helper._test_type)
+ self.assertEqual(helper.test_type, expected)
+ self.assertEqual(helper._test_type, expected)
+ self.assertEqual(helper.test_type, expected)
+
+ def test_collect_collectd_kpi(self):
+ helper = ProxResourceHelper(mock.MagicMock())
+ helper.resource = resource = mock.MagicMock()
+
+ resource.check_if_system_agent_running.return_value = 0, '1234'
+ resource.amqp_collect_nfvi_kpi.return_value = 543
+ resource.check_if_system_agent_running.return_value = (0, None)
+
+ expected = {'core': 543}
+ result = helper.collect_collectd_kpi()
+ self.assertDictEqual(result, expected)
+
+ def test_collect_kpi(self):
+ helper = ProxResourceHelper(mock.MagicMock())
+ helper._queue = queue = mock.MagicMock()
+ helper._result = {'z': 123}
+ helper.resource = resource = mock.MagicMock()
+
+ resource.check_if_system_agent_running.return_value = 0, '1234'
+ resource.amqp_collect_nfvi_kpi.return_value = 543
+ resource.check_if_system_agent_running.return_value = (0, None)
+
+ queue.empty.return_value = False
+ queue.get.return_value = {'a': 789}
+
+ expected = {'z': 123, 'a': 789, 'collect_stats': {'core': 543}}
+ result = helper.collect_kpi()
+ self.assertDictEqual(result, expected)
+
+ @mock.patch('yardstick.network_services.vnf_generic.vnf.prox_helpers.time')
+ @mock.patch('yardstick.network_services.vnf_generic.vnf.prox_helpers.ProxSocketHelper')
+ def test__connect(self, mock_socket_helper_type, *args):
+ client = mock_socket_helper_type()
+ client.connect.side_effect = chain(repeat(socket.error, 5), [None])
+
setup_helper = mock.MagicMock()
- setup_helper.prox_config_data = []
+ setup_helper.vnfd_helper.interfaces = []
helper = ProxResourceHelper(setup_helper)
- helper._cpu_topology = []
- expected = []
- result = helper.test_cores
- self.assertEqual(result, expected)
+ result = helper._connect()
+ self.assertIs(result, client)
- setup_helper.prox_config_data = [
- ('section1', []),
- ('section2', [
- ('a', 'b'),
- ('c', 'd'),
- ]),
- ('core 1s3', []),
- ('core 2s5', [
- ('index', 8),
- ('mode', ''),
- ]),
- ('core 3s1', [
- ('index', 5),
- ('mode', 'gen'),
- ]),
- ('core 4s9h', [
- ('index', 7),
- ('mode', 'gen'),
- ]),
- ]
+ client.connect.side_effect = chain(repeat(socket.error, 65), [None])
+ with self.assertRaises(Exception):
+ helper._connect()
+
+ def test_run_traffic(self):
+ setup_helper = mock.MagicMock()
helper = ProxResourceHelper(setup_helper)
- helper._cpu_topology = {
- 1: {
- 3: {
- 'key1': (23, 32),
- 'key2': (12, 21),
- 'key3': (44, 33),
- },
- },
- 9: {
- 4: {
- 'key1': (44, 32),
- 'key2': (23, 21),
- 'key3': (12, 33),
- },
- },
- }
+ traffic_profile = mock.MagicMock(**{"done": True})
+ helper.run_traffic(traffic_profile)
+ self.assertEqual(helper._terminated.value, 1)
- self.assertIsNone(helper._test_cores)
- expected = [12, 23]
- result = helper.test_cores
- self.assertEqual(result, expected)
- self.assertIs(result, helper._test_cores)
- self.assertIs(result, helper.test_cores)
+ def test__run_traffic_once(self):
+ setup_helper = mock.MagicMock()
+ helper = ProxResourceHelper(setup_helper)
+ traffic_profile = mock.MagicMock(**{"done": True})
+ helper._run_traffic_once(traffic_profile)
+ self.assertEqual(helper._terminated.value, 1)
- def test_get_test_type(self):
+ def test_start_collect(self):
setup_helper = mock.MagicMock()
helper = ProxResourceHelper(setup_helper)
+ helper.resource = resource = mock.MagicMock()
+ self.assertIsNone(helper.start_collect())
+ resource.start.assert_called_once()
- setup_helper.prox_config_data = [
- ('global', [
- ('name', helper.PROX_CORE_MPLS_TEST)
- ]),
- ('section2', [
- ('a', 'b'),
- ('c', 'd'),
- ]),
- ('core 1', []),
- ('core 2', [
- ('index', 8),
- ('mode', ''),
- ]),
- ('core 3', [
- ('index', 5),
- ('mode', 'gen'),
- ]),
- ('core 4', [
- ('index', 7),
- ('mode', 'gen'),
- ]),
- ]
- test_type = helper.get_test_type()
- self.assertEqual(test_type, helper.PROX_CORE_MPLS_TEST)
+ def test_terminate(self):
+ setup_helper = mock.MagicMock()
+ helper = ProxResourceHelper(setup_helper)
+ with self.assertRaises(NotImplementedError):
+ helper.terminate()
- def test_get_cores(self):
+ def test_up_post(self):
setup_helper = mock.MagicMock()
- setup_helper.prox_config_data = [
- ('section1', []),
- ('section2', [
- ('a', 'b'),
- ('c', 'd'),
- ]),
- ('core 1', []),
- ('core 2', [
- ('index', 8),
- ('mode', ''),
- ]),
- ('core 3', [
- ('index', 5),
- ('mode', 'gen'),
- ]),
- ('core 4', [
- ('index', 7),
- ('mode', 'gen'),
- ]),
- ]
+ helper = ProxResourceHelper(setup_helper)
+ helper.client = expected = mock.MagicMock()
+ result = helper.up_post()
+ self.assertEqual(result, expected)
+ def test_execute(self):
+ setup_helper = mock.MagicMock()
helper = ProxResourceHelper(setup_helper)
- helper._cpu_topology = {
- 0: {
- 1: {
- 5: (5, 1, 0)
- },
- 2: {
- 6: (6, 2, 0)
- },
- 3: {
- 7: (7, 3, 0)
- },
- 4: {
- 8: (8, 3, 0)
- },
- }
+ helper.client = mock.MagicMock()
+
+ expected = helper.client.my_command()
+ result = helper.execute('my_command')
+ self.assertEqual(result, expected)
+
+ helper.client = object()
+
+ result = helper.execute('my_command')
+ self.assertIsNone(result)
+
+
+class TestProxDataHelper(unittest.TestCase):
+
+ def test_totals_and_pps(self):
+ pkt_size = 180
+ vnfd_helper = mock.MagicMock()
+ vnfd_helper.port_pairs.all_ports = list(range(4))
+
+ sut = mock.MagicMock()
+ sut.port_stats.return_value = list(range(10))
+
+ data_helper = ProxDataHelper(vnfd_helper, sut, pkt_size, 25, None)
+
+ self.assertEqual(data_helper.rx_total, 6)
+ self.assertEqual(data_helper.tx_total, 7)
+ self.assertEqual(data_helper.pps, 6.25e6)
+
+ def test_samples(self):
+ vnfd_helper = mock.MagicMock()
+ vnfd_helper.port_pairs.all_ports = list(range(4))
+ vnfd_helper.ports_iter.return_value = [('xe1', 3), ('xe2', 7)]
+
+ sut = mock.MagicMock()
+ sut.port_stats.return_value = list(range(10))
+
+ data_helper = ProxDataHelper(vnfd_helper, sut, None, None, None)
+
+ expected = {
+ 'xe1': {
+ 'in_packets': 6,
+ 'out_packets': 7,
+ },
+ 'xe2': {
+ 'in_packets': 6,
+ 'out_packets': 7,
+ },
}
+ result = data_helper.samples
+ self.assertDictEqual(result, expected)
- expected = [7, 8]
- result = helper.get_cores(helper.PROX_CORE_GEN_MODE)
+ def test___enter__(self):
+ vnfd_helper = mock.MagicMock()
+ vnfd_helper.port_pairs.all_ports = list(range(4))
+ vnfd_helper.ports_iter.return_value = [('xe1', 3), ('xe2', 7)]
+
+ sut = mock.MagicMock()
+ sut.port_stats.return_value = list(range(10))
+
+ data_helper = ProxDataHelper(vnfd_helper, sut, None, None, 5.4)
+ data_helper._totals_and_pps = 12, 32, 4.5
+ data_helper.tsc_hz = 9.8
+ data_helper.measured_stats = {'delta': TotStatsTuple(6.1, 6.2, 6.3, 6.4)}
+ data_helper.latency = 7
+
+ self.assertIsNone(data_helper.result_tuple)
+
+ expected = ProxTestDataTuple(5.4, 9.8, 6.1, 6.2, 6.3, 7, 12, 32, 4.5)
+ with data_helper:
+ pass
+
+ result = data_helper.result_tuple
self.assertEqual(result, expected)
- def test_get_cores_mpls(self):
- setup_helper = mock.MagicMock()
- setup_helper.prox_config_data = [
+ data_helper.make_tuple()
+ self.assertIs(data_helper.result_tuple, result)
+
+ def test___enter___negative(self):
+ vnfd_helper = mock.MagicMock()
+
+ data_helper = ProxDataHelper(vnfd_helper, None, None, None, None)
+
+ vnfd_helper.port_pairs.all_ports = []
+ with self.assertRaises(AssertionError):
+ with data_helper:
+ pass
+
+ vnfd_helper.port_pairs.all_ports = [0, 1, 2]
+ with self.assertRaises(AssertionError):
+ with data_helper:
+ pass
+
+ def test_measure_tot_stats(self):
+ vnfd_helper = mock.MagicMock()
+ vnfd_helper.port_pairs.all_ports = list(range(4))
+
+ start = (3, 4, 1, 2)
+ end = (9, 7, 6, 8)
+
+ sut = ProxSocketHelper(mock.MagicMock())
+ sut.get_all_tot_stats = mock.MagicMock(side_effect=[start, end])
+
+ data_helper = ProxDataHelper(vnfd_helper, sut, None, None, 5.4)
+
+ self.assertIsNone(data_helper.measured_stats)
+
+ expected = {
+ 'start_tot': start,
+ 'end_tot': end,
+ 'delta': TotStatsTuple(6, 3, 5, 6),
+ }
+ with data_helper.measure_tot_stats():
+ pass
+
+ self.assertEqual(data_helper.measured_stats, expected)
+
+ def test_capture_tsc_hz(self):
+ vnfd_helper = mock.MagicMock()
+ vnfd_helper.port_pairs.all_ports = list(range(4))
+
+ sut = mock.MagicMock()
+ sut.hz.return_value = '54.6'
+
+ data_helper = ProxDataHelper(vnfd_helper, sut, None, None, None)
+
+ self.assertIsNone(data_helper.tsc_hz)
+
+ expected = 54.6
+ data_helper.capture_tsc_hz()
+ self.assertEqual(data_helper.tsc_hz, expected)
+
+
+class TestProxProfileHelper(unittest.TestCase):
+
+ @mock.patch('yardstick.network_services.vnf_generic.vnf.prox_helpers.utils')
+ def test_get_cls(self, mock_utils):
+ mock_type1 = mock.MagicMock()
+ mock_type1.__prox_profile_type__ = 'another_type'
+ mock_type2 = mock.MagicMock()
+ mock_type2.__prox_profile_type__ = 'my_type'
+ mock_utils.itersubclasses.return_value = [mock_type1, mock_type2]
+
+ self.assertEqual(ProxProfileHelper.get_cls('my_type'), mock_type2)
+
+ @mock.patch('yardstick.network_services.vnf_generic.vnf.prox_helpers.utils')
+ def test_get_cls_default(self, mock_utils):
+ mock_utils.itersubclasses.return_value = []
+ ProxProfileHelper.get_cls('my_type')
+
+ @mock.patch('yardstick.network_services.vnf_generic.vnf.prox_helpers.SocketTopology')
+ def test_cpu_topology(self, mock_socket_topology):
+ mock_socket_topology.parse_cpuinfo.return_value = 432
+
+ resource_helper = mock.MagicMock()
+ resource_helper.setup_helper.ssh_helper.execute.return_value = 0, 'output', ''
+
+ helper = ProxProfileHelper(resource_helper)
+ self.assertIsNone(helper._cpu_topology)
+ result = helper.cpu_topology
+ self.assertEqual(result, 432)
+ self.assertIs(result, helper._cpu_topology)
+ self.assertIs(result, helper.cpu_topology)
+
+ def test_test_cores(self):
+ resource_helper = mock.MagicMock()
+ resource_helper.setup_helper.prox_config_data = []
+
+ helper = ProxProfileHelper(resource_helper)
+ helper._cpu_topology = []
+
+ expected = []
+ result = helper.test_cores
+ self.assertEqual(result, expected)
+
+ resource_helper.setup_helper.prox_config_data = [
('section1', []),
('section2', [
('a', 'b'),
('c', 'd'),
]),
- ('core 1', []),
- ('core 2', [
+ ('core 1s3', []),
+ ('core 2s5', [
('index', 8),
('mode', ''),
]),
- ('core 3', [
+ ('core 3s1', [
('index', 5),
('mode', 'gen'),
- ('name', 'tagged'),
]),
- ('core 4', [
+ ('core 4s9h', [
('index', 7),
('mode', 'gen'),
- ('name', 'udp'),
]),
]
- helper = ProxResourceHelper(setup_helper)
+ helper = ProxProfileHelper(resource_helper)
helper._cpu_topology = {
- 0: {
- 1: {
- 5: (5, 1, 0)
- },
- 2: {
- 6: (6, 2, 0)
- },
+ 1: {
3: {
- 7: (7, 3, 0)
+ 'key1': (23, 32),
+ 'key2': (12, 21),
+ 'key3': (44, 33),
},
+ },
+ 9: {
4: {
- 8: (8, 3, 0)
+ 'key1': (44, 32),
+ 'key2': (23, 21),
+ 'key3': (12, 33),
},
- }
+ },
}
- expected_tagged = [7]
- expected_plain = [8]
- result_tagged, result_plain = helper.get_cores_mpls(helper.PROX_CORE_GEN_MODE)
- self.assertEqual(result_tagged, expected_tagged)
- self.assertEqual(result_plain, expected_plain)
+ self.assertIsNone(helper._test_cores)
+ expected = [12, 23]
+ result = helper.test_cores
+ self.assertEqual(result, expected)
+ self.assertIs(result, helper._test_cores)
+ self.assertIs(result, helper.test_cores)
def test_latency_cores(self):
- setup_helper = mock.MagicMock()
- setup_helper.prox_config_data= []
+ resource_helper = mock.MagicMock()
+ resource_helper.setup_helper.prox_config_data = []
- helper = ProxResourceHelper(setup_helper)
+ helper = ProxProfileHelper(resource_helper)
helper._cpu_topology = []
expected = []
result = helper.latency_cores
self.assertEqual(result, expected)
- setup_helper.prox_config_data = [
+ resource_helper.setup_helper.prox_config_data = [
('section1', []),
('section2', [
('a', 'b'),
]),
]
- helper = ProxResourceHelper(setup_helper)
+ helper = ProxProfileHelper(resource_helper)
helper._cpu_topology = {
1: {
3: {
self.assertIs(result, helper._latency_cores)
self.assertIs(result, helper.latency_cores)
- def test_run_traffic(self):
- setup_helper = mock.MagicMock()
- helper = ProxResourceHelper(setup_helper)
- traffic_proifle = mock.MagicMock(**{"done": True})
- helper.run_traffic(traffic_proifle)
- self.assertEqual(helper._terminated.value, 1)
+ def test_all_rx_cores(self):
+ helper = ProxBngProfileHelper(mock.MagicMock())
+ helper._latency_cores = expected = [3, 4, 6]
+ helper._test_cores = [5, 2, 1]
- def test__run_traffic_once(self):
- setup_helper = mock.MagicMock()
- helper = ProxResourceHelper(setup_helper)
- traffic_proifle = mock.MagicMock(**{"done": True})
- helper._run_traffic_once(traffic_proifle)
- self.assertEqual(helper._terminated.value, 1)
+ result = helper.all_rx_cores
+ self.assertEqual(result, expected)
- def test_start_collect(self):
- setup_helper = mock.MagicMock()
- helper = ProxResourceHelper(setup_helper)
- helper.resource = resource = mock.MagicMock()
- self.assertIsNone(helper.start_collect())
- resource.start.assert_called_once()
+ def test_get_cores(self):
+ resource_helper = mock.MagicMock()
+ resource_helper.setup_helper.prox_config_data = [
+ ('section1', []),
+ ('section2', [
+ ('a', 'b'),
+ ('c', 'd'),
+ ]),
+ ('core 1', []),
+ ('core 2', [
+ ('index', 8),
+ ('mode', ''),
+ ]),
+ ('core 3', [
+ ('index', 5),
+ ('mode', 'gen'),
+ ]),
+ ('core 4', [
+ ('index', 7),
+ ('mode', 'gen'),
+ ]),
+ ]
- def test_terminate(self):
- setup_helper = mock.MagicMock()
- helper = ProxResourceHelper(setup_helper)
- with self.assertRaises(NotImplementedError):
- helper.terminate()
+ helper = ProxProfileHelper(resource_helper)
+ helper._cpu_topology = {
+ 0: {
+ 1: {
+ 5: (5, 1, 0)
+ },
+ 2: {
+ 6: (6, 2, 0)
+ },
+ 3: {
+ 7: (7, 3, 0)
+ },
+ 4: {
+ 8: (8, 3, 0)
+ },
+ }
+ }
- def test_up_post(self):
- setup_helper = mock.MagicMock()
- helper = ProxResourceHelper(setup_helper)
- helper.client = expected = mock.MagicMock()
- result = helper.up_post()
+ expected = [7, 8]
+ result = helper.get_cores(helper.PROX_CORE_GEN_MODE)
self.assertEqual(result, expected)
- def test_execute(self):
- setup_helper = mock.MagicMock()
- helper = ProxResourceHelper(setup_helper)
- helper.client = mock.MagicMock()
+ def test_get_latency(self):
+ resource_helper = mock.MagicMock()
+ resource_helper.setup_helper.vnfd_helper.interfaces = []
- expected = helper.client.my_command()
- result = helper.execute('my_command')
+ helper = ProxProfileHelper(resource_helper)
+ helper._latency_cores = []
+
+ expected = []
+ result = helper.get_latency()
self.assertEqual(result, expected)
- helper.client = object()
+ helper._latency_cores = [1, 2]
+ helper.client = mock.MagicMock()
- result = helper.execute('my_command')
- self.assertIsNone(result)
+ expected = helper.sut.lat_stats()
+ result = helper.get_latency()
+ self.assertIs(result, expected)
@mock.patch('yardstick.network_services.vnf_generic.vnf.prox_helpers.time')
- def test_traffic_context(self, mock_time):
+ def test_traffic_context(self, *args):
setup_helper = mock.MagicMock()
setup_helper.vnfd_helper.interfaces = []
- helper = ProxResourceHelper(setup_helper)
+ helper = ProxProfileHelper(setup_helper)
helper._cpu_topology = {
0: {
1: {
setup_helper.prox_config_data = [
('global', [
- ('name', helper.PROX_CORE_MPLS_TEST)
+ ('not_name', 'other data'),
+ ('name_not', 'more data'),
+ ('name', helper.__prox_profile_type__),
]),
('section1', []),
('section2', [
]),
]
-
client = mock.MagicMock()
client.hz.return_value = 2
client.port_stats.return_value = tuple(range(12))
pass
@mock.patch('yardstick.network_services.vnf_generic.vnf.prox_helpers.time')
- def test_run_test(self, mock_time):
- @contextmanager
- def measure(*args, **kwargs):
- yield stats
-
- bad_vnfd = copy.deepcopy(self.VNFD0)
- bad_vnfd['vdu'][0]['external-interface'].append({
- 'virtual-interface': {
- 'dst_mac': '00:00:00:00:00:05',
- 'vpci': '0000:06:00.0',
- 'local_ip': '152.16.100.20',
- 'type': 'PCI-PASSTHROUGH',
- 'vld_id': 'uplink_1',
- 'netmask': '255.255.255.0',
- 'dpdk_port_num': 0,
- 'bandwidth': '10 Gbps',
- 'driver': "i40e",
- 'dst_ip': '152.16.100.20',
- 'local_iface_name': 'xe2',
- 'local_mac': '00:00:00:00:00:07',
- 'ifname': 'xe2',
- },
- 'vnfd-connection-point-ref': 'xe2',
- 'name': 'xe2',
- })
+ def test_run_test(self, _):
+ resource_helper = mock.MagicMock()
+ resource_helper.step_delta = 0.4
+ resource_helper.vnfd_helper.port_pairs.all_ports = list(range(2))
+ resource_helper.sut.port_stats.return_value = list(range(10))
+
+ helper = ProxProfileHelper(resource_helper)
+
+ helper.run_test(120, 5, 6.5)
- bad_vnfd_helper = VnfdHelper(bad_vnfd)
+
+class TestProxMplsProfileHelper(unittest.TestCase):
+
+ def test_mpls_cores(self):
+ resource_helper = mock.MagicMock()
+ resource_helper.setup_helper.prox_config_data = [
+ ('section1', []),
+ ('section2', [
+ ('a', 'b'),
+ ('c', 'd'),
+ ]),
+ ('core 1', []),
+ ('core 2', [
+ ('index', 8),
+ ('mode', ''),
+ ]),
+ ('core 3', [
+ ('index', 5),
+ ('mode', 'gen'),
+ ('name', 'tagged'),
+ ]),
+ ('core 4', [
+ ('index', 7),
+ ('mode', 'gen'),
+ ('name', 'udp'),
+ ]),
+ ]
+
+ helper = ProxMplsProfileHelper(resource_helper)
+ helper._cpu_topology = {
+ 0: {
+ 1: {
+ 5: (5, 1, 0)
+ },
+ 2: {
+ 6: (6, 2, 0)
+ },
+ 3: {
+ 7: (7, 3, 0)
+ },
+ 4: {
+ 8: (8, 3, 0)
+ },
+ }
+ }
+
+ expected_tagged = [7]
+ expected_plain = [8]
+ self.assertIsNone(helper._cores_tuple)
+ self.assertEqual(helper.tagged_cores, expected_tagged)
+ self.assertEqual(helper.plain_cores, expected_plain)
+ self.assertEqual(helper._cores_tuple, (expected_tagged, expected_plain))
+
+ def test_traffic_context(self):
setup_helper = mock.MagicMock()
- setup_helper.vnfd_helper = bad_vnfd_helper
+ helper = ProxMplsProfileHelper(setup_helper)
+
+ with helper.traffic_context(120, 5.4):
+ pass
+
+
+class TestProxBngProfileHelper(unittest.TestCase):
- stats = {
- 'delta': TotStatsTuple(6, 7, 8, 9),
+ def test_bng_cores(self):
+ resource_helper = mock.MagicMock()
+ resource_helper.setup_helper.prox_config_data = [
+ ('section1', []),
+ ('section2', [
+ ('a', 'b'),
+ ('c', 'd'),
+ ]),
+ ('core 1', []),
+ ('core 2', [
+ ('index', 8),
+ ('mode', ''),
+ ]),
+ ('core 3', [
+ ('index', 5),
+ ('mode', 'gen'),
+ ('name', 'cpe'),
+ ]),
+ ('core 4', [
+ ('index', 7),
+ ('mode', 'gen'),
+ ('name', 'inet'),
+ ]),
+ ('core 6', [
+ ('index', 3),
+ ('mode', 'gen'),
+ ('name', 'arp_task'),
+ ]),
+ ('core 9', [
+ ('index', 2),
+ ('mode', 'gen'),
+ ('name', 'arp'),
+ ]),
+ ]
+
+ helper = ProxBngProfileHelper(resource_helper)
+ helper._cpu_topology = {
+ 0: {
+ 1: {
+ 5: (5, 1, 0)
+ },
+ 2: {
+ 6: (6, 2, 0)
+ },
+ 3: {
+ 7: (7, 3, 0)
+ },
+ 4: {
+ 8: (8, 3, 0)
+ },
+ 6: {
+ 1: (4, 8, 0)
+ },
+ 9: {
+ 2: (3, 7, 0)
+ },
+ }
}
- client = mock.MagicMock()
- client.hz.return_value = 2
- client.measure_tot_stats = measure
- client.port_stats.return_value = tuple(range(12))
+ expected_cpe = [7]
+ expected_inet = [8]
+ expected_arp = [4, 3]
+ expected_arp_task = [0, 4]
+ expected_combined = (expected_cpe, expected_inet, expected_arp, expected_arp_task)
- helper = ProxResourceHelper(setup_helper)
- helper.client = client
- helper.get_latency = mock.MagicMock(return_value=[3.3, 3.6, 3.8])
+ self.assertIsNone(helper._cores_tuple)
+ self.assertEqual(helper.cpe_cores, expected_cpe)
+ self.assertEqual(helper.inet_cores, expected_inet)
+ self.assertEqual(helper.arp_cores, expected_arp)
+ self.assertEqual(helper.arp_task_cores, expected_arp_task)
+ self.assertEqual(helper._cores_tuple, expected_combined)
- with self.assertRaises(AssertionError):
- helper.run_test(980, 15, 45)
+ @mock.patch('yardstick.network_services.vnf_generic.vnf.prox_helpers.time')
+ def test_run_test(self, _):
+ resource_helper = mock.MagicMock()
+ resource_helper.step_delta = 0.4
+ resource_helper.vnfd_helper.port_pairs.all_ports = list(range(2))
+ resource_helper.sut.port_stats.return_value = list(range(10))
- vnfd_helper = VnfdHelper(self.VNFD0)
- setup_helper.vnfd_helper = vnfd_helper
- helper = ProxResourceHelper(setup_helper)
- helper.client = client
- helper.get_latency = mock.MagicMock(return_value=[3.3, 3.6, 3.8])
- helper._test_cores = [3, 4]
+ helper = ProxBngProfileHelper(resource_helper)
+
+ helper.run_test(120, 5, 6.5)
+ helper.run_test(-1000, 5, 6.5) # negative pkt_size is the only way to make ratio > 1
+
+
+class TestProxVpeProfileHelper(unittest.TestCase):
+
+ def test_vpe_cores(self):
+ resource_helper = mock.MagicMock()
+ resource_helper.setup_helper.prox_config_data = [
+ ('section1', []),
+ ('section2', [
+ ('a', 'b'),
+ ('c', 'd'),
+ ]),
+ ('core 1', []),
+ ('core 2', [
+ ('index', 8),
+ ('mode', ''),
+ ]),
+ ('core 3', [
+ ('index', 5),
+ ('mode', 'gen'),
+ ('name', 'cpe'),
+ ]),
+ ('core 4', [
+ ('index', 7),
+ ('mode', 'gen'),
+ ('name', 'inet'),
+ ]),
+ ]
- expected_test_data = ProxTestDataTuple(0.0, 2.0, 6, 7, 8, [3.3, 3.6, 3.8], 6, 7, 6.5e6)
- expected_port_samples = {
- 'xe0': {'in_packets': 6, 'out_packets': 7},
- 'xe1': {'in_packets': 6, 'out_packets': 7},
+ helper = ProxVpeProfileHelper(resource_helper)
+ helper._cpu_topology = {
+ 0: {
+ 1: {
+ 5: (5, 1, 0)
+ },
+ 2: {
+ 6: (6, 2, 0)
+ },
+ 3: {
+ 7: (7, 3, 0)
+ },
+ 4: {
+ 8: (8, 3, 0)
+ },
+ }
}
- test_data, port_samples = helper.run_test(230, 60, 65)
- self.assertTupleEqual(test_data, expected_test_data)
- self.assertDictEqual(port_samples, expected_port_samples)
- def test_get_latency(self):
- setup_helper = mock.MagicMock()
- setup_helper.vnfd_helper.interfaces = []
+ expected_cpe = [7]
+ expected_inet = [8]
+ expected_combined = (expected_cpe, expected_inet)
- helper = ProxResourceHelper(setup_helper)
- helper._latency_cores = []
+ self.assertIsNone(helper._cores_tuple)
+ self.assertEqual(helper.cpe_cores, expected_cpe)
+ self.assertEqual(helper.inet_cores, expected_inet)
+ self.assertEqual(helper._cores_tuple, expected_combined)
- expected = []
- result = helper.get_latency()
- self.assertEqual(result, expected)
+ def test_vpe_ports(self):
+ resource_helper = mock.MagicMock()
+ resource_helper.setup_helper.prox_config_data = [
+ ('section1', []),
+ ('section2', [
+ ('a', 'b'),
+ ('c', 'd'),
+ ]),
+ ('port 3', [
+ ('index', '5'),
+ ('name', 'cpe'),
+ ('mac', 'hardware'),
+ ]),
+ ('port 4', [
+ ('index', '7'),
+ ('name', 'inet'),
+ ('mac', 'hardware'),
+ ]),
+ ]
- helper._latency_cores = [1, 2]
- helper.client = mock.MagicMock()
+ helper = ProxVpeProfileHelper(resource_helper)
+ helper._port_list = {
+ 0: {
+ 1: {
+ 5: 'cpe'
+ },
+ 2: {
+ 6: 'inet'
+ },
+ 3: {
+ 7: 'cpe'
+ },
+ 4: {
+ 8: 'inet'
+ },
+ }
+ }
- expected = helper.sut.lat_stats()
- result = helper.get_latency()
- self.assertIs(result, expected)
+ expected_cpe = [3]
+ expected_inet = [4]
+ expected_combined = (expected_cpe, expected_inet)
+
+ self.assertIsNone(helper._ports_tuple)
+ self.assertEqual(helper.cpe_ports, expected_cpe)
+ self.assertEqual(helper.inet_ports, expected_inet)
+ self.assertEqual(helper._ports_tuple, expected_combined)
@mock.patch('yardstick.network_services.vnf_generic.vnf.prox_helpers.time')
- @mock.patch('yardstick.network_services.vnf_generic.vnf.prox_helpers.ProxSocketHelper')
- def test__connect(self, mock_socket_helper_type, mock_time):
- client = mock_socket_helper_type()
- client.connect.side_effect = chain(repeat(socket.error, 5), [None])
+ def test_run_test(self, _):
+ resource_helper = mock.MagicMock()
+ resource_helper.step_delta = 0.4
+ resource_helper.vnfd_helper.port_pairs.all_ports = list(range(2))
+ resource_helper.sut.port_stats.return_value = list(range(10))
- setup_helper = mock.MagicMock()
- setup_helper.vnfd_helper.interfaces = []
+ helper = ProxVpeProfileHelper(resource_helper)
- helper = ProxResourceHelper(setup_helper)
+ helper.run_test(120, 5, 6.5)
+ helper.run_test(-1000, 5, 6.5) # negative pkt_size is the only way to make ratio > 1
- result = helper._connect()
- self.assertIs(result, client)
- client.connect.side_effect = chain(repeat(socket.error, 65), [None])
+class TestProxlwAFTRProfileHelper(unittest.TestCase):
- with self.assertRaises(Exception):
- helper._connect()
+ def test_lwaftr_cores(self):
+ resource_helper = mock.MagicMock()
+ resource_helper.setup_helper.prox_config_data = [
+ ('section1', []),
+ ('section2', [
+ ('a', 'b'),
+ ('c', 'd'),
+ ]),
+ ('core 1', []),
+ ('core 2', [
+ ('index', 8),
+ ('mode', ''),
+ ]),
+ ('core 3', [
+ ('index', 5),
+ ('mode', 'gen'),
+ ('name', 'tun'),
+ ]),
+ ('core 4', [
+ ('index', 7),
+ ('mode', 'gen'),
+ ('name', 'inet'),
+ ]),
+ ]
+
+ helper = ProxlwAFTRProfileHelper(resource_helper)
+ helper._cpu_topology = {
+ 0: {
+ 1: {
+ 5: (5, 1, 0)
+ },
+ 2: {
+ 6: (6, 2, 0)
+ },
+ 3: {
+ 7: (7, 3, 0)
+ },
+ 4: {
+ 8: (8, 3, 0)
+ },
+ }
+ }
+
+ expected_tun = [7]
+ expected_inet = [8]
+ expected_combined = (expected_tun, expected_inet)
+
+ self.assertIsNone(helper._cores_tuple)
+ self.assertEqual(helper.tun_cores, expected_tun)
+ self.assertEqual(helper.inet_cores, expected_inet)
+ self.assertEqual(helper._cores_tuple, expected_combined)
+
+ def test_tun_ports(self):
+ resource_helper = mock.MagicMock()
+ resource_helper.setup_helper.prox_config_data = [
+ ('section1', []),
+ ('section2', [
+ ('a', 'b'),
+ ('c', 'd'),
+ ]),
+ ('port 3', [
+ ('index', '5'),
+ ('name', 'lwB4'),
+ ('mac', 'hardware'),
+ ]),
+ ('port 4', [
+ ('index', '7'),
+ ('name', 'inet'),
+ ('mac', 'hardware'),
+ ]),
+ ]
+
+ helper = ProxlwAFTRProfileHelper(resource_helper)
+ helper._port_list = {
+ 0: {
+ 1: {
+ 5: 'lwB4'
+ },
+ 2: {
+ 6: 'inet'
+ },
+ 3: {
+ 7: 'lwB4'
+ },
+ 4: {
+ 8: 'inet'
+ },
+ }
+ }
+
+ expected_tun = [3]
+ expected_inet = [4]
+ expected_combined = (expected_tun, expected_inet)
+
+ self.assertIsNone(helper._ports_tuple)
+ self.assertEqual(helper.tun_ports, expected_tun)
+ self.assertEqual(helper.inet_ports, expected_inet)
+ self.assertEqual(helper._ports_tuple, expected_combined)
+
+ @mock.patch('yardstick.network_services.vnf_generic.vnf.prox_helpers.time')
+ def test_run_test(self, _):
+ resource_helper = mock.MagicMock()
+ resource_helper.step_delta = 0.4
+ resource_helper.vnfd_helper.port_pairs.all_ports = list(range(2))
+ resource_helper.sut.port_stats.return_value = list(range(10))
+
+ helper = ProxlwAFTRProfileHelper(resource_helper)
+
+ helper.run_test(120, 5, 6.5)
+ helper.run_test(-1000, 5, 6.5) # negative pkt_size is the only way to make ratio > 1
Code Review / yardstick.git / blobdiff