diff options
Diffstat (limited to 'yardstick/tests/unit/network_services/vnf_generic/vnf/test_vpp_helpers.py')
-rw-r--r-- | yardstick/tests/unit/network_services/vnf_generic/vnf/test_vpp_helpers.py | 1723 |
1 files changed, 1723 insertions, 0 deletions
diff --git a/yardstick/tests/unit/network_services/vnf_generic/vnf/test_vpp_helpers.py b/yardstick/tests/unit/network_services/vnf_generic/vnf/test_vpp_helpers.py new file mode 100644 index 000000000..cca604f43 --- /dev/null +++ b/yardstick/tests/unit/network_services/vnf_generic/vnf/test_vpp_helpers.py @@ -0,0 +1,1723 @@ +# Copyright (c) 2019 Viosoft 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 ipaddress +import unittest + +import mock + +from yardstick.common import exceptions +from yardstick.network_services.helpers import cpu +from yardstick.network_services.vnf_generic.vnf import vpp_helpers +from yardstick.network_services.vnf_generic.vnf.base import VnfdHelper +from yardstick.network_services.vnf_generic.vnf.vpp_helpers import \ + VppSetupEnvHelper, VppConfigGenerator, VatTerminal + + +class TestVppConfigGenerator(unittest.TestCase): + + def test_add_config_item(self): + test_item = {} + vpp_config_generator = VppConfigGenerator() + vpp_config_generator.add_config_item(test_item, '/tmp/vpe.log', + ['unix', 'log']) + self.assertEqual({'unix': {'log': '/tmp/vpe.log'}}, test_item) + + def test_add_config_item_str(self): + test_item = {'unix': ''} + vpp_config_generator = VppConfigGenerator() + vpp_config_generator.add_config_item(test_item, '/tmp/vpe.log', + ['unix', 'log']) + self.assertEqual({'unix': {'log': '/tmp/vpe.log'}}, test_item) + + def test_add_unix_log(self): + vpp_config_generator = VppConfigGenerator() + vpp_config_generator.add_unix_log() + self.assertEqual('unix\n{\n log /tmp/vpe.log\n}\n', + vpp_config_generator.dump_config()) + + def test_add_unix_cli_listen(self): + vpp_config_generator = VppConfigGenerator() + vpp_config_generator.add_unix_cli_listen() + self.assertEqual('unix\n{\n cli-listen /run/vpp/cli.sock\n}\n', + vpp_config_generator.dump_config()) + + def test_add_unix_nodaemon(self): + vpp_config_generator = VppConfigGenerator() + vpp_config_generator.add_unix_nodaemon() + self.assertEqual('unix\n{\n nodaemon \n}\n', + vpp_config_generator.dump_config()) + + def test_add_unix_coredump(self): + vpp_config_generator = VppConfigGenerator() + vpp_config_generator.add_unix_coredump() + self.assertEqual('unix\n{\n full-coredump \n}\n', + vpp_config_generator.dump_config()) + + def test_add_dpdk_dev(self): + vpp_config_generator = VppConfigGenerator() + vpp_config_generator.add_dpdk_dev('0000:00:00.0') + self.assertEqual('dpdk\n{\n dev 0000:00:00.0 \n}\n', + vpp_config_generator.dump_config()) + + def test_add_dpdk_cryptodev(self): + vpp_config_generator = VppConfigGenerator() + vpp_config_generator.add_dpdk_cryptodev(2, '0000:00:00.0') + self.assertEqual( + 'dpdk\n{\n dev 0000:00:01.0 \n dev 0000:00:01.1 \n uio-driver igb_uio\n}\n', + vpp_config_generator.dump_config()) + + def test_add_dpdk_sw_cryptodev(self): + vpp_config_generator = VppConfigGenerator() + vpp_config_generator.add_dpdk_sw_cryptodev('aesni_gcm', 0, 2) + self.assertEqual( + 'dpdk\n{\n vdev cryptodev_aesni_gcm_pmd,socket_id=0 \n}\n', + vpp_config_generator.dump_config()) + + def test_add_dpdk_dev_default_rxq(self): + vpp_config_generator = VppConfigGenerator() + vpp_config_generator.add_dpdk_dev_default_rxq(1) + self.assertEqual( + 'dpdk\n{\n dev default\n {\n num-rx-queues 1\n }\n}\n', + vpp_config_generator.dump_config()) + + def test_add_dpdk_dev_default_rxd(self): + vpp_config_generator = VppConfigGenerator() + vpp_config_generator.add_dpdk_dev_default_rxd(2048) + self.assertEqual( + 'dpdk\n{\n dev default\n {\n num-rx-desc 2048\n }\n}\n', + vpp_config_generator.dump_config()) + + def test_add_dpdk_dev_default_txd(self): + vpp_config_generator = VppConfigGenerator() + vpp_config_generator.add_dpdk_dev_default_txd(2048) + self.assertEqual( + 'dpdk\n{\n dev default\n {\n num-tx-desc 2048\n }\n}\n', + vpp_config_generator.dump_config()) + + def test_add_dpdk_log_level(self): + vpp_config_generator = VppConfigGenerator() + vpp_config_generator.add_dpdk_log_level('debug') + self.assertEqual('dpdk\n{\n log-level debug\n}\n', + vpp_config_generator.dump_config()) + + def test_add_dpdk_socketmem(self): + vpp_config_generator = VppConfigGenerator() + vpp_config_generator.add_dpdk_socketmem('1024,1024') + self.assertEqual('dpdk\n{\n socket-mem 1024,1024\n}\n', + vpp_config_generator.dump_config()) + + def test_add_dpdk_num_mbufs(self): + vpp_config_generator = VppConfigGenerator() + vpp_config_generator.add_dpdk_num_mbufs(32768) + self.assertEqual('dpdk\n{\n num-mbufs 32768\n}\n', + vpp_config_generator.dump_config()) + + def test_add_dpdk_uio_driver(self): + vpp_config_generator = VppConfigGenerator() + vpp_config_generator.add_dpdk_uio_driver('igb_uio') + self.assertEqual('dpdk\n{\n uio-driver igb_uio\n}\n', + vpp_config_generator.dump_config()) + + def test_add_cpu_main_core(self): + vpp_config_generator = VppConfigGenerator() + vpp_config_generator.add_cpu_main_core('1,2') + self.assertEqual('cpu\n{\n main-core 1,2\n}\n', + vpp_config_generator.dump_config()) + + def test_add_cpu_corelist_workers(self): + vpp_config_generator = VppConfigGenerator() + vpp_config_generator.add_cpu_corelist_workers('1,2') + self.assertEqual('cpu\n{\n corelist-workers 1,2\n}\n', + vpp_config_generator.dump_config()) + + def test_add_heapsize(self): + vpp_config_generator = VppConfigGenerator() + vpp_config_generator.add_heapsize('4G') + self.assertEqual('heapsize 4G\n', vpp_config_generator.dump_config()) + + def test_add_ip6_hash_buckets(self): + vpp_config_generator = VppConfigGenerator() + vpp_config_generator.add_ip6_hash_buckets(2000000) + self.assertEqual('ip6\n{\n hash-buckets 2000000\n}\n', + vpp_config_generator.dump_config()) + + def test_add_ip6_heap_size(self): + vpp_config_generator = VppConfigGenerator() + vpp_config_generator.add_ip6_heap_size('4G') + self.assertEqual('ip6\n{\n heap-size 4G\n}\n', + vpp_config_generator.dump_config()) + + def test_add_ip_heap_size(self): + vpp_config_generator = VppConfigGenerator() + vpp_config_generator.add_ip_heap_size('4G') + self.assertEqual('ip\n{\n heap-size 4G\n}\n', + vpp_config_generator.dump_config()) + + def test_add_statseg_size(self): + vpp_config_generator = VppConfigGenerator() + vpp_config_generator.add_statseg_size('4G') + self.assertEqual('statseg\n{\n size 4G\n}\n', + vpp_config_generator.dump_config()) + + def test_add_plugin(self): + vpp_config_generator = VppConfigGenerator() + vpp_config_generator.add_plugin('enable', ['dpdk_plugin.so']) + self.assertEqual( + 'plugins\n{\n plugin [\'dpdk_plugin.so\']\n {\n enable \n }\n}\n', + vpp_config_generator.dump_config()) + + def test_add_dpdk_no_multi_seg(self): + vpp_config_generator = VppConfigGenerator() + vpp_config_generator.add_dpdk_no_multi_seg() + self.assertEqual('dpdk\n{\n no-multi-seg \n}\n', + vpp_config_generator.dump_config()) + + def test_add_dpdk_no_tx_checksum_offload(self): + vpp_config_generator = VppConfigGenerator() + vpp_config_generator.add_dpdk_no_tx_checksum_offload() + self.assertEqual('dpdk\n{\n no-tx-checksum-offload \n}\n', + vpp_config_generator.dump_config()) + + def test_dump_config(self): + vpp_config_generator = VppConfigGenerator() + vpp_config_generator.add_unix_log() + self.assertEqual('unix\n{\n log /tmp/vpe.log\n}\n', + vpp_config_generator.dump_config()) + + def test_pci_dev_check(self): + self.assertTrue(VppConfigGenerator.pci_dev_check('0000:00:00.0')) + + def test_pci_dev_check_error(self): + with self.assertRaises(ValueError) as raised: + VppConfigGenerator.pci_dev_check('0000:00:0.0') + self.assertIn( + 'PCI address 0000:00:0.0 is not in valid format xxxx:xx:xx.x', + str(raised.exception)) + + +class TestVppSetupEnvHelper(unittest.TestCase): + VNFD_0 = { + "benchmark": { + "kpi": [ + "packets_in", + "packets_fwd", + "packets_dropped" + ] + }, + "connection-point": [ + { + "name": "xe0", + "type": "VPORT" + }, + { + "name": "xe1", + "type": "VPORT" + } + ], + "description": "VPP IPsec", + "id": "VipsecApproxVnf", + "mgmt-interface": { + "ip": "10.10.10.101", + "password": "r00t", + "user": "root", + "vdu-id": "ipsecvnf-baremetal" + }, + "name": "IpsecVnf", + "short-name": "IpsecVnf", + "vdu": [ + { + "description": "VPP Ipsec", + "external-interface": [ + { + "name": "xe0", + "virtual-interface": { + "driver": "igb_uio", + "dst_ip": "192.168.100.1", + "dst_mac": "90:e2:ba:7c:30:e8", + "ifname": "xe0", + "local_ip": "192.168.100.2", + "local_mac": "90:e2:ba:7c:41:a8", + "netmask": "255.255.255.0", + "network": {}, + "node_name": "vnf__0", + "peer_ifname": "xe0", + "peer_intf": { + "dpdk_port_num": 0, + "driver": "igb_uio", + "dst_ip": "192.168.100.2", + "dst_mac": "90:e2:ba:7c:41:a8", + "ifname": "xe0", + "local_ip": "192.168.100.1", + "local_mac": "90:e2:ba:7c:30:e8", + "netmask": "255.255.255.0", + "network": {}, + "node_name": "tg__0", + "peer_ifname": "xe0", + "peer_name": "vnf__0", + "vld_id": "uplink_0", + "vpci": "0000:81:00.0" + }, + "peer_name": "tg__0", + "vld_id": "uplink_0", + "vpci": "0000:ff:06.0" + }, + "vnfd-connection-point-ref": "xe0" + }, + { + "name": "xe1", + "virtual-interface": { + "driver": "igb_uio", + "dst_ip": "1.1.1.2", + "dst_mac": "0a:b1:ec:fd:a2:66", + "ifname": "xe1", + "local_ip": "1.1.1.1", + "local_mac": "4e:90:85:d3:c5:13", + "netmask": "255.255.255.0", + "network": {}, + "node_name": "vnf__0", + "peer_ifname": "xe1", + "peer_intf": { + "driver": "igb_uio", + "dst_ip": "1.1.1.1", + "dst_mac": "4e:90:85:d3:c5:13", + "ifname": "xe1", + "local_ip": "1.1.1.2", + "local_mac": "0a:b1:ec:fd:a2:66", + "netmask": "255.255.255.0", + "network": {}, + "node_name": "vnf__1", + "peer_ifname": "xe1", + "peer_name": "vnf__0", + "vld_id": "ciphertext", + "vpci": "0000:00:07.0" + }, + "peer_name": "vnf__1", + "vld_id": "ciphertext", + "vpci": "0000:ff:07.0" + }, + "vnfd-connection-point-ref": "xe1" + } + ], + "id": "ipsecvnf-baremetal", + "name": "ipsecvnf-baremetal", + "routing_table": [] + } + ] + } + + VNFD_1 = { + "benchmark": { + "kpi": [ + "packets_in", + "packets_fwd", + "packets_dropped" + ] + }, + "connection-point": [ + { + "name": "xe0", + "type": "VPORT" + }, + { + "name": "xe1", + "type": "VPORT" + } + ], + "description": "VPP IPsec", + "id": "VipsecApproxVnf", + "mgmt-interface": { + "ip": "10.10.10.101", + "password": "r00t", + "user": "root", + "vdu-id": "ipsecvnf-baremetal" + }, + "name": "IpsecVnf", + "short-name": "IpsecVnf", + "vdu": [ + { + "description": "VPP Ipsec", + "external-interface": [ + { + "name": "xe0", + "virtual-interface": { + "driver": "igb_uio", + "dst_ip": "192.168.100.1", + "dst_mac": "90:e2:ba:7c:30:e8", + "ifname": "xe0", + "local_ip": "192.168.100.2", + "local_mac": "90:e2:ba:7c:41:a8", + "netmask": "255.255.255.0", + "network": {}, + "node_name": "vnf__0", + "peer_ifname": "xe0", + "peer_intf": { + "dpdk_port_num": 0, + "driver": "igb_uio", + "dst_ip": "192.168.100.2", + "dst_mac": "90:e2:ba:7c:41:a8", + "ifname": "xe0", + "local_ip": "192.168.100.1", + "local_mac": "90:e2:ba:7c:30:e8", + "netmask": "255.255.255.0", + "network": {}, + "node_name": "tg__0", + "peer_ifname": "xe0", + "peer_name": "vnf__0", + "vld_id": "uplink_0", + "vpci": "0000:81:00.0" + }, + "peer_name": "tg__0", + "vld_id": "uplink_0", + "vpci": "0000:ff:06.0" + }, + "vnfd-connection-point-ref": "xe0" + }, + { + "name": "xe1", + "virtual-interface": { + "driver": "igb_uio", + "dst_ip": "1.1.1.2", + "dst_mac": "0a:b1:ec:fd:a2:66", + "ifname": "xe1", + "local_ip": "1.1.1.1", + "local_mac": "4e:90:85:d3:c5:13", + "netmask": "255.255.255.0", + "network": {}, + "node_name": "vnf__0", + "peer_ifname": "xe1", + "peer_intf": { + "driver": "igb_uio", + "dst_ip": "1.1.1.1", + "dst_mac": "4e:90:85:d3:c5:13", + "ifname": "xe1", + "local_ip": "1.1.1.2", + "local_mac": "0a:b1:ec:fd:a2:66", + "netmask": "255.255.255.0", + "network": {}, + "node_name": "vnf__1", + "peer_ifname": "xe1", + "peer_name": "vnf__0", + "vld_id": "ciphertext", + "vpci": "0000:00:07.0" + }, + "peer_name": "vnf__1", + "vld_id": "ciphertext", + "vpci": "0000:ff:07.0" + }, + "vnfd-connection-point-ref": "xe1" + } + ], + "id": "ipsecvnf-baremetal", + "name": "ipsecvnf-baremetal", + "routing_table": [] + } + ] + } + + VNFD_2 = { + "benchmark": { + "kpi": [ + "packets_in", + "packets_fwd", + "packets_dropped" + ] + }, + "connection-point": [ + { + "name": "xe0", + "type": "VPORT" + }, + { + "name": "xe1", + "type": "VPORT" + } + ], + "description": "VPP IPsec", + "id": "VipsecApproxVnf", + "mgmt-interface": { + "ip": "10.10.10.101", + "password": "r00t", + "user": "root", + "vdu-id": "ipsecvnf-baremetal" + }, + "name": "IpsecVnf", + "short-name": "IpsecVnf", + "vdu": [ + { + "description": "VPP Ipsec", + "external-interface": [ + { + "name": "xe0", + "virtual-interface": { + "driver": "igb_uio", + "dst_ip": "192.168.100.1", + "dst_mac": "90:e2:ba:7c:30:e8", + "ifname": "xe0", + "local_ip": "192.168.100.2", + "local_mac": "90:e2:ba:7c:41:a8", + "netmask": "255.255.255.0", + "network": {}, + "node_name": "vnf__0", + "peer_ifname": "xe0", + "peer_intf": { + "dpdk_port_num": 0, + "driver": "igb_uio", + "dst_ip": "192.168.100.2", + "dst_mac": "90:e2:ba:7c:41:a8", + "ifname": "xe0", + "local_ip": "192.168.100.1", + "local_mac": "90:e2:ba:7c:30:e8", + "netmask": "255.255.255.0", + "network": {}, + "node_name": "tg__0", + "peer_ifname": "xe0", + "peer_name": "vnf__0", + "vld_id": "uplink_0", + "vpci": "0000:81:00.0" + }, + "peer_name": "tg__0", + "vld_id": "uplink_0", + "vpci": "0000:ff:06.0" + }, + "vnfd-connection-point-ref": "xe0" + }, + { + "name": "xe1", + "virtual-interface": { + "driver": "igb_uio", + "dst_ip": "1.1.1.2", + "dst_mac": "0a:b1:ec:fd:a2:66", + "ifname": "xe1", + "local_ip": "1.1.1.1", + "local_mac": "4e:90:85:d3:c5:13", + "netmask": "255.255.255.0", + "network": {}, + "node_name": "vnf__0", + "peer_ifname": "xe1", + "peer_intf": { + "driver": "igb_uio", + "dst_ip": "1.1.1.1", + "dst_mac": "4e:90:85:d3:c5:13", + "ifname": "xe1", + "local_ip": "1.1.1.2", + "local_mac": "0a:b1:ec:fd:a2:66", + "netmask": "255.255.255.0", + "network": {}, + "node_name": "vnf__1", + "peer_ifname": "xe1", + "peer_name": "vnf__0", + "vld_id": "ciphertext", + "vpci": "0000:00:07.0" + }, + "peer_name": "vnf__1", + "vld_id": "ciphertext", + "vpci": "0000:ff:07.0" + }, + "vnfd-connection-point-ref": "xe1" + } + ], + "id": "ipsecvnf-baremetal", + "name": "ipsecvnf-baremetal", + "routing_table": [] + } + ] + } + + VNFD = { + 'vnfd:vnfd-catalog': { + 'vnfd': [ + VNFD_0, + ], + }, + } + + VPP_INTERFACES_DUMP = [ + { + "sw_if_index": 0, + "sup_sw_if_index": 0, + "l2_address_length": 0, + "l2_address": [0, 0, 0, 0, 0, 0, 0, 0], + "interface_name": "local0", + "admin_up_down": 0, + "link_up_down": 0, + "link_duplex": 0, + "link_speed": 0, + "mtu": 0, + "sub_id": 0, + "sub_dot1ad": 0, + "sub_number_of_tags": 0, + "sub_outer_vlan_id": 0, + "sub_inner_vlan_id": 0, + "sub_exact_match": 0, + "sub_default": 0, + "sub_outer_vlan_id_any": 0, + "sub_inner_vlan_id_any": 0, + "vtr_op": 0, + "vtr_push_dot1q": 0, + "vtr_tag1": 0, + "vtr_tag2": 0 + }, + { + "sw_if_index": 1, + "sup_sw_if_index": 1, + "l2_address_length": 6, + "l2_address": [144, 226, 186, 124, 65, 168, 0, 0], + "interface_name": "TenGigabitEthernetff/6/0", + "admin_up_down": 0, + "link_up_down": 0, + "link_duplex": 2, + "link_speed": 32, + "mtu": 9202, + "sub_id": 0, + "sub_dot1ad": 0, + "sub_number_of_tags": 0, + "sub_outer_vlan_id": 0, + "sub_inner_vlan_id": 0, + "sub_exact_match": 0, + "sub_default": 0, + "sub_outer_vlan_id_any": 0, + "sub_inner_vlan_id_any": 0, + "vtr_op": 0, + "vtr_push_dot1q": 0, + "vtr_tag1": 0, + "vtr_tag2": 0 + }, + { + "sw_if_index": 2, + "sup_sw_if_index": 2, + "l2_address_length": 6, + "l2_address": [78, 144, 133, 211, 197, 19, 0, 0], + "interface_name": "VirtualFunctionEthernetff/7/0", + "admin_up_down": 0, + "link_up_down": 0, + "link_duplex": 2, + "link_speed": 32, + "mtu": 9206, + "sub_id": 0, + "sub_dot1ad": 0, + "sub_number_of_tags": 0, + "sub_outer_vlan_id": 0, + "sub_inner_vlan_id": 0, + "sub_exact_match": 0, + "sub_default": 0, + "sub_outer_vlan_id_any": 0, + "sub_inner_vlan_id_any": 0, + "vtr_op": 0, + "vtr_push_dot1q": 0, + "vtr_tag1": 0, + "vtr_tag2": 0 + } + ] + + VPP_INTERFACES_DUMP_MAC_ERR = [ + { + "sw_if_index": 0, + "sup_sw_if_index": 0, + "l2_address_length": 0, + "l2_address": [0, 0, 0, 0, 0, 0, 0, 0], + "interface_name": "local0", + "admin_up_down": 0, + "link_up_down": 0, + "link_duplex": 0, + "link_speed": 0, + "mtu": 0, + "sub_id": 0, + "sub_dot1ad": 0, + "sub_number_of_tags": 0, + "sub_outer_vlan_id": 0, + "sub_inner_vlan_id": 0, + "sub_exact_match": 0, + "sub_default": 0, + "sub_outer_vlan_id_any": 0, + "sub_inner_vlan_id_any": 0, + "vtr_op": 0, + "vtr_push_dot1q": 0, + "vtr_tag1": 0, + "vtr_tag2": 0 + }, + { + "sw_if_index": 1, + "sup_sw_if_index": 1, + "l2_address_length": 6, + "l2_address": [144, 226, 186, 124, 65, 169, 0, 0], + "interface_name": "TenGigabitEthernetff/6/0", + "admin_up_down": 0, + "link_up_down": 0, + "link_duplex": 2, + "link_speed": 32, + "mtu": 9202, + "sub_id": 0, + "sub_dot1ad": 0, + "sub_number_of_tags": 0, + "sub_outer_vlan_id": 0, + "sub_inner_vlan_id": 0, + "sub_exact_match": 0, + "sub_default": 0, + "sub_outer_vlan_id_any": 0, + "sub_inner_vlan_id_any": 0, + "vtr_op": 0, + "vtr_push_dot1q": 0, + "vtr_tag1": 0, + "vtr_tag2": 0 + }, + { + "sw_if_index": 2, + "sup_sw_if_index": 2, + "l2_address_length": 6, + "l2_address": [78, 144, 133, 211, 197, 20, 0, 0], + "interface_name": "VirtualFunctionEthernetff/7/0", + "admin_up_down": 0, + "link_up_down": 0, + "link_duplex": 2, + "link_speed": 32, + "mtu": 9206, + "sub_id": 0, + "sub_dot1ad": 0, + "sub_number_of_tags": 0, + "sub_outer_vlan_id": 0, + "sub_inner_vlan_id": 0, + "sub_exact_match": 0, + "sub_default": 0, + "sub_outer_vlan_id_any": 0, + "sub_inner_vlan_id_any": 0, + "vtr_op": 0, + "vtr_push_dot1q": 0, + "vtr_tag1": 0, + "vtr_tag2": 0 + } + ] + + CPU_LAYOUT = {'cpuinfo': [[0, 0, 0, 0, 0, 0, 0, 0], + [1, 0, 0, 0, 0, 1, 1, 0]]} + CPU_SMT = {'cpuinfo': [[0, 0, 0, 0, 0, 0, 0, 0], + [1, 0, 0, 0, 0, 1, 1, 0], + [2, 1, 0, 0, 0, 2, 2, 1], + [3, 1, 0, 0, 0, 3, 3, 1], + [4, 2, 0, 0, 0, 4, 4, 2], + [5, 2, 0, 0, 0, 5, 5, 2], + [6, 3, 0, 0, 0, 6, 6, 3], + [7, 3, 0, 0, 0, 7, 7, 3], + [8, 4, 0, 0, 0, 8, 8, 4], + [9, 5, 0, 1, 0, 0, 0, 0], + [10, 6, 0, 1, 0, 1, 1, 0], + [11, 6, 0, 1, 0, 2, 2, 1], + [12, 7, 0, 1, 0, 3, 3, 1], + [13, 7, 0, 1, 0, 4, 4, 2], + [14, 8, 0, 1, 0, 5, 5, 2], + [15, 8, 0, 1, 0, 6, 6, 3], + [16, 9, 0, 1, 0, 7, 7, 3], + [17, 9, 0, 1, 0, 8, 8, 4]]} + + def test_kill_vnf(self): + vnfd_helper = VnfdHelper(self.VNFD_0) + ssh_helper = mock.Mock() + ssh_helper.execute.return_value = 0, 0, 0 + scenario_helper = mock.Mock() + vpp_setup_env_helper = VppSetupEnvHelper(vnfd_helper, ssh_helper, + scenario_helper) + vpp_setup_env_helper.kill_vnf() + + def test_kill_vnf_error(self): + vnfd_helper = VnfdHelper(self.VNFD_0) + ssh_helper = mock.Mock() + ssh_helper.execute.return_value = 1, 0, 0 + scenario_helper = mock.Mock() + vpp_setup_env_helper = VppSetupEnvHelper(vnfd_helper, ssh_helper, + scenario_helper) + with self.assertRaises(RuntimeError) as raised: + vpp_setup_env_helper.kill_vnf() + + self.assertIn('Failed to stop service vpp', str(raised.exception)) + + def test_tear_down(self): + vnfd_helper = VnfdHelper(self.VNFD_0) + ssh_helper = mock.Mock() + scenario_helper = mock.Mock() + vpp_setup_env_helper = VppSetupEnvHelper(vnfd_helper, ssh_helper, + scenario_helper) + vpp_setup_env_helper.tear_down() + + def test_start_vpp_service(self): + vnfd_helper = VnfdHelper(self.VNFD_0) + ssh_helper = mock.Mock() + ssh_helper.execute.return_value = 0, 0, 0 + scenario_helper = mock.Mock() + vpp_setup_env_helper = VppSetupEnvHelper(vnfd_helper, ssh_helper, + scenario_helper) + vpp_setup_env_helper.start_vpp_service() + + def test_start_vpp_service_error(self): + vnfd_helper = VnfdHelper(self.VNFD_0) + ssh_helper = mock.Mock() + ssh_helper.execute.return_value = 1, 0, 0 + scenario_helper = mock.Mock() + vpp_setup_env_helper = VppSetupEnvHelper(vnfd_helper, ssh_helper, + scenario_helper) + with self.assertRaises(RuntimeError) as raised: + vpp_setup_env_helper.start_vpp_service() + + self.assertIn('Failed to start service vpp', str(raised.exception)) + + def test__update_vnfd_helper(self): + vnfd_helper = VnfdHelper(self.VNFD_0) + ssh_helper = mock.Mock() + scenario_helper = mock.Mock() + vpp_setup_env_helper = VppSetupEnvHelper(vnfd_helper, ssh_helper, + scenario_helper) + vpp_setup_env_helper._update_vnfd_helper( + {'vpp-data': {'vpp-key': 'vpp-value'}}) + + self.assertEqual({'vpp-key': 'vpp-value'}, + vpp_setup_env_helper.vnfd_helper.get('vpp-data', {})) + + def test__update_vnfd_helper_with_key(self): + vnfd_helper = VnfdHelper(self.VNFD_0) + ssh_helper = mock.Mock() + scenario_helper = mock.Mock() + vpp_setup_env_helper = VppSetupEnvHelper(vnfd_helper, ssh_helper, + scenario_helper) + vpp_setup_env_helper._update_vnfd_helper({'driver': 'qat'}, 'xe0') + + self.assertEqual('qat', + vpp_setup_env_helper.get_value_by_interface_key( + 'xe0', 'driver')) + + def test__update_vnfd_helper_dict_without_key(self): + vnfd_helper = VnfdHelper(self.VNFD_0) + ssh_helper = mock.Mock() + scenario_helper = mock.Mock() + vpp_setup_env_helper = VppSetupEnvHelper(vnfd_helper, ssh_helper, + scenario_helper) + vpp_setup_env_helper._update_vnfd_helper( + {'mgmt-interface': {'name': 'net'}}) + + self.assertEqual({'ip': '10.10.10.101', + 'name': 'net', + 'password': 'r00t', + 'user': 'root', + 'vdu-id': 'ipsecvnf-baremetal'}, + vpp_setup_env_helper.vnfd_helper.get('mgmt-interface', + {})) + + def test_get_value_by_interface_key(self): + vnfd_helper = VnfdHelper(self.VNFD_0) + ssh_helper = mock.Mock() + + scenario_helper = mock.Mock() + vpp_setup_env_helper = VppSetupEnvHelper(vnfd_helper, ssh_helper, + scenario_helper) + vpp_setup_env_helper._update_vnfd_helper( + {'vpp-data': {'vpp-key': 'vpp-value'}}, 'xe0') + + self.assertEqual({'vpp-key': 'vpp-value'}, + vpp_setup_env_helper.get_value_by_interface_key( + 'xe0', 'vpp-data')) + + def test_get_value_by_interface_key_error(self): + vnfd_helper = VnfdHelper(self.VNFD_0) + ssh_helper = mock.Mock() + + scenario_helper = mock.Mock() + vpp_setup_env_helper = VppSetupEnvHelper(vnfd_helper, ssh_helper, + scenario_helper) + vpp_setup_env_helper._update_vnfd_helper( + {'vpp-data': {'vpp-key': 'vpp-value'}}, 'xe0') + + self.assertIsNone(vpp_setup_env_helper.get_value_by_interface_key( + 'xe2', 'vpp-err')) + + def test_crypto_device_init(self): + vnfd_helper = VnfdHelper(self.VNFD_0) + ssh_helper = mock.Mock() + + scenario_helper = mock.Mock() + vpp_setup_env_helper = VppSetupEnvHelper(vnfd_helper, ssh_helper, + scenario_helper) + vpp_setup_env_helper.dpdk_bind_helper.load_dpdk_driver = mock.Mock() + vpp_setup_env_helper.dpdk_bind_helper.bind = mock.Mock() + + vpp_setup_env_helper.kill_vnf = mock.Mock() + vpp_setup_env_helper.pci_driver_unbind = mock.Mock() + + with mock.patch.object(vpp_setup_env_helper, 'get_pci_dev_driver') as \ + mock_get_pci_dev_driver, \ + mock.patch.object(vpp_setup_env_helper, 'set_sriov_numvfs') as \ + mock_set_sriov_numvfs: + mock_get_pci_dev_driver.return_value = 'igb_uio' + self.assertIsNone( + vpp_setup_env_helper.crypto_device_init('0000:ff:06.0', 32)) + mock_set_sriov_numvfs.assert_called() + + def test_get_sriov_numvfs(self): + vnfd_helper = VnfdHelper(self.VNFD_0) + ssh_helper = mock.Mock() + ssh_helper.execute.return_value = 0, '32', '' + + scenario_helper = mock.Mock() + vpp_setup_env_helper = VppSetupEnvHelper(vnfd_helper, ssh_helper, + scenario_helper) + self.assertEqual(32, + vpp_setup_env_helper.get_sriov_numvfs('0000:ff:06.0')) + + def test_get_sriov_numvfs_error(self): + vnfd_helper = VnfdHelper(self.VNFD_0) + ssh_helper = mock.Mock() + ssh_helper.execute.return_value = 0, 'err', '' + + scenario_helper = mock.Mock() + vpp_setup_env_helper = VppSetupEnvHelper(vnfd_helper, ssh_helper, + scenario_helper) + self.assertEqual(0, + vpp_setup_env_helper.get_sriov_numvfs('0000:ff:06.0')) + + def test_set_sriov_numvfs(self): + vnfd_helper = VnfdHelper(self.VNFD_0) + ssh_helper = mock.Mock() + + scenario_helper = mock.Mock() + vpp_setup_env_helper = VppSetupEnvHelper(vnfd_helper, ssh_helper, + scenario_helper) + vpp_setup_env_helper.set_sriov_numvfs('0000:ff:06.0') + self.assertEqual(ssh_helper.execute.call_count, 1) + + def test_pci_driver_unbind(self): + vnfd_helper = VnfdHelper(self.VNFD_0) + ssh_helper = mock.Mock() + + scenario_helper = mock.Mock() + vpp_setup_env_helper = VppSetupEnvHelper(vnfd_helper, ssh_helper, + scenario_helper) + vpp_setup_env_helper.pci_driver_unbind('0000:ff:06.0') + self.assertEqual(ssh_helper.execute.call_count, 1) + + def test_get_pci_dev_driver(self): + vnfd_helper = VnfdHelper(self.VNFD_0) + ssh_helper = mock.Mock() + ssh_helper.execute.return_value = \ + 0, 'Slot: ff:07.0\n' \ + 'Class: Ethernet controller\n' \ + 'Vendor: Intel Corporation\n' \ + 'Device: 82599 Ethernet Controller Virtual Function\n' \ + 'SVendor: Intel Corporation\n' \ + 'SDevice: 82599 Ethernet Controller Virtual Function\n' \ + 'Rev: 01\n' \ + 'Driver: igb_uio\n' \ + 'Module: ixgbevf', '' + scenario_helper = mock.Mock() + vpp_setup_env_helper = VppSetupEnvHelper(vnfd_helper, ssh_helper, + scenario_helper) + self.assertEqual('igb_uio', vpp_setup_env_helper.get_pci_dev_driver( + '0000:ff:06.0')) + + def test_get_pci_dev_driver_error(self): + vnfd_helper = VnfdHelper(self.VNFD_0) + ssh_helper = mock.Mock() + ssh_helper.execute.return_value = 1, 'err', '' + scenario_helper = mock.Mock() + vpp_setup_env_helper = VppSetupEnvHelper(vnfd_helper, ssh_helper, + scenario_helper) + with self.assertRaises(RuntimeError) as raised: + vpp_setup_env_helper.get_pci_dev_driver( + '0000:ff:06.0') + + self.assertIn("'lspci -vmmks 0000:ff:06.0' failed", + str(raised.exception)) + + def test_get_pci_dev_driver_output_error(self): + vnfd_helper = VnfdHelper(self.VNFD_0) + ssh_helper = mock.Mock() + ssh_helper.execute.return_value = \ + 0, 'Slot: ff:07.0\n' \ + '\n\t' \ + 'Vendor: Intel Corporation\n' \ + 'Device: 82599 Ethernet Controller Virtual Function\n' \ + 'SVendor: Intel Corporation\n' \ + 'SDevice: 82599 Ethernet Controller Virtual Function\n' \ + 'Rev: 01\n' \ + 'Driver_err: igb_uio\n' \ + 'Module: ixgbevf', '' + scenario_helper = mock.Mock() + vpp_setup_env_helper = VppSetupEnvHelper(vnfd_helper, ssh_helper, + scenario_helper) + self.assertIsNone( + vpp_setup_env_helper.get_pci_dev_driver('0000:ff:06.0')) + + def test_vpp_create_ipsec_tunnels(self): + vnfd_helper = VnfdHelper(self.VNFD_0) + ssh_helper = mock.Mock() + ssh_helper.execute.return_value = 0, '', '' + + scenario_helper = mock.Mock() + vpp_setup_env_helper = VppSetupEnvHelper(vnfd_helper, ssh_helper, + scenario_helper) + + self.assertIsNone( + vpp_setup_env_helper.vpp_create_ipsec_tunnels('10.10.10.2', + '10.10.10.1', 'xe0', + 1, 1, mock.Mock(), + 'crypto_key', + mock.Mock(), + 'integ_key', + '20.20.20.0')) + self.assertGreaterEqual(ssh_helper.execute.call_count, 2) + + def test_vpp_create_ipsec_1000_tunnels(self): + vnfd_helper = VnfdHelper(self.VNFD_0) + ssh_helper = mock.Mock() + ssh_helper.execute.return_value = 0, '', '' + + scenario_helper = mock.Mock() + vpp_setup_env_helper = VppSetupEnvHelper(vnfd_helper, ssh_helper, + scenario_helper) + + self.assertIsNone( + vpp_setup_env_helper.vpp_create_ipsec_tunnels('10.10.10.2', + '10.10.10.1', 'xe0', + 1000, 128000, + mock.Mock(), + 'crypto_key', + mock.Mock(), + 'integ_key', + '20.20.20.0')) + self.assertGreaterEqual(ssh_helper.execute.call_count, 2) + + def test_apply_config(self): + vnfd_helper = VnfdHelper(self.VNFD_0) + ssh_helper = mock.Mock() + ssh_helper.execute.return_value = 0, '', '' + + scenario_helper = mock.Mock() + vpp_setup_env_helper = VppSetupEnvHelper(vnfd_helper, ssh_helper, + scenario_helper) + self.assertIsNone(vpp_setup_env_helper.apply_config(mock.Mock())) + self.assertGreaterEqual(ssh_helper.execute.call_count, 2) + + def test_apply_config_error(self): + vnfd_helper = VnfdHelper(self.VNFD_0) + ssh_helper = mock.Mock() + ssh_helper.execute.return_value = 1, '', '' + + scenario_helper = mock.Mock() + vpp_setup_env_helper = VppSetupEnvHelper(vnfd_helper, ssh_helper, + scenario_helper) + with self.assertRaises(RuntimeError) as raised: + vpp_setup_env_helper.apply_config(mock.Mock()) + + self.assertIn('Writing config file failed', str(raised.exception)) + + def test_vpp_route_add(self): + vnfd_helper = VnfdHelper(self.VNFD_0) + ssh_helper = mock.Mock() + scenario_helper = mock.Mock() + vpp_setup_env_helper = VppSetupEnvHelper(vnfd_helper, ssh_helper, + scenario_helper) + + with mock.patch.object(vpp_helpers.VatTerminal, + 'vat_terminal_exec_cmd_from_template') as \ + mock_vat_terminal_exec_cmd_from_template: + mock_vat_terminal_exec_cmd_from_template.return_value = '' + self.assertIsNone( + vpp_setup_env_helper.vpp_route_add('xe0', '10.10.10.1', 24)) + + def test_vpp_route_add_without_index(self): + vnfd_helper = VnfdHelper(self.VNFD_0) + ssh_helper = mock.Mock() + scenario_helper = mock.Mock() + vpp_setup_env_helper = VppSetupEnvHelper(vnfd_helper, ssh_helper, + scenario_helper) + + with mock.patch.object(vpp_helpers.VatTerminal, + 'vat_terminal_exec_cmd_from_template') as \ + mock_vat_terminal_exec_cmd_from_template: + mock_vat_terminal_exec_cmd_from_template.return_value = '' + self.assertIsNone( + vpp_setup_env_helper.vpp_route_add('xe0', '10.10.10.1', 24, + interface='xe0', + use_sw_index=False)) + + def test_add_arp_on_dut(self): + vnfd_helper = VnfdHelper(self.VNFD_0) + ssh_helper = mock.Mock() + scenario_helper = mock.Mock() + vpp_setup_env_helper = VppSetupEnvHelper(vnfd_helper, ssh_helper, + scenario_helper) + + with mock.patch.object(vpp_helpers.VatTerminal, + 'vat_terminal_exec_cmd_from_template') as \ + mock_vat_terminal_exec_cmd_from_template: + mock_vat_terminal_exec_cmd_from_template.return_value = '' + self.assertEqual('', vpp_setup_env_helper.add_arp_on_dut('xe0', + '10.10.10.1', + '00:00:00:00:00:00')) + + def test_set_ip(self): + vnfd_helper = VnfdHelper(self.VNFD_0) + ssh_helper = mock.Mock() + scenario_helper = mock.Mock() + vpp_setup_env_helper = VppSetupEnvHelper(vnfd_helper, ssh_helper, + scenario_helper) + + with mock.patch.object(vpp_helpers.VatTerminal, + 'vat_terminal_exec_cmd_from_template') as \ + mock_vat_terminal_exec_cmd_from_template: + mock_vat_terminal_exec_cmd_from_template.return_value = '' + self.assertEqual('', + vpp_setup_env_helper.set_ip('xe0', '10.10.10.1', + 24)) + + def test_set_interface_state(self): + vnfd_helper = VnfdHelper(self.VNFD_0) + ssh_helper = mock.Mock() + scenario_helper = mock.Mock() + vpp_setup_env_helper = VppSetupEnvHelper(vnfd_helper, ssh_helper, + scenario_helper) + + with mock.patch.object(vpp_helpers.VatTerminal, + 'vat_terminal_exec_cmd_from_template') as \ + mock_vat_terminal_exec_cmd_from_template: + mock_vat_terminal_exec_cmd_from_template.return_value = '' + self.assertEqual('', + vpp_setup_env_helper.set_interface_state('xe0', + 'up')) + + def test_set_interface_state_error(self): + vnfd_helper = VnfdHelper(self.VNFD_0) + ssh_helper = mock.Mock() + scenario_helper = mock.Mock() + vpp_setup_env_helper = VppSetupEnvHelper(vnfd_helper, ssh_helper, + scenario_helper) + + with mock.patch.object(vpp_helpers.VatTerminal, + 'vat_terminal_exec_cmd_from_template') as \ + mock_vat_terminal_exec_cmd_from_template: + mock_vat_terminal_exec_cmd_from_template.return_value = '' + with self.assertRaises(ValueError) as raised: + vpp_setup_env_helper.set_interface_state('xe0', 'error') + self.assertIn('Unexpected interface state: error', + str(raised.exception)) + + def test_set_interface_down_state(self): + vnfd_helper = VnfdHelper(self.VNFD_0) + ssh_helper = mock.Mock() + scenario_helper = mock.Mock() + vpp_setup_env_helper = VppSetupEnvHelper(vnfd_helper, ssh_helper, + scenario_helper) + + with mock.patch.object(vpp_helpers.VatTerminal, + 'vat_terminal_exec_cmd_from_template') as \ + mock_vat_terminal_exec_cmd_from_template: + mock_vat_terminal_exec_cmd_from_template.return_value = '' + self.assertEqual('', + vpp_setup_env_helper.set_interface_state('xe0', + 'down')) + + def test_vpp_set_interface_mtu(self): + vnfd_helper = VnfdHelper(self.VNFD_0) + ssh_helper = mock.Mock() + scenario_helper = mock.Mock() + vpp_setup_env_helper = VppSetupEnvHelper(vnfd_helper, ssh_helper, + scenario_helper) + + with mock.patch.object(vpp_helpers.VatTerminal, + 'vat_terminal_exec_cmd_from_template') as \ + mock_vat_terminal_exec_cmd_from_template: + mock_vat_terminal_exec_cmd_from_template.return_value = '' + self.assertIsNone( + vpp_setup_env_helper.vpp_set_interface_mtu('xe0', 9200)) + + def test_vpp_interfaces_ready_wait(self): + json_output = [self.VPP_INTERFACES_DUMP] + vnfd_helper = VnfdHelper(self.VNFD_0) + ssh_helper = mock.Mock() + scenario_helper = mock.Mock() + vpp_setup_env_helper = VppSetupEnvHelper(vnfd_helper, ssh_helper, + scenario_helper) + + with mock.patch.object(vpp_helpers.VatTerminal, + 'vat_terminal_exec_cmd_from_template') as \ + mock_vat_terminal_exec_cmd_from_template: + mock_vat_terminal_exec_cmd_from_template.return_value = json_output + self.assertIsNone(vpp_setup_env_helper.vpp_interfaces_ready_wait()) + + def test_vpp_interfaces_ready_wait_timeout(self): + json_output = [[ + { + "sw_if_index": 0, + "sup_sw_if_index": 0, + "l2_address_length": 0, + "l2_address": [0, 0, 0, 0, 0, 0, 0, 0], + "interface_name": "xe0", + "admin_up_down": 1, + "link_up_down": 0, + "link_duplex": 0, + "link_speed": 0, + "mtu": 0, + "sub_id": 0, + "sub_dot1ad": 0, + "sub_number_of_tags": 0, + "sub_outer_vlan_id": 0, + "sub_inner_vlan_id": 0, + "sub_exact_match": 0, + "sub_default": 0, + "sub_outer_vlan_id_any": 0, + "sub_inner_vlan_id_any": 0, + "vtr_op": 0, + "vtr_push_dot1q": 0, + "vtr_tag1": 0, + "vtr_tag2": 0 + }]] + vnfd_helper = VnfdHelper(self.VNFD_0) + ssh_helper = mock.Mock() + scenario_helper = mock.Mock() + vpp_setup_env_helper = VppSetupEnvHelper(vnfd_helper, ssh_helper, + scenario_helper) + + with mock.patch.object(vpp_helpers.VatTerminal, + 'vat_terminal_exec_cmd_from_template') as \ + mock_vat_terminal_exec_cmd_from_template: + mock_vat_terminal_exec_cmd_from_template.return_value = json_output + with self.assertRaises(RuntimeError) as raised: + vpp_setup_env_helper.vpp_interfaces_ready_wait(5) + self.assertIn('timeout, not up [\'xe0\']', str(raised.exception)) + + def test_vpp_get_interface_data(self): + json_output = [self.VPP_INTERFACES_DUMP] + vnfd_helper = VnfdHelper(self.VNFD_0) + ssh_helper = mock.Mock() + scenario_helper = mock.Mock() + vpp_setup_env_helper = VppSetupEnvHelper(vnfd_helper, ssh_helper, + scenario_helper) + + with mock.patch.object(vpp_helpers.VatTerminal, + 'vat_terminal_exec_cmd_from_template') as \ + mock_vat_terminal_exec_cmd_from_template: + mock_vat_terminal_exec_cmd_from_template.return_value = json_output + self.assertEqual(json_output[0], + vpp_setup_env_helper.vpp_get_interface_data()) + + def test_vpp_get_interface_data_ifname(self): + json_output = [self.VPP_INTERFACES_DUMP] + vnfd_helper = VnfdHelper(self.VNFD_0) + ssh_helper = mock.Mock() + scenario_helper = mock.Mock() + vpp_setup_env_helper = VppSetupEnvHelper(vnfd_helper, ssh_helper, + scenario_helper) + + with mock.patch.object(vpp_helpers.VatTerminal, + 'vat_terminal_exec_cmd_from_template') as \ + mock_vat_terminal_exec_cmd_from_template: + mock_vat_terminal_exec_cmd_from_template.return_value = json_output + self.assertEqual(json_output[0][2], + vpp_setup_env_helper.vpp_get_interface_data( + 'VirtualFunctionEthernetff/7/0')) + + def test_vpp_get_interface_data_ifname_error(self): + json_output = [self.VPP_INTERFACES_DUMP] + vnfd_helper = VnfdHelper(self.VNFD_0) + ssh_helper = mock.Mock() + scenario_helper = mock.Mock() + vpp_setup_env_helper = VppSetupEnvHelper(vnfd_helper, ssh_helper, + scenario_helper) + + with mock.patch.object(vpp_helpers.VatTerminal, + 'vat_terminal_exec_cmd_from_template') as \ + mock_vat_terminal_exec_cmd_from_template: + mock_vat_terminal_exec_cmd_from_template.return_value = json_output + self.assertEqual({}, vpp_setup_env_helper.vpp_get_interface_data( + 'error')) + + def test_vpp_get_interface_data_ifindex(self): + json_output = [self.VPP_INTERFACES_DUMP] + vnfd_helper = VnfdHelper(self.VNFD_0) + ssh_helper = mock.Mock() + scenario_helper = mock.Mock() + vpp_setup_env_helper = VppSetupEnvHelper(vnfd_helper, ssh_helper, + scenario_helper) + + with mock.patch.object(vpp_helpers.VatTerminal, + 'vat_terminal_exec_cmd_from_template') as \ + mock_vat_terminal_exec_cmd_from_template: + mock_vat_terminal_exec_cmd_from_template.return_value = json_output + self.assertEqual(json_output[0][1], + vpp_setup_env_helper.vpp_get_interface_data(1)) + + def test_vpp_get_interface_data_error(self): + json_output = [self.VPP_INTERFACES_DUMP] + vnfd_helper = VnfdHelper(self.VNFD_0) + ssh_helper = mock.Mock() + scenario_helper = mock.Mock() + vpp_setup_env_helper = VppSetupEnvHelper(vnfd_helper, ssh_helper, + scenario_helper) + + with mock.patch.object(vpp_helpers.VatTerminal, + 'vat_terminal_exec_cmd_from_template') as \ + mock_vat_terminal_exec_cmd_from_template: + mock_vat_terminal_exec_cmd_from_template.return_value = json_output + with self.assertRaises(TypeError) as raised: + vpp_setup_env_helper.vpp_get_interface_data(1.0) + self.assertEqual('', str(raised.exception)) + + def test_update_vpp_interface_data(self): + output = '{}\n{}'.format(self.VPP_INTERFACES_DUMP, + 'dump_interface_table:6019: JSON output ' \ + 'supported only for VPE API calls and dump_stats_table\n' \ + '/opt/nsb_bin/vpp/templates/dump_interfaces.vat(2): \n' \ + 'dump_interface_table error: Misc') + vnfd_helper = VnfdHelper(self.VNFD_0) + ssh_helper = mock.Mock() + ssh_helper.execute.return_value = 0, output.replace("\'", "\""), '' + ssh_helper.join_bin_path.return_value = '/opt/nsb_bin/vpp/templates' + + scenario_helper = mock.Mock() + vpp_setup_env_helper = VppSetupEnvHelper(vnfd_helper, ssh_helper, + scenario_helper) + self.assertIsNone(vpp_setup_env_helper.update_vpp_interface_data()) + self.assertGreaterEqual(ssh_helper.execute.call_count, 1) + self.assertEqual('TenGigabitEthernetff/6/0', + vpp_setup_env_helper.get_value_by_interface_key( + 'xe0', 'vpp_name')) + self.assertEqual(1, vpp_setup_env_helper.get_value_by_interface_key( + 'xe0', 'vpp_sw_index')) + self.assertEqual('VirtualFunctionEthernetff/7/0', + vpp_setup_env_helper.get_value_by_interface_key( + 'xe1', 'vpp_name')) + self.assertEqual(2, vpp_setup_env_helper.get_value_by_interface_key( + 'xe1', 'vpp_sw_index')) + + def test_update_vpp_interface_data_error(self): + output = '{}\n{}'.format(self.VPP_INTERFACES_DUMP_MAC_ERR, + 'dump_interface_table:6019: JSON output ' \ + 'supported only for VPE API calls and dump_stats_table\n' \ + '/opt/nsb_bin/vpp/templates/dump_interfaces.vat(2): \n' \ + 'dump_interface_table error: Misc') + vnfd_helper = VnfdHelper(self.VNFD_0) + ssh_helper = mock.Mock() + ssh_helper.execute.return_value = 0, output.replace("\'", "\""), '' + ssh_helper.join_bin_path.return_value = '/opt/nsb_bin/vpp/templates' + + scenario_helper = mock.Mock() + vpp_setup_env_helper = VppSetupEnvHelper(vnfd_helper, ssh_helper, + scenario_helper) + self.assertIsNone(vpp_setup_env_helper.update_vpp_interface_data()) + self.assertGreaterEqual(ssh_helper.execute.call_count, 1) + + def test_iface_update_numa(self): + vnfd_helper = VnfdHelper(self.VNFD_0) + ssh_helper = mock.Mock() + ssh_helper.execute.return_value = 0, '0', '' + + scenario_helper = mock.Mock() + vpp_setup_env_helper = VppSetupEnvHelper(vnfd_helper, ssh_helper, + scenario_helper) + self.assertIsNone(vpp_setup_env_helper.iface_update_numa()) + self.assertGreaterEqual(ssh_helper.execute.call_count, 2) + self.assertEqual(0, vpp_setup_env_helper.get_value_by_interface_key( + 'xe0', 'numa_node')) + self.assertEqual(0, vpp_setup_env_helper.get_value_by_interface_key( + 'xe1', 'numa_node')) + + def test_iface_update_numa_error(self): + vnfd_helper = VnfdHelper(self.VNFD_1) + ssh_helper = mock.Mock() + ssh_helper.execute.return_value = 0, '-1', '' + + scenario_helper = mock.Mock() + vpp_setup_env_helper = VppSetupEnvHelper(vnfd_helper, ssh_helper, + scenario_helper) + with mock.patch.object(cpu.CpuSysCores, 'get_cpu_layout') as \ + mock_get_cpu_layout: + mock_get_cpu_layout.return_value = self.CPU_LAYOUT + sys_cores = cpu.CpuSysCores(ssh_helper) + vpp_setup_env_helper._update_vnfd_helper( + sys_cores.get_cpu_layout()) + self.assertIsNone(vpp_setup_env_helper.iface_update_numa()) + self.assertGreaterEqual(ssh_helper.execute.call_count, 2) + self.assertEqual(0, vpp_setup_env_helper.get_value_by_interface_key( + 'xe0', 'numa_node')) + self.assertEqual(0, vpp_setup_env_helper.get_value_by_interface_key( + 'xe1', 'numa_node')) + + def test_iface_update_without_numa(self): + vnfd_helper = VnfdHelper(self.VNFD_2) + ssh_helper = mock.Mock() + ssh_helper.execute.return_value = 0, '-1', '' + + scenario_helper = mock.Mock() + vpp_setup_env_helper = VppSetupEnvHelper(vnfd_helper, ssh_helper, + scenario_helper) + with mock.patch.object(cpu.CpuSysCores, 'get_cpu_layout') as \ + mock_get_cpu_layout: + mock_get_cpu_layout.return_value = self.CPU_SMT + sys_cores = cpu.CpuSysCores(ssh_helper) + vpp_setup_env_helper._update_vnfd_helper( + sys_cores.get_cpu_layout()) + self.assertIsNone(vpp_setup_env_helper.iface_update_numa()) + self.assertGreaterEqual(ssh_helper.execute.call_count, 2) + self.assertIsNone(vpp_setup_env_helper.get_value_by_interface_key( + 'xe0', 'numa_node')) + self.assertIsNone(vpp_setup_env_helper.get_value_by_interface_key( + 'xe1', 'numa_node')) + + def test_execute_script(self): + vnfd_helper = VnfdHelper(self.VNFD_0) + ssh_helper = mock.Mock() + + scenario_helper = mock.Mock() + vpp_setup_env_helper = VppSetupEnvHelper(vnfd_helper, ssh_helper, + scenario_helper) + vpp_setup_env_helper.execute_script('dump_interfaces.vat', True, True) + self.assertGreaterEqual(ssh_helper.put_file.call_count, 1) + self.assertGreaterEqual(ssh_helper.execute.call_count, 1) + + def test_execute_script_error(self): + vnfd_helper = VnfdHelper(self.VNFD_0) + ssh_helper = mock.Mock() + ssh_helper.execute.side_effect = Exception + + scenario_helper = mock.Mock() + vpp_setup_env_helper = VppSetupEnvHelper(vnfd_helper, ssh_helper, + scenario_helper) + with self.assertRaises(Exception) as raised: + vpp_setup_env_helper.execute_script('dump_interfaces.vat', True, + True) + self.assertIn( + 'VAT script execution failed: vpp_api_test json in dump_interfaces.vat script', + str(raised.exception)) + self.assertGreaterEqual(ssh_helper.put_file.call_count, 1) + + def test_execute_script_json_out(self): + json_output = [ + { + "sw_if_index": 0, + "sup_sw_if_index": 0 + }, + { + "l2_address_length": 6, + "l2_address": [144, 226, 186, 124, 65, 168, 0, 0] + }, + { + "interface_name": "VirtualFunctionEthernetff/7/0", + "admin_up_down": 0 + } + ] + output = '{}\n{}'.format(json_output, + 'dump_interface_table:6019: JSON output ' \ + 'supported only for VPE API calls and dump_stats_table\n' \ + '/opt/nsb_bin/vpp/templates/dump_interfaces.vat(2): \n' \ + 'dump_interface_table error: Misc') + vnfd_helper = VnfdHelper(self.VNFD_0) + ssh_helper = mock.Mock() + ssh_helper.execute.return_value = 0, output, '' + ssh_helper.join_bin_path.return_value = '/opt/nsb_bin/vpp/templates' + scenario_helper = mock.Mock() + vpp_setup_env_helper = VppSetupEnvHelper(vnfd_helper, ssh_helper, + scenario_helper) + self.assertEqual(str(json_output), + vpp_setup_env_helper.execute_script_json_out( + 'dump_interfaces.vat')) + + def test_self_cleanup_vat_json_output(self): + json_output = [ + { + "sw_if_index": 0, + "sup_sw_if_index": 0 + }, + { + "l2_address_length": 6, + "l2_address": [144, 226, 186, 124, 65, 168, 0, 0] + }, + { + "interface_name": "VirtualFunctionEthernetff/7/0", + "admin_up_down": 0 + } + ] + + output = '{}\n{}'.format(json_output, + 'dump_interface_table:6019: JSON output ' \ + 'supported only for VPE API calls and dump_stats_table\n' \ + '/opt/nsb_bin/vpp/templates/dump_interfaces.vat(2): \n' \ + 'dump_interface_table error: Misc') + self.assertEqual(str(json_output), + VppSetupEnvHelper.cleanup_vat_json_output(output, + '/opt/nsb_bin/vpp/templates/dump_interfaces.vat')) + + def test__convert_mac_to_number_list(self): + self.assertEqual([144, 226, 186, 124, 65, 168], + VppSetupEnvHelper._convert_mac_to_number_list( + '90:e2:ba:7c:41:a8')) + + def test_get_vpp_interface_by_mac(self): + mac_address = '90:e2:ba:7c:41:a8' + self.assertEqual({'admin_up_down': 0, + 'interface_name': 'TenGigabitEthernetff/6/0', + 'l2_address': [144, 226, 186, 124, 65, 168, 0, 0], + 'l2_address_length': 6, + 'link_duplex': 2, + 'link_speed': 32, + 'link_up_down': 0, + 'mtu': 9202, + 'sub_default': 0, + 'sub_dot1ad': 0, + 'sub_exact_match': 0, + 'sub_id': 0, + 'sub_inner_vlan_id': 0, + 'sub_inner_vlan_id_any': 0, + 'sub_number_of_tags': 0, + 'sub_outer_vlan_id': 0, + 'sub_outer_vlan_id_any': 0, + 'sup_sw_if_index': 1, + 'sw_if_index': 1, + 'vtr_op': 0, + 'vtr_push_dot1q': 0, + 'vtr_tag1': 0, + 'vtr_tag2': 0}, + VppSetupEnvHelper.get_vpp_interface_by_mac( + self.VPP_INTERFACES_DUMP, mac_address)) + + def test_get_vpp_interface_by_mac_error(self): + mac_address = '90:e2:ba:7c:41:a9' + with self.assertRaises(ValueError) as raised: + VppSetupEnvHelper.get_vpp_interface_by_mac( + [{ + "sw_if_index": 1, + "sup_sw_if_index": 1, + "l2_address_length": 7, + "l2_address": [144, 226, 186, 124, 65, 169, 0, 0], + "interface_name": "TenGigabitEthernetff/6/0", + "admin_up_down": 0, + "link_up_down": 0, + "link_duplex": 2, + "link_speed": 32, + "mtu": 9202, + "sub_id": 0, + "sub_dot1ad": 0, + "sub_number_of_tags": 0, + "sub_outer_vlan_id": 0, + "sub_inner_vlan_id": 0, + "sub_exact_match": 0, + "sub_default": 0, + "sub_outer_vlan_id_any": 0, + "sub_inner_vlan_id_any": 0, + "vtr_op": 0, + "vtr_push_dot1q": 0, + "vtr_tag1": 0, + "vtr_tag2": 0 + }], mac_address) + + self.assertIn('l2_address_length value is not 6.', + str(raised.exception)) + + def test_get_vpp_interface_by_mac_l2_error(self): + mac_address = '90:e2:ba:7c:41:a7' + with self.assertRaises(KeyError) as raised: + VppSetupEnvHelper.get_vpp_interface_by_mac( + [{ + "sw_if_index": 1, + "sup_sw_if_index": 1, + "l2_address_length": 6, + "l2_address_err": [144, 226, 186, 124, 65, 167, 0, 0], + "interface_name": "TenGigabitEthernetff/6/0", + "admin_up_down": 0, + "link_up_down": 0, + "link_duplex": 2, + "link_speed": 32, + "mtu": 9202, + "sub_id": 0, + "sub_dot1ad": 0, + "sub_number_of_tags": 0, + "sub_outer_vlan_id": 0, + "sub_inner_vlan_id": 0, + "sub_exact_match": 0, + "sub_default": 0, + "sub_outer_vlan_id_any": 0, + "sub_inner_vlan_id_any": 0, + "vtr_op": 0, + "vtr_push_dot1q": 0, + "vtr_tag1": 0, + "vtr_tag2": 0 + }], mac_address) + + self.assertIn( + 'key l2_address not found in interface dict.Probably input list ' \ + 'is not parsed from correct VAT json output.', + str(raised.exception)) + + def test_get_vpp_interface_by_mac_l2_length_error(self): + mac_address = '90:e2:ba:7c:41:a6' + with self.assertRaises(KeyError) as raised: + VppSetupEnvHelper.get_vpp_interface_by_mac( + [{ + "sw_if_index": 1, + "sup_sw_if_index": 1, + "l2_address_length_err": 6, + "l2_address": [144, 226, 186, 124, 65, 166, 0, 0], + "interface_name": "TenGigabitEthernetff/6/0", + "admin_up_down": 0, + "link_up_down": 0, + "link_duplex": 2, + "link_speed": 32, + "mtu": 9202, + "sub_id": 0, + "sub_dot1ad": 0, + "sub_number_of_tags": 0, + "sub_outer_vlan_id": 0, + "sub_inner_vlan_id": 0, + "sub_exact_match": 0, + "sub_default": 0, + "sub_outer_vlan_id_any": 0, + "sub_inner_vlan_id_any": 0, + "vtr_op": 0, + "vtr_push_dot1q": 0, + "vtr_tag1": 0, + "vtr_tag2": 0 + }], mac_address) + + self.assertIn( + 'key l2_address_length not found in interface dict. Probably ' \ + 'input list is not parsed from correct VAT json output.', + str(raised.exception)) + + def test_get_prefix_length(self): + start_ip = '10.10.10.0' + end_ip = '10.10.10.127' + ips = [ipaddress.ip_address(ip) for ip in + [str(ipaddress.ip_address(start_ip)), str(end_ip)]] + lowest_ip, highest_ip = min(ips), max(ips) + + self.assertEqual(25, + VppSetupEnvHelper.get_prefix_length(int(lowest_ip), + int(highest_ip), + lowest_ip.max_prefixlen)) + + def test_get_prefix_length_zero_prefix(self): + start_ip = '10.0.0.0' + end_ip = '10.0.0.0' + ips = [ipaddress.ip_address(ip) for ip in + [str(ipaddress.ip_address(start_ip)), str(end_ip)]] + lowest_ip, highest_ip = min(ips), max(ips) + + self.assertEqual(0, + VppSetupEnvHelper.get_prefix_length(int(lowest_ip), + int(highest_ip), + 0)) + + +class TestVatTerminal(unittest.TestCase): + + def test___init___error(self): + ssh_helper = mock.Mock() + ssh_helper.interactive_terminal_open.side_effect = exceptions.SSHTimeout + + with self.assertRaises(RuntimeError) as raised: + VatTerminal(ssh_helper, json_param=True) + self.assertIn('Cannot open interactive terminal', + str(raised.exception)) + + def test___init___exec_error(self): + ssh_helper = mock.Mock() + ssh_helper.interactive_terminal_exec_command.side_effect = exceptions.SSHTimeout + VatTerminal(ssh_helper, json_param=True) + + def test_vat_terminal_exec_cmd(self): + ssh_helper = mock.Mock() + ssh_helper.interactive_terminal_exec_command.return_value = str( + {'empty': 'value'}).replace("\'", "\"") + vat_terminal = VatTerminal(ssh_helper, json_param=True) + + self.assertEqual({'empty': 'value'}, + vat_terminal.vat_terminal_exec_cmd( + "hw_interface_set_mtu sw_if_index 1 mtu 9200")) + + def test_vat_terminal_exec_cmd_array(self): + ssh_helper = mock.Mock() + ssh_helper.interactive_terminal_exec_command.return_value = str( + [{'empty': 'value'}]).replace("\'", "\"") + vat_terminal = VatTerminal(ssh_helper, json_param=True) + + self.assertEqual([{'empty': 'value'}], + vat_terminal.vat_terminal_exec_cmd( + "hw_interface_set_mtu sw_if_index 1 mtu 9200")) + + def test_vat_terminal_exec_cmd_without_output(self): + ssh_helper = mock.Mock() + ssh_helper.interactive_terminal_exec_command.return_value = str( + {'empty': 'value'}).replace("\'", "\"") + vat_terminal = VatTerminal(ssh_helper, json_param=False) + + self.assertIsNone(vat_terminal.vat_terminal_exec_cmd( + "hw_interface_set_mtu sw_if_index 1 mtu 9200")) + + def test_vat_terminal_exec_cmd_error(self): + ssh_helper = mock.Mock() + ssh_helper.interactive_terminal_exec_command.return_value = str( + {'empty': 'value'}).replace("\'", "\"") + ssh_helper.interactive_terminal_exec_command.side_effect = exceptions.SSHTimeout + + vat_terminal = VatTerminal(ssh_helper, json_param=True) + + with self.assertRaises(RuntimeError) as raised: + vat_terminal.vat_terminal_exec_cmd( + "hw_interface_set_mtu sw_if_index 1 mtu 9200") + self.assertIn( + 'VPP is not running on node. VAT command hw_interface_set_mtu ' \ + 'sw_if_index 1 mtu 9200 execution failed', + str(raised.exception)) + + def test_vat_terminal_exec_cmd_output_error(self): + ssh_helper = mock.Mock() + ssh_helper.interactive_terminal_exec_command.return_value = str( + 'empty: value').replace("\'", "\"") + + vat_terminal = VatTerminal(ssh_helper, json_param=True) + + with self.assertRaises(RuntimeError) as raised: + vat_terminal.vat_terminal_exec_cmd( + "hw_interface_set_mtu sw_if_index 1 mtu 9200") + self.assertIn( + 'VAT command hw_interface_set_mtu sw_if_index 1 mtu 9200: no JSON data.', + str(raised.exception)) + + def test_vat_terminal_close(self): + ssh_helper = mock.Mock() + vat_terminal = VatTerminal(ssh_helper, json_param=False) + self.assertIsNone(vat_terminal.vat_terminal_close()) + + def test_vat_terminal_close_error(self): + ssh_helper = mock.Mock() + ssh_helper.interactive_terminal_exec_command.side_effect = exceptions.SSHTimeout + vat_terminal = VatTerminal(ssh_helper, json_param=False) + with self.assertRaises(RuntimeError) as raised: + vat_terminal.vat_terminal_close() + self.assertIn('Failed to close VAT console', str(raised.exception)) + + def test_vat_terminal_close_vat_error(self): + ssh_helper = mock.Mock() + ssh_helper.interactive_terminal_close.side_effect = exceptions.SSHTimeout + vat_terminal = VatTerminal(ssh_helper, json_param=False) + with self.assertRaises(RuntimeError) as raised: + vat_terminal.vat_terminal_close() + self.assertIn('Cannot close interactive terminal', + str(raised.exception)) + + def test_vat_terminal_exec_cmd_from_template(self): + ssh_helper = mock.Mock() + vat_terminal = VatTerminal(ssh_helper, json_param=False) + + with mock.patch.object(vat_terminal, 'vat_terminal_exec_cmd') as \ + mock_vat_terminal_exec_cmd: + mock_vat_terminal_exec_cmd.return_value = 'empty' + self.assertEqual(['empty'], + vat_terminal.vat_terminal_exec_cmd_from_template( + "hw_interface_set_mtu.vat", sw_if_index=1, + mtu=9200)) |