diff options
Diffstat (limited to 'tests/unit/network_services/helpers')
6 files changed, 0 insertions, 2130 deletions
diff --git a/tests/unit/network_services/helpers/__init__.py b/tests/unit/network_services/helpers/__init__.py deleted file mode 100644 index e69de29bb..000000000 --- a/tests/unit/network_services/helpers/__init__.py +++ /dev/null diff --git a/tests/unit/network_services/helpers/acl_vnf_topology_ixia.yaml b/tests/unit/network_services/helpers/acl_vnf_topology_ixia.yaml deleted file mode 100644 index f60834fbd..000000000 --- a/tests/unit/network_services/helpers/acl_vnf_topology_ixia.yaml +++ /dev/null @@ -1,50 +0,0 @@ -# Copyright (c) 2016-2017 Intel Corporation -# -# Licensed under the Apache License, Version 2.0 (the "License"); -# you may not use this file except in compliance with the License. -# You may obtain a copy of the License at -# -# http://www.apache.org/licenses/LICENSE-2.0 -# -# Unless required by applicable law or agreed to in writing, software -# distributed under the License is distributed on an "AS IS" BASIS, -# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -# See the License for the specific language governing permissions and -# limitations under the License. - -nsd:nsd-catalog: - nsd: - - id: VACL - name: VACL - short-name: VACL - description: scenario with VACL,L3fwd and VNF - constituent-vnfd: - - member-vnf-index: '1' - vnfd-id-ref: tg__1 - VNF model: ../../vnf_descriptors/ixia_rfc2544_tpl.yaml - - member-vnf-index: '2' - vnfd-id-ref: vnf__1 - VNF model: ../../vnf_descriptors/acl_vnf.yaml - - vld: - - id: uplink_1 - name: tg__1 to vnf__1 link 1 - type: ELAN - vnfd-connection-point-ref: - - member-vnf-index-ref: '1' - vnfd-connection-point-ref: xe0 - vnfd-id-ref: tg__1 #TREX - - member-vnf-index-ref: '2' - vnfd-connection-point-ref: xe0 - vnfd-id-ref: vnf__1 #VNF - - - id: downlink_1 - name: vnf__1 to tg__1 link 2 - type: ELAN - vnfd-connection-point-ref: - - member-vnf-index-ref: '2' - vnfd-connection-point-ref: xe1 - vnfd-id-ref: vnf__1 #L3fwd - - member-vnf-index-ref: '1' - vnfd-connection-point-ref: xe1 - vnfd-id-ref: tg__1 #VACL VNF diff --git a/tests/unit/network_services/helpers/test_cpu.py b/tests/unit/network_services/helpers/test_cpu.py deleted file mode 100644 index 1f9d3f219..000000000 --- a/tests/unit/network_services/helpers/test_cpu.py +++ /dev/null @@ -1,117 +0,0 @@ -# Copyright (c) 2016-2017 Intel Corporation -# -# Licensed under the Apache License, Version 2.0 (the "License"); -# you may not use this file except in compliance with the License. -# You may obtain a copy of the License at -# -# http://www.apache.org/licenses/LICENSE-2.0 -# -# Unless required by applicable law or agreed to in writing, software -# distributed under the License is distributed on an "AS IS" BASIS, -# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -# See the License for the specific language governing permissions and -# limitations under the License. -# - -from __future__ import absolute_import -from __future__ import division -import unittest -import mock -import subprocess - -from yardstick.network_services.helpers.cpu import \ - CpuSysCores - - -class TestCpuSysCores(unittest.TestCase): - - def test___init__(self): - with mock.patch("yardstick.ssh.SSH") as ssh: - ssh_mock = mock.Mock(autospec=ssh.SSH) - ssh_mock.execute = \ - mock.Mock(return_value=(1, "", "")) - ssh_mock.put = \ - mock.Mock(return_value=(1, "", "")) - cpu_topo = CpuSysCores(ssh_mock) - self.assertIsNotNone(cpu_topo.connection) - - def test__get_core_details(self): - with mock.patch("yardstick.ssh.SSH") as ssh: - ssh_mock = mock.Mock(autospec=ssh.SSH) - ssh_mock.execute = \ - mock.Mock(return_value=(1, "", "")) - ssh_mock.put = \ - mock.Mock(return_value=(1, "", "")) - cpu_topo = CpuSysCores(ssh_mock) - subprocess.check_output = mock.Mock(return_value=0) - lines = ["cpu:1", "topo:2", ""] - self.assertEqual([{'topo': '2', 'cpu': '1'}], - cpu_topo._get_core_details(lines)) - - def test_get_core_socket(self): - with mock.patch("yardstick.ssh.SSH") as ssh: - ssh_mock = mock.Mock(autospec=ssh.SSH) - ssh_mock.execute = \ - mock.Mock(return_value=(1, "cpu:1\ntest:2\n \n", "")) - ssh_mock.put = \ - mock.Mock(return_value=(1, "", "")) - cpu_topo = CpuSysCores(ssh_mock) - subprocess.check_output = mock.Mock(return_value=0) - cpu_topo._get_core_details = \ - mock.Mock(side_effect=[[{'Core(s) per socket': '2', 'Thread(s) per core': '1'}], - [{'physical id': '2', 'processor': '1'}]]) - self.assertEqual({'thread_per_core': '1', '2': ['1'], - 'cores_per_socket': '2'}, - cpu_topo.get_core_socket()) - - def test_validate_cpu_cfg(self): - with mock.patch("yardstick.ssh.SSH") as ssh: - ssh_mock = mock.Mock(autospec=ssh.SSH) - ssh_mock.execute = \ - mock.Mock(return_value=(1, "cpu:1\ntest:2\n \n", "")) - ssh_mock.put = \ - mock.Mock(return_value=(1, "", "")) - cpu_topo = CpuSysCores(ssh_mock) - subprocess.check_output = mock.Mock(return_value=0) - cpu_topo._get_core_details = \ - mock.Mock(side_effect=[[{'Core(s) per socket': '2', 'Thread(s) per core': '1'}], - [{'physical id': '2', 'processor': '1'}]]) - cpu_topo.core_map = \ - {'thread_per_core': '1', '2':['1'], 'cores_per_socket': '2'} - self.assertEqual(-1, cpu_topo.validate_cpu_cfg()) - - def test_validate_cpu_cfg_2t(self): - with mock.patch("yardstick.ssh.SSH") as ssh: - ssh_mock = mock.Mock(autospec=ssh.SSH) - ssh_mock.execute = \ - mock.Mock(return_value=(1, "cpu:1\ntest:2\n \n", "")) - ssh_mock.put = \ - mock.Mock(return_value=(1, "", "")) - cpu_topo = CpuSysCores(ssh_mock) - subprocess.check_output = mock.Mock(return_value=0) - cpu_topo._get_core_details = \ - mock.Mock(side_effect=[[{'Core(s) per socket': '2', 'Thread(s) per core': '1'}], - [{'physical id': '2', 'processor': '1'}]]) - cpu_topo.core_map = \ - {'thread_per_core': 1, '2':['1'], 'cores_per_socket': '2'} - vnf_cfg = {'lb_config': 'SW', 'lb_count': 1, 'worker_config': - '1C/2T', 'worker_threads': 1} - self.assertEqual(-1, cpu_topo.validate_cpu_cfg(vnf_cfg)) - - def test_validate_cpu_cfg_fail(self): - with mock.patch("yardstick.ssh.SSH") as ssh: - ssh_mock = mock.Mock(autospec=ssh.SSH) - ssh_mock.execute = \ - mock.Mock(return_value=(1, "cpu:1\ntest:2\n \n", "")) - ssh_mock.put = \ - mock.Mock(return_value=(1, "", "")) - cpu_topo = CpuSysCores(ssh_mock) - subprocess.check_output = mock.Mock(return_value=0) - cpu_topo._get_core_details = \ - mock.Mock(side_effect=[[{'Core(s) per socket': '2', 'Thread(s) per core': '1'}], - [{'physical id': '2', 'processor': '1'}]]) - cpu_topo.core_map = \ - {'thread_per_core': 1, '2':[1], 'cores_per_socket': 2} - vnf_cfg = {'lb_config': 'SW', 'lb_count': 1, 'worker_config': - '1C/1T', 'worker_threads': 1} - self.assertEqual(-1, cpu_topo.validate_cpu_cfg(vnf_cfg)) diff --git a/tests/unit/network_services/helpers/test_dpdkbindnic_helper.py b/tests/unit/network_services/helpers/test_dpdkbindnic_helper.py deleted file mode 100644 index 367072e84..000000000 --- a/tests/unit/network_services/helpers/test_dpdkbindnic_helper.py +++ /dev/null @@ -1,633 +0,0 @@ -# Copyright (c) 2016-2017 Intel Corporation -# -# Licensed under the Apache License, Version 2.0 (the "License"); -# you may not use this file except in compliance with the License. -# You may obtain a copy of the License at -# -# http://www.apache.org/licenses/LICENSE-2.0 -# -# Unless required by applicable law or agreed to in writing, software -# distributed under the License is distributed on an "AS IS" BASIS, -# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -# See the License for the specific language governing permissions and -# limitations under the License. - -import mock -import unittest - -import os - -from yardstick.error import IncorrectConfig, SSHError -from yardstick.error import IncorrectNodeSetup -from yardstick.error import IncorrectSetup -from yardstick.network_services.helpers.dpdkbindnic_helper import DpdkInterface -from yardstick.network_services.helpers.dpdkbindnic_helper import DpdkNode -from yardstick.network_services.helpers.dpdkbindnic_helper import DpdkBindHelper -from yardstick.network_services.helpers.dpdkbindnic_helper import DpdkBindHelperException -from yardstick.network_services.helpers.dpdkbindnic_helper import NETWORK_KERNEL -from yardstick.network_services.helpers.dpdkbindnic_helper import NETWORK_DPDK -from yardstick.network_services.helpers.dpdkbindnic_helper import CRYPTO_KERNEL -from yardstick.network_services.helpers.dpdkbindnic_helper import CRYPTO_DPDK -from yardstick.network_services.helpers.dpdkbindnic_helper import NETWORK_OTHER -from yardstick.network_services.helpers.dpdkbindnic_helper import CRYPTO_OTHER - - -NAME = "tg_0" - - -class TestDpdkInterface(unittest.TestCase): - - SAMPLE_NETDEVS = { - 'enp11s0': { - 'address': '0a:de:ad:be:ef:f5', - 'device': '0x1533', - 'driver': 'igb', - 'ifindex': '2', - 'interface_name': 'enp11s0', - 'operstate': 'down', - 'pci_bus_id': '0000:0b:00.0', - 'subsystem_device': '0x1533', - 'subsystem_vendor': '0x15d9', - 'vendor': '0x8086' - }, - 'lan': { - 'address': '0a:de:ad:be:ef:f4', - 'device': '0x153a', - 'driver': 'e1000e', - 'ifindex': '3', - 'interface_name': 'lan', - 'operstate': 'up', - 'pci_bus_id': '0000:00:19.0', - 'subsystem_device': '0x153a', - 'subsystem_vendor': '0x15d9', - 'vendor': '0x8086' - } - } - - SAMPLE_VM_NETDEVS = { - 'eth1': { - 'address': 'fa:de:ad:be:ef:5b', - 'device': '0x0001', - 'driver': 'virtio_net', - 'ifindex': '3', - 'interface_name': 'eth1', - 'operstate': 'down', - 'pci_bus_id': '0000:00:04.0', - 'vendor': '0x1af4' - } - } - - def test_parse_netdev_info(self): - output = """\ -/sys/devices/pci0000:00/0000:00:1c.3/0000:0b:00.0/net/enp11s0/ifindex:2 -/sys/devices/pci0000:00/0000:00:1c.3/0000:0b:00.0/net/enp11s0/address:0a:de:ad:be:ef:f5 -/sys/devices/pci0000:00/0000:00:1c.3/0000:0b:00.0/net/enp11s0/operstate:down -/sys/devices/pci0000:00/0000:00:1c.3/0000:0b:00.0/net/enp11s0/device/vendor:0x8086 -/sys/devices/pci0000:00/0000:00:1c.3/0000:0b:00.0/net/enp11s0/device/device:0x1533 -/sys/devices/pci0000:00/0000:00:1c.3/0000:0b:00.0/net/enp11s0/device/subsystem_vendor:0x15d9 -/sys/devices/pci0000:00/0000:00:1c.3/0000:0b:00.0/net/enp11s0/device/subsystem_device:0x1533 -/sys/devices/pci0000:00/0000:00:1c.3/0000:0b:00.0/net/enp11s0/driver:igb -/sys/devices/pci0000:00/0000:00:1c.3/0000:0b:00.0/net/enp11s0/pci_bus_id:0000:0b:00.0 -/sys/devices/pci0000:00/0000:00:19.0/net/lan/ifindex:3 -/sys/devices/pci0000:00/0000:00:19.0/net/lan/address:0a:de:ad:be:ef:f4 -/sys/devices/pci0000:00/0000:00:19.0/net/lan/operstate:up -/sys/devices/pci0000:00/0000:00:19.0/net/lan/device/vendor:0x8086 -/sys/devices/pci0000:00/0000:00:19.0/net/lan/device/device:0x153a -/sys/devices/pci0000:00/0000:00:19.0/net/lan/device/subsystem_vendor:0x15d9 -/sys/devices/pci0000:00/0000:00:19.0/net/lan/device/subsystem_device:0x153a -/sys/devices/pci0000:00/0000:00:19.0/net/lan/driver:e1000e -/sys/devices/pci0000:00/0000:00:19.0/net/lan/pci_bus_id:0000:00:19.0 -""" - res = DpdkBindHelper.parse_netdev_info(output) - self.assertDictEqual(res, self.SAMPLE_NETDEVS) - - def test_parse_netdev_info_virtio(self): - output = """\ -/sys/devices/pci0000:00/0000:00:04.0/virtio1/net/eth1/ifindex:3 -/sys/devices/pci0000:00/0000:00:04.0/virtio1/net/eth1/address:fa:de:ad:be:ef:5b -/sys/devices/pci0000:00/0000:00:04.0/virtio1/net/eth1/operstate:down -/sys/devices/pci0000:00/0000:00:04.0/virtio1/net/eth1/device/vendor:0x1af4 -/sys/devices/pci0000:00/0000:00:04.0/virtio1/net/eth1/device/device:0x0001 -/sys/devices/pci0000:00/0000:00:04.0/virtio1/net/eth1/driver:virtio_net -""" - res = DpdkBindHelper.parse_netdev_info(output) - self.assertDictEqual(res, self.SAMPLE_VM_NETDEVS) - - def test_probe_missing_values(self): - mock_dpdk_node = mock.Mock() - mock_dpdk_node.netdevs = self.SAMPLE_NETDEVS.copy() - - interface = {'local_mac': '0a:de:ad:be:ef:f5'} - dpdk_intf = DpdkInterface(mock_dpdk_node, interface) - - dpdk_intf.probe_missing_values() - self.assertEqual(interface['vpci'], '0000:0b:00.0') - - interface['local_mac'] = '0a:de:ad:be:ef:f4' - dpdk_intf.probe_missing_values() - self.assertEqual(interface['vpci'], '0000:00:19.0') - - def test_probe_missing_values_no_update(self): - mock_dpdk_node = mock.Mock() - mock_dpdk_node.netdevs = self.SAMPLE_NETDEVS.copy() - del mock_dpdk_node.netdevs['enp11s0']['driver'] - del mock_dpdk_node.netdevs['lan']['driver'] - - interface = {'local_mac': '0a:de:ad:be:ef:f5'} - dpdk_intf = DpdkInterface(mock_dpdk_node, interface) - - dpdk_intf.probe_missing_values() - self.assertNotIn('vpci', interface) - self.assertNotIn('driver', interface) - - def test_probe_missing_values_negative(self): - mock_dpdk_node = mock.Mock() - mock_dpdk_node.netdevs.values.side_effect = IncorrectNodeSetup - - interface = {'local_mac': '0a:de:ad:be:ef:f5'} - dpdk_intf = DpdkInterface(mock_dpdk_node, interface) - - with self.assertRaises(IncorrectConfig): - dpdk_intf.probe_missing_values() - - -class TestDpdkNode(unittest.TestCase): - - INTERFACES = [ - {'name': 'name1', - 'virtual-interface': { - 'local_mac': 404, - 'vpci': 'pci10', - }}, - {'name': 'name2', - 'virtual-interface': { - 'local_mac': 404, - 'vpci': 'pci2', - }}, - {'name': 'name3', - 'virtual-interface': { - 'local_mac': 404, - 'vpci': 'some-pci1', - }}, - ] - - def test_probe_dpdk_drivers(self): - mock_ssh_helper = mock.Mock() - mock_ssh_helper.execute.return_value = 0, '', '' - - interfaces = [ - {'name': 'name1', - 'virtual-interface': { - 'local_mac': 404, - 'vpci': 'pci10', - }}, - {'name': 'name2', - 'virtual-interface': { - 'local_mac': 404, - 'vpci': 'pci2', - }}, - {'name': 'name3', - 'virtual-interface': { - 'local_mac': 404, - 'vpci': 'some-pci1', - }}, - ] - - dpdk_node = DpdkNode(NAME, interfaces, mock_ssh_helper) - dpdk_helper = dpdk_node.dpdk_helper - - dpdk_helper.probe_real_kernel_drivers = mock.Mock() - dpdk_helper.real_kernel_interface_driver_map = { - 'pci1': 'driver1', - 'pci2': 'driver2', - 'pci3': 'driver3', - 'pci4': 'driver1', - 'pci6': 'driver3', - } - - dpdk_node._probe_dpdk_drivers() - self.assertNotIn('driver', interfaces[0]['virtual-interface']) - self.assertEqual(interfaces[1]['virtual-interface']['driver'], 'driver2') - self.assertEqual(interfaces[2]['virtual-interface']['driver'], 'driver1') - - def test_check(self): - def update(): - if not mock_force_rebind.called: - raise IncorrectConfig - - interfaces[0]['virtual-interface'].update({ - 'vpci': '0000:01:02.1', - 'local_ip': '10.20.30.40', - 'netmask': '255.255.0.0', - 'driver': 'ixgbe', - }) - - mock_ssh_helper = mock.Mock() - mock_ssh_helper.execute.return_value = 0, '', '' - - interfaces = [ - {'name': 'name1', - 'virtual-interface': { - 'local_mac': 404, - }}, - ] - - dpdk_node = DpdkNode(NAME, interfaces, mock_ssh_helper) - dpdk_node._probe_missing_values = mock_probe_missing = mock.Mock(side_effect=update) - dpdk_node._force_rebind = mock_force_rebind = mock.Mock() - - self.assertIsNone(dpdk_node.check()) - self.assertEqual(mock_probe_missing.call_count, 2) - - @mock.patch('yardstick.network_services.helpers.dpdkbindnic_helper.DpdkInterface') - def test_check_negative(self, mock_intf_type): - mock_ssh_helper = mock.Mock() - mock_ssh_helper.execute.return_value = 0, '', '' - - mock_intf_type().check.side_effect = SSHError - - dpdk_node = DpdkNode(NAME, self.INTERFACES, mock_ssh_helper) - - with self.assertRaises(IncorrectSetup): - dpdk_node.check() - - def test_probe_netdevs(self): - mock_ssh_helper = mock.Mock() - mock_ssh_helper.execute.return_value = 0, '', '' - - expected = {'key1': 500, 'key2': 'hello world'} - update = {'key1': 1000, 'key3': []} - - dpdk_node = DpdkNode(NAME, self.INTERFACES, mock_ssh_helper) - dpdk_helper = dpdk_node.dpdk_helper - dpdk_helper.find_net_devices = mock.Mock(side_effect=[expected, update]) - - self.assertDictEqual(dpdk_node.netdevs, {}) - dpdk_node._probe_netdevs() - self.assertDictEqual(dpdk_node.netdevs, expected) - - expected = {'key1': 1000, 'key2': 'hello world', 'key3': []} - dpdk_node._probe_netdevs() - self.assertDictEqual(dpdk_node.netdevs, expected) - - def test_probe_netdevs_setup_negative(self): - mock_ssh_helper = mock.Mock() - mock_ssh_helper.execute.return_value = 0, '', '' - - dpdk_node = DpdkNode(NAME, self.INTERFACES, mock_ssh_helper) - dpdk_helper = dpdk_node.dpdk_helper - dpdk_helper.find_net_devices = mock.Mock(side_effect=DpdkBindHelperException) - - with self.assertRaises(DpdkBindHelperException): - dpdk_node._probe_netdevs() - - def test_force_rebind(self): - mock_ssh_helper = mock.Mock() - mock_ssh_helper.execute.return_value = 0, '', '' - - dpdk_node = DpdkNode(NAME, self.INTERFACES, mock_ssh_helper) - dpdk_helper = dpdk_node.dpdk_helper - dpdk_helper.force_dpdk_rebind = mock_helper_func = mock.Mock() - - dpdk_node._force_rebind() - self.assertEqual(mock_helper_func.call_count, 1) - - -class TestDpdkBindHelper(unittest.TestCase): - bin_path = "/opt/nsb_bin" - EXAMPLE_OUTPUT = """ - -Network devices using DPDK-compatible driver -============================================ -0000:00:04.0 'Virtio network device' drv=igb_uio unused= -0000:00:05.0 'Virtio network device' drv=igb_uio unused= - -Network devices using kernel driver -=================================== -0000:00:03.0 'Virtio network device' if=ens3 drv=virtio-pci unused=igb_uio *Active* - -Other network devices -===================== -<none> - -Crypto devices using DPDK-compatible driver -=========================================== -<none> - -Crypto devices using kernel driver -================================== -<none> - -Other crypto devices -==================== -<none> -""" - - PARSED_EXAMPLE = { - NETWORK_DPDK: [ - {'active': False, - 'dev_type': 'Virtio network device', - 'driver': 'igb_uio', - 'iface': None, - 'unused': '', - 'vpci': '0000:00:04.0', - }, - {'active': False, - 'dev_type': 'Virtio network device', - 'driver': 'igb_uio', - 'iface': None, - 'unused': '', - 'vpci': '0000:00:05.0', - } - ], - NETWORK_KERNEL: [ - {'active': True, - 'dev_type': 'Virtio network device', - 'driver': 'virtio-pci', - 'iface': 'ens3', - 'unused': 'igb_uio', - 'vpci': '0000:00:03.0', - } - ], - CRYPTO_KERNEL: [], - CRYPTO_DPDK: [], - NETWORK_OTHER: [], - CRYPTO_OTHER: [], - } - - CLEAN_STATUS = { - NETWORK_KERNEL: [], - NETWORK_DPDK: [], - CRYPTO_KERNEL: [], - CRYPTO_DPDK: [], - NETWORK_OTHER: [], - CRYPTO_OTHER: [], - } - - ONE_INPUT_LINE = ("0000:00:03.0 'Virtio network device' if=ens3 " - "drv=virtio-pci unused=igb_uio *Active*") - - ONE_INPUT_LINE_PARSED = [{ - 'vpci': '0000:00:03.0', - 'dev_type': 'Virtio network device', - 'iface': 'ens3', - 'driver': 'virtio-pci', - 'unused': 'igb_uio', - 'active': True, - }] - - def test___init__(self): - conn = mock.Mock() - conn.provision_tool = mock.Mock(return_value='path_to_tool') - conn.join_bin_path.return_value = os.path.join(self.bin_path, DpdkBindHelper.DPDK_DEVBIND) - - dpdk_bind_helper = DpdkBindHelper(conn) - - self.assertEqual(conn, dpdk_bind_helper.ssh_helper) - self.assertEqual(self.CLEAN_STATUS, dpdk_bind_helper.dpdk_status) - self.assertIsNone(dpdk_bind_helper.status_nic_row_re) - self.assertEqual(dpdk_bind_helper.dpdk_devbind, - os.path.join(self.bin_path, dpdk_bind_helper.DPDK_DEVBIND)) - self.assertIsNone(dpdk_bind_helper._status_cmd_attr) - - def test__dpdk_execute(self): - conn = mock.Mock() - conn.execute = mock.Mock(return_value=(0, 'output', 'error')) - conn.provision_tool = mock.Mock(return_value='tool_path') - dpdk_bind_helper = DpdkBindHelper(conn) - self.assertEqual((0, 'output', 'error'), dpdk_bind_helper._dpdk_execute('command')) - - def test__dpdk_execute_failure(self): - conn = mock.Mock() - conn.execute = mock.Mock(return_value=(1, 'output', 'error')) - conn.provision_tool = mock.Mock(return_value='tool_path') - dpdk_bind_helper = DpdkBindHelper(conn) - with self.assertRaises(DpdkBindHelperException): - dpdk_bind_helper._dpdk_execute('command') - - def test__addline(self): - conn = mock.Mock() - - dpdk_bind_helper = DpdkBindHelper(conn) - - dpdk_bind_helper._add_line(NETWORK_KERNEL, self.ONE_INPUT_LINE) - - self.assertIsNotNone(dpdk_bind_helper.dpdk_status) - self.assertEqual(self.ONE_INPUT_LINE_PARSED, dpdk_bind_helper.dpdk_status[NETWORK_KERNEL]) - - def test__switch_active_dict_by_header(self): - line = "Crypto devices using DPDK-compatible driver" - olddict = 'olddict' - self.assertEqual(CRYPTO_DPDK, DpdkBindHelper._switch_active_dict(line, olddict)) - - def test__switch_active_dict_by_header_empty(self): - line = "<none>" - olddict = 'olddict' - self.assertEqual(olddict, DpdkBindHelper._switch_active_dict(line, olddict)) - - def test_parse_dpdk_status_output(self): - conn = mock.Mock() - - dpdk_bind_helper = DpdkBindHelper(conn) - - dpdk_bind_helper._parse_dpdk_status_output(self.EXAMPLE_OUTPUT) - - self.maxDiff = None - self.assertEqual(self.PARSED_EXAMPLE, dpdk_bind_helper.dpdk_status) - - def test_kernel_bound_pci_addresses(self): - mock_ssh_helper = mock.Mock() - mock_ssh_helper.execute.return_value = 0, '', '' - - expected = ['a', 'b', 3] - - dpdk_helper = DpdkBindHelper(mock_ssh_helper) - dpdk_helper.dpdk_status = { - NETWORK_DPDK: [{'vpci': 4}, {'vpci': 5}, {'vpci': 'g'}], - NETWORK_KERNEL: [{'vpci': 'a'}, {'vpci': 'b'}, {'vpci': 3}], - CRYPTO_DPDK: [], - } - - result = dpdk_helper.kernel_bound_pci_addresses - self.assertEqual(result, expected) - - def test_find_net_devices_negative(self): - mock_ssh_helper = mock.Mock() - mock_ssh_helper.execute.return_value = 1, 'error', 'debug' - - dpdk_helper = DpdkBindHelper(mock_ssh_helper) - - self.assertDictEqual(dpdk_helper.find_net_devices(), {}) - - def test_read_status(self): - conn = mock.Mock() - conn.execute = mock.Mock(return_value=(0, self.EXAMPLE_OUTPUT, '')) - conn.provision_tool = mock.Mock(return_value='path_to_tool') - - dpdk_bind_helper = DpdkBindHelper(conn) - - self.assertEqual(self.PARSED_EXAMPLE, dpdk_bind_helper.read_status()) - - def test__get_bound_pci_addresses(self): - conn = mock.Mock() - - dpdk_bind_helper = DpdkBindHelper(conn) - - dpdk_bind_helper._parse_dpdk_status_output(self.EXAMPLE_OUTPUT) - - self.assertEqual(['0000:00:04.0', '0000:00:05.0'], - dpdk_bind_helper._get_bound_pci_addresses(NETWORK_DPDK)) - self.assertEqual(['0000:00:03.0'], - dpdk_bind_helper._get_bound_pci_addresses(NETWORK_KERNEL)) - - def test_interface_driver_map(self): - conn = mock.Mock() - - dpdk_bind_helper = DpdkBindHelper(conn) - - dpdk_bind_helper._parse_dpdk_status_output(self.EXAMPLE_OUTPUT) - - self.assertEqual({'0000:00:04.0': 'igb_uio', - '0000:00:03.0': 'virtio-pci', - '0000:00:05.0': 'igb_uio', - }, - dpdk_bind_helper.interface_driver_map) - - def test_bind(self): - conn = mock.Mock() - conn.execute = mock.Mock(return_value=(0, '', '')) - conn.join_bin_path.return_value = os.path.join(self.bin_path, DpdkBindHelper.DPDK_DEVBIND) - - dpdk_bind_helper = DpdkBindHelper(conn) - dpdk_bind_helper.read_status = mock.Mock() - - dpdk_bind_helper.bind(['0000:00:03.0', '0000:00:04.0'], 'my_driver') - - conn.execute.assert_called_with('sudo /opt/nsb_bin/dpdk-devbind.py --force ' - '-b my_driver 0000:00:03.0 0000:00:04.0') - dpdk_bind_helper.read_status.assert_called_once() - - def test_bind_single_pci(self): - conn = mock.Mock() - conn.execute = mock.Mock(return_value=(0, '', '')) - conn.join_bin_path.return_value = os.path.join(self.bin_path, DpdkBindHelper.DPDK_DEVBIND) - - dpdk_bind_helper = DpdkBindHelper(conn) - dpdk_bind_helper.read_status = mock.Mock() - - dpdk_bind_helper.bind('0000:00:03.0', 'my_driver') - - conn.execute.assert_called_with('sudo /opt/nsb_bin/dpdk-devbind.py --force ' - '-b my_driver 0000:00:03.0') - dpdk_bind_helper.read_status.assert_called_once() - - def test_rebind_drivers(self): - conn = mock.Mock() - - dpdk_bind_helper = DpdkBindHelper(conn) - - dpdk_bind_helper.bind = mock.Mock() - dpdk_bind_helper.used_drivers = { - 'd1': ['0000:05:00.0'], - 'd3': ['0000:05:01.0', '0000:05:02.0'], - } - - dpdk_bind_helper.rebind_drivers() - - dpdk_bind_helper.bind.assert_any_call(['0000:05:00.0'], 'd1', True) - dpdk_bind_helper.bind.assert_any_call(['0000:05:01.0', '0000:05:02.0'], 'd3', True) - - def test_save_used_drivers(self): - conn = mock.Mock() - dpdk_bind_helper = DpdkBindHelper(conn) - dpdk_bind_helper.dpdk_status = self.PARSED_EXAMPLE - - dpdk_bind_helper.save_used_drivers() - - expected = { - 'igb_uio': ['0000:00:04.0', '0000:00:05.0'], - 'virtio-pci': ['0000:00:03.0'], - } - - self.assertDictEqual(expected, dpdk_bind_helper.used_drivers) - - def test_force_dpdk_rebind(self): - mock_ssh_helper = mock.Mock() - mock_ssh_helper.execute.return_value = 0, '', '' - - dpdk_helper = DpdkBindHelper(mock_ssh_helper, 'driver2') - dpdk_helper.dpdk_status = { - NETWORK_DPDK: [ - { - 'vpci': 'pci1', - }, - { - 'vpci': 'pci3', - }, - { - 'vpci': 'pci6', - }, - { - 'vpci': 'pci3', - }, - ] - } - dpdk_helper.real_kernel_interface_driver_map = { - 'pci1': 'real_driver1', - 'pci2': 'real_driver2', - 'pci3': 'real_driver1', - 'pci4': 'real_driver4', - 'pci6': 'real_driver6', - } - dpdk_helper.load_dpdk_driver = mock.Mock() - dpdk_helper.read_status = mock.Mock() - dpdk_helper.save_real_kernel_interface_driver_map = mock.Mock() - dpdk_helper.save_used_drivers = mock.Mock() - dpdk_helper.bind = mock_bind = mock.Mock() - - dpdk_helper.force_dpdk_rebind() - self.assertEqual(mock_bind.call_count, 2) - - def test_save_real_kernel_drivers(self): - mock_ssh_helper = mock.Mock() - mock_ssh_helper.execute.return_value = 0, '', '' - - dpdk_helper = DpdkBindHelper(mock_ssh_helper) - dpdk_helper.real_kernel_drivers = { - 'abc': '123', - } - dpdk_helper.real_kernel_interface_driver_map = { - 'abc': 'AAA', - 'def': 'DDD', - 'abs': 'AAA', - 'ghi': 'GGG', - } - - # save_used_drivers must be called before save_real_kernel_drivers can be - with self.assertRaises(AttributeError): - dpdk_helper.save_real_kernel_drivers() - - dpdk_helper.save_used_drivers() - - expected_used_drivers = { - 'AAA': ['abc', 'abs'], - 'DDD': ['def'], - 'GGG': ['ghi'], - } - dpdk_helper.save_real_kernel_drivers() - self.assertDictEqual(dpdk_helper.used_drivers, expected_used_drivers) - self.assertDictEqual(dpdk_helper.real_kernel_drivers, {}) - - def test_get_real_kernel_driver(self): - mock_ssh_helper = mock.Mock() - mock_ssh_helper.execute.side_effect = [ - (0, 'non-matching text', ''), - (0, 'pre Kernel modules: real_driver1', ''), - (0, 'before Ethernet middle Virtio network device after', ''), - ] - - dpdk_helper = DpdkBindHelper(mock_ssh_helper) - - self.assertIsNone(dpdk_helper.get_real_kernel_driver('abc')) - self.assertEqual(dpdk_helper.get_real_kernel_driver('abc'), 'real_driver1') - self.assertEqual(dpdk_helper.get_real_kernel_driver('abc'), DpdkBindHelper.VIRTIO_DRIVER) diff --git a/tests/unit/network_services/helpers/test_iniparser.py b/tests/unit/network_services/helpers/test_iniparser.py deleted file mode 100644 index bd27b497e..000000000 --- a/tests/unit/network_services/helpers/test_iniparser.py +++ /dev/null @@ -1,225 +0,0 @@ -# Copyright (c) 2017 Intel Corporation -# -# Licensed under the Apache License, Version 2.0 (the "License"); -# you may not use this file except in compliance with the License. -# You may obtain a copy of the License at -# -# http://www.apache.org/licenses/LICENSE-2.0 -# -# Unless required by applicable law or agreed to in writing, software -# distributed under the License is distributed on an "AS IS" BASIS, -# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -# See the License for the specific language governing permissions and -# limitations under the License. -# - -from __future__ import absolute_import - -import unittest -from contextlib import contextmanager -import mock - -from tests.unit import STL_MOCKS - - -STLClient = mock.MagicMock() -stl_patch = mock.patch.dict("sys.modules", STL_MOCKS) -stl_patch.start() - -if stl_patch: - from yardstick.network_services.helpers.iniparser import ParseError - from yardstick.network_services.helpers.iniparser import LineParser - from yardstick.network_services.helpers.iniparser import BaseParser - from yardstick.network_services.helpers.iniparser import ConfigParser - -PARSE_TEXT_1 = """\ - -[section1] -key1=value1 -list1: value2 - value3 - value4 -key3='single quote value' ; comment here -key4= - -[section2] ; comment with #2 other symbol -# here is a comment line -list2: value5 -key with no value # mixed comment ; symbols -; another comment line -key5= - -[section1] ; reopen a section! -key2="double quote value" -""" - -PARSE_TEXT_2 = """\ -[section1] -list1 = item1 - item2 - ended by eof""" - -PARSE_TEXT_BAD_1 = """\ -key1=value1 -""" - -PARSE_TEXT_BAD_2 = """\ -[section1 -""" - -PARSE_TEXT_BAD_3 = """\ -[] -""" - -PARSE_TEXT_BAD_4 = """\ -[section1] - bad continuation -""" - -PARSE_TEXT_BAD_5 = """\ -[section1] -=value with no key -""" - - -class TestParseError(unittest.TestCase): - - def test___str__(self): - error = ParseError('a', 2, 'c') - self.assertEqual(str(error), "at line 2, a: 'c'") - - -class TestLineParser(unittest.TestCase): - - def test___repr__(self): - line_parser = LineParser('', 101) - self.assertIsNotNone(repr(line_parser)) - - def test_error_invalid_assignment(self): - line_parser = LineParser('', 101) - self.assertIsNotNone(line_parser.error_invalid_assignment()) - - -class TestBaseParser(unittest.TestCase): - - @staticmethod - def make_open(text_blob): - @contextmanager - def internal_open(*args, **kwargs): - yield text_blob.split('\n') - - return internal_open - - def test_parse(self): - parser = BaseParser() - parser.parse() - - def test_parse_empty_string(self): - parser = BaseParser() - self.assertIsNone(parser.parse('')) - - def test_not_implemented_methods(self): - parser = BaseParser() - - with self.assertRaises(NotImplementedError): - parser.assignment('key', 'value', LineParser('', 100)) - - with self.assertRaises(NotImplementedError): - parser.new_section('section') - - with self.assertRaises(NotImplementedError): - parser.comment('comment') - - -class TestConfigParser(unittest.TestCase): - - @staticmethod - def make_open(text_blob): - @contextmanager - def internal_open(*args, **kwargs): - yield text_blob.split('\n') - - return internal_open - - @mock.patch('yardstick.network_services.helpers.iniparser.open') - def test_parse(self, mock_open): - mock_open.side_effect = self.make_open(PARSE_TEXT_1) - - existing_data = [['section0', [['key0', 'value0']]]] - config_parser = ConfigParser('my_file', existing_data) - config_parser.parse() - - expected = [ - [ - 'section0', - [ - ['key0', 'value0'], - ], - ], - [ - 'section1', - [ - ['key1', 'value1'], - ['list1', 'value2\nvalue3\nvalue4'], - ['key3', 'single quote value'], - ['key4', ''], - ['key2', 'double quote value'], - ], - ], - [ - 'section2', - [ - ['list2', 'value5'], - ['key with no value', '@'], - ['key5', ''], - ], - ], - ] - - self.assertEqual(config_parser.sections, expected) - self.assertIsNotNone(config_parser.find_section('section1')) - self.assertIsNone(config_parser.find_section('section3')) - self.assertEqual(config_parser.find_section_index('section1'), 1) - self.assertEqual(config_parser.find_section_index('section3'), -1) - - @mock.patch('yardstick.network_services.helpers.iniparser.open') - def test_parse_2(self, mock_open): - mock_open.side_effect = self.make_open(PARSE_TEXT_2) - - config_parser = ConfigParser('my_file') - config_parser.parse() - - expected = [ - [ - 'section1', - [ - ['list1', 'item1\nitem2\nended by eof'], - ], - ], - ] - - self.assertEqual(config_parser.sections, expected) - - @mock.patch('yardstick.network_services.helpers.iniparser.open') - def test_parse_negative(self, mock_open): - bad_text_dict = { - 'no section': PARSE_TEXT_BAD_1, - 'incomplete section': PARSE_TEXT_BAD_2, - 'empty section name': PARSE_TEXT_BAD_3, - 'bad_continuation': PARSE_TEXT_BAD_4, - 'value with no key': PARSE_TEXT_BAD_5, - } - - for bad_reason, bad_text in bad_text_dict.items(): - mock_open.side_effect = self.make_open(bad_text) - - config_parser = ConfigParser('my_file', []) - - try: - # TODO: replace with assertRaises, when the UT framework supports - # advanced messages when exceptions fail to occur - config_parser.parse() - except ParseError: - pass - else: - self.fail('\n'.join([bad_reason, bad_text, str(config_parser.sections)])) diff --git a/tests/unit/network_services/helpers/test_samplevnf_helper.py b/tests/unit/network_services/helpers/test_samplevnf_helper.py deleted file mode 100644 index dc74b1859..000000000 --- a/tests/unit/network_services/helpers/test_samplevnf_helper.py +++ /dev/null @@ -1,1105 +0,0 @@ -# Copyright (c) 2016-2017 Intel Corporation -# -# Licensed under the Apache License, Version 2.0 (the "License"); -# you may not use this file except in compliance with the License. -# You may obtain a copy of the License at -# -# http://www.apache.org/licenses/LICENSE-2.0 -# -# Unless required by applicable law or agreed to in writing, software -# distributed under the License is distributed on an "AS IS" BASIS, -# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -# See the License for the specific language governing permissions and -# limitations under the License. -# - -import mock -import os -import six -import unittest - -from yardstick.network_services.helpers import samplevnf_helper -from yardstick.network_services.vnf_generic.vnf.base import VnfdHelper - - -class TestPortPairs(unittest.TestCase): - def test_port_pairs_list(self): - vnfd = TestMultiPortConfig.VNFD['vnfd:vnfd-catalog']['vnfd'][0] - interfaces = vnfd['vdu'][0]['external-interface'] - port_pairs = samplevnf_helper.PortPairs(interfaces) - self.assertEqual(port_pairs.port_pair_list, [("xe0", "xe1")]) - - def test_valid_networks(self): - vnfd = TestMultiPortConfig.VNFD['vnfd:vnfd-catalog']['vnfd'][0] - interfaces = vnfd['vdu'][0]['external-interface'] - port_pairs = samplevnf_helper.PortPairs(interfaces) - self.assertEqual(port_pairs.valid_networks, [ - ("uplink_0", "downlink_0")]) - - def test_all_ports(self): - vnfd = TestMultiPortConfig.VNFD['vnfd:vnfd-catalog']['vnfd'][0] - interfaces = vnfd['vdu'][0]['external-interface'] - port_pairs = samplevnf_helper.PortPairs(interfaces) - self.assertEqual(set(port_pairs.all_ports), {"xe0", "xe1"}) - - def test_uplink_ports(self): - vnfd = TestMultiPortConfig.VNFD['vnfd:vnfd-catalog']['vnfd'][0] - interfaces = vnfd['vdu'][0]['external-interface'] - port_pairs = samplevnf_helper.PortPairs(interfaces) - self.assertEqual(port_pairs.uplink_ports, ["xe0"]) - - def test_downlink_ports(self): - vnfd = TestMultiPortConfig.VNFD['vnfd:vnfd-catalog']['vnfd'][0] - interfaces = vnfd['vdu'][0]['external-interface'] - port_pairs = samplevnf_helper.PortPairs(interfaces) - self.assertEqual(port_pairs.downlink_ports, ["xe1"]) - - -class TestMultiPortConfig(unittest.TestCase): - - VNFD_0 = {'short-name': 'VpeVnf', - 'vdu': - [{'routing_table': - [{'network': '152.16.100.20', - 'netmask': '255.255.255.0', - 'gateway': '152.16.100.20', - 'if': 'xe0'}, - {'network': '152.16.40.20', - 'netmask': '255.255.255.0', - 'gateway': '152.16.40.20', - 'if': 'xe1'}], - 'description': 'VPE approximation using DPDK', - 'name': 'vpevnf-baremetal', - 'nd_route_tbl': - [{'network': '0064:ff9b:0:0:0:0:9810:6414', - 'netmask': '112', - 'gateway': '0064:ff9b:0:0:0:0:9810:6414', - 'if': 'xe0'}, - {'network': '0064:ff9b:0:0:0:0:9810:2814', - 'netmask': '112', - 'gateway': '0064:ff9b:0:0:0:0:9810:2814', - 'if': 'xe1'}], - 'id': 'vpevnf-baremetal', - 'external-interface': - [ - {'virtual-interface': - { - 'dst_mac': '00:00:00:00:00:04', - 'vpci': '0000:05:00.0', - 'local_ip': '152.16.100.19', - 'type': 'PCI-PASSTHROUGH', - 'netmask': '255.255.255.0', - 'dpdk_port_num': 0, - 'bandwidth': '10 Gbps', - 'driver': "i40e", - 'dst_ip': '152.16.100.20', - 'ifname': 'xe0', - 'local_iface_name': 'eth0', - 'local_mac': '00:00:00:00:00:02', - 'vld_id': 'uplink_0', - }, - 'vnfd-connection-point-ref': 'xe0', - 'name': 'xe0'}, - {'virtual-interface': - { - 'dst_mac': '00:00:00:00:00:03', - 'vpci': '0000:05:00.1', - 'local_ip': '152.16.40.19', - 'type': 'PCI-PASSTHROUGH', - 'driver': "i40e", - 'netmask': '255.255.255.0', - 'dpdk_port_num': 1, - 'bandwidth': '10 Gbps', - 'dst_ip': '152.16.40.20', - 'ifname': 'xe1', - 'local_iface_name': 'eth1', - 'local_mac': '00:00:00:00:00:01', - 'vld_id': 'downlink_0', - }, - 'vnfd-connection-point-ref': 'xe1', - 'name': 'xe1'} - ]}], - 'description': 'Vpe approximation using DPDK', - 'mgmt-interface': - {'vdu-id': 'vpevnf-baremetal', - 'host': '1.2.1.1', - 'password': 'r00t', - 'user': 'root', - 'ip': '1.2.1.1'}, - 'benchmark': - {'kpi': ['packets_in', 'packets_fwd', 'packets_dropped']}, - 'connection-point': [{'type': 'VPORT', 'name': 'xe0'}, - {'type': 'VPORT', 'name': 'xe1'}], - 'id': 'AclApproxVnf', 'name': 'VPEVnfSsh'} - - VNFD = { - 'vnfd:vnfd-catalog': { - 'vnfd': [ - VNFD_0, - ] - } - } - - def setUp(self): - self._mock_open = mock.patch.object(six.moves.builtins, 'open') - self.mock_open = self._mock_open.start() - self._mock_config_parser = mock.patch.object( - samplevnf_helper, 'ConfigParser') - self.mock_config_parser = self._mock_config_parser.start() - - self.addCleanup(self._cleanup) - - def _cleanup(self): - self._mock_open.stop() - self._mock_config_parser.stop() - - def test_validate_ip_and_prefixlen(self): - ip_addr, prefix_len = ( - samplevnf_helper.MultiPortConfig.validate_ip_and_prefixlen( - '10.20.30.40', '16')) - self.assertEqual(ip_addr, '10.20.30.40') - self.assertEqual(prefix_len, 16) - - ip_addr, prefix_len = ( - samplevnf_helper.MultiPortConfig.validate_ip_and_prefixlen( - '::1', '40')) - self.assertEqual(ip_addr, '0000:0000:0000:0000:0000:0000:0000:0001') - self.assertEqual(prefix_len, 40) - - def test_validate_ip_and_prefixlen_negative(self): - with self.assertRaises(AttributeError): - samplevnf_helper.MultiPortConfig.validate_ip_and_prefixlen('', '') - - with self.assertRaises(AttributeError): - samplevnf_helper.MultiPortConfig.validate_ip_and_prefixlen( - '10.20.30.400', '16') - - with self.assertRaises(AttributeError): - samplevnf_helper.MultiPortConfig.validate_ip_and_prefixlen( - '10.20.30.40', '33') - - with self.assertRaises(AttributeError): - samplevnf_helper.MultiPortConfig.validate_ip_and_prefixlen( - '::1', '129') - - @mock.patch.object(os.path, 'isfile', return_value=False) - def test___init__(self, *args): - topology_file = mock.Mock() - config_tpl = mock.Mock() - tmp_file = mock.Mock() - vnfd_mock = mock.MagicMock() - opnfv_vnf = samplevnf_helper.MultiPortConfig( - topology_file, config_tpl, tmp_file, vnfd_mock) - self.assertEqual(0, opnfv_vnf.swq) - opnfv_vnf = samplevnf_helper.MultiPortConfig( - topology_file, config_tpl, tmp_file, vnfd_mock) - self.assertEqual(0, opnfv_vnf.swq) - - def test_update_timer(self): - topology_file = mock.Mock() - config_tpl = mock.Mock() - tmp_file = mock.Mock() - vnfd_mock = mock.MagicMock() - opnfv_vnf = samplevnf_helper.MultiPortConfig( - topology_file, config_tpl, tmp_file, vnfd_mock) - opnfv_vnf.get_config_tpl_data = mock.MagicMock() - opnfv_vnf.socket = 0 - opnfv_vnf.start_core = 0 - opnfv_vnf.update_write_parser = mock.MagicMock() - self.assertIsNone(opnfv_vnf.update_timer()) - - def test_generate_script(self): - topology_file = mock.Mock() - config_tpl = mock.Mock() - tmp_file = mock.Mock() - vnfd_mock = VnfdHelper(self.VNFD_0) - opnfv_vnf = samplevnf_helper.MultiPortConfig( - topology_file, config_tpl, tmp_file, vnfd_mock) - opnfv_vnf.get_config_tpl_data = mock.MagicMock() - opnfv_vnf.socket = 0 - opnfv_vnf.start_core = 0 - opnfv_vnf.update_write_parser = mock.MagicMock() - opnfv_vnf.generate_script_data = \ - mock.Mock(return_value={'link_config': 0, 'arp_config': '', - 'arp_config6': '', 'actions': '', - 'arp_route_tbl': '', 'arp_route_tbl6': '', - 'rules': ''}) - opnfv_vnf.port_pair_list = [("xe0", "xe1")] - self.assertIsNotNone(opnfv_vnf.generate_script(self.VNFD)) - opnfv_vnf.lb_config = 'HW' - self.assertIsNotNone(opnfv_vnf.generate_script(self.VNFD)) - - def test_generate_script_data(self): - topology_file = mock.Mock() - config_tpl = mock.Mock() - tmp_file = mock.Mock() - vnfd_mock = mock.MagicMock() - opnfv_vnf = samplevnf_helper.MultiPortConfig( - topology_file, config_tpl, tmp_file, vnfd_mock) - opnfv_vnf.get_config_tpl_data = mock.MagicMock() - opnfv_vnf.socket = 0 - opnfv_vnf.start_core = 0 - opnfv_vnf.update_write_parser = mock.MagicMock() - opnfv_vnf.port_pair_list = [("xe0", "xe1")] - opnfv_vnf.vnf_type = 'ACL' - opnfv_vnf.generate_link_config = mock.Mock() - opnfv_vnf.generate_arp_config = mock.Mock() - opnfv_vnf.generate_arp_config6 = mock.Mock() - opnfv_vnf.generate_action_config = mock.Mock() - opnfv_vnf.generate_rule_config = mock.Mock() - self.assertIsNotNone(opnfv_vnf.generate_script_data()) - - def test_generate_rule_config(self): - topology_file = mock.Mock() - config_tpl = mock.Mock() - tmp_file = mock.Mock() - vnfd_mock = mock.MagicMock() - opnfv_vnf = samplevnf_helper.MultiPortConfig( - topology_file, config_tpl, tmp_file, vnfd_mock) - opnfv_vnf.get_config_tpl_data = mock.MagicMock() - opnfv_vnf.socket = 0 - opnfv_vnf.start_core = 0 - opnfv_vnf.update_write_parser = mock.MagicMock() - opnfv_vnf.generate_script_data = \ - mock.Mock(return_value={'link_config': 0, 'arp_config': '', - 'arp_config6': '', 'actions': '', - 'rules': ''}) - opnfv_vnf.port_pair_list = [("xe0", "xe1")] - opnfv_vnf.get_port_pairs = mock.Mock() - opnfv_vnf.vnf_type = 'ACL' - opnfv_vnf.get_ports_gateway = mock.Mock(return_value=u'1.1.1.1') - opnfv_vnf.get_netmask_gateway = mock.Mock( - return_value=u'255.255.255.0') - opnfv_vnf.get_ports_gateway6 = mock.Mock(return_value=u'1.1.1.1') - opnfv_vnf.get_netmask_gateway6 = mock.Mock( - return_value=u'255.255.255.0') - opnfv_vnf.txrx_pipeline = '' - opnfv_vnf.vnfd = self.VNFD['vnfd:vnfd-catalog']['vnfd'][0] - opnfv_vnf.interfaces = opnfv_vnf.vnfd['vdu'][0]['external-interface'] - opnfv_vnf.rules = '' - self.assertIsNotNone(opnfv_vnf.generate_rule_config()) - opnfv_vnf.rules = 'new' - self.assertIsNotNone(opnfv_vnf.generate_rule_config()) - - def test_generate_action_config(self): - topology_file = mock.Mock() - config_tpl = mock.Mock() - tmp_file = mock.Mock() - vnfd_mock = mock.MagicMock() - opnfv_vnf = samplevnf_helper.MultiPortConfig( - topology_file, config_tpl, tmp_file, vnfd_mock) - opnfv_vnf.get_config_tpl_data = mock.MagicMock() - opnfv_vnf.socket = 0 - opnfv_vnf.start_core = 0 - opnfv_vnf.update_write_parser = mock.MagicMock() - opnfv_vnf.generate_script_data = \ - mock.Mock(return_value={'link_config': 0, 'arp_config': '', - 'arp_config6': '', 'actions': '', - 'rules': ''}) - opnfv_vnf.port_pair_list = [("xe0", "xe1")] - opnfv_vnf.get_port_pairs = mock.Mock() - opnfv_vnf.vnf_type = 'VFW' - opnfv_vnf.get_ports_gateway = mock.Mock(return_value=u'1.1.1.1') - opnfv_vnf.get_netmask_gateway = mock.Mock( - return_value=u'255.255.255.0') - opnfv_vnf.get_ports_gateway6 = mock.Mock(return_value=u'1.1.1.1') - opnfv_vnf.get_netmask_gateway6 = mock.Mock( - return_value=u'255.255.255.0') - opnfv_vnf.txrx_pipeline = '' - opnfv_vnf.rules = '' - self.assertIsNotNone(opnfv_vnf.generate_action_config()) - - def test_generate_arp_config6(self): - topology_file = mock.Mock() - config_tpl = mock.Mock() - tmp_file = mock.Mock() - vnfd_mock = mock.MagicMock() - opnfv_vnf = samplevnf_helper.MultiPortConfig( - topology_file, config_tpl, tmp_file, vnfd_mock) - opnfv_vnf.get_config_tpl_data = mock.MagicMock() - opnfv_vnf.socket = 0 - opnfv_vnf.start_core = 0 - opnfv_vnf.update_write_parser = mock.MagicMock() - opnfv_vnf.generate_script_data = \ - mock.Mock(return_value={'link_config': 0, 'arp_config': '', - 'arp_config6': '', 'actions': '', - 'rules': ''}) - opnfv_vnf.port_pair_list = [("xe0", "xe1")] - opnfv_vnf.get_port_pairs = mock.Mock() - opnfv_vnf.vnf_type = 'VFW' - opnfv_vnf.get_ports_gateway = mock.Mock(return_value=u'1.1.1.1') - opnfv_vnf.get_netmask_gateway = mock.Mock( - return_value=u'255.255.255.0') - opnfv_vnf.get_ports_gateway6 = mock.Mock(return_value=u'1.1.1.1') - opnfv_vnf.get_netmask_gateway6 = mock.Mock( - return_value=u'255.255.255.0') - opnfv_vnf.txrx_pipeline = '' - opnfv_vnf.rules = '' - opnfv_vnf.interfaces = mock.MagicMock() - opnfv_vnf.get_ports_gateway6 = mock.Mock() - self.assertIsNotNone(opnfv_vnf.generate_arp_config6()) - - def test_generate_arp_config(self): - topology_file = mock.Mock() - config_tpl = mock.Mock() - tmp_file = mock.Mock() - vnfd_mock = mock.MagicMock() - opnfv_vnf = samplevnf_helper.MultiPortConfig( - topology_file, config_tpl, tmp_file, vnfd_mock) - opnfv_vnf.get_config_tpl_data = mock.MagicMock() - opnfv_vnf.socket = 0 - opnfv_vnf.start_core = 0 - opnfv_vnf.update_write_parser = mock.MagicMock() - opnfv_vnf.generate_script_data = \ - mock.Mock(return_value={'link_config': 0, 'arp_config': '', - 'arp_config6': '', 'actions': '', - 'rules': ''}) - opnfv_vnf.port_pair_list = [("xe0", "xe1")] - opnfv_vnf.get_port_pairs = mock.Mock() - opnfv_vnf.vnf_type = 'VFW' - opnfv_vnf.get_ports_gateway = mock.Mock(return_value=u'1.1.1.1') - opnfv_vnf.get_netmask_gateway = mock.Mock( - return_value=u'255.255.255.0') - opnfv_vnf.get_ports_gateway6 = mock.Mock(return_value=u'1.1.1.1') - opnfv_vnf.get_netmask_gateway6 = mock.Mock( - return_value=u'255.255.255.0') - opnfv_vnf.txrx_pipeline = '' - opnfv_vnf.rules = '' - opnfv_vnf.interfaces = mock.MagicMock() - opnfv_vnf.get_ports_gateway6 = mock.Mock() - self.assertIsNotNone(opnfv_vnf.generate_arp_config()) - - def test_get_ports_gateway(self): - topology_file = mock.Mock() - config_tpl = mock.Mock() - tmp_file = mock.Mock() - vnfd_mock = mock.MagicMock() - opnfv_vnf = samplevnf_helper.MultiPortConfig( - topology_file, config_tpl, tmp_file, vnfd_mock) - opnfv_vnf.get_config_tpl_data = mock.MagicMock() - opnfv_vnf.socket = 0 - opnfv_vnf.start_core = 0 - opnfv_vnf.update_write_parser = mock.MagicMock() - opnfv_vnf.generate_script_data = \ - mock.Mock(return_value={'link_config': 0, 'arp_config': '', - 'arp_config6': '', 'actions': '', - 'rules': ''}) - opnfv_vnf.port_pair_list = [("xe0", "xe1")] - opnfv_vnf.get_port_pairs = mock.Mock() - opnfv_vnf.vnf_type = 'VFW' - opnfv_vnf.txrx_pipeline = '' - opnfv_vnf.rules = '' - opnfv_vnf.interfaces = mock.MagicMock() - opnfv_vnf.get_ports_gateway6 = mock.Mock() - opnfv_vnf.vnfd = self.VNFD['vnfd:vnfd-catalog']['vnfd'][0] - self.assertIsNotNone(opnfv_vnf.get_ports_gateway('xe0')) - - def test_get_ports_gateway6(self): - topology_file = mock.Mock() - config_tpl = mock.Mock() - tmp_file = mock.Mock() - vnfd_mock = mock.MagicMock() - opnfv_vnf = samplevnf_helper.MultiPortConfig( - topology_file, config_tpl, tmp_file, vnfd_mock) - opnfv_vnf.get_config_tpl_data = mock.MagicMock() - opnfv_vnf.socket = 0 - opnfv_vnf.start_core = 0 - opnfv_vnf.update_write_parser = mock.MagicMock() - opnfv_vnf.generate_script_data = \ - mock.Mock(return_value={'link_config': 0, 'arp_config': '', - 'arp_config6': '', 'actions': '', - 'rules': ''}) - opnfv_vnf.port_pair_list = [("xe0", "xe1")] - opnfv_vnf.get_port_pairs = mock.Mock() - opnfv_vnf.vnf_type = 'VFW' - opnfv_vnf.txrx_pipeline = '' - opnfv_vnf.rules = '' - opnfv_vnf.interfaces = mock.MagicMock() - opnfv_vnf.get_ports_gateway6 = mock.Mock() - opnfv_vnf.vnfd = self.VNFD['vnfd:vnfd-catalog']['vnfd'][0] - self.assertIsNotNone(opnfv_vnf.get_ports_gateway6('xe0')) - - def test_get_netmask_gateway(self): - topology_file = mock.Mock() - config_tpl = mock.Mock() - tmp_file = mock.Mock() - vnfd_mock = mock.MagicMock() - opnfv_vnf = samplevnf_helper.MultiPortConfig( - topology_file, config_tpl, tmp_file, vnfd_mock) - opnfv_vnf.get_config_tpl_data = mock.MagicMock() - opnfv_vnf.socket = 0 - opnfv_vnf.start_core = 0 - opnfv_vnf.update_write_parser = mock.MagicMock() - opnfv_vnf.generate_script_data = \ - mock.Mock(return_value={'link_config': 0, 'arp_config': '', - 'arp_config6': '', 'actions': '', - 'rules': ''}) - opnfv_vnf.port_pair_list = [("xe0", "xe1")] - opnfv_vnf.get_port_pairs = mock.Mock() - opnfv_vnf.vnf_type = 'VFW' - opnfv_vnf.txrx_pipeline = '' - opnfv_vnf.rules = '' - opnfv_vnf.interfaces = mock.MagicMock() - opnfv_vnf.get_ports_gateway6 = mock.Mock() - opnfv_vnf.vnfd = self.VNFD['vnfd:vnfd-catalog']['vnfd'][0] - self.assertIsNotNone(opnfv_vnf.get_netmask_gateway('xe0')) - - def test_get_netmask_gateway6(self): - topology_file = mock.Mock() - config_tpl = mock.Mock() - tmp_file = mock.Mock() - vnfd_mock = mock.MagicMock() - opnfv_vnf = samplevnf_helper.MultiPortConfig( - topology_file, config_tpl, tmp_file, vnfd_mock) - opnfv_vnf.get_config_tpl_data = mock.MagicMock() - opnfv_vnf.socket = 0 - opnfv_vnf.start_core = 0 - opnfv_vnf.update_write_parser = mock.MagicMock() - opnfv_vnf.generate_script_data = \ - mock.Mock(return_value={'link_config': 0, 'arp_config': '', - 'arp_config6': '', 'actions': '', - 'rules': ''}) - opnfv_vnf.port_pair_list = [("xe0", "xe1")] - opnfv_vnf.get_port_pairs = mock.Mock() - opnfv_vnf.vnf_type = 'VFW' - opnfv_vnf.txrx_pipeline = '' - opnfv_vnf.rules = '' - opnfv_vnf.interfaces = mock.MagicMock() - opnfv_vnf.get_ports_gateway6 = mock.Mock() - opnfv_vnf.vnfd = self.VNFD['vnfd:vnfd-catalog']['vnfd'][0] - self.assertIsNotNone(opnfv_vnf.get_netmask_gateway6('xe0')) - - def test_generate_link_config(self): - topology_file = mock.Mock() - config_tpl = mock.Mock() - tmp_file = mock.Mock() - vnfd_mock = mock.MagicMock() - - opnfv_vnf = samplevnf_helper.MultiPortConfig( - topology_file, config_tpl, tmp_file, vnfd_mock) - opnfv_vnf.get_config_tpl_data = mock.MagicMock() - opnfv_vnf.socket = 0 - opnfv_vnf.start_core = 0 - opnfv_vnf.update_write_parser = mock.MagicMock() - opnfv_vnf.generate_script_data = \ - mock.Mock(return_value={'link_config': 0, 'arp_config': '', - 'arp_config6': '', 'actions': '', - 'rules': ''}) - opnfv_vnf.port_pair_list = [("xe0", "xe1")] - opnfv_vnf.get_port_pairs = mock.Mock() - opnfv_vnf.vnf_type = 'VFW' - opnfv_vnf.txrx_pipeline = '' - opnfv_vnf.rules = '' - opnfv_vnf.get_ports_gateway6 = mock.Mock() - opnfv_vnf.vnfd = self.VNFD['vnfd:vnfd-catalog']['vnfd'][0] - opnfv_vnf.interfaces = opnfv_vnf.vnfd['vdu'][0]['external-interface'] - opnfv_vnf.all_ports = ['32', '1', '987'] - opnfv_vnf.validate_ip_and_prefixlen = mock.Mock( - return_value=('10.20.30.40', 16)) - - result = opnfv_vnf.generate_link_config() - self.assertEqual(len(result.splitlines()), 9) - - def test_generate_config(self): - topology_file = mock.Mock() - config_tpl = mock.Mock() - tmp_file = mock.Mock() - vnfd_mock = mock.MagicMock() - opnfv_vnf = samplevnf_helper.MultiPortConfig( - topology_file, config_tpl, tmp_file, vnfd_mock) - opnfv_vnf.get_config_tpl_data = mock.MagicMock() - opnfv_vnf.socket = 0 - opnfv_vnf.start_core = 0 - opnfv_vnf.update_write_parser = mock.MagicMock() - opnfv_vnf.generate_script_data = \ - mock.Mock(return_value={'link_config': 0, 'arp_config': '', - 'arp_config6': '', 'actions': '', - 'rules': ''}) - opnfv_vnf.port_pair_list = [("xe0", "xe1")] - opnfv_vnf.txrx_pipeline = '' - opnfv_vnf.rules = '' - opnfv_vnf.get_ports_gateway6 = mock.Mock() - opnfv_vnf.vnfd = self.VNFD['vnfd:vnfd-catalog']['vnfd'][0] - opnfv_vnf.interfaces = opnfv_vnf.vnfd['vdu'][0]['external-interface'] - opnfv_vnf.generate_lb_to_port_pair_mapping = mock.Mock() - opnfv_vnf.generate_config_data = mock.Mock() - opnfv_vnf.write_parser = mock.MagicMock() - opnfv_vnf.is_openstack = True - self.assertIsNone(opnfv_vnf.generate_config()) - opnfv_vnf.is_openstack = False - self.assertIsNone(opnfv_vnf.generate_config()) - - def test_get_config_tpl_data(self): - topology_file = mock.Mock() - config_tpl = mock.Mock() - tmp_file = mock.Mock() - vnfd_mock = mock.MagicMock() - opnfv_vnf = samplevnf_helper.MultiPortConfig( - topology_file, config_tpl, tmp_file, vnfd_mock) - opnfv_vnf.socket = 0 - opnfv_vnf.start_core = 0 - opnfv_vnf.port_pair_list = [("xe0", "xe1")] - opnfv_vnf.txrx_pipeline = '' - opnfv_vnf.rules = '' - opnfv_vnf.write_parser = mock.MagicMock() - opnfv_vnf.read_parser = mock.MagicMock() - opnfv_vnf.read_parser.sections = mock.Mock(return_value=['MASTER']) - opnfv_vnf.read_parser.has_option = mock.Mock(return_value=True) - opnfv_vnf.read_parser.get = mock.Mock(return_value='filename') - - self.assertIsNotNone(opnfv_vnf.get_config_tpl_data('filename')) - - def test_get_txrx_tpl_data(self): - topology_file = mock.Mock() - config_tpl = mock.Mock() - tmp_file = mock.Mock() - vnfd_mock = mock.MagicMock() - opnfv_vnf = samplevnf_helper.MultiPortConfig( - topology_file, config_tpl, tmp_file, vnfd_mock) - opnfv_vnf.socket = 0 - opnfv_vnf.start_core = 0 - opnfv_vnf.port_pair_list = [("xe0", "xe1")] - opnfv_vnf.txrx_pipeline = '' - opnfv_vnf.rules = '' - opnfv_vnf.write_parser = mock.MagicMock() - opnfv_vnf.read_parser = mock.MagicMock() - opnfv_vnf.read_parser.sections = mock.Mock(return_value=['MASTER']) - opnfv_vnf.read_parser.has_option = mock.Mock(return_value=True) - opnfv_vnf.read_parser.get = mock.Mock(return_value='filename') - - self.assertIsNotNone(opnfv_vnf.get_txrx_tpl_data('filename')) - - def test_init_write_parser_template(self): - topology_file = mock.Mock() - config_tpl = mock.Mock() - tmp_file = mock.Mock() - vnfd_mock = mock.MagicMock() - opnfv_vnf = samplevnf_helper.MultiPortConfig( - topology_file, config_tpl, tmp_file, vnfd_mock) - opnfv_vnf.socket = 0 - opnfv_vnf.start_core = 0 - opnfv_vnf.port_pair_list = [("xe0", "xe1")] - opnfv_vnf.txrx_pipeline = '' - opnfv_vnf.rules = '' - opnfv_vnf.write_parser = mock.MagicMock() - opnfv_vnf.read_parser = mock.MagicMock() - opnfv_vnf.read_parser.sections = mock.Mock(return_value=['MASTER']) - opnfv_vnf.read_parser.has_option = mock.Mock(return_value=True) - opnfv_vnf.read_parser.get = mock.Mock(return_value='filename') - - self.assertIsNone(opnfv_vnf.init_write_parser_template('filename')) - opnfv_vnf.write_parser.add_section = mock.MagicMock() - opnfv_vnf.read_parser.item = mock.Mock(return_value=[1, 2, 3]) - opnfv_vnf.read_parser.has_option = mock.Mock(return_value=False) - opnfv_vnf.write_parser.set = mock.Mock() - self.assertIsNone(opnfv_vnf.init_write_parser_template('filename')) - - def test_init_write_parser_template_2(self): - topology_file = mock.Mock() - config_tpl = mock.Mock() - tmp_file = mock.Mock() - vnfd_mock = mock.MagicMock() - opnfv_vnf = samplevnf_helper.MultiPortConfig( - topology_file, config_tpl, tmp_file, vnfd_mock) - opnfv_vnf.socket = 0 - opnfv_vnf.start_core = 0 - opnfv_vnf.port_pair_list = [("xe0", "xe1")] - opnfv_vnf.txrx_pipeline = '' - opnfv_vnf.rules = '' - opnfv_vnf.write_parser = mock.MagicMock() - opnfv_vnf.read_parser = mock.MagicMock() - opnfv_vnf.read_parser.sections = mock.Mock(return_value=['MASTER']) - opnfv_vnf.read_parser.has_option = mock.Mock(return_value=[]) - opnfv_vnf.write_parser.set = mock.Mock() - opnfv_vnf.read_parser.items = mock.MagicMock() - self.assertIsNone(opnfv_vnf.init_write_parser_template('filename')) - - def test_update_write_parser(self): - topology_file = mock.Mock() - config_tpl = mock.Mock() - tmp_file = mock.Mock() - vnfd_mock = mock.MagicMock() - opnfv_vnf = samplevnf_helper.MultiPortConfig( - topology_file, config_tpl, tmp_file, vnfd_mock) - opnfv_vnf.socket = 0 - opnfv_vnf.start_core = 0 - opnfv_vnf.port_pair_list = [("xe0", "xe1")] - opnfv_vnf.txrx_pipeline = '' - opnfv_vnf.rules = '' - opnfv_vnf.write_parser = mock.MagicMock() - opnfv_vnf.read_parser = mock.MagicMock() - opnfv_vnf.read_parser.sections = mock.Mock(return_value=['MASTER']) - opnfv_vnf.read_parser.has_option = mock.Mock(return_value=[]) - opnfv_vnf.write_parser.set = mock.Mock() - opnfv_vnf.write_parser.add_section = mock.Mock() - opnfv_vnf.read_parser.items = mock.MagicMock() - opnfv_vnf.pipeline_counter = 0 - self.assertIsNone(opnfv_vnf.update_write_parser({'filename': 1})) - - def test_get_worker_threads(self): - topology_file = mock.Mock() - config_tpl = mock.Mock() - tmp_file = mock.Mock() - vnfd_mock = mock.MagicMock() - opnfv_vnf = samplevnf_helper.MultiPortConfig( - topology_file, config_tpl, tmp_file, vnfd_mock) - opnfv_vnf.socket = 0 - opnfv_vnf.start_core = 0 - opnfv_vnf.port_pair_list = [("xe0", "xe1")] - opnfv_vnf.txrx_pipeline = '' - opnfv_vnf.rules = '' - opnfv_vnf.write_parser = mock.MagicMock() - opnfv_vnf.read_parser = mock.MagicMock() - opnfv_vnf.read_parser.sections = mock.Mock(return_value=['MASTER']) - opnfv_vnf.read_parser.has_option = mock.Mock(return_value=[]) - opnfv_vnf.write_parser.set = mock.Mock() - opnfv_vnf.write_parser.add_section = mock.Mock() - opnfv_vnf.read_parser.items = mock.MagicMock() - opnfv_vnf.pipeline_counter = 0 - opnfv_vnf.worker_config = '1t' - result = opnfv_vnf.get_worker_threads(1) - self.assertEqual(1, result) - opnfv_vnf.worker_config = '2t' - result = opnfv_vnf.get_worker_threads(2) - self.assertEqual(2, result) - opnfv_vnf.worker_config = '2t' - result = opnfv_vnf.get_worker_threads(3) - self.assertEqual(2, result) - - # TODO(elfoley): Split this test into smaller tests - def test_generate_next_core_id(self): - topology_file = mock.Mock() - config_tpl = mock.Mock() - tmp_file = mock.Mock() - vnfd_mock = mock.MagicMock() - opnfv_vnf = samplevnf_helper.MultiPortConfig( - topology_file, config_tpl, tmp_file, vnfd_mock) - opnfv_vnf.socket = 0 - opnfv_vnf.start_core = 0 - opnfv_vnf.port_pair_list = [("xe0", "xe1")] - opnfv_vnf.txrx_pipeline = '' - opnfv_vnf.rules = '' - opnfv_vnf.write_parser = mock.MagicMock() - opnfv_vnf.read_parser = mock.MagicMock() - opnfv_vnf.read_parser.sections = mock.Mock(return_value=['MASTER']) - opnfv_vnf.read_parser.has_option = mock.Mock(return_value=[]) - opnfv_vnf.write_parser.set = mock.Mock() - opnfv_vnf.write_parser.add_section = mock.Mock() - opnfv_vnf.read_parser.items = mock.MagicMock() - opnfv_vnf.pipeline_counter = 0 - opnfv_vnf.worker_config = '1t' - opnfv_vnf.start_core = 0 - result = opnfv_vnf.generate_next_core_id() - self.assertIsNone(result) - opnfv_vnf.worker_config = '2t' - opnfv_vnf.start_core = 'a' - self.assertRaises(ValueError, opnfv_vnf.generate_next_core_id) - opnfv_vnf.worker_config = '2t' - opnfv_vnf.start_core = 1 - result = opnfv_vnf.generate_next_core_id() - self.assertIsNone(result) - - def test_generate_lb_to_port_pair_mapping(self): - topology_file = mock.Mock() - config_tpl = mock.Mock() - tmp_file = mock.Mock() - vnfd_mock = VnfdHelper(self.VNFD_0) - opnfv_vnf = samplevnf_helper.MultiPortConfig( - topology_file, config_tpl, tmp_file, vnfd_mock) - opnfv_vnf.socket = 0 - opnfv_vnf.start_core = 0 - opnfv_vnf.txrx_pipeline = '' - opnfv_vnf.rules = '' - opnfv_vnf.write_parser = mock.MagicMock() - opnfv_vnf.read_parser = mock.MagicMock() - opnfv_vnf.read_parser.sections = mock.Mock(return_value=['MASTER']) - opnfv_vnf.read_parser.has_option = mock.Mock(return_value=[]) - opnfv_vnf.write_parser.set = mock.Mock() - opnfv_vnf.write_parser.add_section = mock.Mock() - opnfv_vnf.read_parser.items = mock.MagicMock() - opnfv_vnf.pipeline_counter = 0 - opnfv_vnf.worker_config = '1t' - opnfv_vnf.start_core = 0 - opnfv_vnf.lb_count = 1 - opnfv_vnf._port_pairs = samplevnf_helper.PortPairs(vnfd_mock.interfaces) - opnfv_vnf.port_pair_list = opnfv_vnf._port_pairs.port_pair_list - result = opnfv_vnf.generate_lb_to_port_pair_mapping() - self.assertIsNone(result) - result = opnfv_vnf.set_priv_to_pub_mapping() - self.assertEqual('(0,1)', result) - - def test_set_priv_que_handler(self): - topology_file = mock.Mock() - config_tpl = mock.Mock() - tmp_file = mock.Mock() - vnfd_mock = VnfdHelper(self.VNFD_0) - opnfv_vnf = samplevnf_helper.MultiPortConfig( - topology_file, config_tpl, tmp_file, vnfd_mock) - opnfv_vnf.socket = 0 - opnfv_vnf.start_core = 0 - opnfv_vnf.port_pair_list = [("xe0", "xe1")] - opnfv_vnf.port_pairs = [("xe0", "xe1")] - opnfv_vnf.txrx_pipeline = '' - opnfv_vnf.rules = '' - opnfv_vnf.write_parser = mock.MagicMock() - opnfv_vnf.read_parser = mock.MagicMock() - opnfv_vnf.read_parser.sections = mock.Mock(return_value=['MASTER']) - opnfv_vnf.read_parser.has_option = mock.Mock(return_value=[]) - opnfv_vnf.write_parser.set = mock.Mock() - opnfv_vnf.write_parser.add_section = mock.Mock() - opnfv_vnf.read_parser.items = mock.MagicMock() - opnfv_vnf.pipeline_counter = 0 - opnfv_vnf.worker_config = '1t' - opnfv_vnf.start_core = 0 - opnfv_vnf.lb_count = 1 - result = opnfv_vnf.set_priv_que_handler() - self.assertIsNone(result) - - def test_generate_arp_route_tbl(self): - # ELF: could n=do this in setup - topology_file = mock.Mock() - config_tpl = mock.Mock() - tmp_file = "" - vnfd_mock = mock.MagicMock() - vnfd_mock.port_num.side_effect = ['32', '1', '987'] - vnfd_mock.find_interface.side_effect = [ - { - 'virtual-interface': { - 'dst_ip': '10.20.30.40', - 'netmask': '20', - }, - }, - { - 'virtual-interface': { - 'dst_ip': '10.200.30.40', - 'netmask': '24', - }, - }, - { - 'virtual-interface': { - 'dst_ip': '10.20.3.40', - 'netmask': '8', - }, - }, - ] - - opnfv_vnf = samplevnf_helper.MultiPortConfig( - topology_file, config_tpl, tmp_file, vnfd_mock) - opnfv_vnf.all_ports = [3, 2, 5] - - expected = 'routeadd net 32 10.20.30.40 0xfffff000\n' \ - 'routeadd net 1 10.200.30.40 0xffffff00\n' \ - 'routeadd net 987 10.20.3.40 0xff000000' - result = opnfv_vnf.generate_arp_route_tbl() - self.assertEqual(result, expected) - - def test_generate_arpicmp_data(self): - topology_file = mock.Mock() - config_tpl = mock.Mock() - tmp_file = mock.Mock() - vnfd_mock = mock.MagicMock() - opnfv_vnf = samplevnf_helper.MultiPortConfig( - topology_file, config_tpl, tmp_file, vnfd_mock) - opnfv_vnf.socket = 0 - opnfv_vnf.start_core = 0 - opnfv_vnf.port_pair_list = [("xe0", "xe1")] - opnfv_vnf.port_pairs = [("xe0", "xe1")] - opnfv_vnf.txrx_pipeline = '' - opnfv_vnf.rules = '' - opnfv_vnf.write_parser = mock.MagicMock() - opnfv_vnf.read_parser = mock.MagicMock() - opnfv_vnf.read_parser.sections = mock.Mock(return_value=['MASTER']) - opnfv_vnf.read_parser.has_option = mock.Mock(return_value=[]) - opnfv_vnf.write_parser.set = mock.Mock() - opnfv_vnf.write_parser.add_section = mock.Mock() - opnfv_vnf.read_parser.items = mock.MagicMock() - opnfv_vnf.pipeline_counter = 0 - opnfv_vnf.worker_config = '1t' - opnfv_vnf.start_core = 0 - opnfv_vnf.lb_count = 1 - opnfv_vnf.vnfd = self.VNFD['vnfd:vnfd-catalog']['vnfd'][0] - opnfv_vnf.interfaces = opnfv_vnf.vnfd['vdu'][0]['external-interface'] - result = opnfv_vnf.generate_arpicmp_data() - self.assertIsNotNone(result) - opnfv_vnf.nfv_type = 'ovs' - opnfv_vnf.lb_to_port_pair_mapping = [0, 1] - result = opnfv_vnf.generate_arpicmp_data() - self.assertIsNotNone(result) - opnfv_vnf.nfv_type = 'openstack' - opnfv_vnf.lb_to_port_pair_mapping = [0, 1] - result = opnfv_vnf.generate_arpicmp_data() - self.assertIsNotNone(result) - opnfv_vnf.lb_config = 'HW' - opnfv_vnf.lb_to_port_pair_mapping = [0, 1] - result = opnfv_vnf.generate_arpicmp_data() - self.assertIsNotNone(result) - - def test_generate_final_txrx_data(self): - topology_file = mock.Mock() - config_tpl = mock.Mock() - tmp_file = mock.Mock() - vnfd_mock = mock.MagicMock() - opnfv_vnf = samplevnf_helper.MultiPortConfig( - topology_file, config_tpl, tmp_file, vnfd_mock) - opnfv_vnf.socket = 0 - opnfv_vnf.start_core = 0 - opnfv_vnf.port_pair_list = [("xe0", "xe1")] - opnfv_vnf.port_pairs = [("xe0", "xe1")] - opnfv_vnf.txrx_pipeline = '' - opnfv_vnf.rules = '' - opnfv_vnf.write_parser = mock.MagicMock() - opnfv_vnf.read_parser = mock.MagicMock() - opnfv_vnf.read_parser.sections = mock.Mock(return_value=['MASTER']) - opnfv_vnf.read_parser.has_option = mock.Mock(return_value=[]) - opnfv_vnf.write_parser.set = mock.Mock() - opnfv_vnf.write_parser.add_section = mock.Mock() - opnfv_vnf.read_parser.items = mock.MagicMock() - opnfv_vnf.pipeline_counter = 0 - opnfv_vnf.worker_config = '1t' - opnfv_vnf.start_core = 0 - opnfv_vnf.lb_count = 1 - opnfv_vnf.vnfd = self.VNFD['vnfd:vnfd-catalog']['vnfd'][0] - opnfv_vnf.interfaces = opnfv_vnf.vnfd['vdu'][0]['external-interface'] - opnfv_vnf.lb_to_port_pair_mapping = [0, 1] - opnfv_vnf.ports_len = 2 - opnfv_vnf.lb_index = 1 - opnfv_vnf.pktq_out_os = [1, 2] - result = opnfv_vnf.generate_final_txrx_data() - self.assertIsNotNone(result) - opnfv_vnf.nfv_type = 'openstack' - opnfv_vnf.pktq_out_os = [1, 2] - opnfv_vnf.lb_index = 1 - result = opnfv_vnf.generate_final_txrx_data() - self.assertIsNotNone(result) - - def test_generate_initial_txrx_data(self): - topology_file = mock.Mock() - config_tpl = mock.Mock() - tmp_file = mock.Mock() - vnfd_mock = mock.MagicMock() - opnfv_vnf = samplevnf_helper.MultiPortConfig( - topology_file, config_tpl, tmp_file, vnfd_mock) - opnfv_vnf.socket = 0 - opnfv_vnf.start_core = 0 - opnfv_vnf.port_pair_list = [("xe0", "xe1")] - opnfv_vnf.port_pairs = [("xe0", "xe1")] - opnfv_vnf.txrx_pipeline = '' - opnfv_vnf.rules = '' - opnfv_vnf.write_parser = mock.MagicMock() - opnfv_vnf.read_parser = mock.MagicMock() - opnfv_vnf.read_parser.sections = mock.Mock(return_value=['MASTER']) - opnfv_vnf.read_parser.has_option = mock.Mock(return_value=[]) - opnfv_vnf.write_parser.set = mock.Mock() - opnfv_vnf.write_parser.add_section = mock.Mock() - opnfv_vnf.read_parser.items = mock.MagicMock() - opnfv_vnf.pipeline_counter = 0 - opnfv_vnf.worker_config = '1t' - opnfv_vnf.start_core = 0 - opnfv_vnf.lb_count = 1 - opnfv_vnf.vnfd = self.VNFD['vnfd:vnfd-catalog']['vnfd'][0] - opnfv_vnf.interfaces = opnfv_vnf.vnfd['vdu'][0]['external-interface'] - opnfv_vnf.lb_to_port_pair_mapping = [0, 1] - opnfv_vnf.lb_index = 1 - opnfv_vnf.ports_len = 2 - result = opnfv_vnf.generate_initial_txrx_data() - self.assertIsNotNone(result) - opnfv_vnf.nfv_type = 'openstack' - opnfv_vnf.pktq_out_os = [1, 2] - result = opnfv_vnf.generate_initial_txrx_data() - self.assertIsNotNone(result) - opnfv_vnf.nfv_type = 'ovs' - opnfv_vnf.init_ovs = False - opnfv_vnf.ovs_pktq_out = '' - opnfv_vnf.pktq_out_os = [1, 2] - opnfv_vnf.lb_index = 1 - result = opnfv_vnf.generate_initial_txrx_data() - self.assertIsNotNone(result) - opnfv_vnf.nfv_type = 'ovs' - opnfv_vnf.init_ovs = True - opnfv_vnf.pktq_out_os = [1, 2] - opnfv_vnf.ovs_pktq_out = '' - opnfv_vnf.lb_index = 1 - result = opnfv_vnf.generate_initial_txrx_data() - self.assertIsNotNone(result) - - def test_generate_lb_data(self): - topology_file = mock.Mock() - config_tpl = mock.Mock() - tmp_file = mock.Mock() - vnfd_mock = mock.MagicMock() - opnfv_vnf = samplevnf_helper.MultiPortConfig( - topology_file, config_tpl, tmp_file, vnfd_mock) - opnfv_vnf.socket = 0 - opnfv_vnf.start_core = 0 - opnfv_vnf.port_pair_list = [("xe0", "xe1")] - opnfv_vnf.port_pairs = [("xe0", "xe1")] - opnfv_vnf.txrx_pipeline = '' - opnfv_vnf.rules = '' - opnfv_vnf.write_parser = mock.MagicMock() - opnfv_vnf.read_parser = mock.MagicMock() - opnfv_vnf.read_parser.sections = mock.Mock(return_value=['MASTER']) - opnfv_vnf.read_parser.has_option = mock.Mock(return_value=[]) - opnfv_vnf.write_parser.set = mock.Mock() - opnfv_vnf.write_parser.add_section = mock.Mock() - opnfv_vnf.read_parser.items = mock.MagicMock() - opnfv_vnf.pipeline_counter = 0 - opnfv_vnf.worker_config = '1t' - opnfv_vnf.start_core = 0 - opnfv_vnf.lb_count = 1 - opnfv_vnf.vnfd = self.VNFD['vnfd:vnfd-catalog']['vnfd'][0] - opnfv_vnf.interfaces = opnfv_vnf.vnfd['vdu'][0]['external-interface'] - opnfv_vnf.lb_to_port_pair_mapping = [0, 1] - opnfv_vnf.lb_index = 1 - opnfv_vnf.ports_len = 2 - opnfv_vnf.prv_que_handler = 0 - result = opnfv_vnf.generate_lb_data() - self.assertIsNotNone(result) - - def test_generate_vnf_data(self): - topology_file = mock.Mock() - config_tpl = mock.Mock() - tmp_file = mock.Mock() - vnfd_mock = mock.MagicMock() - opnfv_vnf = samplevnf_helper.MultiPortConfig( - topology_file, config_tpl, tmp_file, vnfd_mock) - opnfv_vnf.socket = 0 - opnfv_vnf.start_core = 0 - opnfv_vnf.port_pair_list = [("xe0", "xe1")] - opnfv_vnf.port_pairs = [("xe0", "xe1")] - opnfv_vnf.txrx_pipeline = '' - opnfv_vnf.rules = '' - opnfv_vnf.write_parser = mock.MagicMock() - opnfv_vnf.read_parser = mock.MagicMock() - opnfv_vnf.read_parser.sections = mock.Mock(return_value=['MASTER']) - opnfv_vnf.read_parser.has_option = mock.Mock(return_value=[]) - opnfv_vnf.write_parser.set = mock.Mock() - opnfv_vnf.write_parser.add_section = mock.Mock() - opnfv_vnf.read_parser.items = mock.MagicMock() - opnfv_vnf.pipeline_counter = 0 - opnfv_vnf.worker_config = '1t' - opnfv_vnf.start_core = 0 - opnfv_vnf.lb_count = 1 - opnfv_vnf.vnfd = self.VNFD['vnfd:vnfd-catalog']['vnfd'][0] - opnfv_vnf.interfaces = opnfv_vnf.vnfd['vdu'][0]['external-interface'] - opnfv_vnf.lb_to_port_pair_mapping = [0, 1] - opnfv_vnf.lb_index = 1 - opnfv_vnf.ports_len = 1 - opnfv_vnf.pktq_out = ['1', '2'] - opnfv_vnf.vnf_tpl = {'public_ip_port_range': '98164810', - 'vnf_set': '(2,4,5)'} - opnfv_vnf.prv_que_handler = 0 - result = opnfv_vnf.generate_vnf_data() - self.assertIsNotNone(result) - opnfv_vnf.lb_config = 'HW' - opnfv_vnf.mul = 0.1 - result = opnfv_vnf.generate_vnf_data() - self.assertIsNotNone(result) - opnfv_vnf.lb_config = 'HW' - opnfv_vnf.mul = 0.1 - opnfv_vnf.vnf_type = 'ACL' - result = opnfv_vnf.generate_vnf_data() - self.assertIsNotNone(result) - - def test_generate_config_data(self): - topology_file = mock.Mock() - config_tpl = mock.Mock() - tmp_file = mock.Mock() - vnfd_mock = VnfdHelper(self.VNFD_0) - opnfv_vnf = samplevnf_helper.MultiPortConfig( - topology_file, config_tpl, tmp_file, vnfd_mock) - opnfv_vnf.socket = 0 - opnfv_vnf.start_core = 0 - opnfv_vnf.port_pair_list = [("xe0", "xe1")] - opnfv_vnf.port_pairs = [("xe0", "xe1")] - opnfv_vnf.txrx_pipeline = '' - opnfv_vnf.rules = '' - opnfv_vnf.write_parser = mock.MagicMock() - opnfv_vnf.read_parser = mock.MagicMock() - opnfv_vnf.read_parser.sections = mock.Mock(return_value=['MASTER']) - opnfv_vnf.read_parser.has_option = mock.Mock(return_value=[]) - opnfv_vnf.write_parser.set = mock.Mock() - opnfv_vnf.write_parser.add_section = mock.Mock() - opnfv_vnf.read_parser.items = mock.MagicMock() - opnfv_vnf.pipeline_counter = 0 - opnfv_vnf.worker_config = '1t' - opnfv_vnf.start_core = 0 - opnfv_vnf.lb_count = 1 - opnfv_vnf.vnfd = self.VNFD['vnfd:vnfd-catalog']['vnfd'][0] - opnfv_vnf.interfaces = opnfv_vnf.vnfd['vdu'][0]['external-interface'] - opnfv_vnf.lb_to_port_pair_mapping = [0, 1] - opnfv_vnf.lb_index = 1 - opnfv_vnf.ports_len = 1 - opnfv_vnf.pktq_out = ['1', '2'] - opnfv_vnf.prv_que_handler = 0 - opnfv_vnf.init_write_parser_template = mock.Mock() - opnfv_vnf.arpicmp_tpl = mock.MagicMock() - opnfv_vnf.txrx_tpl = mock.MagicMock() - opnfv_vnf.loadb_tpl = mock.MagicMock() - opnfv_vnf.vnf_tpl = {'public_ip_port_range': '98164810 (1,65535)', - 'vnf_set': "(2,4,5)"} - opnfv_vnf.generate_vnf_data = mock.Mock(return_value={}) - opnfv_vnf.update_write_parser = mock.Mock() - result = opnfv_vnf.generate_config_data() - self.assertIsNone(result) - opnfv_vnf.generate_final_txrx_data = mock.Mock() - opnfv_vnf.update_write_parser = mock.Mock() - result = opnfv_vnf.generate_config_data() - self.assertIsNone(result) - opnfv_vnf.lb_to_port_pair_mapping = [0, 1] - opnfv_vnf.lb_index = 1 - opnfv_vnf.ports_len = 1 - opnfv_vnf.pktq_out = ['1', '2'] - opnfv_vnf.prv_que_handler = 0 - opnfv_vnf.init_write_parser_template = mock.Mock() - opnfv_vnf.arpicmp_tpl = mock.MagicMock() - opnfv_vnf.txrx_tpl = mock.MagicMock() - opnfv_vnf.loadb_tpl = mock.MagicMock() - opnfv_vnf.vnf_type = 'CGNAPT' - opnfv_vnf.update_timer = mock.Mock() - opnfv_vnf.port_pair_list = [("xe0", "xe1"), ("xe0", "xe2")] - opnfv_vnf.lb_to_port_pair_mapping = [0, 1] - opnfv_vnf.generate_arpicmp_data = mock.Mock() - result = opnfv_vnf.generate_config_data() - self.assertIsNone(result) - - def test_init_eal(self): - topology_file = mock.Mock() - config_tpl = mock.Mock() - tmp_file = mock.Mock() - vnfd_mock = mock.MagicMock() - opnfv_vnf = samplevnf_helper.MultiPortConfig( - topology_file, config_tpl, tmp_file, vnfd_mock) - opnfv_vnf.socket = 0 - opnfv_vnf.start_core = 0 - opnfv_vnf.port_pair_list = [("xe0", "xe1")] - opnfv_vnf.port_pairs = [("xe0", "xe1")] - opnfv_vnf.txrx_pipeline = '' - opnfv_vnf.rules = '' - opnfv_vnf.write_parser = mock.MagicMock() - opnfv_vnf.read_parser = mock.MagicMock() - opnfv_vnf.read_parser.sections = mock.Mock(return_value=['MASTER']) - opnfv_vnf.read_parser.has_option = mock.Mock(return_value=[]) - opnfv_vnf.write_parser.set = mock.Mock() - opnfv_vnf.write_parser.add_section = mock.Mock() - opnfv_vnf.read_parser.items = mock.MagicMock() - opnfv_vnf.pipeline_counter = 0 - opnfv_vnf.worker_config = '1t' - opnfv_vnf.start_core = 0 - opnfv_vnf.lb_count = 1 - opnfv_vnf.vnfd = self.VNFD['vnfd:vnfd-catalog']['vnfd'][0] - opnfv_vnf.interfaces = opnfv_vnf.vnfd['vdu'][0]['external-interface'] - opnfv_vnf.lb_to_port_pair_mapping = [0, 1] - opnfv_vnf.lb_index = 1 - opnfv_vnf.ports_len = 1 - opnfv_vnf.pktq_out = ['1', '2'] - opnfv_vnf.prv_que_handler = 0 - opnfv_vnf.init_write_parser_template = mock.Mock() - opnfv_vnf.arpicmp_tpl = mock.MagicMock() - opnfv_vnf.txrx_tpl = mock.MagicMock() - opnfv_vnf.loadb_tpl = mock.MagicMock() - opnfv_vnf.vnf_tpl = {'public_ip_port_range': '98164810 (1,65535)'} - opnfv_vnf.generate_vnf_data = mock.Mock(return_value={}) - opnfv_vnf.update_write_parser = mock.Mock() - opnfv_vnf.tmp_file = "/tmp/config" - result = opnfv_vnf.init_eal() - self.assertIsNone(result) |