diff options
author | Emma Foley <emma.l.foley@intel.com> | 2018-04-27 10:33:17 +0000 |
---|---|---|
committer | Emma Foley <emma.l.foley@intel.com> | 2018-05-15 17:42:47 +0000 |
commit | 065c00d3147d12c692d43179d9c1c9afadb97825 (patch) | |
tree | 6c8d0e86b0e419b9667de9da424a260883809c66 /yardstick/tests/unit | |
parent | e12f98c2bbed5191714bdc553e7b269b1437cecd (diff) |
Move tests: unit/network_services/helpers
* Fix pylint errors
* Add TODOs
Some errors are ignored locally, as they were a symptom of other problems.
These issues have been flagged with a TODO, and should be fixed later.
JIRA: YARDSTICK-837
Signed-off-by: Emma Foley <emma.l.foley@intel.com>
Change-Id: If0b77a6e0b102071ecfb212362647c62a621e4f9
Diffstat (limited to 'yardstick/tests/unit')
6 files changed, 2131 insertions, 0 deletions
diff --git a/yardstick/tests/unit/network_services/helpers/__init__.py b/yardstick/tests/unit/network_services/helpers/__init__.py new file mode 100644 index 000000000..e69de29bb --- /dev/null +++ b/yardstick/tests/unit/network_services/helpers/__init__.py diff --git a/yardstick/tests/unit/network_services/helpers/acl_vnf_topology_ixia.yaml b/yardstick/tests/unit/network_services/helpers/acl_vnf_topology_ixia.yaml new file mode 100644 index 000000000..f60834fbd --- /dev/null +++ b/yardstick/tests/unit/network_services/helpers/acl_vnf_topology_ixia.yaml @@ -0,0 +1,50 @@ +# 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/yardstick/tests/unit/network_services/helpers/test_cpu.py b/yardstick/tests/unit/network_services/helpers/test_cpu.py new file mode 100644 index 000000000..871fbf8c9 --- /dev/null +++ b/yardstick/tests/unit/network_services/helpers/test_cpu.py @@ -0,0 +1,121 @@ +# 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 division +import unittest +import mock +import subprocess + +from yardstick.network_services.helpers.cpu import \ + CpuSysCores + + +class TestCpuSysCores(unittest.TestCase): + + def setUp(self): + self._mock_ssh = mock.patch("yardstick.ssh.SSH") + self.mock_ssh = self._mock_ssh.start() + + self.addCleanup(self._cleanup) + + def _cleanup(self): + self._mock_ssh.stop() + + def test___init__(self): + self.mock_ssh.execute.return_value = 1, "", "" + self.mock_ssh.put.return_value = 1, "", "" + cpu_topo = CpuSysCores(self.mock_ssh) + 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/yardstick/tests/unit/network_services/helpers/test_dpdkbindnic_helper.py b/yardstick/tests/unit/network_services/helpers/test_dpdkbindnic_helper.py new file mode 100644 index 000000000..367072e84 --- /dev/null +++ b/yardstick/tests/unit/network_services/helpers/test_dpdkbindnic_helper.py @@ -0,0 +1,633 @@ +# 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/yardstick/tests/unit/network_services/helpers/test_iniparser.py b/yardstick/tests/unit/network_services/helpers/test_iniparser.py new file mode 100644 index 000000000..1a09f0761 --- /dev/null +++ b/yardstick/tests/unit/network_services/helpers/test_iniparser.py @@ -0,0 +1,223 @@ +# 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. +# + +import unittest +from contextlib import contextmanager +import mock + +from yardstick.tests 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): + 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): + 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/yardstick/tests/unit/network_services/helpers/test_samplevnf_helper.py b/yardstick/tests/unit/network_services/helpers/test_samplevnf_helper.py new file mode 100644 index 000000000..6d5e1da60 --- /dev/null +++ b/yardstick/tests/unit/network_services/helpers/test_samplevnf_helper.py @@ -0,0 +1,1104 @@ +# 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): + 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) |