# limitations under the License.
#
-from __future__ import absolute_import
-
+from itertools import repeat, chain
import os
import socket
-import unittest
-from itertools import repeat, chain
+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 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()
'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 = []
@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
helper = ProxResourceHelper(mock.MagicMock())
helper.resource = resource = mock.MagicMock()
- resource.check_if_sa_running.return_value = 0, '1234'
+ resource.check_if_system_agent_running.return_value = 0, '1234'
resource.amqp_collect_nfvi_kpi.return_value = 543
- resource.check_if_sa_running.return_value = (0, None)
+ resource.check_if_system_agent_running.return_value = (0, None)
expected = {'core': 543}
result = helper.collect_collectd_kpi()
helper._result = {'z': 123}
helper.resource = resource = mock.MagicMock()
- resource.check_if_sa_running.return_value = 0, '1234'
+ resource.check_if_system_agent_running.return_value = 0, '1234'
resource.amqp_collect_nfvi_kpi.return_value = 543
- resource.check_if_sa_running.return_value = (0, None)
+ resource.check_if_system_agent_running.return_value = (0, None)
queue.empty.return_value = False
queue.get.return_value = {'a': 789}
@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):
+ def test__connect(self, mock_socket_helper_type, *args):
client = mock_socket_helper_type()
client.connect.side_effect = chain(repeat(socket.error, 5), [None])
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.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'),
+ ]),
+ ]
+
+ 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)
+ },
+ }
+ }
+
+ expected_cpe = [7]
+ expected_inet = [8]
+ expected_combined = (expected_cpe, expected_inet)
+
+ 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)
+
+ 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 = ProxVpeProfileHelper(resource_helper)
+ helper._port_list = {
+ 0: {
+ 1: {
+ 5: 'cpe'
+ },
+ 2: {
+ 6: 'inet'
+ },
+ 3: {
+ 7: 'cpe'
+ },
+ 4: {
+ 8: 'inet'
+ },
+ }
+ }
+
+ 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')
+ 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 = ProxVpeProfileHelper(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 TestProxlwAFTRProfileHelper(unittest.TestCase):
+
+ 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