# See the License for the specific language governing permissions and
# limitations under the License.
+from __future__ import absolute_import
import unittest
import mock
from experimental_framework.constants import conf_file_sections as conf_file
mock_init_input_validation.return_value = None
mock_get_core_nics.return_value = "{corenics}"
mock_path.isfile.return_value = True
- expected = 'pktgen_dir/program -c coremask -n memchannel ' \
+ expected = 'sudo pktgen_dir/program -c coremask -n memchannel ' \
'--proc-type auto --file-prefix pg -- -T -P -m ' \
'"{corenics}" -f base_dir/experimental_framework/' \
'packet_generators/dpdk_pktgen/lua_file ' \
mock_init_input_validation.return_value = None
mock_get_core_nics.return_value = "{corenics}"
mock_path.isfile.return_value = True
- expected = 'pktgen_dir/program -c coremask -n memchannel ' \
+ expected = 'sudo pktgen_dir/program -c coremask -n memchannel ' \
'--proc-type auto --file-prefix pg -- -T -P -m ' \
'"{corenics}" -f base_dir/experimental_framework/' \
'packet_generators/dpdk_pktgen/lua_file ' \
mock_dir_name,
mock_os_path,
mock_os_system):
- '''
+ """
Calls the packet generator and starts to send traffic
Blocking call
- '''
+ """
mock_get_core_nics.return_value = "{corenics}"
mock_os_path.realpath.return_value = 'pktgen_dir_test'
mock_os_path.dirname.return_value = 'current_directory'
@staticmethod
def mock_run_command(command=None):
- if command == 'ifconfig interface_1 down':
+ if command == 'sudo ifconfig interface_1 down':
MockRunCommand.ret_val[0] = True
- if command == 'dpdk_directory/tools/dpdk_nic_bind.py --unbind 1:00.0':
+ if command == 'sudo dpdk_directory/tools/dpdk_nic_bind.py ' \
+ '--unbind 1:00.0':
MockRunCommand.ret_val[1] = True
- if command == 'dpdk_directory/tools/dpdk_nic_bind.py ' \
+ if command == 'sudo dpdk_directory/tools/dpdk_nic_bind.py ' \
'--bind=igb_uio 1:00.0':
MockRunCommand.ret_val[2] = True
- if command == 'ifconfig interface_2 down':
+ if command == 'sudo ifconfig interface_2 down':
MockRunCommand.ret_val[3] = True
- if command == 'dpdk_directory/tools/dpdk_nic_bind.py ' \
+ if command == 'sudo dpdk_directory/tools/dpdk_nic_bind.py ' \
'--unbind 1:00.1':
MockRunCommand.ret_val[4] = True
- if command == 'dpdk_directory/tools/dpdk_nic_bind.py ' \
+ if command == 'sudo dpdk_directory/tools/dpdk_nic_bind.py ' \
'--bind=igb_uio 1:00.1':
MockRunCommand.ret_val[5] = True
else:
@staticmethod
def mock_run_command_finalization(command=None):
- if command == 'dpdk_directory/tools/dpdk_nic_bind.py --unbind 1:00.0':
+ if command == 'sudo dpdk_directory/tools/dpdk_nic_bind.py ' \
+ '--unbind 1:00.0':
MockRunCommand.ret_val_finalization[0] = True
- if command == 'dpdk_directory/tools/dpdk_nic_bind.py ' \
+ if command == 'sudo dpdk_directory/tools/dpdk_nic_bind.py ' \
'--bind=ixgbe 1:00.0':
MockRunCommand.ret_val_finalization[1] = True
- if command == 'ifconfig interface_1 up':
+ if command == 'sudo ifconfig interface_1 up':
MockRunCommand.ret_val_finalization[2] = True
- if command == 'dpdk_directory/tools/dpdk_nic_bind.py --unbind 1:00.1':
+ if command == 'sudo dpdk_directory/tools/dpdk_nic_bind.py ' \
+ '--unbind 1:00.1':
MockRunCommand.ret_val_finalization[3] = True
- if command == 'dpdk_directory/tools/dpdk_nic_bind.py ' \
+ if command == 'sudo dpdk_directory/tools/dpdk_nic_bind.py ' \
'--bind=ixgbe 1:00.1':
MockRunCommand.ret_val_finalization[4] = True
- if command == 'ifconfig interface_2 up':
+ if command == 'sudo ifconfig interface_2 up':
MockRunCommand.ret_val_finalization[5] = True
else:
return MockRunCommand.ret_val_finalization
+@mock.patch('experimental_framework.packet_generators.dpdk_packet_generator.time')
class TestDpdkPacketGenOthers(unittest.TestCase):
def setUp(self):
@mock.patch('experimental_framework.packet_generators.'
'dpdk_packet_generator.DpdkPacketGenerator.'
'_cores_configuration')
- def test__get_core_nics_for_failure(self, mock_cores_configuration):
+ def test__get_core_nics_for_failure(self, mock_cores_configuration, mock_time):
mock_cores_configuration.return_value = None
self.assertRaises(ValueError, mut.DpdkPacketGenerator._get_core_nics,
'', '')
'dpdk_packet_generator.DpdkPacketGenerator.'
'_cores_configuration')
def test__get_core_nics_one_nic_for_success(self,
- mock_cores_configuration):
+ mock_cores_configuration, mock_time):
mock_cores_configuration.return_value = 'ret_val'
expected = 'ret_val'
output = mut.DpdkPacketGenerator._get_core_nics(1, 'coremask')
'dpdk_packet_generator.DpdkPacketGenerator.'
'_cores_configuration')
def test__get_core_nics_two_nics_for_success(self,
- mock_cores_configuration):
+ mock_cores_configuration, mock_time):
mock_cores_configuration.return_value = 'ret_val'
expected = 'ret_val'
output = mut.DpdkPacketGenerator._get_core_nics(2, 'coremask')
mock_cores_configuration.assert_called_once_with('coremask', 1, 2, 2)
@mock.patch('os.path.isfile')
- def test__init_input_validation_for_success(self, mock_is_file):
+ def test__init_input_validation_for_success(self, mock_is_file, mock_time):
mock_is_file.return_value = True
pcap_file_0 = 'pcap_file_0'
variables), None)
@mock.patch('os.path.isfile')
- def test__init_input_validation_for_failure(self, mock_is_file):
+ def test__init_input_validation_for_failure(self, mock_is_file, mock_time):
mock_is_file.return_value = True
pcap_file_0 = 'pcap_file_0'
lua_script, pcap_directory, lua_directory, variables)
@mock.patch('os.path.isfile')
- def test__init_input_validation_for_failure_2(self, mock_is_file):
+ def test__init_input_validation_for_failure_2(self, mock_is_file, mock_time):
mock_is_file.return_value = True
pcap_directory = None
lua_script, pcap_directory, lua_directory, variables)
@mock.patch('os.path.isfile')
- def test__init_input_validation_for_failure_3(self, mock_is_file):
+ def test__init_input_validation_for_failure_3(self, mock_is_file, mock_time):
mock_is_file.return_value = True
pcap_directory = 'directory'
lua_script, pcap_directory, lua_directory, variables)
@mock.patch('os.path.isfile')
- def test__init_input_validation_for_failure_4(self, mock_is_file):
+ def test__init_input_validation_for_failure_4(self, mock_is_file, mock_time):
mock_is_file.return_value = True
pcap_directory = 'directory'
lua_script, pcap_directory, lua_directory, variables)
@mock.patch('os.path.isfile')
- def test__init_input_validation_for_failure_5(self, mock_is_file):
+ def test__init_input_validation_for_failure_5(self, mock_is_file, mock_time):
mock_is_file.return_value = True
pcap_directory = 'directory'
lua_script, pcap_directory, lua_directory, variables)
@mock.patch('os.path.isfile', side_effect=[False])
- def test__init_input_validation_for_failure_6(self, mock_is_file):
+ def test__init_input_validation_for_failure_6(self, mock_is_file, mock_time):
# mock_is_file.return_value = False
pcap_directory = 'directory'
lua_script, pcap_directory, lua_directory, variables)
@mock.patch('os.path.isfile', side_effect=[True, False])
- def test__init_input_validation_for_failure_7(self, mock_is_file):
+ def test__init_input_validation_for_failure_7(self, mock_is_file, mock_time):
pcap_directory = 'directory'
pcap_file_0 = 'pcap_file_0'
pcap_file_1 = 'pcap_file_1'
lua_script, pcap_directory, lua_directory, variables)
@mock.patch('os.path.isfile', side_effect=[True, True, False])
- def test__init_input_validation_for_failure_8(self, mock_is_file):
+ def test__init_input_validation_for_failure_8(self, mock_is_file, mock_time):
pcap_directory = 'directory'
pcap_file_0 = 'pcap_file_0'
pcap_file_1 = 'pcap_file_1'
lua_script, pcap_directory, lua_directory, variables)
@mock.patch('os.chdir')
- def test__chdir_for_success(self, mock_os_chdir):
+ def test__chdir_for_success(self, mock_os_chdir, mock_time):
mut.DpdkPacketGenerator._chdir('directory')
mock_os_chdir.assert_called_once_with('directory')
@mock.patch('experimental_framework.common.run_command',
side_effect=MockRunCommand.mock_run_command)
- def test__init_physical_nics_for_success(self, mock_run_command):
+ def test__init_physical_nics_for_success(self, mock_run_command, mock_time):
dpdk_interfaces = 1
dpdk_vars = dict()
@mock.patch('experimental_framework.common.run_command',
side_effect=MockRunCommand.mock_run_command)
- def test__init_physical_nics_for_success_2(self, mock_run_command):
+ def test__init_physical_nics_for_success_2(self, mock_run_command, mock_time):
dpdk_interfaces = 2
dpdk_vars = dict()
[True, True, True, True, True, True])
@mock.patch('experimental_framework.common.run_command')
- def test__init_physical_nics_for_failure(self, mock_run_command):
+ def test__init_physical_nics_for_failure(self, mock_run_command, mock_time):
dpdk_interfaces = 3
dpdk_vars = dict()
self.assertRaises(ValueError, self.mut._init_physical_nics,
@mock.patch('experimental_framework.common.run_command',
side_effect=MockRunCommand.mock_run_command_finalization)
- def test__finalize_physical_nics_for_success(self, mock_run_command):
+ def test__finalize_physical_nics_for_success(self, mock_run_command, mock_time):
dpdk_interfaces = 1
dpdk_vars = dict()
dpdk_vars[conf_file.CFSP_DPDK_DPDK_DIRECTORY] = 'dpdk_directory/'
@mock.patch('experimental_framework.common.run_command',
side_effect=MockRunCommand.mock_run_command_finalization)
- def test__finalize_physical_nics_for_success_2(self, mock_run_command):
+ def test__finalize_physical_nics_for_success_2(self, mock_run_command, mock_time):
dpdk_interfaces = 2
dpdk_vars = dict()
dpdk_vars[conf_file.CFSP_DPDK_DPDK_DIRECTORY] = 'dpdk_directory/'
self.assertEqual(MockRunCommand.mock_run_command_finalization(),
[True, True, True, True, True, True])
- def test__finalize_physical_nics_for_failure(self):
+ def test__finalize_physical_nics_for_failure(self, mock_time):
dpdk_interfaces = 0
dpdk_vars = dict()
self.assertRaises(ValueError, self.mut._finalize_physical_nics,
dpdk_interfaces, dpdk_vars)
- def test__cores_configuration_for_success(self):
+ def test__cores_configuration_for_success(self, mock_time):
coremask = '1f'
expected = '[2:1].0,[4:3].1'
output = mut.DpdkPacketGenerator._cores_configuration(coremask,
1, 2, 2)
self.assertEqual(expected, output)
- def test__cores_configuration_for_success_2(self):
+ def test__cores_configuration_for_success_2(self, mock_time):
coremask = '1f'
expected = '2.0,[4:3].1'
output = mut.DpdkPacketGenerator._cores_configuration(coremask,
1, 1, 2)
self.assertEqual(expected, output)
- def test__cores_configuration_for_success_3(self):
+ def test__cores_configuration_for_success_3(self, mock_time):
coremask = '1f'
expected = '[3:2].0,4.1'
output = mut.DpdkPacketGenerator._cores_configuration(coremask,
1, 2, 1)
self.assertEqual(expected, output)
- def test__cores_configuration_for_failure(self):
+ def test__cores_configuration_for_failure(self, mock_time):
coremask = '1'
self.assertRaises(ValueError,
mut.DpdkPacketGenerator._cores_configuration,
@mock.patch('experimental_framework.common.LOG')
@mock.patch('experimental_framework.common.run_command')
- def test__change_vlan_for_success(self, mock_run_command, mock_log):
+ def test__change_vlan_for_success(self, mock_run_command, mock_log, mock_time):
mut.DpdkPacketGenerator._change_vlan('/directory/', 'pcap_file', '10')
expected_param = '/directory/vlan_tag.sh /directory/pcap_file 10'
- mock_run_command.assert_called_once_with(expected_param)
+ mock_run_command.assert_called_with(expected_param)
Code Review / yardstick.git / blobdiff