summaryrefslogtreecommitdiffstats
path: root/tests/unit/network_services
diff options
context:
space:
mode:
Diffstat (limited to 'tests/unit/network_services')
-rw-r--r--tests/unit/network_services/collector/__init__.py0
-rw-r--r--tests/unit/network_services/collector/test_publisher.py39
-rw-r--r--tests/unit/network_services/collector/test_subscriber.py96
-rw-r--r--tests/unit/network_services/helpers/__init__.py0
-rw-r--r--tests/unit/network_services/helpers/acl_vnf_topology_ixia.yaml50
-rw-r--r--tests/unit/network_services/helpers/test_cpu.py117
-rw-r--r--tests/unit/network_services/helpers/test_dpdkbindnic_helper.py633
-rw-r--r--tests/unit/network_services/helpers/test_iniparser.py225
-rw-r--r--tests/unit/network_services/helpers/test_samplevnf_helper.py1109
-rw-r--r--tests/unit/network_services/libs/__init__.py0
-rw-r--r--tests/unit/network_services/libs/ixia_libs/__init__.py0
-rw-r--r--tests/unit/network_services/libs/ixia_libs/test_IxNet.py874
-rw-r--r--tests/unit/network_services/nfvi/__init__.py0
-rw-r--r--tests/unit/network_services/nfvi/test_collectd.py151
-rw-r--r--tests/unit/network_services/nfvi/test_resource.py273
-rw-r--r--tests/unit/network_services/test_utils.py143
-rw-r--r--tests/unit/network_services/test_yang_model.py135
-rw-r--r--tests/unit/network_services/traffic_profile/__init__.py0
-rw-r--r--tests/unit/network_services/traffic_profile/test_base.py73
-rw-r--r--tests/unit/network_services/traffic_profile/test_fixed.py120
-rw-r--r--tests/unit/network_services/traffic_profile/test_http.py41
-rw-r--r--tests/unit/network_services/traffic_profile/test_http_ixload.py272
-rw-r--r--tests/unit/network_services/traffic_profile/test_ixia_rfc2544.py611
-rw-r--r--tests/unit/network_services/traffic_profile/test_prox_acl.py78
-rw-r--r--tests/unit/network_services/traffic_profile/test_prox_binsearch.py185
-rw-r--r--tests/unit/network_services/traffic_profile/test_prox_profile.py130
-rw-r--r--tests/unit/network_services/traffic_profile/test_prox_ramp.py97
-rw-r--r--tests/unit/network_services/traffic_profile/test_rfc2544.py276
-rw-r--r--tests/unit/network_services/traffic_profile/test_trex_traffic_profile.py309
-rw-r--r--tests/unit/network_services/vnf_generic/vnf/test_acl_vnf.py25
-rw-r--r--tests/unit/network_services/vnf_generic/vnf/test_base.py8
-rw-r--r--tests/unit/network_services/vnf_generic/vnf/test_cgnapt_vnf.py23
-rw-r--r--tests/unit/network_services/vnf_generic/vnf/test_sample_vnf.py107
-rw-r--r--tests/unit/network_services/vnf_generic/vnf/test_vfw_vnf.py23
-rw-r--r--tests/unit/network_services/vnf_generic/vnf/test_vpe_vnf.py11
35 files changed, 145 insertions, 6089 deletions
diff --git a/tests/unit/network_services/collector/__init__.py b/tests/unit/network_services/collector/__init__.py
deleted file mode 100644
index e69de29bb..000000000
--- a/tests/unit/network_services/collector/__init__.py
+++ /dev/null
diff --git a/tests/unit/network_services/collector/test_publisher.py b/tests/unit/network_services/collector/test_publisher.py
deleted file mode 100644
index 4a175841d..000000000
--- a/tests/unit/network_services/collector/test_publisher.py
+++ /dev/null
@@ -1,39 +0,0 @@
-# Copyright (c) 2016-2017 Intel Corporation
-#
-# Licensed under the Apache License, Version 2.0 (the "License");
-# you may not use this file except in compliance with the License.
-# You may obtain a copy of the License at
-#
-# http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing, software
-# distributed under the License is distributed on an "AS IS" BASIS,
-# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-# See the License for the specific language governing permissions and
-# limitations under the License.
-#
-
-# Unittest for yardstick.network_services.collector.publisher
-
-from __future__ import absolute_import
-import unittest
-
-from yardstick.network_services.collector import publisher
-
-
-class PublisherTestCase(unittest.TestCase):
-
- def setUp(self):
- self.test_publisher = publisher.Publisher()
-
- def test_successful_init(self):
- pass
-
- def test_unsuccessful_init(self):
- pass
-
- def test_start(self):
- self.assertIsNone(self.test_publisher.start())
-
- def test_stop(self):
- self.assertIsNone(self.test_publisher.stop())
diff --git a/tests/unit/network_services/collector/test_subscriber.py b/tests/unit/network_services/collector/test_subscriber.py
deleted file mode 100644
index d4b4ecf7a..000000000
--- a/tests/unit/network_services/collector/test_subscriber.py
+++ /dev/null
@@ -1,96 +0,0 @@
-# Copyright (c) 2016-2017 Intel Corporation
-#
-# Licensed under the Apache License, Version 2.0 (the "License");
-# you may not use this file except in compliance with the License.
-# You may obtain a copy of the License at
-#
-# http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing, software
-# distributed under the License is distributed on an "AS IS" BASIS,
-# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-# See the License for the specific language governing permissions and
-# limitations under the License.
-#
-
-# Unittest for yardstick.network_services.collector.subscriber
-
-from __future__ import absolute_import
-import unittest
-import mock
-
-from yardstick.network_services.collector import subscriber
-
-
-class MockVnfAprrox(object):
-
- def __init__(self):
- self.result = {}
- self.name = "vnf__1"
-
- def collect_kpi(self):
- self.result = {
- 'pkt_in_up_stream': 100,
- 'pkt_drop_up_stream': 5,
- 'pkt_in_down_stream': 50,
- 'pkt_drop_down_stream': 40
- }
- return self.result
-
-
-class CollectorTestCase(unittest.TestCase):
-
- NODES = {
- 'node1': {},
- 'node2': {
- 'ip': '1.2.3.4',
- 'collectd': {
- 'plugins': {'abc': 12, 'def': 34},
- 'interval': 987,
- },
- },
- }
- TRAFFIC_PROFILE = {
- 'key1': 'value1',
- }
-
- def setUp(self):
- vnf = MockVnfAprrox()
- self.ssh_patch = mock.patch('yardstick.network_services.nfvi.resource.ssh', autospec=True)
- mock_ssh = self.ssh_patch.start()
- mock_instance = mock.Mock()
- mock_instance.execute.return_value = 0, '', ''
- mock_ssh.AutoConnectSSH.from_node.return_value = mock_instance
- self.collector = subscriber.Collector([vnf], self.NODES, self.TRAFFIC_PROFILE, 1800)
-
- def tearDown(self):
- self.ssh_patch.stop()
-
- def test___init__(self, *_):
- vnf = MockVnfAprrox()
- collector = subscriber.Collector([vnf], {}, {})
- self.assertEqual(len(collector.vnfs), 1)
- self.assertEqual(collector.traffic_profile, {})
-
- def test___init___with_data(self, *_):
- self.assertEqual(len(self.collector.vnfs), 1)
- self.assertDictEqual(self.collector.traffic_profile, self.TRAFFIC_PROFILE)
- self.assertEqual(len(self.collector.resource_profiles), 1)
-
- def test___init___negative(self, *_):
- pass
-
- def test_start(self, *_):
- self.assertIsNone(self.collector.start())
-
- def test_stop(self, *_):
- self.assertIsNone(self.collector.stop())
-
- def test_get_kpi(self, *_):
- result = self.collector.get_kpi()
-
- self.assertEqual(result["vnf__1"]["pkt_in_up_stream"], 100)
- self.assertEqual(result["vnf__1"]["pkt_drop_up_stream"], 5)
- self.assertEqual(result["vnf__1"]["pkt_in_down_stream"], 50)
- self.assertEqual(result["vnf__1"]["pkt_drop_down_stream"], 40)
- self.assertIn('node2', result)
diff --git a/tests/unit/network_services/helpers/__init__.py b/tests/unit/network_services/helpers/__init__.py
deleted file mode 100644
index e69de29bb..000000000
--- a/tests/unit/network_services/helpers/__init__.py
+++ /dev/null
diff --git a/tests/unit/network_services/helpers/acl_vnf_topology_ixia.yaml b/tests/unit/network_services/helpers/acl_vnf_topology_ixia.yaml
deleted file mode 100644
index f60834fbd..000000000
--- a/tests/unit/network_services/helpers/acl_vnf_topology_ixia.yaml
+++ /dev/null
@@ -1,50 +0,0 @@
-# Copyright (c) 2016-2017 Intel Corporation
-#
-# Licensed under the Apache License, Version 2.0 (the "License");
-# you may not use this file except in compliance with the License.
-# You may obtain a copy of the License at
-#
-# http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing, software
-# distributed under the License is distributed on an "AS IS" BASIS,
-# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-# See the License for the specific language governing permissions and
-# limitations under the License.
-
-nsd:nsd-catalog:
- nsd:
- - id: VACL
- name: VACL
- short-name: VACL
- description: scenario with VACL,L3fwd and VNF
- constituent-vnfd:
- - member-vnf-index: '1'
- vnfd-id-ref: tg__1
- VNF model: ../../vnf_descriptors/ixia_rfc2544_tpl.yaml
- - member-vnf-index: '2'
- vnfd-id-ref: vnf__1
- VNF model: ../../vnf_descriptors/acl_vnf.yaml
-
- vld:
- - id: uplink_1
- name: tg__1 to vnf__1 link 1
- type: ELAN
- vnfd-connection-point-ref:
- - member-vnf-index-ref: '1'
- vnfd-connection-point-ref: xe0
- vnfd-id-ref: tg__1 #TREX
- - member-vnf-index-ref: '2'
- vnfd-connection-point-ref: xe0
- vnfd-id-ref: vnf__1 #VNF
-
- - id: downlink_1
- name: vnf__1 to tg__1 link 2
- type: ELAN
- vnfd-connection-point-ref:
- - member-vnf-index-ref: '2'
- vnfd-connection-point-ref: xe1
- vnfd-id-ref: vnf__1 #L3fwd
- - member-vnf-index-ref: '1'
- vnfd-connection-point-ref: xe1
- vnfd-id-ref: tg__1 #VACL VNF
diff --git a/tests/unit/network_services/helpers/test_cpu.py b/tests/unit/network_services/helpers/test_cpu.py
deleted file mode 100644
index 1f9d3f219..000000000
--- a/tests/unit/network_services/helpers/test_cpu.py
+++ /dev/null
@@ -1,117 +0,0 @@
-# Copyright (c) 2016-2017 Intel Corporation
-#
-# Licensed under the Apache License, Version 2.0 (the "License");
-# you may not use this file except in compliance with the License.
-# You may obtain a copy of the License at
-#
-# http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing, software
-# distributed under the License is distributed on an "AS IS" BASIS,
-# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-# See the License for the specific language governing permissions and
-# limitations under the License.
-#
-
-from __future__ import absolute_import
-from __future__ import division
-import unittest
-import mock
-import subprocess
-
-from yardstick.network_services.helpers.cpu import \
- CpuSysCores
-
-
-class TestCpuSysCores(unittest.TestCase):
-
- def test___init__(self):
- with mock.patch("yardstick.ssh.SSH") as ssh:
- ssh_mock = mock.Mock(autospec=ssh.SSH)
- ssh_mock.execute = \
- mock.Mock(return_value=(1, "", ""))
- ssh_mock.put = \
- mock.Mock(return_value=(1, "", ""))
- cpu_topo = CpuSysCores(ssh_mock)
- self.assertIsNotNone(cpu_topo.connection)
-
- def test__get_core_details(self):
- with mock.patch("yardstick.ssh.SSH") as ssh:
- ssh_mock = mock.Mock(autospec=ssh.SSH)
- ssh_mock.execute = \
- mock.Mock(return_value=(1, "", ""))
- ssh_mock.put = \
- mock.Mock(return_value=(1, "", ""))
- cpu_topo = CpuSysCores(ssh_mock)
- subprocess.check_output = mock.Mock(return_value=0)
- lines = ["cpu:1", "topo:2", ""]
- self.assertEqual([{'topo': '2', 'cpu': '1'}],
- cpu_topo._get_core_details(lines))
-
- def test_get_core_socket(self):
- with mock.patch("yardstick.ssh.SSH") as ssh:
- ssh_mock = mock.Mock(autospec=ssh.SSH)
- ssh_mock.execute = \
- mock.Mock(return_value=(1, "cpu:1\ntest:2\n \n", ""))
- ssh_mock.put = \
- mock.Mock(return_value=(1, "", ""))
- cpu_topo = CpuSysCores(ssh_mock)
- subprocess.check_output = mock.Mock(return_value=0)
- cpu_topo._get_core_details = \
- mock.Mock(side_effect=[[{'Core(s) per socket': '2', 'Thread(s) per core': '1'}],
- [{'physical id': '2', 'processor': '1'}]])
- self.assertEqual({'thread_per_core': '1', '2': ['1'],
- 'cores_per_socket': '2'},
- cpu_topo.get_core_socket())
-
- def test_validate_cpu_cfg(self):
- with mock.patch("yardstick.ssh.SSH") as ssh:
- ssh_mock = mock.Mock(autospec=ssh.SSH)
- ssh_mock.execute = \
- mock.Mock(return_value=(1, "cpu:1\ntest:2\n \n", ""))
- ssh_mock.put = \
- mock.Mock(return_value=(1, "", ""))
- cpu_topo = CpuSysCores(ssh_mock)
- subprocess.check_output = mock.Mock(return_value=0)
- cpu_topo._get_core_details = \
- mock.Mock(side_effect=[[{'Core(s) per socket': '2', 'Thread(s) per core': '1'}],
- [{'physical id': '2', 'processor': '1'}]])
- cpu_topo.core_map = \
- {'thread_per_core': '1', '2':['1'], 'cores_per_socket': '2'}
- self.assertEqual(-1, cpu_topo.validate_cpu_cfg())
-
- def test_validate_cpu_cfg_2t(self):
- with mock.patch("yardstick.ssh.SSH") as ssh:
- ssh_mock = mock.Mock(autospec=ssh.SSH)
- ssh_mock.execute = \
- mock.Mock(return_value=(1, "cpu:1\ntest:2\n \n", ""))
- ssh_mock.put = \
- mock.Mock(return_value=(1, "", ""))
- cpu_topo = CpuSysCores(ssh_mock)
- subprocess.check_output = mock.Mock(return_value=0)
- cpu_topo._get_core_details = \
- mock.Mock(side_effect=[[{'Core(s) per socket': '2', 'Thread(s) per core': '1'}],
- [{'physical id': '2', 'processor': '1'}]])
- cpu_topo.core_map = \
- {'thread_per_core': 1, '2':['1'], 'cores_per_socket': '2'}
- vnf_cfg = {'lb_config': 'SW', 'lb_count': 1, 'worker_config':
- '1C/2T', 'worker_threads': 1}
- self.assertEqual(-1, cpu_topo.validate_cpu_cfg(vnf_cfg))
-
- def test_validate_cpu_cfg_fail(self):
- with mock.patch("yardstick.ssh.SSH") as ssh:
- ssh_mock = mock.Mock(autospec=ssh.SSH)
- ssh_mock.execute = \
- mock.Mock(return_value=(1, "cpu:1\ntest:2\n \n", ""))
- ssh_mock.put = \
- mock.Mock(return_value=(1, "", ""))
- cpu_topo = CpuSysCores(ssh_mock)
- subprocess.check_output = mock.Mock(return_value=0)
- cpu_topo._get_core_details = \
- mock.Mock(side_effect=[[{'Core(s) per socket': '2', 'Thread(s) per core': '1'}],
- [{'physical id': '2', 'processor': '1'}]])
- cpu_topo.core_map = \
- {'thread_per_core': 1, '2':[1], 'cores_per_socket': 2}
- vnf_cfg = {'lb_config': 'SW', 'lb_count': 1, 'worker_config':
- '1C/1T', 'worker_threads': 1}
- self.assertEqual(-1, cpu_topo.validate_cpu_cfg(vnf_cfg))
diff --git a/tests/unit/network_services/helpers/test_dpdkbindnic_helper.py b/tests/unit/network_services/helpers/test_dpdkbindnic_helper.py
deleted file mode 100644
index 367072e84..000000000
--- a/tests/unit/network_services/helpers/test_dpdkbindnic_helper.py
+++ /dev/null
@@ -1,633 +0,0 @@
-# Copyright (c) 2016-2017 Intel Corporation
-#
-# Licensed under the Apache License, Version 2.0 (the "License");
-# you may not use this file except in compliance with the License.
-# You may obtain a copy of the License at
-#
-# http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing, software
-# distributed under the License is distributed on an "AS IS" BASIS,
-# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-# See the License for the specific language governing permissions and
-# limitations under the License.
-
-import mock
-import unittest
-
-import os
-
-from yardstick.error import IncorrectConfig, SSHError
-from yardstick.error import IncorrectNodeSetup
-from yardstick.error import IncorrectSetup
-from yardstick.network_services.helpers.dpdkbindnic_helper import DpdkInterface
-from yardstick.network_services.helpers.dpdkbindnic_helper import DpdkNode
-from yardstick.network_services.helpers.dpdkbindnic_helper import DpdkBindHelper
-from yardstick.network_services.helpers.dpdkbindnic_helper import DpdkBindHelperException
-from yardstick.network_services.helpers.dpdkbindnic_helper import NETWORK_KERNEL
-from yardstick.network_services.helpers.dpdkbindnic_helper import NETWORK_DPDK
-from yardstick.network_services.helpers.dpdkbindnic_helper import CRYPTO_KERNEL
-from yardstick.network_services.helpers.dpdkbindnic_helper import CRYPTO_DPDK
-from yardstick.network_services.helpers.dpdkbindnic_helper import NETWORK_OTHER
-from yardstick.network_services.helpers.dpdkbindnic_helper import CRYPTO_OTHER
-
-
-NAME = "tg_0"
-
-
-class TestDpdkInterface(unittest.TestCase):
-
- SAMPLE_NETDEVS = {
- 'enp11s0': {
- 'address': '0a:de:ad:be:ef:f5',
- 'device': '0x1533',
- 'driver': 'igb',
- 'ifindex': '2',
- 'interface_name': 'enp11s0',
- 'operstate': 'down',
- 'pci_bus_id': '0000:0b:00.0',
- 'subsystem_device': '0x1533',
- 'subsystem_vendor': '0x15d9',
- 'vendor': '0x8086'
- },
- 'lan': {
- 'address': '0a:de:ad:be:ef:f4',
- 'device': '0x153a',
- 'driver': 'e1000e',
- 'ifindex': '3',
- 'interface_name': 'lan',
- 'operstate': 'up',
- 'pci_bus_id': '0000:00:19.0',
- 'subsystem_device': '0x153a',
- 'subsystem_vendor': '0x15d9',
- 'vendor': '0x8086'
- }
- }
-
- SAMPLE_VM_NETDEVS = {
- 'eth1': {
- 'address': 'fa:de:ad:be:ef:5b',
- 'device': '0x0001',
- 'driver': 'virtio_net',
- 'ifindex': '3',
- 'interface_name': 'eth1',
- 'operstate': 'down',
- 'pci_bus_id': '0000:00:04.0',
- 'vendor': '0x1af4'
- }
- }
-
- def test_parse_netdev_info(self):
- output = """\
-/sys/devices/pci0000:00/0000:00:1c.3/0000:0b:00.0/net/enp11s0/ifindex:2
-/sys/devices/pci0000:00/0000:00:1c.3/0000:0b:00.0/net/enp11s0/address:0a:de:ad:be:ef:f5
-/sys/devices/pci0000:00/0000:00:1c.3/0000:0b:00.0/net/enp11s0/operstate:down
-/sys/devices/pci0000:00/0000:00:1c.3/0000:0b:00.0/net/enp11s0/device/vendor:0x8086
-/sys/devices/pci0000:00/0000:00:1c.3/0000:0b:00.0/net/enp11s0/device/device:0x1533
-/sys/devices/pci0000:00/0000:00:1c.3/0000:0b:00.0/net/enp11s0/device/subsystem_vendor:0x15d9
-/sys/devices/pci0000:00/0000:00:1c.3/0000:0b:00.0/net/enp11s0/device/subsystem_device:0x1533
-/sys/devices/pci0000:00/0000:00:1c.3/0000:0b:00.0/net/enp11s0/driver:igb
-/sys/devices/pci0000:00/0000:00:1c.3/0000:0b:00.0/net/enp11s0/pci_bus_id:0000:0b:00.0
-/sys/devices/pci0000:00/0000:00:19.0/net/lan/ifindex:3
-/sys/devices/pci0000:00/0000:00:19.0/net/lan/address:0a:de:ad:be:ef:f4
-/sys/devices/pci0000:00/0000:00:19.0/net/lan/operstate:up
-/sys/devices/pci0000:00/0000:00:19.0/net/lan/device/vendor:0x8086
-/sys/devices/pci0000:00/0000:00:19.0/net/lan/device/device:0x153a
-/sys/devices/pci0000:00/0000:00:19.0/net/lan/device/subsystem_vendor:0x15d9
-/sys/devices/pci0000:00/0000:00:19.0/net/lan/device/subsystem_device:0x153a
-/sys/devices/pci0000:00/0000:00:19.0/net/lan/driver:e1000e
-/sys/devices/pci0000:00/0000:00:19.0/net/lan/pci_bus_id:0000:00:19.0
-"""
- res = DpdkBindHelper.parse_netdev_info(output)
- self.assertDictEqual(res, self.SAMPLE_NETDEVS)
-
- def test_parse_netdev_info_virtio(self):
- output = """\
-/sys/devices/pci0000:00/0000:00:04.0/virtio1/net/eth1/ifindex:3
-/sys/devices/pci0000:00/0000:00:04.0/virtio1/net/eth1/address:fa:de:ad:be:ef:5b
-/sys/devices/pci0000:00/0000:00:04.0/virtio1/net/eth1/operstate:down
-/sys/devices/pci0000:00/0000:00:04.0/virtio1/net/eth1/device/vendor:0x1af4
-/sys/devices/pci0000:00/0000:00:04.0/virtio1/net/eth1/device/device:0x0001
-/sys/devices/pci0000:00/0000:00:04.0/virtio1/net/eth1/driver:virtio_net
-"""
- res = DpdkBindHelper.parse_netdev_info(output)
- self.assertDictEqual(res, self.SAMPLE_VM_NETDEVS)
-
- def test_probe_missing_values(self):
- mock_dpdk_node = mock.Mock()
- mock_dpdk_node.netdevs = self.SAMPLE_NETDEVS.copy()
-
- interface = {'local_mac': '0a:de:ad:be:ef:f5'}
- dpdk_intf = DpdkInterface(mock_dpdk_node, interface)
-
- dpdk_intf.probe_missing_values()
- self.assertEqual(interface['vpci'], '0000:0b:00.0')
-
- interface['local_mac'] = '0a:de:ad:be:ef:f4'
- dpdk_intf.probe_missing_values()
- self.assertEqual(interface['vpci'], '0000:00:19.0')
-
- def test_probe_missing_values_no_update(self):
- mock_dpdk_node = mock.Mock()
- mock_dpdk_node.netdevs = self.SAMPLE_NETDEVS.copy()
- del mock_dpdk_node.netdevs['enp11s0']['driver']
- del mock_dpdk_node.netdevs['lan']['driver']
-
- interface = {'local_mac': '0a:de:ad:be:ef:f5'}
- dpdk_intf = DpdkInterface(mock_dpdk_node, interface)
-
- dpdk_intf.probe_missing_values()
- self.assertNotIn('vpci', interface)
- self.assertNotIn('driver', interface)
-
- def test_probe_missing_values_negative(self):
- mock_dpdk_node = mock.Mock()
- mock_dpdk_node.netdevs.values.side_effect = IncorrectNodeSetup
-
- interface = {'local_mac': '0a:de:ad:be:ef:f5'}
- dpdk_intf = DpdkInterface(mock_dpdk_node, interface)
-
- with self.assertRaises(IncorrectConfig):
- dpdk_intf.probe_missing_values()
-
-
-class TestDpdkNode(unittest.TestCase):
-
- INTERFACES = [
- {'name': 'name1',
- 'virtual-interface': {
- 'local_mac': 404,
- 'vpci': 'pci10',
- }},
- {'name': 'name2',
- 'virtual-interface': {
- 'local_mac': 404,
- 'vpci': 'pci2',
- }},
- {'name': 'name3',
- 'virtual-interface': {
- 'local_mac': 404,
- 'vpci': 'some-pci1',
- }},
- ]
-
- def test_probe_dpdk_drivers(self):
- mock_ssh_helper = mock.Mock()
- mock_ssh_helper.execute.return_value = 0, '', ''
-
- interfaces = [
- {'name': 'name1',
- 'virtual-interface': {
- 'local_mac': 404,
- 'vpci': 'pci10',
- }},
- {'name': 'name2',
- 'virtual-interface': {
- 'local_mac': 404,
- 'vpci': 'pci2',
- }},
- {'name': 'name3',
- 'virtual-interface': {
- 'local_mac': 404,
- 'vpci': 'some-pci1',
- }},
- ]
-
- dpdk_node = DpdkNode(NAME, interfaces, mock_ssh_helper)
- dpdk_helper = dpdk_node.dpdk_helper
-
- dpdk_helper.probe_real_kernel_drivers = mock.Mock()
- dpdk_helper.real_kernel_interface_driver_map = {
- 'pci1': 'driver1',
- 'pci2': 'driver2',
- 'pci3': 'driver3',
- 'pci4': 'driver1',
- 'pci6': 'driver3',
- }
-
- dpdk_node._probe_dpdk_drivers()
- self.assertNotIn('driver', interfaces[0]['virtual-interface'])
- self.assertEqual(interfaces[1]['virtual-interface']['driver'], 'driver2')
- self.assertEqual(interfaces[2]['virtual-interface']['driver'], 'driver1')
-
- def test_check(self):
- def update():
- if not mock_force_rebind.called:
- raise IncorrectConfig
-
- interfaces[0]['virtual-interface'].update({
- 'vpci': '0000:01:02.1',
- 'local_ip': '10.20.30.40',
- 'netmask': '255.255.0.0',
- 'driver': 'ixgbe',
- })
-
- mock_ssh_helper = mock.Mock()
- mock_ssh_helper.execute.return_value = 0, '', ''
-
- interfaces = [
- {'name': 'name1',
- 'virtual-interface': {
- 'local_mac': 404,
- }},
- ]
-
- dpdk_node = DpdkNode(NAME, interfaces, mock_ssh_helper)
- dpdk_node._probe_missing_values = mock_probe_missing = mock.Mock(side_effect=update)
- dpdk_node._force_rebind = mock_force_rebind = mock.Mock()
-
- self.assertIsNone(dpdk_node.check())
- self.assertEqual(mock_probe_missing.call_count, 2)
-
- @mock.patch('yardstick.network_services.helpers.dpdkbindnic_helper.DpdkInterface')
- def test_check_negative(self, mock_intf_type):
- mock_ssh_helper = mock.Mock()
- mock_ssh_helper.execute.return_value = 0, '', ''
-
- mock_intf_type().check.side_effect = SSHError
-
- dpdk_node = DpdkNode(NAME, self.INTERFACES, mock_ssh_helper)
-
- with self.assertRaises(IncorrectSetup):
- dpdk_node.check()
-
- def test_probe_netdevs(self):
- mock_ssh_helper = mock.Mock()
- mock_ssh_helper.execute.return_value = 0, '', ''
-
- expected = {'key1': 500, 'key2': 'hello world'}
- update = {'key1': 1000, 'key3': []}
-
- dpdk_node = DpdkNode(NAME, self.INTERFACES, mock_ssh_helper)
- dpdk_helper = dpdk_node.dpdk_helper
- dpdk_helper.find_net_devices = mock.Mock(side_effect=[expected, update])
-
- self.assertDictEqual(dpdk_node.netdevs, {})
- dpdk_node._probe_netdevs()
- self.assertDictEqual(dpdk_node.netdevs, expected)
-
- expected = {'key1': 1000, 'key2': 'hello world', 'key3': []}
- dpdk_node._probe_netdevs()
- self.assertDictEqual(dpdk_node.netdevs, expected)
-
- def test_probe_netdevs_setup_negative(self):
- mock_ssh_helper = mock.Mock()
- mock_ssh_helper.execute.return_value = 0, '', ''
-
- dpdk_node = DpdkNode(NAME, self.INTERFACES, mock_ssh_helper)
- dpdk_helper = dpdk_node.dpdk_helper
- dpdk_helper.find_net_devices = mock.Mock(side_effect=DpdkBindHelperException)
-
- with self.assertRaises(DpdkBindHelperException):
- dpdk_node._probe_netdevs()
-
- def test_force_rebind(self):
- mock_ssh_helper = mock.Mock()
- mock_ssh_helper.execute.return_value = 0, '', ''
-
- dpdk_node = DpdkNode(NAME, self.INTERFACES, mock_ssh_helper)
- dpdk_helper = dpdk_node.dpdk_helper
- dpdk_helper.force_dpdk_rebind = mock_helper_func = mock.Mock()
-
- dpdk_node._force_rebind()
- self.assertEqual(mock_helper_func.call_count, 1)
-
-
-class TestDpdkBindHelper(unittest.TestCase):
- bin_path = "/opt/nsb_bin"
- EXAMPLE_OUTPUT = """
-
-Network devices using DPDK-compatible driver
-============================================
-0000:00:04.0 'Virtio network device' drv=igb_uio unused=
-0000:00:05.0 'Virtio network device' drv=igb_uio unused=
-
-Network devices using kernel driver
-===================================
-0000:00:03.0 'Virtio network device' if=ens3 drv=virtio-pci unused=igb_uio *Active*
-
-Other network devices
-=====================
-<none>
-
-Crypto devices using DPDK-compatible driver
-===========================================
-<none>
-
-Crypto devices using kernel driver
-==================================
-<none>
-
-Other crypto devices
-====================
-<none>
-"""
-
- PARSED_EXAMPLE = {
- NETWORK_DPDK: [
- {'active': False,
- 'dev_type': 'Virtio network device',
- 'driver': 'igb_uio',
- 'iface': None,
- 'unused': '',
- 'vpci': '0000:00:04.0',
- },
- {'active': False,
- 'dev_type': 'Virtio network device',
- 'driver': 'igb_uio',
- 'iface': None,
- 'unused': '',
- 'vpci': '0000:00:05.0',
- }
- ],
- NETWORK_KERNEL: [
- {'active': True,
- 'dev_type': 'Virtio network device',
- 'driver': 'virtio-pci',
- 'iface': 'ens3',
- 'unused': 'igb_uio',
- 'vpci': '0000:00:03.0',
- }
- ],
- CRYPTO_KERNEL: [],
- CRYPTO_DPDK: [],
- NETWORK_OTHER: [],
- CRYPTO_OTHER: [],
- }
-
- CLEAN_STATUS = {
- NETWORK_KERNEL: [],
- NETWORK_DPDK: [],
- CRYPTO_KERNEL: [],
- CRYPTO_DPDK: [],
- NETWORK_OTHER: [],
- CRYPTO_OTHER: [],
- }
-
- ONE_INPUT_LINE = ("0000:00:03.0 'Virtio network device' if=ens3 "
- "drv=virtio-pci unused=igb_uio *Active*")
-
- ONE_INPUT_LINE_PARSED = [{
- 'vpci': '0000:00:03.0',
- 'dev_type': 'Virtio network device',
- 'iface': 'ens3',
- 'driver': 'virtio-pci',
- 'unused': 'igb_uio',
- 'active': True,
- }]
-
- def test___init__(self):
- conn = mock.Mock()
- conn.provision_tool = mock.Mock(return_value='path_to_tool')
- conn.join_bin_path.return_value = os.path.join(self.bin_path, DpdkBindHelper.DPDK_DEVBIND)
-
- dpdk_bind_helper = DpdkBindHelper(conn)
-
- self.assertEqual(conn, dpdk_bind_helper.ssh_helper)
- self.assertEqual(self.CLEAN_STATUS, dpdk_bind_helper.dpdk_status)
- self.assertIsNone(dpdk_bind_helper.status_nic_row_re)
- self.assertEqual(dpdk_bind_helper.dpdk_devbind,
- os.path.join(self.bin_path, dpdk_bind_helper.DPDK_DEVBIND))
- self.assertIsNone(dpdk_bind_helper._status_cmd_attr)
-
- def test__dpdk_execute(self):
- conn = mock.Mock()
- conn.execute = mock.Mock(return_value=(0, 'output', 'error'))
- conn.provision_tool = mock.Mock(return_value='tool_path')
- dpdk_bind_helper = DpdkBindHelper(conn)
- self.assertEqual((0, 'output', 'error'), dpdk_bind_helper._dpdk_execute('command'))
-
- def test__dpdk_execute_failure(self):
- conn = mock.Mock()
- conn.execute = mock.Mock(return_value=(1, 'output', 'error'))
- conn.provision_tool = mock.Mock(return_value='tool_path')
- dpdk_bind_helper = DpdkBindHelper(conn)
- with self.assertRaises(DpdkBindHelperException):
- dpdk_bind_helper._dpdk_execute('command')
-
- def test__addline(self):
- conn = mock.Mock()
-
- dpdk_bind_helper = DpdkBindHelper(conn)
-
- dpdk_bind_helper._add_line(NETWORK_KERNEL, self.ONE_INPUT_LINE)
-
- self.assertIsNotNone(dpdk_bind_helper.dpdk_status)
- self.assertEqual(self.ONE_INPUT_LINE_PARSED, dpdk_bind_helper.dpdk_status[NETWORK_KERNEL])
-
- def test__switch_active_dict_by_header(self):
- line = "Crypto devices using DPDK-compatible driver"
- olddict = 'olddict'
- self.assertEqual(CRYPTO_DPDK, DpdkBindHelper._switch_active_dict(line, olddict))
-
- def test__switch_active_dict_by_header_empty(self):
- line = "<none>"
- olddict = 'olddict'
- self.assertEqual(olddict, DpdkBindHelper._switch_active_dict(line, olddict))
-
- def test_parse_dpdk_status_output(self):
- conn = mock.Mock()
-
- dpdk_bind_helper = DpdkBindHelper(conn)
-
- dpdk_bind_helper._parse_dpdk_status_output(self.EXAMPLE_OUTPUT)
-
- self.maxDiff = None
- self.assertEqual(self.PARSED_EXAMPLE, dpdk_bind_helper.dpdk_status)
-
- def test_kernel_bound_pci_addresses(self):
- mock_ssh_helper = mock.Mock()
- mock_ssh_helper.execute.return_value = 0, '', ''
-
- expected = ['a', 'b', 3]
-
- dpdk_helper = DpdkBindHelper(mock_ssh_helper)
- dpdk_helper.dpdk_status = {
- NETWORK_DPDK: [{'vpci': 4}, {'vpci': 5}, {'vpci': 'g'}],
- NETWORK_KERNEL: [{'vpci': 'a'}, {'vpci': 'b'}, {'vpci': 3}],
- CRYPTO_DPDK: [],
- }
-
- result = dpdk_helper.kernel_bound_pci_addresses
- self.assertEqual(result, expected)
-
- def test_find_net_devices_negative(self):
- mock_ssh_helper = mock.Mock()
- mock_ssh_helper.execute.return_value = 1, 'error', 'debug'
-
- dpdk_helper = DpdkBindHelper(mock_ssh_helper)
-
- self.assertDictEqual(dpdk_helper.find_net_devices(), {})
-
- def test_read_status(self):
- conn = mock.Mock()
- conn.execute = mock.Mock(return_value=(0, self.EXAMPLE_OUTPUT, ''))
- conn.provision_tool = mock.Mock(return_value='path_to_tool')
-
- dpdk_bind_helper = DpdkBindHelper(conn)
-
- self.assertEqual(self.PARSED_EXAMPLE, dpdk_bind_helper.read_status())
-
- def test__get_bound_pci_addresses(self):
- conn = mock.Mock()
-
- dpdk_bind_helper = DpdkBindHelper(conn)
-
- dpdk_bind_helper._parse_dpdk_status_output(self.EXAMPLE_OUTPUT)
-
- self.assertEqual(['0000:00:04.0', '0000:00:05.0'],
- dpdk_bind_helper._get_bound_pci_addresses(NETWORK_DPDK))
- self.assertEqual(['0000:00:03.0'],
- dpdk_bind_helper._get_bound_pci_addresses(NETWORK_KERNEL))
-
- def test_interface_driver_map(self):
- conn = mock.Mock()
-
- dpdk_bind_helper = DpdkBindHelper(conn)
-
- dpdk_bind_helper._parse_dpdk_status_output(self.EXAMPLE_OUTPUT)
-
- self.assertEqual({'0000:00:04.0': 'igb_uio',
- '0000:00:03.0': 'virtio-pci',
- '0000:00:05.0': 'igb_uio',
- },
- dpdk_bind_helper.interface_driver_map)
-
- def test_bind(self):
- conn = mock.Mock()
- conn.execute = mock.Mock(return_value=(0, '', ''))
- conn.join_bin_path.return_value = os.path.join(self.bin_path, DpdkBindHelper.DPDK_DEVBIND)
-
- dpdk_bind_helper = DpdkBindHelper(conn)
- dpdk_bind_helper.read_status = mock.Mock()
-
- dpdk_bind_helper.bind(['0000:00:03.0', '0000:00:04.0'], 'my_driver')
-
- conn.execute.assert_called_with('sudo /opt/nsb_bin/dpdk-devbind.py --force '
- '-b my_driver 0000:00:03.0 0000:00:04.0')
- dpdk_bind_helper.read_status.assert_called_once()
-
- def test_bind_single_pci(self):
- conn = mock.Mock()
- conn.execute = mock.Mock(return_value=(0, '', ''))
- conn.join_bin_path.return_value = os.path.join(self.bin_path, DpdkBindHelper.DPDK_DEVBIND)
-
- dpdk_bind_helper = DpdkBindHelper(conn)
- dpdk_bind_helper.read_status = mock.Mock()
-
- dpdk_bind_helper.bind('0000:00:03.0', 'my_driver')
-
- conn.execute.assert_called_with('sudo /opt/nsb_bin/dpdk-devbind.py --force '
- '-b my_driver 0000:00:03.0')
- dpdk_bind_helper.read_status.assert_called_once()
-
- def test_rebind_drivers(self):
- conn = mock.Mock()
-
- dpdk_bind_helper = DpdkBindHelper(conn)
-
- dpdk_bind_helper.bind = mock.Mock()
- dpdk_bind_helper.used_drivers = {
- 'd1': ['0000:05:00.0'],
- 'd3': ['0000:05:01.0', '0000:05:02.0'],
- }
-
- dpdk_bind_helper.rebind_drivers()
-
- dpdk_bind_helper.bind.assert_any_call(['0000:05:00.0'], 'd1', True)
- dpdk_bind_helper.bind.assert_any_call(['0000:05:01.0', '0000:05:02.0'], 'd3', True)
-
- def test_save_used_drivers(self):
- conn = mock.Mock()
- dpdk_bind_helper = DpdkBindHelper(conn)
- dpdk_bind_helper.dpdk_status = self.PARSED_EXAMPLE
-
- dpdk_bind_helper.save_used_drivers()
-
- expected = {
- 'igb_uio': ['0000:00:04.0', '0000:00:05.0'],
- 'virtio-pci': ['0000:00:03.0'],
- }
-
- self.assertDictEqual(expected, dpdk_bind_helper.used_drivers)
-
- def test_force_dpdk_rebind(self):
- mock_ssh_helper = mock.Mock()
- mock_ssh_helper.execute.return_value = 0, '', ''
-
- dpdk_helper = DpdkBindHelper(mock_ssh_helper, 'driver2')
- dpdk_helper.dpdk_status = {
- NETWORK_DPDK: [
- {
- 'vpci': 'pci1',
- },
- {
- 'vpci': 'pci3',
- },
- {
- 'vpci': 'pci6',
- },
- {
- 'vpci': 'pci3',
- },
- ]
- }
- dpdk_helper.real_kernel_interface_driver_map = {
- 'pci1': 'real_driver1',
- 'pci2': 'real_driver2',
- 'pci3': 'real_driver1',
- 'pci4': 'real_driver4',
- 'pci6': 'real_driver6',
- }
- dpdk_helper.load_dpdk_driver = mock.Mock()
- dpdk_helper.read_status = mock.Mock()
- dpdk_helper.save_real_kernel_interface_driver_map = mock.Mock()
- dpdk_helper.save_used_drivers = mock.Mock()
- dpdk_helper.bind = mock_bind = mock.Mock()
-
- dpdk_helper.force_dpdk_rebind()
- self.assertEqual(mock_bind.call_count, 2)
-
- def test_save_real_kernel_drivers(self):
- mock_ssh_helper = mock.Mock()
- mock_ssh_helper.execute.return_value = 0, '', ''
-
- dpdk_helper = DpdkBindHelper(mock_ssh_helper)
- dpdk_helper.real_kernel_drivers = {
- 'abc': '123',
- }
- dpdk_helper.real_kernel_interface_driver_map = {
- 'abc': 'AAA',
- 'def': 'DDD',
- 'abs': 'AAA',
- 'ghi': 'GGG',
- }
-
- # save_used_drivers must be called before save_real_kernel_drivers can be
- with self.assertRaises(AttributeError):
- dpdk_helper.save_real_kernel_drivers()
-
- dpdk_helper.save_used_drivers()
-
- expected_used_drivers = {
- 'AAA': ['abc', 'abs'],
- 'DDD': ['def'],
- 'GGG': ['ghi'],
- }
- dpdk_helper.save_real_kernel_drivers()
- self.assertDictEqual(dpdk_helper.used_drivers, expected_used_drivers)
- self.assertDictEqual(dpdk_helper.real_kernel_drivers, {})
-
- def test_get_real_kernel_driver(self):
- mock_ssh_helper = mock.Mock()
- mock_ssh_helper.execute.side_effect = [
- (0, 'non-matching text', ''),
- (0, 'pre Kernel modules: real_driver1', ''),
- (0, 'before Ethernet middle Virtio network device after', ''),
- ]
-
- dpdk_helper = DpdkBindHelper(mock_ssh_helper)
-
- self.assertIsNone(dpdk_helper.get_real_kernel_driver('abc'))
- self.assertEqual(dpdk_helper.get_real_kernel_driver('abc'), 'real_driver1')
- self.assertEqual(dpdk_helper.get_real_kernel_driver('abc'), DpdkBindHelper.VIRTIO_DRIVER)
diff --git a/tests/unit/network_services/helpers/test_iniparser.py b/tests/unit/network_services/helpers/test_iniparser.py
deleted file mode 100644
index bd27b497e..000000000
--- a/tests/unit/network_services/helpers/test_iniparser.py
+++ /dev/null
@@ -1,225 +0,0 @@
-# Copyright (c) 2017 Intel Corporation
-#
-# Licensed under the Apache License, Version 2.0 (the "License");
-# you may not use this file except in compliance with the License.
-# You may obtain a copy of the License at
-#
-# http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing, software
-# distributed under the License is distributed on an "AS IS" BASIS,
-# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-# See the License for the specific language governing permissions and
-# limitations under the License.
-#
-
-from __future__ import absolute_import
-
-import unittest
-from contextlib import contextmanager
-import mock
-
-from tests.unit import STL_MOCKS
-
-
-STLClient = mock.MagicMock()
-stl_patch = mock.patch.dict("sys.modules", STL_MOCKS)
-stl_patch.start()
-
-if stl_patch:
- from yardstick.network_services.helpers.iniparser import ParseError
- from yardstick.network_services.helpers.iniparser import LineParser
- from yardstick.network_services.helpers.iniparser import BaseParser
- from yardstick.network_services.helpers.iniparser import ConfigParser
-
-PARSE_TEXT_1 = """\
-
-[section1]
-key1=value1
-list1: value2
- value3
- value4
-key3='single quote value' ; comment here
-key4=
-
-[section2] ; comment with #2 other symbol
-# here is a comment line
-list2: value5
-key with no value # mixed comment ; symbols
-; another comment line
-key5=
-
-[section1] ; reopen a section!
-key2="double quote value"
-"""
-
-PARSE_TEXT_2 = """\
-[section1]
-list1 = item1
- item2
- ended by eof"""
-
-PARSE_TEXT_BAD_1 = """\
-key1=value1
-"""
-
-PARSE_TEXT_BAD_2 = """\
-[section1
-"""
-
-PARSE_TEXT_BAD_3 = """\
-[]
-"""
-
-PARSE_TEXT_BAD_4 = """\
-[section1]
- bad continuation
-"""
-
-PARSE_TEXT_BAD_5 = """\
-[section1]
-=value with no key
-"""
-
-
-class TestParseError(unittest.TestCase):
-
- def test___str__(self):
- error = ParseError('a', 2, 'c')
- self.assertEqual(str(error), "at line 2, a: 'c'")
-
-
-class TestLineParser(unittest.TestCase):
-
- def test___repr__(self):
- line_parser = LineParser('', 101)
- self.assertIsNotNone(repr(line_parser))
-
- def test_error_invalid_assignment(self):
- line_parser = LineParser('', 101)
- self.assertIsNotNone(line_parser.error_invalid_assignment())
-
-
-class TestBaseParser(unittest.TestCase):
-
- @staticmethod
- def make_open(text_blob):
- @contextmanager
- def internal_open(*args, **kwargs):
- yield text_blob.split('\n')
-
- return internal_open
-
- def test_parse(self):
- parser = BaseParser()
- parser.parse()
-
- def test_parse_empty_string(self):
- parser = BaseParser()
- self.assertIsNone(parser.parse(''))
-
- def test_not_implemented_methods(self):
- parser = BaseParser()
-
- with self.assertRaises(NotImplementedError):
- parser.assignment('key', 'value', LineParser('', 100))
-
- with self.assertRaises(NotImplementedError):
- parser.new_section('section')
-
- with self.assertRaises(NotImplementedError):
- parser.comment('comment')
-
-
-class TestConfigParser(unittest.TestCase):
-
- @staticmethod
- def make_open(text_blob):
- @contextmanager
- def internal_open(*args, **kwargs):
- yield text_blob.split('\n')
-
- return internal_open
-
- @mock.patch('yardstick.network_services.helpers.iniparser.open')
- def test_parse(self, mock_open):
- mock_open.side_effect = self.make_open(PARSE_TEXT_1)
-
- existing_data = [['section0', [['key0', 'value0']]]]
- config_parser = ConfigParser('my_file', existing_data)
- config_parser.parse()
-
- expected = [
- [
- 'section0',
- [
- ['key0', 'value0'],
- ],
- ],
- [
- 'section1',
- [
- ['key1', 'value1'],
- ['list1', 'value2\nvalue3\nvalue4'],
- ['key3', 'single quote value'],
- ['key4', ''],
- ['key2', 'double quote value'],
- ],
- ],
- [
- 'section2',
- [
- ['list2', 'value5'],
- ['key with no value', '@'],
- ['key5', ''],
- ],
- ],
- ]
-
- self.assertEqual(config_parser.sections, expected)
- self.assertIsNotNone(config_parser.find_section('section1'))
- self.assertIsNone(config_parser.find_section('section3'))
- self.assertEqual(config_parser.find_section_index('section1'), 1)
- self.assertEqual(config_parser.find_section_index('section3'), -1)
-
- @mock.patch('yardstick.network_services.helpers.iniparser.open')
- def test_parse_2(self, mock_open):
- mock_open.side_effect = self.make_open(PARSE_TEXT_2)
-
- config_parser = ConfigParser('my_file')
- config_parser.parse()
-
- expected = [
- [
- 'section1',
- [
- ['list1', 'item1\nitem2\nended by eof'],
- ],
- ],
- ]
-
- self.assertEqual(config_parser.sections, expected)
-
- @mock.patch('yardstick.network_services.helpers.iniparser.open')
- def test_parse_negative(self, mock_open):
- bad_text_dict = {
- 'no section': PARSE_TEXT_BAD_1,
- 'incomplete section': PARSE_TEXT_BAD_2,
- 'empty section name': PARSE_TEXT_BAD_3,
- 'bad_continuation': PARSE_TEXT_BAD_4,
- 'value with no key': PARSE_TEXT_BAD_5,
- }
-
- for bad_reason, bad_text in bad_text_dict.items():
- mock_open.side_effect = self.make_open(bad_text)
-
- config_parser = ConfigParser('my_file', [])
-
- try:
- # TODO: replace with assertRaises, when the UT framework supports
- # advanced messages when exceptions fail to occur
- config_parser.parse()
- except ParseError:
- pass
- else:
- self.fail('\n'.join([bad_reason, bad_text, str(config_parser.sections)]))
diff --git a/tests/unit/network_services/helpers/test_samplevnf_helper.py b/tests/unit/network_services/helpers/test_samplevnf_helper.py
deleted file mode 100644
index 3b6c89d3a..000000000
--- a/tests/unit/network_services/helpers/test_samplevnf_helper.py
+++ /dev/null
@@ -1,1109 +0,0 @@
-# Copyright (c) 2016-2017 Intel Corporation
-#
-# Licensed under the Apache License, Version 2.0 (the "License");
-# you may not use this file except in compliance with the License.
-# You may obtain a copy of the License at
-#
-# http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing, software
-# distributed under the License is distributed on an "AS IS" BASIS,
-# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-# See the License for the specific language governing permissions and
-# limitations under the License.
-#
-
-import mock
-import os
-import six
-import unittest
-
-from yardstick.network_services.helpers import samplevnf_helper
-from yardstick.network_services.vnf_generic.vnf.base import VnfdHelper
-
-
-class TestPortPairs(unittest.TestCase):
- def test_port_pairs_list(self):
- vnfd = TestMultiPortConfig.VNFD['vnfd:vnfd-catalog']['vnfd'][0]
- interfaces = vnfd['vdu'][0]['external-interface']
- port_pairs = samplevnf_helper.PortPairs(interfaces)
- self.assertEqual(port_pairs.port_pair_list, [("xe0", "xe1")])
-
- def test_valid_networks(self):
- vnfd = TestMultiPortConfig.VNFD['vnfd:vnfd-catalog']['vnfd'][0]
- interfaces = vnfd['vdu'][0]['external-interface']
- port_pairs = samplevnf_helper.PortPairs(interfaces)
- self.assertEqual(port_pairs.valid_networks, [
- ("uplink_0", "downlink_0")])
-
- def test_all_ports(self):
- vnfd = TestMultiPortConfig.VNFD['vnfd:vnfd-catalog']['vnfd'][0]
- interfaces = vnfd['vdu'][0]['external-interface']
- port_pairs = samplevnf_helper.PortPairs(interfaces)
- self.assertEqual(set(port_pairs.all_ports), {"xe0", "xe1"})
-
- def test_uplink_ports(self):
- vnfd = TestMultiPortConfig.VNFD['vnfd:vnfd-catalog']['vnfd'][0]
- interfaces = vnfd['vdu'][0]['external-interface']
- port_pairs = samplevnf_helper.PortPairs(interfaces)
- self.assertEqual(port_pairs.uplink_ports, ["xe0"])
-
- def test_downlink_ports(self):
- vnfd = TestMultiPortConfig.VNFD['vnfd:vnfd-catalog']['vnfd'][0]
- interfaces = vnfd['vdu'][0]['external-interface']
- port_pairs = samplevnf_helper.PortPairs(interfaces)
- self.assertEqual(port_pairs.downlink_ports, ["xe1"])
-
-
-class TestMultiPortConfig(unittest.TestCase):
-
- VNFD_0 = {'short-name': 'VpeVnf',
- 'vdu':
- [{'routing_table':
- [{'network': '152.16.100.20',
- 'netmask': '255.255.255.0',
- 'gateway': '152.16.100.20',
- 'if': 'xe0'},
- {'network': '152.16.40.20',
- 'netmask': '255.255.255.0',
- 'gateway': '152.16.40.20',
- 'if': 'xe1'}],
- 'description': 'VPE approximation using DPDK',
- 'name': 'vpevnf-baremetal',
- 'nd_route_tbl':
- [{'network': '0064:ff9b:0:0:0:0:9810:6414',
- 'netmask': '112',
- 'gateway': '0064:ff9b:0:0:0:0:9810:6414',
- 'if': 'xe0'},
- {'network': '0064:ff9b:0:0:0:0:9810:2814',
- 'netmask': '112',
- 'gateway': '0064:ff9b:0:0:0:0:9810:2814',
- 'if': 'xe1'}],
- 'id': 'vpevnf-baremetal',
- 'external-interface':
- [
- {'virtual-interface':
- {
- 'dst_mac': '00:00:00:00:00:04',
- 'vpci': '0000:05:00.0',
- 'local_ip': '152.16.100.19',
- 'type': 'PCI-PASSTHROUGH',
- 'netmask': '255.255.255.0',
- 'dpdk_port_num': 0,
- 'bandwidth': '10 Gbps',
- 'driver': "i40e",
- 'dst_ip': '152.16.100.20',
- 'ifname': 'xe0',
- 'local_iface_name': 'eth0',
- 'local_mac': '00:00:00:00:00:02',
- 'vld_id': 'uplink_0',
- },
- 'vnfd-connection-point-ref': 'xe0',
- 'name': 'xe0'},
- {'virtual-interface':
- {
- 'dst_mac': '00:00:00:00:00:03',
- 'vpci': '0000:05:00.1',
- 'local_ip': '152.16.40.19',
- 'type': 'PCI-PASSTHROUGH',
- 'driver': "i40e",
- 'netmask': '255.255.255.0',
- 'dpdk_port_num': 1,
- 'bandwidth': '10 Gbps',
- 'dst_ip': '152.16.40.20',
- 'ifname': 'xe1',
- 'local_iface_name': 'eth1',
- 'local_mac': '00:00:00:00:00:01',
- 'vld_id': 'downlink_0',
- },
- 'vnfd-connection-point-ref': 'xe1',
- 'name': 'xe1'}
- ]}],
- 'description': 'Vpe approximation using DPDK',
- 'mgmt-interface':
- {'vdu-id': 'vpevnf-baremetal',
- 'host': '1.2.1.1',
- 'password': 'r00t',
- 'user': 'root',
- 'ip': '1.2.1.1'},
- 'benchmark':
- {'kpi': ['packets_in', 'packets_fwd', 'packets_dropped']},
- 'connection-point': [{'type': 'VPORT', 'name': 'xe0'},
- {'type': 'VPORT', 'name': 'xe1'}],
- 'id': 'AclApproxVnf', 'name': 'VPEVnfSsh'}
-
- VNFD = {
- 'vnfd:vnfd-catalog': {
- 'vnfd': [
- VNFD_0,
- ]
- }
- }
-
- def setUp(self):
- self._mock_open = mock.patch.object(six.moves.builtins, 'open')
- self.mock_open = self._mock_open.start()
- self._mock_os = mock.patch.object(os, 'path')
- self.mock_os = self._mock_os.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_os.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')
-
- def test___init__(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)
- self.assertEqual(0, opnfv_vnf.swq)
- self.mock_os.path = mock.MagicMock()
- self.mock_os.path.isfile = mock.Mock(return_value=False)
- opnfv_vnf = samplevnf_helper.MultiPortConfig(
- topology_file, config_tpl, tmp_file, vnfd_mock)
- self.assertEqual(0, opnfv_vnf.swq)
-
- def test_update_timer(self):
- topology_file = mock.Mock()
- config_tpl = mock.Mock()
- tmp_file = mock.Mock()
- vnfd_mock = mock.MagicMock()
- opnfv_vnf = samplevnf_helper.MultiPortConfig(
- topology_file, config_tpl, tmp_file, vnfd_mock)
- opnfv_vnf.get_config_tpl_data = mock.MagicMock()
- opnfv_vnf.socket = 0
- opnfv_vnf.start_core = 0
- opnfv_vnf.update_write_parser = mock.MagicMock()
- self.assertIsNone(opnfv_vnf.update_timer())
-
- def test_generate_script(self):
- topology_file = mock.Mock()
- config_tpl = mock.Mock()
- tmp_file = mock.Mock()
- vnfd_mock = VnfdHelper(self.VNFD_0)
- opnfv_vnf = samplevnf_helper.MultiPortConfig(
- topology_file, config_tpl, tmp_file, vnfd_mock)
- opnfv_vnf.get_config_tpl_data = mock.MagicMock()
- opnfv_vnf.socket = 0
- opnfv_vnf.start_core = 0
- opnfv_vnf.update_write_parser = mock.MagicMock()
- opnfv_vnf.generate_script_data = \
- mock.Mock(return_value={'link_config': 0, 'arp_config': '',
- 'arp_config6': '', 'actions': '',
- 'arp_route_tbl': '', 'arp_route_tbl6': '',
- 'rules': ''})
- opnfv_vnf.port_pair_list = [("xe0", "xe1")]
- self.assertIsNotNone(opnfv_vnf.generate_script(self.VNFD))
- opnfv_vnf.lb_config = 'HW'
- self.assertIsNotNone(opnfv_vnf.generate_script(self.VNFD))
-
- def test_generate_script_data(self):
- topology_file = mock.Mock()
- config_tpl = mock.Mock()
- tmp_file = mock.Mock()
- vnfd_mock = mock.MagicMock()
- opnfv_vnf = samplevnf_helper.MultiPortConfig(
- topology_file, config_tpl, tmp_file, vnfd_mock)
- opnfv_vnf.get_config_tpl_data = mock.MagicMock()
- opnfv_vnf.socket = 0
- opnfv_vnf.start_core = 0
- opnfv_vnf.update_write_parser = mock.MagicMock()
- opnfv_vnf.port_pair_list = [("xe0", "xe1")]
- opnfv_vnf.vnf_type = 'ACL'
- opnfv_vnf.generate_link_config = mock.Mock()
- opnfv_vnf.generate_arp_config = mock.Mock()
- opnfv_vnf.generate_arp_config6 = mock.Mock()
- opnfv_vnf.generate_action_config = mock.Mock()
- opnfv_vnf.generate_rule_config = mock.Mock()
- self.assertIsNotNone(opnfv_vnf.generate_script_data())
-
- def test_generate_rule_config(self):
- topology_file = mock.Mock()
- config_tpl = mock.Mock()
- tmp_file = mock.Mock()
- vnfd_mock = mock.MagicMock()
- opnfv_vnf = samplevnf_helper.MultiPortConfig(
- topology_file, config_tpl, tmp_file, vnfd_mock)
- opnfv_vnf.get_config_tpl_data = mock.MagicMock()
- opnfv_vnf.socket = 0
- opnfv_vnf.start_core = 0
- opnfv_vnf.update_write_parser = mock.MagicMock()
- opnfv_vnf.generate_script_data = \
- mock.Mock(return_value={'link_config': 0, 'arp_config': '',
- 'arp_config6': '', 'actions': '',
- 'rules': ''})
- opnfv_vnf.port_pair_list = [("xe0", "xe1")]
- opnfv_vnf.get_port_pairs = mock.Mock()
- opnfv_vnf.vnf_type = 'ACL'
- opnfv_vnf.get_ports_gateway = mock.Mock(return_value=u'1.1.1.1')
- opnfv_vnf.get_netmask_gateway = mock.Mock(
- return_value=u'255.255.255.0')
- opnfv_vnf.get_ports_gateway6 = mock.Mock(return_value=u'1.1.1.1')
- opnfv_vnf.get_netmask_gateway6 = mock.Mock(
- return_value=u'255.255.255.0')
- opnfv_vnf.txrx_pipeline = ''
- opnfv_vnf.vnfd = self.VNFD['vnfd:vnfd-catalog']['vnfd'][0]
- opnfv_vnf.interfaces = opnfv_vnf.vnfd['vdu'][0]['external-interface']
- opnfv_vnf.rules = ''
- self.assertIsNotNone(opnfv_vnf.generate_rule_config())
- opnfv_vnf.rules = 'new'
- self.assertIsNotNone(opnfv_vnf.generate_rule_config())
-
- def test_generate_action_config(self):
- topology_file = mock.Mock()
- config_tpl = mock.Mock()
- tmp_file = mock.Mock()
- vnfd_mock = mock.MagicMock()
- opnfv_vnf = samplevnf_helper.MultiPortConfig(
- topology_file, config_tpl, tmp_file, vnfd_mock)
- opnfv_vnf.get_config_tpl_data = mock.MagicMock()
- opnfv_vnf.socket = 0
- opnfv_vnf.start_core = 0
- opnfv_vnf.update_write_parser = mock.MagicMock()
- opnfv_vnf.generate_script_data = \
- mock.Mock(return_value={'link_config': 0, 'arp_config': '',
- 'arp_config6': '', 'actions': '',
- 'rules': ''})
- opnfv_vnf.port_pair_list = [("xe0", "xe1")]
- opnfv_vnf.get_port_pairs = mock.Mock()
- opnfv_vnf.vnf_type = 'VFW'
- opnfv_vnf.get_ports_gateway = mock.Mock(return_value=u'1.1.1.1')
- opnfv_vnf.get_netmask_gateway = mock.Mock(
- return_value=u'255.255.255.0')
- opnfv_vnf.get_ports_gateway6 = mock.Mock(return_value=u'1.1.1.1')
- opnfv_vnf.get_netmask_gateway6 = mock.Mock(
- return_value=u'255.255.255.0')
- opnfv_vnf.txrx_pipeline = ''
- opnfv_vnf.rules = ''
- self.assertIsNotNone(opnfv_vnf.generate_action_config())
-
- def test_generate_arp_config6(self):
- topology_file = mock.Mock()
- config_tpl = mock.Mock()
- tmp_file = mock.Mock()
- vnfd_mock = mock.MagicMock()
- opnfv_vnf = samplevnf_helper.MultiPortConfig(
- topology_file, config_tpl, tmp_file, vnfd_mock)
- opnfv_vnf.get_config_tpl_data = mock.MagicMock()
- opnfv_vnf.socket = 0
- opnfv_vnf.start_core = 0
- opnfv_vnf.update_write_parser = mock.MagicMock()
- opnfv_vnf.generate_script_data = \
- mock.Mock(return_value={'link_config': 0, 'arp_config': '',
- 'arp_config6': '', 'actions': '',
- 'rules': ''})
- opnfv_vnf.port_pair_list = [("xe0", "xe1")]
- opnfv_vnf.get_port_pairs = mock.Mock()
- opnfv_vnf.vnf_type = 'VFW'
- opnfv_vnf.get_ports_gateway = mock.Mock(return_value=u'1.1.1.1')
- opnfv_vnf.get_netmask_gateway = mock.Mock(
- return_value=u'255.255.255.0')
- opnfv_vnf.get_ports_gateway6 = mock.Mock(return_value=u'1.1.1.1')
- opnfv_vnf.get_netmask_gateway6 = mock.Mock(
- return_value=u'255.255.255.0')
- opnfv_vnf.txrx_pipeline = ''
- opnfv_vnf.rules = ''
- opnfv_vnf.interfaces = mock.MagicMock()
- opnfv_vnf.get_ports_gateway6 = mock.Mock()
- self.assertIsNotNone(opnfv_vnf.generate_arp_config6())
-
- def test_generate_arp_config(self):
- topology_file = mock.Mock()
- config_tpl = mock.Mock()
- tmp_file = mock.Mock()
- vnfd_mock = mock.MagicMock()
- opnfv_vnf = samplevnf_helper.MultiPortConfig(
- topology_file, config_tpl, tmp_file, vnfd_mock)
- opnfv_vnf.get_config_tpl_data = mock.MagicMock()
- opnfv_vnf.socket = 0
- opnfv_vnf.start_core = 0
- opnfv_vnf.update_write_parser = mock.MagicMock()
- opnfv_vnf.generate_script_data = \
- mock.Mock(return_value={'link_config': 0, 'arp_config': '',
- 'arp_config6': '', 'actions': '',
- 'rules': ''})
- opnfv_vnf.port_pair_list = [("xe0", "xe1")]
- opnfv_vnf.get_port_pairs = mock.Mock()
- opnfv_vnf.vnf_type = 'VFW'
- opnfv_vnf.get_ports_gateway = mock.Mock(return_value=u'1.1.1.1')
- opnfv_vnf.get_netmask_gateway = mock.Mock(
- return_value=u'255.255.255.0')
- opnfv_vnf.get_ports_gateway6 = mock.Mock(return_value=u'1.1.1.1')
- opnfv_vnf.get_netmask_gateway6 = mock.Mock(
- return_value=u'255.255.255.0')
- opnfv_vnf.txrx_pipeline = ''
- opnfv_vnf.rules = ''
- opnfv_vnf.interfaces = mock.MagicMock()
- opnfv_vnf.get_ports_gateway6 = mock.Mock()
- self.assertIsNotNone(opnfv_vnf.generate_arp_config())
-
- def test_get_ports_gateway(self):
- topology_file = mock.Mock()
- config_tpl = mock.Mock()
- tmp_file = mock.Mock()
- vnfd_mock = mock.MagicMock()
- opnfv_vnf = samplevnf_helper.MultiPortConfig(
- topology_file, config_tpl, tmp_file, vnfd_mock)
- opnfv_vnf.get_config_tpl_data = mock.MagicMock()
- opnfv_vnf.socket = 0
- opnfv_vnf.start_core = 0
- opnfv_vnf.update_write_parser = mock.MagicMock()
- opnfv_vnf.generate_script_data = \
- mock.Mock(return_value={'link_config': 0, 'arp_config': '',
- 'arp_config6': '', 'actions': '',
- 'rules': ''})
- opnfv_vnf.port_pair_list = [("xe0", "xe1")]
- opnfv_vnf.get_port_pairs = mock.Mock()
- opnfv_vnf.vnf_type = 'VFW'
- opnfv_vnf.txrx_pipeline = ''
- opnfv_vnf.rules = ''
- opnfv_vnf.interfaces = mock.MagicMock()
- opnfv_vnf.get_ports_gateway6 = mock.Mock()
- opnfv_vnf.vnfd = self.VNFD['vnfd:vnfd-catalog']['vnfd'][0]
- self.assertIsNotNone(opnfv_vnf.get_ports_gateway('xe0'))
-
- def test_get_ports_gateway6(self):
- topology_file = mock.Mock()
- config_tpl = mock.Mock()
- tmp_file = mock.Mock()
- vnfd_mock = mock.MagicMock()
- opnfv_vnf = samplevnf_helper.MultiPortConfig(
- topology_file, config_tpl, tmp_file, vnfd_mock)
- opnfv_vnf.get_config_tpl_data = mock.MagicMock()
- opnfv_vnf.socket = 0
- opnfv_vnf.start_core = 0
- opnfv_vnf.update_write_parser = mock.MagicMock()
- opnfv_vnf.generate_script_data = \
- mock.Mock(return_value={'link_config': 0, 'arp_config': '',
- 'arp_config6': '', 'actions': '',
- 'rules': ''})
- opnfv_vnf.port_pair_list = [("xe0", "xe1")]
- opnfv_vnf.get_port_pairs = mock.Mock()
- opnfv_vnf.vnf_type = 'VFW'
- opnfv_vnf.txrx_pipeline = ''
- opnfv_vnf.rules = ''
- opnfv_vnf.interfaces = mock.MagicMock()
- opnfv_vnf.get_ports_gateway6 = mock.Mock()
- opnfv_vnf.vnfd = self.VNFD['vnfd:vnfd-catalog']['vnfd'][0]
- self.assertIsNotNone(opnfv_vnf.get_ports_gateway6('xe0'))
-
- def test_get_netmask_gateway(self):
- topology_file = mock.Mock()
- config_tpl = mock.Mock()
- tmp_file = mock.Mock()
- vnfd_mock = mock.MagicMock()
- opnfv_vnf = samplevnf_helper.MultiPortConfig(
- topology_file, config_tpl, tmp_file, vnfd_mock)
- opnfv_vnf.get_config_tpl_data = mock.MagicMock()
- opnfv_vnf.socket = 0
- opnfv_vnf.start_core = 0
- opnfv_vnf.update_write_parser = mock.MagicMock()
- opnfv_vnf.generate_script_data = \
- mock.Mock(return_value={'link_config': 0, 'arp_config': '',
- 'arp_config6': '', 'actions': '',
- 'rules': ''})
- opnfv_vnf.port_pair_list = [("xe0", "xe1")]
- opnfv_vnf.get_port_pairs = mock.Mock()
- opnfv_vnf.vnf_type = 'VFW'
- opnfv_vnf.txrx_pipeline = ''
- opnfv_vnf.rules = ''
- opnfv_vnf.interfaces = mock.MagicMock()
- opnfv_vnf.get_ports_gateway6 = mock.Mock()
- opnfv_vnf.vnfd = self.VNFD['vnfd:vnfd-catalog']['vnfd'][0]
- self.assertIsNotNone(opnfv_vnf.get_netmask_gateway('xe0'))
-
- def test_get_netmask_gateway6(self):
- topology_file = mock.Mock()
- config_tpl = mock.Mock()
- tmp_file = mock.Mock()
- vnfd_mock = mock.MagicMock()
- opnfv_vnf = samplevnf_helper.MultiPortConfig(
- topology_file, config_tpl, tmp_file, vnfd_mock)
- opnfv_vnf.get_config_tpl_data = mock.MagicMock()
- opnfv_vnf.socket = 0
- opnfv_vnf.start_core = 0
- opnfv_vnf.update_write_parser = mock.MagicMock()
- opnfv_vnf.generate_script_data = \
- mock.Mock(return_value={'link_config': 0, 'arp_config': '',
- 'arp_config6': '', 'actions': '',
- 'rules': ''})
- opnfv_vnf.port_pair_list = [("xe0", "xe1")]
- opnfv_vnf.get_port_pairs = mock.Mock()
- opnfv_vnf.vnf_type = 'VFW'
- opnfv_vnf.txrx_pipeline = ''
- opnfv_vnf.rules = ''
- opnfv_vnf.interfaces = mock.MagicMock()
- opnfv_vnf.get_ports_gateway6 = mock.Mock()
- opnfv_vnf.vnfd = self.VNFD['vnfd:vnfd-catalog']['vnfd'][0]
- self.assertIsNotNone(opnfv_vnf.get_netmask_gateway6('xe0'))
-
- def test_generate_link_config(self):
- topology_file = mock.Mock()
- config_tpl = mock.Mock()
- tmp_file = mock.Mock()
- vnfd_mock = mock.MagicMock()
-
- opnfv_vnf = samplevnf_helper.MultiPortConfig(
- topology_file, config_tpl, tmp_file, vnfd_mock)
- opnfv_vnf.get_config_tpl_data = mock.MagicMock()
- opnfv_vnf.socket = 0
- opnfv_vnf.start_core = 0
- opnfv_vnf.update_write_parser = mock.MagicMock()
- opnfv_vnf.generate_script_data = \
- mock.Mock(return_value={'link_config': 0, 'arp_config': '',
- 'arp_config6': '', 'actions': '',
- 'rules': ''})
- opnfv_vnf.port_pair_list = [("xe0", "xe1")]
- opnfv_vnf.get_port_pairs = mock.Mock()
- opnfv_vnf.vnf_type = 'VFW'
- opnfv_vnf.txrx_pipeline = ''
- opnfv_vnf.rules = ''
- opnfv_vnf.get_ports_gateway6 = mock.Mock()
- opnfv_vnf.vnfd = self.VNFD['vnfd:vnfd-catalog']['vnfd'][0]
- opnfv_vnf.interfaces = opnfv_vnf.vnfd['vdu'][0]['external-interface']
- opnfv_vnf.all_ports = ['32', '1', '987']
- opnfv_vnf.validate_ip_and_prefixlen = mock.Mock(
- return_value=('10.20.30.40', 16))
-
- result = opnfv_vnf.generate_link_config()
- self.assertEqual(len(result.splitlines()), 9)
-
- def test_generate_config(self):
- topology_file = mock.Mock()
- config_tpl = mock.Mock()
- tmp_file = mock.Mock()
- vnfd_mock = mock.MagicMock()
- opnfv_vnf = samplevnf_helper.MultiPortConfig(
- topology_file, config_tpl, tmp_file, vnfd_mock)
- opnfv_vnf.get_config_tpl_data = mock.MagicMock()
- opnfv_vnf.socket = 0
- opnfv_vnf.start_core = 0
- opnfv_vnf.update_write_parser = mock.MagicMock()
- opnfv_vnf.generate_script_data = \
- mock.Mock(return_value={'link_config': 0, 'arp_config': '',
- 'arp_config6': '', 'actions': '',
- 'rules': ''})
- opnfv_vnf.port_pair_list = [("xe0", "xe1")]
- opnfv_vnf.txrx_pipeline = ''
- opnfv_vnf.rules = ''
- opnfv_vnf.get_ports_gateway6 = mock.Mock()
- opnfv_vnf.vnfd = self.VNFD['vnfd:vnfd-catalog']['vnfd'][0]
- opnfv_vnf.interfaces = opnfv_vnf.vnfd['vdu'][0]['external-interface']
- opnfv_vnf.generate_lb_to_port_pair_mapping = mock.Mock()
- opnfv_vnf.generate_config_data = mock.Mock()
- opnfv_vnf.write_parser = mock.MagicMock()
- opnfv_vnf.is_openstack = True
- self.assertIsNone(opnfv_vnf.generate_config())
- opnfv_vnf.is_openstack = False
- self.assertIsNone(opnfv_vnf.generate_config())
-
- def test_get_config_tpl_data(self):
- topology_file = mock.Mock()
- config_tpl = mock.Mock()
- tmp_file = mock.Mock()
- vnfd_mock = mock.MagicMock()
- opnfv_vnf = samplevnf_helper.MultiPortConfig(
- topology_file, config_tpl, tmp_file, vnfd_mock)
- opnfv_vnf.socket = 0
- opnfv_vnf.start_core = 0
- opnfv_vnf.port_pair_list = [("xe0", "xe1")]
- opnfv_vnf.txrx_pipeline = ''
- opnfv_vnf.rules = ''
- opnfv_vnf.write_parser = mock.MagicMock()
- opnfv_vnf.read_parser = mock.MagicMock()
- opnfv_vnf.read_parser.sections = mock.Mock(return_value=['MASTER'])
- opnfv_vnf.read_parser.has_option = mock.Mock(return_value=True)
- opnfv_vnf.read_parser.get = mock.Mock(return_value='filename')
-
- self.assertIsNotNone(opnfv_vnf.get_config_tpl_data('filename'))
-
- def test_get_txrx_tpl_data(self):
- topology_file = mock.Mock()
- config_tpl = mock.Mock()
- tmp_file = mock.Mock()
- vnfd_mock = mock.MagicMock()
- opnfv_vnf = samplevnf_helper.MultiPortConfig(
- topology_file, config_tpl, tmp_file, vnfd_mock)
- opnfv_vnf.socket = 0
- opnfv_vnf.start_core = 0
- opnfv_vnf.port_pair_list = [("xe0", "xe1")]
- opnfv_vnf.txrx_pipeline = ''
- opnfv_vnf.rules = ''
- opnfv_vnf.write_parser = mock.MagicMock()
- opnfv_vnf.read_parser = mock.MagicMock()
- opnfv_vnf.read_parser.sections = mock.Mock(return_value=['MASTER'])
- opnfv_vnf.read_parser.has_option = mock.Mock(return_value=True)
- opnfv_vnf.read_parser.get = mock.Mock(return_value='filename')
-
- self.assertIsNotNone(opnfv_vnf.get_txrx_tpl_data('filename'))
-
- def test_init_write_parser_template(self):
- topology_file = mock.Mock()
- config_tpl = mock.Mock()
- tmp_file = mock.Mock()
- vnfd_mock = mock.MagicMock()
- opnfv_vnf = samplevnf_helper.MultiPortConfig(
- topology_file, config_tpl, tmp_file, vnfd_mock)
- opnfv_vnf.socket = 0
- opnfv_vnf.start_core = 0
- opnfv_vnf.port_pair_list = [("xe0", "xe1")]
- opnfv_vnf.txrx_pipeline = ''
- opnfv_vnf.rules = ''
- opnfv_vnf.write_parser = mock.MagicMock()
- opnfv_vnf.read_parser = mock.MagicMock()
- opnfv_vnf.read_parser.sections = mock.Mock(return_value=['MASTER'])
- opnfv_vnf.read_parser.has_option = mock.Mock(return_value=True)
- opnfv_vnf.read_parser.get = mock.Mock(return_value='filename')
-
- self.assertIsNone(opnfv_vnf.init_write_parser_template('filename'))
- opnfv_vnf.write_parser.add_section = mock.MagicMock()
- opnfv_vnf.read_parser.item = mock.Mock(return_value=[1, 2, 3])
- opnfv_vnf.read_parser.has_option = mock.Mock(return_value=False)
- opnfv_vnf.write_parser.set = mock.Mock()
- self.assertIsNone(opnfv_vnf.init_write_parser_template('filename'))
-
- def test_init_write_parser_template_2(self):
- topology_file = mock.Mock()
- config_tpl = mock.Mock()
- tmp_file = mock.Mock()
- vnfd_mock = mock.MagicMock()
- opnfv_vnf = samplevnf_helper.MultiPortConfig(
- topology_file, config_tpl, tmp_file, vnfd_mock)
- opnfv_vnf.socket = 0
- opnfv_vnf.start_core = 0
- opnfv_vnf.port_pair_list = [("xe0", "xe1")]
- opnfv_vnf.txrx_pipeline = ''
- opnfv_vnf.rules = ''
- opnfv_vnf.write_parser = mock.MagicMock()
- opnfv_vnf.read_parser = mock.MagicMock()
- opnfv_vnf.read_parser.sections = mock.Mock(return_value=['MASTER'])
- opnfv_vnf.read_parser.has_option = mock.Mock(return_value=[])
- opnfv_vnf.write_parser.set = mock.Mock()
- opnfv_vnf.read_parser.items = mock.MagicMock()
- self.assertIsNone(opnfv_vnf.init_write_parser_template('filename'))
-
- def test_update_write_parser(self):
- topology_file = mock.Mock()
- config_tpl = mock.Mock()
- tmp_file = mock.Mock()
- vnfd_mock = mock.MagicMock()
- opnfv_vnf = samplevnf_helper.MultiPortConfig(
- topology_file, config_tpl, tmp_file, vnfd_mock)
- opnfv_vnf.socket = 0
- opnfv_vnf.start_core = 0
- opnfv_vnf.port_pair_list = [("xe0", "xe1")]
- opnfv_vnf.txrx_pipeline = ''
- opnfv_vnf.rules = ''
- opnfv_vnf.write_parser = mock.MagicMock()
- opnfv_vnf.read_parser = mock.MagicMock()
- opnfv_vnf.read_parser.sections = mock.Mock(return_value=['MASTER'])
- opnfv_vnf.read_parser.has_option = mock.Mock(return_value=[])
- opnfv_vnf.write_parser.set = mock.Mock()
- opnfv_vnf.write_parser.add_section = mock.Mock()
- opnfv_vnf.read_parser.items = mock.MagicMock()
- opnfv_vnf.pipeline_counter = 0
- self.assertIsNone(opnfv_vnf.update_write_parser({'filename': 1}))
-
- def test_get_worker_threads(self):
- topology_file = mock.Mock()
- config_tpl = mock.Mock()
- tmp_file = mock.Mock()
- vnfd_mock = mock.MagicMock()
- opnfv_vnf = samplevnf_helper.MultiPortConfig(
- topology_file, config_tpl, tmp_file, vnfd_mock)
- opnfv_vnf.socket = 0
- opnfv_vnf.start_core = 0
- opnfv_vnf.port_pair_list = [("xe0", "xe1")]
- opnfv_vnf.txrx_pipeline = ''
- opnfv_vnf.rules = ''
- opnfv_vnf.write_parser = mock.MagicMock()
- opnfv_vnf.read_parser = mock.MagicMock()
- opnfv_vnf.read_parser.sections = mock.Mock(return_value=['MASTER'])
- opnfv_vnf.read_parser.has_option = mock.Mock(return_value=[])
- opnfv_vnf.write_parser.set = mock.Mock()
- opnfv_vnf.write_parser.add_section = mock.Mock()
- opnfv_vnf.read_parser.items = mock.MagicMock()
- opnfv_vnf.pipeline_counter = 0
- opnfv_vnf.worker_config = '1t'
- result = opnfv_vnf.get_worker_threads(1)
- self.assertEqual(1, result)
- opnfv_vnf.worker_config = '2t'
- result = opnfv_vnf.get_worker_threads(2)
- self.assertEqual(2, result)
- opnfv_vnf.worker_config = '2t'
- result = opnfv_vnf.get_worker_threads(3)
- self.assertEqual(2, result)
-
- # TODO(elfoley): Split this test into smaller tests
- def test_generate_next_core_id(self):
- topology_file = mock.Mock()
- config_tpl = mock.Mock()
- tmp_file = mock.Mock()
- vnfd_mock = mock.MagicMock()
- opnfv_vnf = samplevnf_helper.MultiPortConfig(
- topology_file, config_tpl, tmp_file, vnfd_mock)
- opnfv_vnf.socket = 0
- opnfv_vnf.start_core = 0
- opnfv_vnf.port_pair_list = [("xe0", "xe1")]
- opnfv_vnf.txrx_pipeline = ''
- opnfv_vnf.rules = ''
- opnfv_vnf.write_parser = mock.MagicMock()
- opnfv_vnf.read_parser = mock.MagicMock()
- opnfv_vnf.read_parser.sections = mock.Mock(return_value=['MASTER'])
- opnfv_vnf.read_parser.has_option = mock.Mock(return_value=[])
- opnfv_vnf.write_parser.set = mock.Mock()
- opnfv_vnf.write_parser.add_section = mock.Mock()
- opnfv_vnf.read_parser.items = mock.MagicMock()
- opnfv_vnf.pipeline_counter = 0
- opnfv_vnf.worker_config = '1t'
- opnfv_vnf.start_core = 0
- result = opnfv_vnf.generate_next_core_id()
- self.assertIsNone(result)
- opnfv_vnf.worker_config = '2t'
- opnfv_vnf.start_core = 'a'
- self.assertRaises(ValueError, opnfv_vnf.generate_next_core_id)
- opnfv_vnf.worker_config = '2t'
- opnfv_vnf.start_core = 1
- result = opnfv_vnf.generate_next_core_id()
- self.assertIsNone(result)
-
- def test_generate_lb_to_port_pair_mapping(self):
- topology_file = mock.Mock()
- config_tpl = mock.Mock()
- tmp_file = mock.Mock()
- vnfd_mock = VnfdHelper(self.VNFD_0)
- opnfv_vnf = samplevnf_helper.MultiPortConfig(
- topology_file, config_tpl, tmp_file, vnfd_mock)
- opnfv_vnf.socket = 0
- opnfv_vnf.start_core = 0
- opnfv_vnf.txrx_pipeline = ''
- opnfv_vnf.rules = ''
- opnfv_vnf.write_parser = mock.MagicMock()
- opnfv_vnf.read_parser = mock.MagicMock()
- opnfv_vnf.read_parser.sections = mock.Mock(return_value=['MASTER'])
- opnfv_vnf.read_parser.has_option = mock.Mock(return_value=[])
- opnfv_vnf.write_parser.set = mock.Mock()
- opnfv_vnf.write_parser.add_section = mock.Mock()
- opnfv_vnf.read_parser.items = mock.MagicMock()
- opnfv_vnf.pipeline_counter = 0
- opnfv_vnf.worker_config = '1t'
- opnfv_vnf.start_core = 0
- opnfv_vnf.lb_count = 1
- opnfv_vnf._port_pairs = samplevnf_helper.PortPairs(vnfd_mock.interfaces)
- opnfv_vnf.port_pair_list = opnfv_vnf._port_pairs.port_pair_list
- result = opnfv_vnf.generate_lb_to_port_pair_mapping()
- self.assertIsNone(result)
- result = opnfv_vnf.set_priv_to_pub_mapping()
- self.assertEqual('(0,1)', result)
-
- def test_set_priv_que_handler(self):
- topology_file = mock.Mock()
- config_tpl = mock.Mock()
- tmp_file = mock.Mock()
- vnfd_mock = VnfdHelper(self.VNFD_0)
- opnfv_vnf = samplevnf_helper.MultiPortConfig(
- topology_file, config_tpl, tmp_file, vnfd_mock)
- opnfv_vnf.socket = 0
- opnfv_vnf.start_core = 0
- opnfv_vnf.port_pair_list = [("xe0", "xe1")]
- opnfv_vnf.port_pairs = [("xe0", "xe1")]
- opnfv_vnf.txrx_pipeline = ''
- opnfv_vnf.rules = ''
- opnfv_vnf.write_parser = mock.MagicMock()
- opnfv_vnf.read_parser = mock.MagicMock()
- opnfv_vnf.read_parser.sections = mock.Mock(return_value=['MASTER'])
- opnfv_vnf.read_parser.has_option = mock.Mock(return_value=[])
- opnfv_vnf.write_parser.set = mock.Mock()
- opnfv_vnf.write_parser.add_section = mock.Mock()
- opnfv_vnf.read_parser.items = mock.MagicMock()
- opnfv_vnf.pipeline_counter = 0
- opnfv_vnf.worker_config = '1t'
- opnfv_vnf.start_core = 0
- opnfv_vnf.lb_count = 1
- result = opnfv_vnf.set_priv_que_handler()
- self.assertIsNone(result)
-
- def test_generate_arp_route_tbl(self):
- # ELF: could n=do this in setup
- topology_file = mock.Mock()
- config_tpl = mock.Mock()
- tmp_file = ""
- vnfd_mock = mock.MagicMock()
- vnfd_mock.port_num.side_effect = ['32', '1', '987']
- vnfd_mock.find_interface.side_effect = [
- {
- 'virtual-interface': {
- 'dst_ip': '10.20.30.40',
- 'netmask': '20',
- },
- },
- {
- 'virtual-interface': {
- 'dst_ip': '10.200.30.40',
- 'netmask': '24',
- },
- },
- {
- 'virtual-interface': {
- 'dst_ip': '10.20.3.40',
- 'netmask': '8',
- },
- },
- ]
-
- opnfv_vnf = samplevnf_helper.MultiPortConfig(
- topology_file, config_tpl, tmp_file, vnfd_mock)
- opnfv_vnf.all_ports = [3, 2, 5]
-
- expected = 'routeadd net 32 10.20.30.40 0xfffff000\n' \
- 'routeadd net 1 10.200.30.40 0xffffff00\n' \
- 'routeadd net 987 10.20.3.40 0xff000000'
- result = opnfv_vnf.generate_arp_route_tbl()
- self.assertEqual(result, expected)
-
- def test_generate_arpicmp_data(self):
- topology_file = mock.Mock()
- config_tpl = mock.Mock()
- tmp_file = mock.Mock()
- vnfd_mock = mock.MagicMock()
- opnfv_vnf = samplevnf_helper.MultiPortConfig(
- topology_file, config_tpl, tmp_file, vnfd_mock)
- opnfv_vnf.socket = 0
- opnfv_vnf.start_core = 0
- opnfv_vnf.port_pair_list = [("xe0", "xe1")]
- opnfv_vnf.port_pairs = [("xe0", "xe1")]
- opnfv_vnf.txrx_pipeline = ''
- opnfv_vnf.rules = ''
- opnfv_vnf.write_parser = mock.MagicMock()
- opnfv_vnf.read_parser = mock.MagicMock()
- opnfv_vnf.read_parser.sections = mock.Mock(return_value=['MASTER'])
- opnfv_vnf.read_parser.has_option = mock.Mock(return_value=[])
- opnfv_vnf.write_parser.set = mock.Mock()
- opnfv_vnf.write_parser.add_section = mock.Mock()
- opnfv_vnf.read_parser.items = mock.MagicMock()
- opnfv_vnf.pipeline_counter = 0
- opnfv_vnf.worker_config = '1t'
- opnfv_vnf.start_core = 0
- opnfv_vnf.lb_count = 1
- opnfv_vnf.vnfd = self.VNFD['vnfd:vnfd-catalog']['vnfd'][0]
- opnfv_vnf.interfaces = opnfv_vnf.vnfd['vdu'][0]['external-interface']
- result = opnfv_vnf.generate_arpicmp_data()
- self.assertIsNotNone(result)
- opnfv_vnf.nfv_type = 'ovs'
- opnfv_vnf.lb_to_port_pair_mapping = [0, 1]
- result = opnfv_vnf.generate_arpicmp_data()
- self.assertIsNotNone(result)
- opnfv_vnf.nfv_type = 'openstack'
- opnfv_vnf.lb_to_port_pair_mapping = [0, 1]
- result = opnfv_vnf.generate_arpicmp_data()
- self.assertIsNotNone(result)
- opnfv_vnf.lb_config = 'HW'
- opnfv_vnf.lb_to_port_pair_mapping = [0, 1]
- result = opnfv_vnf.generate_arpicmp_data()
- self.assertIsNotNone(result)
-
- def test_generate_final_txrx_data(self):
- topology_file = mock.Mock()
- config_tpl = mock.Mock()
- tmp_file = mock.Mock()
- vnfd_mock = mock.MagicMock()
- opnfv_vnf = samplevnf_helper.MultiPortConfig(
- topology_file, config_tpl, tmp_file, vnfd_mock)
- opnfv_vnf.socket = 0
- opnfv_vnf.start_core = 0
- opnfv_vnf.port_pair_list = [("xe0", "xe1")]
- opnfv_vnf.port_pairs = [("xe0", "xe1")]
- opnfv_vnf.txrx_pipeline = ''
- opnfv_vnf.rules = ''
- opnfv_vnf.write_parser = mock.MagicMock()
- opnfv_vnf.read_parser = mock.MagicMock()
- opnfv_vnf.read_parser.sections = mock.Mock(return_value=['MASTER'])
- opnfv_vnf.read_parser.has_option = mock.Mock(return_value=[])
- opnfv_vnf.write_parser.set = mock.Mock()
- opnfv_vnf.write_parser.add_section = mock.Mock()
- opnfv_vnf.read_parser.items = mock.MagicMock()
- opnfv_vnf.pipeline_counter = 0
- opnfv_vnf.worker_config = '1t'
- opnfv_vnf.start_core = 0
- opnfv_vnf.lb_count = 1
- opnfv_vnf.vnfd = self.VNFD['vnfd:vnfd-catalog']['vnfd'][0]
- opnfv_vnf.interfaces = opnfv_vnf.vnfd['vdu'][0]['external-interface']
- opnfv_vnf.lb_to_port_pair_mapping = [0, 1]
- opnfv_vnf.ports_len = 2
- opnfv_vnf.lb_index = 1
- opnfv_vnf.pktq_out_os = [1, 2]
- result = opnfv_vnf.generate_final_txrx_data()
- self.assertIsNotNone(result)
- opnfv_vnf.nfv_type = 'openstack'
- opnfv_vnf.pktq_out_os = [1, 2]
- opnfv_vnf.lb_index = 1
- result = opnfv_vnf.generate_final_txrx_data()
- self.assertIsNotNone(result)
-
- def test_generate_initial_txrx_data(self):
- topology_file = mock.Mock()
- config_tpl = mock.Mock()
- tmp_file = mock.Mock()
- vnfd_mock = mock.MagicMock()
- opnfv_vnf = samplevnf_helper.MultiPortConfig(
- topology_file, config_tpl, tmp_file, vnfd_mock)
- opnfv_vnf.socket = 0
- opnfv_vnf.start_core = 0
- opnfv_vnf.port_pair_list = [("xe0", "xe1")]
- opnfv_vnf.port_pairs = [("xe0", "xe1")]
- opnfv_vnf.txrx_pipeline = ''
- opnfv_vnf.rules = ''
- opnfv_vnf.write_parser = mock.MagicMock()
- opnfv_vnf.read_parser = mock.MagicMock()
- opnfv_vnf.read_parser.sections = mock.Mock(return_value=['MASTER'])
- opnfv_vnf.read_parser.has_option = mock.Mock(return_value=[])
- opnfv_vnf.write_parser.set = mock.Mock()
- opnfv_vnf.write_parser.add_section = mock.Mock()
- opnfv_vnf.read_parser.items = mock.MagicMock()
- opnfv_vnf.pipeline_counter = 0
- opnfv_vnf.worker_config = '1t'
- opnfv_vnf.start_core = 0
- opnfv_vnf.lb_count = 1
- opnfv_vnf.vnfd = self.VNFD['vnfd:vnfd-catalog']['vnfd'][0]
- opnfv_vnf.interfaces = opnfv_vnf.vnfd['vdu'][0]['external-interface']
- opnfv_vnf.lb_to_port_pair_mapping = [0, 1]
- opnfv_vnf.lb_index = 1
- opnfv_vnf.ports_len = 2
- result = opnfv_vnf.generate_initial_txrx_data()
- self.assertIsNotNone(result)
- opnfv_vnf.nfv_type = 'openstack'
- opnfv_vnf.pktq_out_os = [1, 2]
- result = opnfv_vnf.generate_initial_txrx_data()
- self.assertIsNotNone(result)
- opnfv_vnf.nfv_type = 'ovs'
- opnfv_vnf.init_ovs = False
- opnfv_vnf.ovs_pktq_out = ''
- opnfv_vnf.pktq_out_os = [1, 2]
- opnfv_vnf.lb_index = 1
- result = opnfv_vnf.generate_initial_txrx_data()
- self.assertIsNotNone(result)
- opnfv_vnf.nfv_type = 'ovs'
- opnfv_vnf.init_ovs = True
- opnfv_vnf.pktq_out_os = [1, 2]
- opnfv_vnf.ovs_pktq_out = ''
- opnfv_vnf.lb_index = 1
- result = opnfv_vnf.generate_initial_txrx_data()
- self.assertIsNotNone(result)
-
- def test_generate_lb_data(self):
- topology_file = mock.Mock()
- config_tpl = mock.Mock()
- tmp_file = mock.Mock()
- vnfd_mock = mock.MagicMock()
- opnfv_vnf = samplevnf_helper.MultiPortConfig(
- topology_file, config_tpl, tmp_file, vnfd_mock)
- opnfv_vnf.socket = 0
- opnfv_vnf.start_core = 0
- opnfv_vnf.port_pair_list = [("xe0", "xe1")]
- opnfv_vnf.port_pairs = [("xe0", "xe1")]
- opnfv_vnf.txrx_pipeline = ''
- opnfv_vnf.rules = ''
- opnfv_vnf.write_parser = mock.MagicMock()
- opnfv_vnf.read_parser = mock.MagicMock()
- opnfv_vnf.read_parser.sections = mock.Mock(return_value=['MASTER'])
- opnfv_vnf.read_parser.has_option = mock.Mock(return_value=[])
- opnfv_vnf.write_parser.set = mock.Mock()
- opnfv_vnf.write_parser.add_section = mock.Mock()
- opnfv_vnf.read_parser.items = mock.MagicMock()
- opnfv_vnf.pipeline_counter = 0
- opnfv_vnf.worker_config = '1t'
- opnfv_vnf.start_core = 0
- opnfv_vnf.lb_count = 1
- opnfv_vnf.vnfd = self.VNFD['vnfd:vnfd-catalog']['vnfd'][0]
- opnfv_vnf.interfaces = opnfv_vnf.vnfd['vdu'][0]['external-interface']
- opnfv_vnf.lb_to_port_pair_mapping = [0, 1]
- opnfv_vnf.lb_index = 1
- opnfv_vnf.ports_len = 2
- opnfv_vnf.prv_que_handler = 0
- result = opnfv_vnf.generate_lb_data()
- self.assertIsNotNone(result)
-
- def test_generate_vnf_data(self):
- topology_file = mock.Mock()
- config_tpl = mock.Mock()
- tmp_file = mock.Mock()
- vnfd_mock = mock.MagicMock()
- opnfv_vnf = samplevnf_helper.MultiPortConfig(
- topology_file, config_tpl, tmp_file, vnfd_mock)
- opnfv_vnf.socket = 0
- opnfv_vnf.start_core = 0
- opnfv_vnf.port_pair_list = [("xe0", "xe1")]
- opnfv_vnf.port_pairs = [("xe0", "xe1")]
- opnfv_vnf.txrx_pipeline = ''
- opnfv_vnf.rules = ''
- opnfv_vnf.write_parser = mock.MagicMock()
- opnfv_vnf.read_parser = mock.MagicMock()
- opnfv_vnf.read_parser.sections = mock.Mock(return_value=['MASTER'])
- opnfv_vnf.read_parser.has_option = mock.Mock(return_value=[])
- opnfv_vnf.write_parser.set = mock.Mock()
- opnfv_vnf.write_parser.add_section = mock.Mock()
- opnfv_vnf.read_parser.items = mock.MagicMock()
- opnfv_vnf.pipeline_counter = 0
- opnfv_vnf.worker_config = '1t'
- opnfv_vnf.start_core = 0
- opnfv_vnf.lb_count = 1
- opnfv_vnf.vnfd = self.VNFD['vnfd:vnfd-catalog']['vnfd'][0]
- opnfv_vnf.interfaces = opnfv_vnf.vnfd['vdu'][0]['external-interface']
- opnfv_vnf.lb_to_port_pair_mapping = [0, 1]
- opnfv_vnf.lb_index = 1
- opnfv_vnf.ports_len = 1
- opnfv_vnf.pktq_out = ['1', '2']
- opnfv_vnf.vnf_tpl = {'public_ip_port_range': '98164810',
- 'vnf_set': '(2,4,5)'}
- opnfv_vnf.prv_que_handler = 0
- result = opnfv_vnf.generate_vnf_data()
- self.assertIsNotNone(result)
- opnfv_vnf.lb_config = 'HW'
- opnfv_vnf.mul = 0.1
- result = opnfv_vnf.generate_vnf_data()
- self.assertIsNotNone(result)
- opnfv_vnf.lb_config = 'HW'
- opnfv_vnf.mul = 0.1
- opnfv_vnf.vnf_type = 'ACL'
- result = opnfv_vnf.generate_vnf_data()
- self.assertIsNotNone(result)
-
- def test_generate_config_data(self):
- topology_file = mock.Mock()
- config_tpl = mock.Mock()
- tmp_file = mock.Mock()
- vnfd_mock = VnfdHelper(self.VNFD_0)
- opnfv_vnf = samplevnf_helper.MultiPortConfig(
- topology_file, config_tpl, tmp_file, vnfd_mock)
- opnfv_vnf.socket = 0
- opnfv_vnf.start_core = 0
- opnfv_vnf.port_pair_list = [("xe0", "xe1")]
- opnfv_vnf.port_pairs = [("xe0", "xe1")]
- opnfv_vnf.txrx_pipeline = ''
- opnfv_vnf.rules = ''
- opnfv_vnf.write_parser = mock.MagicMock()
- opnfv_vnf.read_parser = mock.MagicMock()
- opnfv_vnf.read_parser.sections = mock.Mock(return_value=['MASTER'])
- opnfv_vnf.read_parser.has_option = mock.Mock(return_value=[])
- opnfv_vnf.write_parser.set = mock.Mock()
- opnfv_vnf.write_parser.add_section = mock.Mock()
- opnfv_vnf.read_parser.items = mock.MagicMock()
- opnfv_vnf.pipeline_counter = 0
- opnfv_vnf.worker_config = '1t'
- opnfv_vnf.start_core = 0
- opnfv_vnf.lb_count = 1
- opnfv_vnf.vnfd = self.VNFD['vnfd:vnfd-catalog']['vnfd'][0]
- opnfv_vnf.interfaces = opnfv_vnf.vnfd['vdu'][0]['external-interface']
- opnfv_vnf.lb_to_port_pair_mapping = [0, 1]
- opnfv_vnf.lb_index = 1
- opnfv_vnf.ports_len = 1
- opnfv_vnf.pktq_out = ['1', '2']
- opnfv_vnf.prv_que_handler = 0
- opnfv_vnf.init_write_parser_template = mock.Mock()
- opnfv_vnf.arpicmp_tpl = mock.MagicMock()
- opnfv_vnf.txrx_tpl = mock.MagicMock()
- opnfv_vnf.loadb_tpl = mock.MagicMock()
- opnfv_vnf.vnf_tpl = {'public_ip_port_range': '98164810 (1,65535)',
- 'vnf_set': "(2,4,5)"}
- opnfv_vnf.generate_vnf_data = mock.Mock(return_value={})
- opnfv_vnf.update_write_parser = mock.Mock()
- result = opnfv_vnf.generate_config_data()
- self.assertIsNone(result)
- opnfv_vnf.generate_final_txrx_data = mock.Mock()
- opnfv_vnf.update_write_parser = mock.Mock()
- result = opnfv_vnf.generate_config_data()
- self.assertIsNone(result)
- opnfv_vnf.lb_to_port_pair_mapping = [0, 1]
- opnfv_vnf.lb_index = 1
- opnfv_vnf.ports_len = 1
- opnfv_vnf.pktq_out = ['1', '2']
- opnfv_vnf.prv_que_handler = 0
- opnfv_vnf.init_write_parser_template = mock.Mock()
- opnfv_vnf.arpicmp_tpl = mock.MagicMock()
- opnfv_vnf.txrx_tpl = mock.MagicMock()
- opnfv_vnf.loadb_tpl = mock.MagicMock()
- opnfv_vnf.vnf_type = 'CGNAPT'
- opnfv_vnf.update_timer = mock.Mock()
- opnfv_vnf.port_pair_list = [("xe0", "xe1"), ("xe0", "xe2")]
- opnfv_vnf.lb_to_port_pair_mapping = [0, 1]
- opnfv_vnf.generate_arpicmp_data = mock.Mock()
- result = opnfv_vnf.generate_config_data()
- self.assertIsNone(result)
-
- def test_init_eal(self):
- topology_file = mock.Mock()
- config_tpl = mock.Mock()
- tmp_file = mock.Mock()
- vnfd_mock = mock.MagicMock()
- opnfv_vnf = samplevnf_helper.MultiPortConfig(
- topology_file, config_tpl, tmp_file, vnfd_mock)
- opnfv_vnf.socket = 0
- opnfv_vnf.start_core = 0
- opnfv_vnf.port_pair_list = [("xe0", "xe1")]
- opnfv_vnf.port_pairs = [("xe0", "xe1")]
- opnfv_vnf.txrx_pipeline = ''
- opnfv_vnf.rules = ''
- opnfv_vnf.write_parser = mock.MagicMock()
- opnfv_vnf.read_parser = mock.MagicMock()
- opnfv_vnf.read_parser.sections = mock.Mock(return_value=['MASTER'])
- opnfv_vnf.read_parser.has_option = mock.Mock(return_value=[])
- opnfv_vnf.write_parser.set = mock.Mock()
- opnfv_vnf.write_parser.add_section = mock.Mock()
- opnfv_vnf.read_parser.items = mock.MagicMock()
- opnfv_vnf.pipeline_counter = 0
- opnfv_vnf.worker_config = '1t'
- opnfv_vnf.start_core = 0
- opnfv_vnf.lb_count = 1
- opnfv_vnf.vnfd = self.VNFD['vnfd:vnfd-catalog']['vnfd'][0]
- opnfv_vnf.interfaces = opnfv_vnf.vnfd['vdu'][0]['external-interface']
- opnfv_vnf.lb_to_port_pair_mapping = [0, 1]
- opnfv_vnf.lb_index = 1
- opnfv_vnf.ports_len = 1
- opnfv_vnf.pktq_out = ['1', '2']
- opnfv_vnf.prv_que_handler = 0
- opnfv_vnf.init_write_parser_template = mock.Mock()
- opnfv_vnf.arpicmp_tpl = mock.MagicMock()
- opnfv_vnf.txrx_tpl = mock.MagicMock()
- opnfv_vnf.loadb_tpl = mock.MagicMock()
- opnfv_vnf.vnf_tpl = {'public_ip_port_range': '98164810 (1,65535)'}
- opnfv_vnf.generate_vnf_data = mock.Mock(return_value={})
- opnfv_vnf.update_write_parser = mock.Mock()
- opnfv_vnf.tmp_file = "/tmp/config"
- result = opnfv_vnf.init_eal()
- self.assertIsNone(result)
diff --git a/tests/unit/network_services/libs/__init__.py b/tests/unit/network_services/libs/__init__.py
deleted file mode 100644
index e69de29bb..000000000
--- a/tests/unit/network_services/libs/__init__.py
+++ /dev/null
diff --git a/tests/unit/network_services/libs/ixia_libs/__init__.py b/tests/unit/network_services/libs/ixia_libs/__init__.py
deleted file mode 100644
index e69de29bb..000000000
--- a/tests/unit/network_services/libs/ixia_libs/__init__.py
+++ /dev/null
diff --git a/tests/unit/network_services/libs/ixia_libs/test_IxNet.py b/tests/unit/network_services/libs/ixia_libs/test_IxNet.py
deleted file mode 100644
index 2a97048aa..000000000
--- a/tests/unit/network_services/libs/ixia_libs/test_IxNet.py
+++ /dev/null
@@ -1,874 +0,0 @@
-# Copyright (c) 2016-2017 Intel Corporation
-#
-# Licensed under the Apache License, Version 2.0 (the "License");
-# you may not use this file except in compliance with the License.
-# You may obtain a copy of the License at
-#
-# http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing, software
-# distributed under the License is distributed on an "AS IS" BASIS,
-# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-# See the License for the specific language governing permissions and
-# limitations under the License.
-#
-
-# Unittest for yardstick.network_services.libs.ixia_libs.IxNet
-
-from __future__ import absolute_import
-import unittest
-import mock
-
-from yardstick.network_services.libs.ixia_libs.IxNet.IxNet import IxNextgen
-from yardstick.network_services.libs.ixia_libs.IxNet.IxNet import IP_VERSION_4
-from yardstick.network_services.libs.ixia_libs.IxNet.IxNet import IP_VERSION_6
-
-
-UPLINK = "uplink"
-DOWNLINK = "downlink"
-
-class TestIxNextgen(unittest.TestCase):
-
- def test___init__(self):
- ixnet_gen = IxNextgen()
- self.assertIsNone(ixnet_gen._bidir)
-
- @mock.patch("yardstick.network_services.libs.ixia_libs.IxNet.IxNet.sys")
- def test_connect(self, *args):
-
- ixnet_gen = IxNextgen()
- ixnet_gen.get_config = mock.MagicMock()
- ixnet_gen.get_ixnet = mock.MagicMock()
-
- self.assertRaises(ImportError, ixnet_gen._connect, {"py_lib_path": "/tmp"})
-
- def test_clear_ixia_config(self):
- ixnet = mock.MagicMock()
- ixnet.execute = mock.Mock()
-
- ixnet_gen = IxNextgen(ixnet)
-
- result = ixnet_gen.clear_ixia_config()
- self.assertIsNone(result)
- self.assertEqual(ixnet.execute.call_count, 1)
-
- def test_load_ixia_profile(self):
- ixnet = mock.MagicMock()
- ixnet.execute = mock.Mock()
-
- ixnet_gen = IxNextgen(ixnet)
-
- result = ixnet_gen.load_ixia_profile({})
- self.assertIsNone(result)
- self.assertEqual(ixnet.execute.call_count, 1)
-
- def test_load_ixia_config(self):
- ixnet = mock.MagicMock()
- ixnet.execute = mock.Mock()
-
- ixnet_gen = IxNextgen(ixnet)
-
- result = ixnet_gen.ix_load_config({})
- self.assertIsNone(result)
- self.assertEqual(ixnet.execute.call_count, 2)
-
- @mock.patch('yardstick.network_services.libs.ixia_libs.IxNet.IxNet.log')
- def test_ix_assign_ports(self, mock_logger):
- ixnet = mock.MagicMock()
- ixnet.getList.return_value = [0, 1]
- ixnet.getAttribute.side_effect = ['up', 'down']
-
- config = {
- 'chassis': '1.1.1.1',
- 'cards': ['1', '2'],
- 'ports': ['2', '2'],
- }
-
- ixnet_gen = IxNextgen(ixnet)
- ixnet_gen._cfg = config
-
- result = ixnet_gen.ix_assign_ports()
- self.assertIsNone(result)
- self.assertEqual(ixnet.execute.call_count, 1)
- self.assertEqual(ixnet.commit.call_count, 1)
- self.assertEqual(ixnet.getAttribute.call_count, 2)
- self.assertEqual(mock_logger.error.call_count, 1)
-
- def test_ix_update_frame(self):
- static_traffic_params = {
- UPLINK: {
- "id": 1,
- "bidir": "False",
- "duration": 60,
- "iload": "100",
- "outer_l2": {
- "dstmac": "00:00:00:00:00:03",
- "framesPerSecond": True,
- "framesize": {
- "64B": "100",
- "1KB": "0",
- },
- "srcmac": "00:00:00:00:00:01"
- },
- "outer_l3": {
- "dscp": 0,
- "dstip4": "152.16.40.20",
- "proto": "udp",
- "srcip4": "152.16.100.20",
- "ttl": 32
- },
- "outer_l3v4": {
- "dscp": 0,
- "dstip4": "152.16.40.20",
- "proto": "udp",
- "srcip4": "152.16.100.20",
- "ttl": 32
- },
- "outer_l3v6": {
- "count": 1024,
- "dscp": 0,
- "dstip4": "152.16.100.20",
- "proto": "udp",
- "srcip4": "152.16.40.20",
- "ttl": 32
- },
- "outer_l4": {
- "dstport": "2001",
- "srcport": "1234"
- },
- "traffic_type": "continuous"
- },
- DOWNLINK: {
- "id": 2,
- "bidir": "False",
- "duration": 60,
- "iload": "100",
- "outer_l2": {
- "dstmac": "00:00:00:00:00:04",
- "framesPerSecond": False,
- "framesize": {"64B": "100"},
- "srcmac": "00:00:00:00:00:01"
- },
- "outer_l3": {
- "count": 1024,
- "dscp": 0,
- "dstip4": "152.16.100.20",
- "proto": "udp",
- "srcip4": "152.16.40.20",
- "ttl": 32
- },
- "outer_l3v4": {
- "count": 1024,
- "dscp": 0,
- "dstip4": "152.16.100.20",
- "proto": "udp",
- "srcip4": "152.16.40.20",
- "ttl": 32
- },
- "outer_l3v6": {
- "count": 1024,
- "dscp": 0,
- "dstip4": "152.16.100.20",
- "proto": "udp",
- "srcip4": "152.16.40.20",
- "ttl": 32
- },
- "outer_l4": {
- "dstport": "1234",
- "srcport": "2001"
- },
- "traffic_type": "continuous"
- }
- }
-
- ixnet = mock.MagicMock()
- ixnet.remapIds.return_value = ["0"]
- ixnet.setMultiAttribute.return_value = [1]
- ixnet.commit.return_value = [1]
- ixnet.getList.side_effect = [
- [1],
- [1],
- [1],
- [
- "ethernet.header.destinationAddress",
- "ethernet.header.sourceAddress",
- ],
- ]
-
- ixnet_gen = IxNextgen(ixnet)
-
- result = ixnet_gen.ix_update_frame(static_traffic_params)
- self.assertIsNone(result)
- self.assertEqual(ixnet.setMultiAttribute.call_count, 7)
- self.assertEqual(ixnet.commit.call_count, 2)
-
- def test_ix_update_udp(self):
- ixnet = mock.MagicMock()
-
- ixnet_gen = IxNextgen(ixnet)
-
- result = ixnet_gen.ix_update_udp({})
- self.assertIsNone(result)
-
- def test_ix_update_tcp(self):
- ixnet = mock.MagicMock()
- ixnet_gen = IxNextgen(ixnet)
-
- result = ixnet_gen.ix_update_tcp({})
- self.assertIsNone(result)
-
- def test_ix_start_traffic(self):
- ixnet = mock.MagicMock()
- ixnet.getList.return_value = [0]
- ixnet.getAttribute.return_value = 'down'
-
- ixnet_gen = IxNextgen(ixnet)
-
- result = ixnet_gen.ix_start_traffic()
- self.assertIsNone(result)
- self.assertEqual(ixnet.getList.call_count, 1)
- self.assertEqual(ixnet.execute.call_count, 3)
-
- def test_ix_stop_traffic(self):
- ixnet = mock.MagicMock()
- ixnet.getList.return_value = [0]
-
- ixnet_gen = IxNextgen(ixnet)
-
- result = ixnet_gen.ix_stop_traffic()
- self.assertIsNone(result)
- self.assertEqual(ixnet.getList.call_count, 1)
- self.assertEqual(ixnet.execute.call_count, 1)
-
- def test_ix_get_statistics(self):
- ixnet = mock.MagicMock()
- ixnet.execute.return_value = ""
- ixnet.getList.side_effect = [
- [
- '::ixNet::OBJ-/statistics/view:"Traffic Item Statistics"',
- '::ixNet::OBJ-/statistics/view:"Port Statistics"',
- ],
- [
- '::ixNet::OBJ-/statistics/view:"Flow Statistics"',
- ],
- ]
-
- ixnet_gen = IxNextgen(ixnet)
-
- result = ixnet_gen.ix_get_statistics()
- self.assertIsNotNone(result)
- self.assertEqual(ixnet.getList.call_count, 1)
- self.assertEqual(ixnet.execute.call_count, 20)
-
- def test_find_view_obj_no_where(self):
- views = ['here', 'there', 'everywhere']
- result = IxNextgen.find_view_obj('no_where', views)
- self.assertEqual(result, '')
-
- def test_add_ip_header_v4(self):
- static_traffic_params = {
- "uplink_0": {
- "id": 1,
- "bidir": "False",
- "duration": 60,
- "iload": "100",
- "outer_l2": {
- "dstmac": "00:00:00:00:00:03",
- "framesPerSecond": True,
- "framesize": {"64B": "100"},
- "srcmac": "00:00:00:00:00:01"
- },
- "outer_l3": {
- "dscp": 0,
- "dstip4": "152.16.40.20",
- "proto": "udp",
- "srcip4": "152.16.100.20",
- "count": 1024,
- "ttl": 32
- },
- "outer_l3v4": {
- "dscp": 0,
- "dstip4": "152.16.40.20",
- "proto": "udp",
- "srcip4": "152.16.100.20",
- "ttl": 32
- },
- "outer_l3v6": {
- "count": 1024,
- "dscp": 0,
- "dstip4": "152.16.100.20",
- "proto": "udp",
- "srcip4": "152.16.40.20",
- "ttl": 32
- },
- "outer_l4": {
- "dstport": "2001",
- "srcport": "1234"
- },
- "traffic_type": "continuous"
- },
- "downlink_0": {
- "id": 2,
- "bidir": "False",
- "duration": 60,
- "iload": "100",
- "outer_l2": {
- "dstmac": "00:00:00:00:00:04",
- "framesPerSecond": True,
- "framesize": {"64B": "100"},
- "srcmac": "00:00:00:00:00:01"
- },
- "outer_l3": {
- "count": 1024,
- "dscp": 0,
- "dstip4": "152.16.100.20",
- "proto": "udp",
- "srcip4": "152.16.40.20",
- "ttl": 32
- },
- "outer_l3v4": {
- "count": 1024,
- "dscp": 0,
- "dstip4": "152.16.100.20",
- "proto": "udp",
- "srcip4": "152.16.40.20",
- "ttl": 32
- },
- "outer_l3v6": {
- "count": 1024,
- "dscp": 0,
- "dstip4": "152.16.100.20",
- "proto": "udp",
- "srcip4": "152.16.40.20",
- "ttl": 32
- },
- "outer_l4": {
- "dstport": "1234",
- "srcport": "2001"
- },
- "traffic_type": "continuous"
- }
- }
-
- ixnet = mock.MagicMock()
- ixnet.remapIds.return_value = ["0"]
- ixnet.setMultiAttribute.return_value = [1]
- ixnet.commit.return_value = [1]
- ixnet.getList.side_effect = [[1], [0], [0], ["srcIp", "dstIp"]]
-
- ixnet_gen = IxNextgen(ixnet)
-
- result = ixnet_gen.add_ip_header(static_traffic_params, IP_VERSION_4)
- self.assertIsNone(result)
- self.assertGreater(ixnet.setMultiAttribute.call_count, 0)
- self.assertEqual(ixnet.commit.call_count, 1)
-
- def test_add_ip_header_v4_nothing_to_do(self):
- static_traffic_params = {
- "uplink_0": {
- "id": 1,
- "bidir": "False",
- "duration": 60,
- "iload": "100",
- "outer_l2": {
- "dstmac": "00:00:00:00:00:03",
- "framesPerSecond": True,
- "framesize": {"64B": "100"},
- "srcmac": "00:00:00:00:00:01"
- },
- "outer_l3": {
- "dscp": 0,
- "dstip4": "152.16.40.20",
- "proto": "udp",
- "srcip4": "152.16.100.20",
- "count": 1024,
- "ttl": 32
- },
- "outer_l3v4": {
- "dscp": 0,
- "dstip4": "152.16.40.20",
- "proto": "udp",
- "srcip4": "152.16.100.20",
- "ttl": 32
- },
- "outer_l3v6": {
- "count": 1024,
- "dscp": 0,
- "dstip4": "152.16.100.20",
- "proto": "udp",
- "srcip4": "152.16.40.20",
- "ttl": 32
- },
- "outer_l4": {
- "dstport": "2001",
- "srcport": "1234"
- },
- "traffic_type": "continuous"
- },
- "downlink_0": {
- "id": 2,
- "bidir": "False",
- "duration": 60,
- "iload": "100",
- "outer_l2": {
- "dstmac": "00:00:00:00:00:04",
- "framesPerSecond": True,
- "framesize": {"64B": "100"},
- "srcmac": "00:00:00:00:00:01"
- },
- "outer_l3": {
- "count": 1024,
- "dscp": 0,
- "dstip4": "152.16.100.20",
- "proto": "udp",
- "srcip4": "152.16.40.20",
- "ttl": 32
- },
- "outer_l3v4": {
- "count": 1024,
- "dscp": 0,
- "dstip4": "152.16.100.20",
- "proto": "udp",
- "srcip4": "152.16.40.20",
- "ttl": 32
- },
- "outer_l3v6": {
- "count": 1024,
- "dscp": 0,
- "dstip4": "152.16.100.20",
- "proto": "udp",
- "srcip4": "152.16.40.20",
- "ttl": 32
- },
- "outer_l4": {
- "dstport": "1234",
- "srcport": "2001"
- },
- "traffic_type": "continuous"
- }
- }
-
- ixnet = mock.MagicMock()
- ixnet.remapIds.return_value = ["0"]
- ixnet.setMultiAttribute.return_value = [1]
- ixnet.commit.return_value = [1]
- ixnet.getList.side_effect = [[1], [0, 1], [0], ["srcIp", "dstIp"]]
-
- ixnet_gen = IxNextgen(ixnet)
-
- result = ixnet_gen.add_ip_header(static_traffic_params, IP_VERSION_4)
- self.assertIsNone(result)
- self.assertGreater(ixnet.setMultiAttribute.call_count, 0)
- self.assertEqual(ixnet.commit.call_count, 1)
-
- def test_add_ip_header_v6(self):
- static_traffic_profile = {
- "uplink_0": {
- "id": 1,
- "bidir": "False",
- "duration": 60,
- "iload": "100",
- "outer_l2": {
- "dstmac": "00:00:00:00:00:03",
- "framesPerSecond": True,
- "framesize": {"64B": "100"},
- "srcmac": "00:00:00:00:00:01"
- },
- "outer_l3": {
- "count": 1024,
- "dscp": 0,
- "dstip4": "152.16.40.20",
- "proto": "udp",
- "srcip4": "152.16.100.20",
- "ttl": 32
- },
- "outer_l3v6": {
- "count": 1024,
- "dscp": 0,
- "dstip4": "152.16.100.20",
- "proto": "udp",
- "srcip4": "152.16.40.20",
- "ttl": 32
- },
- "outer_l4": {
- "dstport": "2001",
- "srcport": "1234"
- },
- "traffic_type": "continuous"
- },
- "downlink_0": {
- "id": 2,
- "bidir": "False",
- "duration": 60,
- "iload": "100",
- "outer_l2": {
- "dstmac": "00:00:00:00:00:04",
- "framesPerSecond": True,
- "framesize": {"64B": "100"},
- "srcmac": "00:00:00:00:00:01"
- },
- "outer_l3": {
- "count": 1024,
- "dscp": 0,
- "dstip4": "152.16.100.20",
- "proto": "udp",
- "srcip4": "152.16.40.20",
- "ttl": 32
- },
- "outer_l3v6": {
- "count": 1024,
- "dscp": 0,
- "dstip4": "152.16.100.20",
- "proto": "udp",
- "srcip4": "152.16.40.20",
- "ttl": 32
- },
- "outer_l4": {
- "dstport": "1234",
- "srcport": "2001"
- },
- "traffic_type": "continuous"
- }
- }
-
- ixnet = mock.MagicMock()
- ixnet.getList.side_effect = [[1], [1], [1], ["srcIp", "dstIp"]]
- ixnet.remapIds.return_value = ["0"]
- ixnet.setMultiAttribute.return_value = [1]
- ixnet.commit.return_value = [1]
-
- ixnet_gen = IxNextgen(ixnet)
-
- result = ixnet_gen.add_ip_header(static_traffic_profile, IP_VERSION_6)
- self.assertIsNone(result)
- self.assertGreater(ixnet.setMultiAttribute.call_count, 0)
- self.assertEqual(ixnet.commit.call_count, 1)
-
- def test_add_ip_header_v6_nothing_to_do(self):
- static_traffic_params = {
- "uplink_0": {
- "id": 1,
- "bidir": "False",
- "duration": 60,
- "iload": "100",
- "outer_l2": {
- "dstmac": "00:00:00:00:00:03",
- "framesPerSecond": True,
- "framesize": {"64B": "100"},
- "srcmac": "00:00:00:00:00:01"
- },
- "outer_l3": {
- "dscp": 0,
- "dstip4": "152.16.40.20",
- "proto": "udp",
- "srcip4": "152.16.100.20",
- "count": 1024,
- "ttl": 32
- },
- "outer_l3v6": {
- "dscp": 0,
- "dstip4": "152.16.40.20",
- "proto": "udp",
- "srcip4": "152.16.100.20",
- "ttl": 32
- },
- "outer_l4": {
- "dstport": "2001",
- "srcport": "1234"
- },
- "traffic_type": "continuous"
- },
- "downlink_0": {
- "id": 2,
- "bidir": "False",
- "duration": 60,
- "iload": "100",
- "outer_l2": {
- "dstmac": "00:00:00:00:00:04",
- "framesPerSecond": True,
- "framesize": {"64B": "100"},
- "srcmac": "00:00:00:00:00:01"
- },
- "outer_l3": {
- "count": 1024,
- "dscp": 0,
- "dstip4": "152.16.100.20",
- "proto": "udp",
- "srcip4": "152.16.40.20",
- "ttl": 32
- },
- "outer_l3v6": {
- "count": 1024,
- "dscp": 0,
- "dstip4": "152.16.100.20",
- "proto": "udp",
- "srcip4": "152.16.40.20",
- "ttl": 32
- },
- "outer_l4": {
- "dstport": "1234",
- "srcport": "2001"
- },
- "traffic_type": "continuous"
- }
- }
-
- ixnet = mock.MagicMock()
- ixnet.getList.side_effect = [[1], [0, 1], [1], ["srcIP", "dstIP"]]
- ixnet.remapIds.return_value = ["0"]
- ixnet.setMultiAttribute.return_value = [1]
- ixnet.commit.return_value = [1]
-
- ixnet_gen = IxNextgen(ixnet)
-
- result = ixnet_gen.add_ip_header(static_traffic_params, IP_VERSION_6)
- self.assertIsNone(result)
- self.assertEqual(ixnet.setMultiAttribute.call_count, 0)
-
- def test_set_random_ip_multi_attributes_bad_ip_version(self):
- bad_ip_version = object()
- ixnet_gen = IxNextgen(mock.Mock())
- mock1 = mock.Mock()
- mock2 = mock.Mock()
- mock3 = mock.Mock()
- with self.assertRaises(ValueError):
- ixnet_gen.set_random_ip_multi_attributes(mock1, bad_ip_version, mock2, mock3)
-
- def test_get_config(self):
- tg_cfg = {
- "vdu": [
- {
- "external-interface": [
- {
- "virtual-interface": {
- "vpci": "0000:07:00.1",
- },
- },
- {
- "virtual-interface": {
- "vpci": "0001:08:01.2",
- },
- },
- ],
- },
- ],
- "mgmt-interface": {
- "ip": "test1",
- "tg-config": {
- "dut_result_dir": "test2",
- "version": "test3",
- "ixchassis": "test4",
- "tcl_port": "test5",
- "py_lib_path": "test6",
- },
- }
- }
-
- expected = {
- 'py_lib_path': 'test6',
- 'machine': 'test1',
- 'port': 'test5',
- 'chassis': 'test4',
- 'cards': ['0000', '0001'],
- 'ports': ['07', '08'],
- 'output_dir': 'test2',
- 'version': 'test3',
- 'bidir': True,
- }
-
- result = IxNextgen.get_config(tg_cfg)
- self.assertDictEqual(result, expected)
-
- def test_ix_update_ether(self):
- static_traffic_params = {
- "uplink_0": {
- "id": 1,
- "bidir": "False",
- "duration": 60,
- "iload": "100",
- "outer_l2": {
- "dstmac": "00:00:00:00:00:03",
- "framesPerSecond": True,
- "framesize": 64,
- "srcmac": "00:00:00:00:00:01"
- },
- "outer_l3": {
- "dscp": 0,
- "dstip4": "152.16.40.20",
- "proto": "udp",
- "srcip4": "152.16.100.20",
- "ttl": 32
- },
- "outer_l3v4": {
- "dscp": 0,
- "dstip4": "152.16.40.20",
- "proto": "udp",
- "srcip4": "152.16.100.20",
- "ttl": 32
- },
- "outer_l3v6": {
- "count": 1024,
- "dscp": 0,
- "dstip4": "152.16.100.20",
- "proto": "udp",
- "srcip4": "152.16.40.20",
- "ttl": 32
- },
- "outer_l4": {
- "dstport": "2001",
- "srcport": "1234"
- },
- "traffic_type": "continuous"
- },
- "downlink_0": {
- "id": 2,
- "bidir": "False",
- "duration": 60,
- "iload": "100",
- "outer_l2": {
- "dstmac": "00:00:00:00:00:04",
- "framesPerSecond": True,
- "framesize": 64,
- "srcmac": "00:00:00:00:00:01"
- },
- "outer_l3": {
- "count": 1024,
- "dscp": 0,
- "dstip4": "152.16.100.20",
- "proto": "udp",
- "srcip4": "152.16.40.20",
- "ttl": 32
- },
- "outer_l3v4": {
- "count": 1024,
- "dscp": 0,
- "dstip4": "152.16.100.20",
- "proto": "udp",
- "srcip4": "152.16.40.20",
- "ttl": 32
- },
- "outer_l3v6": {
- "count": 1024,
- "dscp": 0,
- "dstip4": "152.16.100.20",
- "proto": "udp",
- "srcip4": "152.16.40.20",
- "ttl": 32
- },
- "outer_l4": {
- "dstport": "1234",
- "srcport": "2001"
- },
- "traffic_type": "continuous"
- }
- }
-
- ixnet = mock.MagicMock()
- ixnet.setMultiAttribute.return_value = [1]
- ixnet.commit.return_value = [1]
- ixnet.getList.side_effect = [
- [1],
- [1],
- [1],
- [
- "ethernet.header.destinationAddress",
- "ethernet.header.sourceAddress",
- ],
- ]
-
- ixnet_gen = IxNextgen(ixnet)
-
- result = ixnet_gen.ix_update_ether(static_traffic_params)
- self.assertIsNone(result)
- self.assertGreater(ixnet.setMultiAttribute.call_count, 0)
-
- def test_ix_update_ether_nothing_to_do(self):
- static_traffic_params = {
- "uplink_0": {
- "id": 1,
- "bidir": "False",
- "duration": 60,
- "iload": "100",
- "outer_l3": {
- "dscp": 0,
- "dstip4": "152.16.40.20",
- "proto": "udp",
- "srcip4": "152.16.100.20",
- "ttl": 32
- },
- "outer_l3v4": {
- "dscp": 0,
- "dstip4": "152.16.40.20",
- "proto": "udp",
- "srcip4": "152.16.100.20",
- "ttl": 32
- },
- "outer_l3v6": {
- "count": 1024,
- "dscp": 0,
- "dstip4": "152.16.100.20",
- "proto": "udp",
- "srcip4": "152.16.40.20",
- "ttl": 32
- },
- "outer_l4": {
- "dstport": "2001",
- "srcport": "1234"
- },
- "traffic_type": "continuous"
- },
- "downlink_0": {
- "id": 2,
- "bidir": "False",
- "duration": 60,
- "iload": "100",
- "outer_l3": {
- "count": 1024,
- "dscp": 0,
- "dstip4": "152.16.100.20",
- "proto": "udp",
- "srcip4": "152.16.40.20",
- "ttl": 32
- },
- "outer_l3v4": {
- "count": 1024,
- "dscp": 0,
- "dstip4": "152.16.100.20",
- "proto": "udp",
- "srcip4": "152.16.40.20",
- "ttl": 32
- },
- "outer_l3v6": {
- "count": 1024,
- "dscp": 0,
- "dstip4": "152.16.100.20",
- "proto": "udp",
- "srcip4": "152.16.40.20",
- "ttl": 32
- },
- "outer_l4": {
- "dstport": "1234",
- "srcport": "2001"
- },
- "traffic_type": "continuous"
- }
- }
-
- ixnet = mock.MagicMock()
- ixnet.setMultiAttribute.return_value = [1]
- ixnet.commit.return_value = [1]
- ixnet.getList.side_effect = [
- [1],
- [1],
- [1],
- [
- "ethernet.header.destinationAddress",
- "ethernet.header.sourceAddress",
- ],
- ]
-
- ixnet_gen = IxNextgen(ixnet)
-
- result = ixnet_gen.ix_update_ether(static_traffic_params)
- self.assertIsNone(result)
- self.assertEqual(ixnet.setMultiAttribute.call_count, 0)
diff --git a/tests/unit/network_services/nfvi/__init__.py b/tests/unit/network_services/nfvi/__init__.py
deleted file mode 100644
index e69de29bb..000000000
--- a/tests/unit/network_services/nfvi/__init__.py
+++ /dev/null
diff --git a/tests/unit/network_services/nfvi/test_collectd.py b/tests/unit/network_services/nfvi/test_collectd.py
deleted file mode 100644
index 0ae175624..000000000
--- a/tests/unit/network_services/nfvi/test_collectd.py
+++ /dev/null
@@ -1,151 +0,0 @@
-# Copyright (c) 2016-2017 Intel Corporation
-#
-# Licensed under the Apache License, Version 2.0 (the "License");
-# you may not use this file except in compliance with the License.
-# You may obtain a copy of the License at
-#
-# http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing, software
-# distributed under the License is distributed on an "AS IS" BASIS,
-# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-# See the License for the specific language governing permissions and
-# limitations under the License.
-
-from __future__ import absolute_import
-import unittest
-import multiprocessing
-import mock
-
-from yardstick.network_services.nfvi.collectd import AmqpConsumer
-
-
-class TestAmqpConsumer(unittest.TestCase):
- def setUp(self):
- self.queue = multiprocessing.Queue()
- self.url = 'amqp://admin:admin@127.0.0.1:5672/%2F'
- self.amqp_consumer = AmqpConsumer(self.url, self.queue)
-
- def test___init__(self):
- self.assertEqual(self.url, self.amqp_consumer._url)
-
- def test_on_connection_open(self):
- self.amqp_consumer._connection = mock.Mock(autospec=AmqpConsumer)
- self.amqp_consumer._connection.add_on_close_callback = \
- mock.Mock(return_value=0)
- self.amqp_consumer._connection.channel = mock.Mock(return_value=0)
- self.assertIsNone(self.amqp_consumer.on_connection_open(10))
-
- def test_on_connection_closed(self):
- self.amqp_consumer._connection = mock.Mock(autospec=AmqpConsumer)
- self.amqp_consumer._connection.ioloop = mock.Mock()
- self.amqp_consumer._connection.ioloop.stop = mock.Mock(return_value=0)
- self.amqp_consumer._connection.add_timeout = mock.Mock(return_value=0)
- self.amqp_consumer._closing = True
- self.assertIsNone(
- self.amqp_consumer.on_connection_closed("", 404, "Not Found"))
- self.amqp_consumer._closing = False
- self.assertIsNone(
- self.amqp_consumer.on_connection_closed("", 404, "Not Found"))
-
- def test_reconnect(self):
- self.amqp_consumer._connection = mock.Mock(autospec=AmqpConsumer)
- self.amqp_consumer._connection.ioloop = mock.Mock()
- self.amqp_consumer._connection.ioloop.stop = mock.Mock(return_value=0)
- self.amqp_consumer.connect = mock.Mock(return_value=0)
- self.amqp_consumer._closing = True
- self.assertIsNone(self.amqp_consumer.reconnect())
-
- def test_on_channel_open(self):
- self.amqp_consumer._connection = mock.Mock(autospec=AmqpConsumer)
- self.amqp_consumer._connection.add_on_close_callback = \
- mock.Mock(return_value=0)
- self.amqp_consumer._channel = mock.Mock()
- self.amqp_consumer.add_on_channel_close_callback = mock.Mock()
- self.amqp_consumer._channel.exchange_declare = \
- mock.Mock(return_value=0)
- self.assertIsNone(
- self.amqp_consumer.on_channel_open(self.amqp_consumer._channel))
-
- def test_add_on_channel_close_callback(self):
- self.amqp_consumer._connection = mock.Mock(autospec=AmqpConsumer)
- self.amqp_consumer._connection.add_on_close_callback = \
- mock.Mock(return_value=0)
- self.amqp_consumer._channel = mock.Mock()
- self.amqp_consumer._channel.add_on_close_callback = mock.Mock()
- self.assertIsNone(self.amqp_consumer.add_on_channel_close_callback())
-
- def test_on_channel_closed(self):
- self.amqp_consumer._connection = mock.Mock(autospec=AmqpConsumer)
- self.amqp_consumer._connection.close = mock.Mock(return_value=0)
- _channel = mock.Mock()
- self.assertIsNone(
- self.amqp_consumer.on_channel_closed(_channel, "", ""))
-
- def test_ion_exchange_declareok(self):
- self.amqp_consumer.setup_queue = mock.Mock(return_value=0)
- self.assertIsNone(self.amqp_consumer.on_exchange_declareok(10))
-
- def test_setup_queue(self):
- self.amqp_consumer._channel = mock.Mock()
- self.amqp_consumer._channel.add_on_close_callback = mock.Mock()
- self.assertIsNone(self.amqp_consumer.setup_queue("collectd"))
-
- def test_on_queue_declareok(self):
- self.amqp_consumer._channel = mock.Mock()
- self.amqp_consumer._channel.queue_bind = mock.Mock()
- self.assertIsNone(self.amqp_consumer.on_queue_declareok(10))
-
- def test__on_bindok(self):
- self.amqp_consumer._channel = mock.Mock()
- self.amqp_consumer._channel.basic_consume = mock.Mock()
- self.amqp_consumer.add_on_cancel_callback = mock.Mock()
- self.assertIsNone(self.amqp_consumer._on_bindok(10))
-
- def test_add_on_cancel_callback(self):
- self.amqp_consumer._channel = mock.Mock()
- self.amqp_consumer._channel.add_on_cancel_callback = mock.Mock()
- self.assertIsNone(self.amqp_consumer.add_on_cancel_callback())
-
- def test_on_consumer_cancelled(self):
- self.amqp_consumer._channel = mock.Mock()
- self.amqp_consumer._channel.close = mock.Mock()
- self.assertIsNone(self.amqp_consumer.on_consumer_cancelled(10))
-
- def test_on_message(self):
- body = "msg {} cpu/cpu-0/ipc 101010:10"
- properties = ""
- basic_deliver = mock.Mock()
- basic_deliver.delivery_tag = mock.Mock(return_value=0)
- self.amqp_consumer.ack_message = mock.Mock()
- self.assertIsNone(
- self.amqp_consumer.on_message(10, basic_deliver, properties, body))
-
- def test_ack_message(self):
- self.amqp_consumer._channel = mock.Mock()
- self.amqp_consumer._channel.basic_ack = mock.Mock()
- self.assertIsNone(self.amqp_consumer.ack_message(10))
-
- def test_on_cancelok(self):
- self.amqp_consumer._channel = mock.Mock()
- self.amqp_consumer._channel.close = mock.Mock()
- self.assertIsNone(self.amqp_consumer.on_cancelok(10))
-
- def test_run(self):
- self.amqp_consumer._connection = mock.Mock(autospec=AmqpConsumer)
- self.amqp_consumer.connect = mock.Mock()
- self.amqp_consumer._connection.ioloop.start = mock.Mock()
- self.assertIsNone(self.amqp_consumer.run())
-
- def test_stop(self):
- self.amqp_consumer._connection = mock.Mock(autospec=AmqpConsumer)
- self.amqp_consumer.connect = mock.Mock()
- self.amqp_consumer._connection.ioloop.start = mock.Mock()
- self.amqp_consumer._channel = mock.Mock()
- self.amqp_consumer._channel.basic_cancel = mock.Mock()
- self.assertIsNone(self.amqp_consumer.stop())
-
- def test_close_connection(self):
- self.amqp_consumer._connection = mock.Mock(autospec=AmqpConsumer)
- self.amqp_consumer._connection.close = mock.Mock()
- self.assertIsNone(self.amqp_consumer.close_connection())
diff --git a/tests/unit/network_services/nfvi/test_resource.py b/tests/unit/network_services/nfvi/test_resource.py
deleted file mode 100644
index f5f7f0fe7..000000000
--- a/tests/unit/network_services/nfvi/test_resource.py
+++ /dev/null
@@ -1,273 +0,0 @@
-# Copyright (c) 2016-2017 Intel Corporation
-#
-# Licensed under the Apache License, Version 2.0 (the "License");
-# you may not use this file except in compliance with the License.
-# You may obtain a copy of the License at
-#
-# http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing, software
-# distributed under the License is distributed on an "AS IS" BASIS,
-# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-# See the License for the specific language governing permissions and
-# limitations under the License.
-
-import errno
-
-import mock
-import unittest
-
-from yardstick.network_services.nfvi.resource import ResourceProfile
-from yardstick.network_services.nfvi import resource, collectd
-
-
-class TestResourceProfile(unittest.TestCase):
- VNFD = {'vnfd:vnfd-catalog':
- {'vnfd':
- [{'short-name': 'VpeVnf',
- 'vdu':
- [{'routing_table':
- [{'network': '172.16.100.20',
- 'netmask': '255.255.255.0',
- 'gateway': '172.16.100.20',
- 'if': 'xe0'},
- {'network': '172.16.40.20',
- 'netmask': '255.255.255.0',
- 'gateway': '172.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': '3c:fd:fe:9e:64:38',
- 'vpci': '0000:05:00.0',
- 'local_ip': '172.16.100.19',
- 'type': 'PCI-PASSTHROUGH',
- 'netmask': '255.255.255.0',
- 'dpdk_port_num': 0,
- 'bandwidth': '10 Gbps',
- 'dst_ip': '172.16.100.20',
- 'local_mac': '3c:fd:fe:a1:2b:80'},
- 'vnfd-connection-point-ref': 'xe0',
- 'name': 'xe0'},
- {'virtual-interface':
- {'dst_mac': '00:1e:67:d0:60:5c',
- 'vpci': '0000:05:00.1',
- 'local_ip': '172.16.40.19',
- 'type': 'PCI-PASSTHROUGH',
- 'netmask': '255.255.255.0',
- 'dpdk_port_num': 1,
- 'bandwidth': '10 Gbps',
- 'dst_ip': '172.16.40.20',
- 'local_mac': '3c:fd:fe:a1:2b:81'},
- 'vnfd-connection-point-ref': 'xe1',
- 'name': 'xe1'}]}],
- 'description': 'Vpe approximation using DPDK',
- 'mgmt-interface':
- {'vdu-id': 'vpevnf-baremetal',
- 'host': '127.0.0.1',
- 'password': 'r00t',
- 'user': 'root',
- 'ip': '127.0.0.1'},
- 'benchmark':
- {'kpi': ['packets_in', 'packets_fwd', 'packets_dropped']},
- 'connection-point': [{'type': 'VPORT', 'name': 'xe0'},
- {'type': 'VPORT', 'name': 'xe1'}],
- 'id': 'VpeApproxVnf', 'name': 'VPEVnfSsh'}]}}
-
- def setUp(self):
- with mock.patch("yardstick.ssh.AutoConnectSSH") as ssh:
- self.ssh_mock = mock.Mock(autospec=ssh.SSH)
- self.ssh_mock.execute = \
- mock.Mock(return_value=(0, "", ""))
- ssh.from_node.return_value = self.ssh_mock
-
- mgmt = self.VNFD['vnfd:vnfd-catalog']['vnfd'][0]['mgmt-interface']
- # interfaces = \
- # self.VNFD['vnfd:vnfd-catalog']['vnfd'][0]['vdu'][0]['external-interface']
- port_names = \
- self.VNFD['vnfd:vnfd-catalog']['vnfd'][0]['vdu'][0]['external-interface']
- self.resource_profile = \
- ResourceProfile(mgmt, port_names)
- self.resource_profile.connection = self.ssh_mock
-
- def test___init__(self):
- self.assertTrue(self.resource_profile.enable)
-
- def test_check_if_system_agent_running(self):
- self.assertEqual(self.resource_profile.check_if_system_agent_running("collectd"),
- (0, ""))
-
- def test_check_if_system_agent_running_excetion(self):
- with mock.patch.object(self.resource_profile.connection, "execute") as mock_execute:
- mock_execute.side_effect = OSError(errno.ECONNRESET, "error")
- self.assertEqual(
- self.resource_profile.check_if_system_agent_running("collectd"),
- (1, None))
-
- def test_get_cpu_data(self):
- reskey = ["", "cpufreq", "cpufreq-0"]
- value = "metric:10"
- val = self.resource_profile.get_cpu_data(reskey[1], reskey[2], value)
- self.assertIsNotNone(val)
-
- def test_get_cpu_data_error(self):
- reskey = ["", "", ""]
- value = "metric:10"
- val = self.resource_profile.get_cpu_data(reskey[0], reskey[1], value)
- self.assertEqual(val, ('error', 'Invalid', '', ''))
-
- def test__start_collectd(self):
- self.assertIsNone(
- self.resource_profile._start_collectd(self.ssh_mock, "/opt/nsb_bin"))
-
- def test__prepare_collectd_conf(self):
- self.assertIsNone(
- self.resource_profile._prepare_collectd_conf("/opt/nsb_bin"))
-
- def test__setup_ovs_stats(self):
- # TODO(elfoley): This method doesn't actually return anything, the side
- # effects should be checked
- self.assertIsNone(
- self.resource_profile._setup_ovs_stats(self.ssh_mock))
-
- def test__provide_config_file(self,):
- loadplugin = range(5)
- port_names = range(5)
- kwargs = {
- "interval": '25',
- "loadplugin": loadplugin,
- "port_names": port_names,
- }
- self.resource_profile._provide_config_file("/opt/nsb_bin", "collectd.conf", kwargs)
- self.ssh_mock.execute.assert_called_once()
-
- def test_initiate_systemagent(self):
- self.resource_profile._start_collectd = mock.Mock()
- self.assertIsNone(
- self.resource_profile.initiate_systemagent("/opt/nsb_bin"))
-
- def test_initiate_systemagent_raise(self):
- self.resource_profile._start_collectd = mock.Mock(side_effect=RuntimeError)
- with self.assertRaises(RuntimeError):
- self.resource_profile.initiate_systemagent("/opt/nsb_bin")
-
- def test__parse_hugepages(self):
- reskey = ["cpu", "cpuFreq"]
- value = "timestamp:12345"
- res = self.resource_profile.parse_hugepages(reskey, value)
- self.assertEqual({'cpu/cpuFreq': '12345'}, res)
-
- def test__parse_dpdkstat(self):
- reskey = ["dpdk0", "0"]
- value = "tx:12345"
- res = self.resource_profile.parse_dpdkstat(reskey, value)
- self.assertEqual({'dpdk0/0': '12345'}, res)
-
- def test__parse_virt(self):
- reskey = ["vm0", "cpu"]
- value = "load:45"
- res = self.resource_profile.parse_virt(reskey, value)
- self.assertEqual({'vm0/cpu': '45'}, res)
-
- def test__parse_ovs_stats(self):
- reskey = ["ovs", "stats"]
- value = "tx:45"
- res = self.resource_profile.parse_ovs_stats(reskey, value)
- self.assertEqual({'ovs/stats': '45'}, res)
-
- def test_parse_collectd_result(self):
- res = self.resource_profile.parse_collectd_result({})
- expected_result = {'cpu': {}, 'dpdkstat': {}, 'hugepages': {},
- 'memory': {}, 'ovs_stats': {}, 'timestamp': '',
- 'virt': {}}
- self.assertDictEqual(res, expected_result)
-
- def test_parse_collectd_result_cpu(self):
- metric = {"nsb_stats/cpu/0/ipc": "101"}
- self.resource_profile.get_cpu_data = mock.Mock(return_value=[1,
- "ipc",
- "1234",
- ""])
- res = self.resource_profile.parse_collectd_result(metric)
- expected_result = {'cpu': {1: {'ipc': '1234'}}, 'dpdkstat': {}, 'hugepages': {},
- 'memory': {}, 'ovs_stats': {}, 'timestamp': '',
- 'virt': {}}
- self.assertDictEqual(res, expected_result)
-
- def test_parse_collectd_result_memory(self):
- metric = {"nsb_stats/memory/bw": "101"}
- res = self.resource_profile.parse_collectd_result(metric)
- expected_result = {'cpu': {}, 'dpdkstat': {}, 'hugepages': {},
- 'memory': {'bw': '101'}, 'ovs_stats': {}, 'timestamp': '',
- 'virt': {}}
- self.assertDictEqual(res, expected_result)
-
- def test_parse_collectd_result_hugepage(self):
- # amqp returns bytes
- metric = {b"nsb_stats/hugepages/free": b"101"}
- self.resource_profile.parse_hugepages = mock.Mock(return_value={"free": "101"})
- res = self.resource_profile.parse_collectd_result(metric)
- expected_result = {'cpu': {}, 'dpdkstat': {}, 'hugepages': {'free': '101'},
- 'memory': {}, 'ovs_stats': {}, 'timestamp': '',
- 'virt': {}}
- self.assertDictEqual(res, expected_result)
-
- def test_parse_collectd_result_dpdk_virt_ovs(self):
- metric = {b"nsb_stats/dpdkstat/tx": b"101",
- b"nsb_stats/ovs_stats/tx": b"101",
- b"nsb_stats/virt/virt/memory": b"101"}
- self.resource_profile.parse_dpdkstat = \
- mock.Mock(return_value={"tx": "101"})
- self.resource_profile.parse_virt = \
- mock.Mock(return_value={"memory": "101"})
- self.resource_profile.parse_ovs_stats = \
- mock.Mock(return_value={"tx": "101"})
- res = self.resource_profile.parse_collectd_result(metric)
- expected_result = {'cpu': {}, 'dpdkstat': {'tx': '101'}, 'hugepages': {},
- 'memory': {}, 'ovs_stats': {'tx': '101'}, 'timestamp': '',
- 'virt': {'memory': '101'}}
- self.assertDictEqual(res, expected_result)
-
- def test_amqp_process_for_nfvi_kpi(self):
- self.resource_profile.amqp_client = \
- mock.MagicMock(side_effect=[None, mock.MagicMock()])
- self.resource_profile.run_collectd_amqp = \
- mock.Mock(return_value=0)
- res = self.resource_profile.amqp_process_for_nfvi_kpi()
- self.assertIsNone(res)
-
- def test_amqp_collect_nfvi_kpi(self):
- self.resource_profile.amqp_client = \
- mock.MagicMock(side_effect=[None, mock.MagicMock()])
- self.resource_profile.run_collectd_amqp = \
- mock.Mock(return_value=0)
- self.resource_profile.parse_collectd_result = mock.Mock()
- res = self.resource_profile.amqp_collect_nfvi_kpi()
- self.assertIsNotNone(res)
-
- def test_run_collectd_amqp(self):
- resource.AmqpConsumer = mock.Mock(autospec=collectd)
- self.assertIsNone(self.resource_profile.run_collectd_amqp())
-
- def test_start(self):
- self.assertIsNone(self.resource_profile.start())
-
- def test_stop(self):
- self.assertIsNone(self.resource_profile.stop())
-
- def test_stop_amqp_not_running(self):
- self.resource_profile.amqp_client = mock.MagicMock()
- # TODO(efoley): Fix this incorrect test.
- # Should check that we don't try to stop amqp when it's not running
- self.assertIsNone(self.resource_profile.stop())
diff --git a/tests/unit/network_services/test_utils.py b/tests/unit/network_services/test_utils.py
deleted file mode 100644
index bf98a4474..000000000
--- a/tests/unit/network_services/test_utils.py
+++ /dev/null
@@ -1,143 +0,0 @@
-# Copyright (c) 2016-2017 Intel Corporation
-#
-# Licensed under the Apache License, Version 2.0 (the "License");
-# you may not use this file except in compliance with the License.
-# You may obtain a copy of the License at
-#
-# http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing, software
-# distributed under the License is distributed on an "AS IS" BASIS,
-# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-# See the License for the specific language governing permissions and
-# limitations under the License.
-#
-
-# Unittest for yardstick.network_services.utils
-
-import os
-import unittest
-import mock
-
-from yardstick.network_services import utils
-
-
-class UtilsTestCase(unittest.TestCase):
- """Test all VNF helper methods."""
-
- DPDK_PATH = os.path.join(utils.NSB_ROOT, "dpdk-devbind.py")
-
- def setUp(self):
- super(UtilsTestCase, self).setUp()
-
- def test_get_nsb_options(self):
- result = utils.get_nsb_option("bin_path", None)
- self.assertEqual(result, utils.NSB_ROOT)
-
- def test_get_nsb_option_is_invalid_key(self):
- result = utils.get_nsb_option("bin", None)
- self.assertEqual(result, None)
-
- def test_get_nsb_option_default(self):
- default = object()
- result = utils.get_nsb_option("nosuch", default)
- self.assertIs(result, default)
-
- def test_provision_tool(self):
- with mock.patch("yardstick.ssh.SSH") as ssh:
- ssh_mock = mock.Mock(autospec=ssh.SSH)
- ssh_mock.execute = \
- mock.Mock(return_value=(0, self.DPDK_PATH, ""))
- ssh.return_value = ssh_mock
- tool_path = utils.provision_tool(ssh_mock, self.DPDK_PATH)
- self.assertEqual(tool_path, self.DPDK_PATH)
-
- def test_provision_tool_no_path(self):
- with mock.patch("yardstick.ssh.SSH") as ssh:
- ssh_mock = mock.Mock(autospec=ssh.SSH)
- ssh_mock.execute = \
- mock.Mock(return_value=(1, self.DPDK_PATH, ""))
- ssh.return_value = ssh_mock
- tool_path = utils.provision_tool(ssh_mock, self.DPDK_PATH)
- self.assertEqual(tool_path, self.DPDK_PATH)
-
-
-class PciAddressTestCase(unittest.TestCase):
-
- PCI_ADDRESS_DBSF = '000A:07:03.2'
- PCI_ADDRESS_BSF = '06:02.1'
- PCI_ADDRESS_DBSF_MULTILINE_1 = '0001:08:04.3\nother text\n'
- PCI_ADDRESS_DBSF_MULTILINE_2 = 'first line\n 0001:08:04.3 \nother text\n'
- # Will match and return the first address found.
- PCI_ADDRESS_DBSF_MULTILINE_3 = ' 0001:08:04.1 \n 05:03.1 \nother\n'
- PCI_ADDRESS_BSF_MULTILINE_1 = 'first line\n 08:04.3 \n 0002:05:03.1\n'
- BAD_INPUT_1 = 'no address found'
- BAD_INPUT_2 = '001:08:04.1'
- BAD_INPUT_3 = '08:4.1'
-
- def test_pciaddress_dbsf(self):
- pci_address = utils.PciAddress(PciAddressTestCase.PCI_ADDRESS_DBSF)
- self.assertEqual('000a', pci_address.domain)
- self.assertEqual('07', pci_address.bus)
- self.assertEqual('03', pci_address.slot)
- self.assertEqual('2', pci_address.function)
-
- def test_pciaddress_bsf(self):
- pci_address = utils.PciAddress(PciAddressTestCase.PCI_ADDRESS_BSF)
- self.assertEqual('0000', pci_address.domain)
- self.assertEqual('06', pci_address.bus)
- self.assertEqual('02', pci_address.slot)
- self.assertEqual('1', pci_address.function)
-
- def test_pciaddress_dbsf_multiline_1(self):
- pci_address = utils.PciAddress(
- PciAddressTestCase.PCI_ADDRESS_DBSF_MULTILINE_1)
- self.assertEqual('0001', pci_address.domain)
- self.assertEqual('08', pci_address.bus)
- self.assertEqual('04', pci_address.slot)
- self.assertEqual('3', pci_address.function)
-
- def test_pciaddress_dbsf_multiline_2(self):
- pci_address = utils.PciAddress(
- PciAddressTestCase.PCI_ADDRESS_DBSF_MULTILINE_2)
- self.assertEqual('0001', pci_address.domain)
- self.assertEqual('08', pci_address.bus)
- self.assertEqual('04', pci_address.slot)
- self.assertEqual('3', pci_address.function)
-
- def test_pciaddress_dbsf_multiline_3(self):
- pci_address = utils.PciAddress(
- PciAddressTestCase.PCI_ADDRESS_DBSF_MULTILINE_3)
- self.assertEqual('0001', pci_address.domain)
- self.assertEqual('08', pci_address.bus)
- self.assertEqual('04', pci_address.slot)
- self.assertEqual('1', pci_address.function)
-
- def test_pciaddress_bsf_multiline_1(self):
- pci_address = utils.PciAddress(
- PciAddressTestCase.PCI_ADDRESS_BSF_MULTILINE_1)
- self.assertEqual('0000', pci_address.domain)
- self.assertEqual('08', pci_address.bus)
- self.assertEqual('04', pci_address.slot)
- self.assertEqual('3', pci_address.function)
-
- def test_pciaddress_bad_input_no_address(self):
- with self.assertRaises(ValueError) as exception:
- utils.PciAddress(PciAddressTestCase.BAD_INPUT_1)
- self.assertEqual('Invalid PCI address: {}'.format(
- PciAddressTestCase.BAD_INPUT_1), str(exception.exception))
-
- def test_pciaddress_bad_input_dbsf_bad_formatted(self):
- # In this test case, the domain has only 3 characters instead of 4.
- pci_address = utils.PciAddress(
- PciAddressTestCase.BAD_INPUT_2)
- self.assertEqual('0000', pci_address.domain)
- self.assertEqual('08', pci_address.bus)
- self.assertEqual('04', pci_address.slot)
- self.assertEqual('1', pci_address.function)
-
- def test_pciaddress_bad_input_bsf_bad_formatted(self):
- with self.assertRaises(ValueError) as exception:
- utils.PciAddress(PciAddressTestCase.BAD_INPUT_3)
- self.assertEqual('Invalid PCI address: {}'.format(
- PciAddressTestCase.BAD_INPUT_3), str(exception.exception))
diff --git a/tests/unit/network_services/test_yang_model.py b/tests/unit/network_services/test_yang_model.py
deleted file mode 100644
index 0b29da701..000000000
--- a/tests/unit/network_services/test_yang_model.py
+++ /dev/null
@@ -1,135 +0,0 @@
-# Copyright (c) 2016-2017 Intel Corporation
-#
-# Licensed under the Apache License, Version 2.0 (the "License");
-# you may not use this file except in compliance with the License.
-# You may obtain a copy of the License at
-#
-# http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing, software
-# distributed under the License is distributed on an "AS IS" BASIS,
-# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-# See the License for the specific language governing permissions and
-# limitations under the License.
-#
-
-# Unittest for yardstick.network_services.utils
-
-from __future__ import absolute_import
-
-import unittest
-import mock
-
-import yaml
-
-from yardstick.network_services.yang_model import YangModel
-
-
-class YangModelTestCase(unittest.TestCase):
- """Test all Yang Model methods."""
-
- ENTRIES = {
- 'access-list1': {
- 'acl': {
- 'access-list-entries': [{
- 'ace': {
- 'ace-oper-data': {
- 'match-counter': 0},
- 'actions': 'drop,count',
- 'matches': {
- 'destination-ipv4-network':
- '152.16.40.20/24',
- 'destination-port-range': {
- 'lower-port': 0,
- 'upper-port': 65535},
- 'source-ipv4-network': '0.0.0.0/0',
- 'source-port-range': {
- 'lower-port': 0,
- 'upper-port': 65535}},
- 'rule-name': 'rule1588'}},
- {
- 'ace': {
- 'ace-oper-data': {
- 'match-counter': 0},
- 'actions': 'drop,count',
- 'matches': {
- 'destination-ipv4-network':
- '0.0.0.0/0',
- 'destination-port-range': {
- 'lower-port': 0,
- 'upper-port': 65535},
- 'source-ipv4-network':
- '152.16.100.20/24',
- 'source-port-range': {
- 'lower-port': 0,
- 'upper-port': 65535}},
- 'rule-name': 'rule1589'}}],
- 'acl-name': 'sample-ipv4-acl',
- 'acl-type': 'ipv4-acl'}
- }
- }
-
- def test__init__(self):
- cfg = "yang.yaml"
- y = YangModel(cfg)
- self.assertEqual(y.config_file, cfg)
-
- def test_config_file_setter(self):
- cfg = "yang.yaml"
- y = YangModel(cfg)
- self.assertEqual(y.config_file, cfg)
- cfg2 = "yang2.yaml"
- y.config_file = cfg2
- self.assertEqual(y.config_file, cfg2)
-
- def test__get_entries(self):
- cfg = "yang.yaml"
- y = YangModel(cfg)
- y._options = self.ENTRIES
- y._get_entries()
- self.assertIn("p acl add", y._rules)
-
- def test__get_entries_no_options(self):
- cfg = "yang.yaml"
- y = YangModel(cfg)
- y._get_entries()
- self.assertEqual(y._rules, '')
-
- @mock.patch('yardstick.network_services.yang_model.yaml_load')
- @mock.patch('yardstick.network_services.yang_model.open')
- def test__read_config(self, mock_open, mock_safe_load):
- cfg = "yang.yaml"
- y = YangModel(cfg)
- mock_safe_load.return_value = expected = {'key1': 'value1', 'key2': 'value2'}
- y._read_config()
- self.assertDictEqual(y._options, expected)
-
- @mock.patch('yardstick.network_services.yang_model.open')
- def test__read_config_open_error(self, mock_open):
- cfg = "yang.yaml"
- y = YangModel(cfg)
- mock_open.side_effect = IOError('my error')
-
- self.assertEqual(y._options, {})
- with self.assertRaises(IOError) as raised:
- y._read_config()
-
- self.assertIn('my error', str(raised.exception))
- self.assertEqual(y._options, {})
-
- def test_get_rules(self):
- cfg = "yang.yaml"
- y = YangModel(cfg)
- y._read_config = read_mock = mock.Mock()
- y._get_entries = get_mock = mock.Mock()
-
- y._rules = None
- self.assertIsNone(y.get_rules())
- self.assertEqual(read_mock.call_count, 1)
- self.assertEqual(get_mock.call_count, 1)
-
- # True value should prevent calling read and get
- y._rules = 999
- self.assertEqual(y.get_rules(), 999)
- self.assertEqual(read_mock.call_count, 1)
- self.assertEqual(get_mock.call_count, 1)
diff --git a/tests/unit/network_services/traffic_profile/__init__.py b/tests/unit/network_services/traffic_profile/__init__.py
deleted file mode 100644
index e69de29bb..000000000
--- a/tests/unit/network_services/traffic_profile/__init__.py
+++ /dev/null
diff --git a/tests/unit/network_services/traffic_profile/test_base.py b/tests/unit/network_services/traffic_profile/test_base.py
deleted file mode 100644
index 3b8804976..000000000
--- a/tests/unit/network_services/traffic_profile/test_base.py
+++ /dev/null
@@ -1,73 +0,0 @@
-# Copyright (c) 2016-2017 Intel Corporation
-#
-# Licensed under the Apache License, Version 2.0 (the "License");
-# you may not use this file except in compliance with the License.
-# You may obtain a copy of the License at
-#
-# http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing, software
-# distributed under the License is distributed on an "AS IS" BASIS,
-# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-# See the License for the specific language governing permissions and
-# limitations under the License.
-
-import sys
-
-import mock
-import unittest
-
-from yardstick.common import exceptions
-from yardstick.network_services import traffic_profile as tprofile_package
-from yardstick.network_services.traffic_profile import base
-from yardstick import tests as y_tests
-
-
-class TestTrafficProfile(unittest.TestCase):
- TRAFFIC_PROFILE = {
- "schema": "isb:traffic_profile:0.1",
- "name": "fixed",
- "description": "Fixed traffic profile to run UDP traffic",
- "traffic_profile": {
- "traffic_type": "FixedTraffic",
- "frame_rate": 100, # pps
- "flow_number": 10,
- "frame_size": 64}}
-
- def _get_res_mock(self, **kw):
- _mock = mock.MagicMock()
- for k, v in kw.items():
- setattr(_mock, k, v)
- return _mock
-
- def test___init__(self):
- traffic_profile = base.TrafficProfile(self.TRAFFIC_PROFILE)
- self.assertEqual(self.TRAFFIC_PROFILE, traffic_profile.params)
-
- def test_execute(self):
- traffic_profile = base.TrafficProfile(self.TRAFFIC_PROFILE)
- self.assertRaises(NotImplementedError,
- traffic_profile.execute_traffic, {})
-
- def test_get_existing_traffic_profile(self):
- traffic_profile_list = [
- 'RFC2544Profile', 'FixedProfile', 'TrafficProfileGenericHTTP',
- 'IXIARFC2544Profile', 'ProxACLProfile', 'ProxBinSearchProfile',
- 'ProxProfile', 'ProxRampProfile']
- with mock.patch.dict(sys.modules, y_tests.STL_MOCKS):
- tprofile_package.register_modules()
-
- for tp in traffic_profile_list:
- traffic_profile = base.TrafficProfile.get(
- {'traffic_profile': {'traffic_type': tp}})
- self.assertEqual(tp, traffic_profile.__class__.__name__)
-
- def test_get_non_existing_traffic_profile(self):
- self.assertRaises(exceptions.TrafficProfileNotImplemented,
- base.TrafficProfile.get, self.TRAFFIC_PROFILE)
-
-
-class TestDummyProfile(unittest.TestCase):
- def test_execute(self):
- dummy_profile = base.DummyProfile(base.TrafficProfile)
- self.assertIsNone(dummy_profile.execute({}))
diff --git a/tests/unit/network_services/traffic_profile/test_fixed.py b/tests/unit/network_services/traffic_profile/test_fixed.py
deleted file mode 100644
index dec94964b..000000000
--- a/tests/unit/network_services/traffic_profile/test_fixed.py
+++ /dev/null
@@ -1,120 +0,0 @@
-# Copyright (c) 2016-2017 Intel Corporation
-#
-# Licensed under the Apache License, Version 2.0 (the "License");
-# you may not use this file except in compliance with the License.
-# You may obtain a copy of the License at
-#
-# http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing, software
-# distributed under the License is distributed on an "AS IS" BASIS,
-# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-# See the License for the specific language governing permissions and
-# limitations under the License.
-#
-
-from __future__ import absolute_import
-
-import unittest
-import mock
-
-from tests.unit import STL_MOCKS
-
-STLClient = mock.MagicMock()
-stl_patch = mock.patch.dict("sys.modules", STL_MOCKS)
-stl_patch.start()
-
-if stl_patch:
- from yardstick.network_services.traffic_profile.base import TrafficProfile
- from yardstick.network_services.traffic_profile.fixed import FixedProfile
-
-
-class TestFixedProfile(unittest.TestCase):
- TRAFFIC_PROFILE = {
- "schema": "isb:traffic_profile:0.1",
- "name": "fixed",
- "description": "Fixed traffic profile to run UDP traffic",
- "traffic_profile": {
- "traffic_type": "FixedTraffic",
- "frame_rate": 100, # pps
- "flow_number": 10,
- "frame_size": 64}}
-
- VNFD = {'vnfd:vnfd-catalog':
- {'vnfd':
- [{'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',
- 'dst_ip': '152.16.100.20',
- 'local_mac': '00:00:00:00:00:01'},
- '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',
- 'netmask': '255.255.255.0',
- 'dpdk_port_num': 1,
- 'bandwidth': '10 Gbps',
- 'dst_ip': '152.16.40.20',
- 'local_mac': '00:00:00:00:00:02'},
- 'vnfd-connection-point-ref': 'xe1',
- 'name': 'xe1'}]}],
- 'description': 'Vpe approximation using DPDK',
- 'mgmt-interface':
- {'vdu-id': 'vpevnf-baremetal',
- 'host': '1.1.1.1',
- 'password': 'r00t',
- 'user': 'root',
- 'ip': '1.1.1.1'},
- 'benchmark':
- {'kpi': ['packets_in', 'packets_fwd', 'packets_dropped']},
- 'connection-point': [{'type': 'VPORT', 'name': 'xe0'},
- {'type': 'VPORT', 'name': 'xe1'}],
- 'id': 'VpeApproxVnf', 'name': 'VPEVnfSsh'}]}}
-
- def test___init__(self):
- fixed_profile = \
- FixedProfile(TrafficProfile)
- self.assertIsNotNone(fixed_profile)
-
- def test_execute(self):
- traffic_generator = mock.Mock(autospec=TrafficProfile)
- traffic_generator.my_ports = [0, 1]
- traffic_generator.vnfd = self.VNFD['vnfd:vnfd-catalog']['vnfd'][0]
- traffic_generator.client = \
- mock.Mock(return_value=True)
- fixed_profile = FixedProfile(self.TRAFFIC_PROFILE)
- fixed_profile.params = self.TRAFFIC_PROFILE
- fixed_profile.first_run = True
- self.assertIsNone(fixed_profile.execute(traffic_generator))
diff --git a/tests/unit/network_services/traffic_profile/test_http.py b/tests/unit/network_services/traffic_profile/test_http.py
deleted file mode 100644
index 5d8029ea0..000000000
--- a/tests/unit/network_services/traffic_profile/test_http.py
+++ /dev/null
@@ -1,41 +0,0 @@
-# Copyright (c) 2016-2017 Intel Corporation
-#
-# Licensed under the Apache License, Version 2.0 (the "License");
-# you may not use this file except in compliance with the License.
-# You may obtain a copy of the License at
-#
-# http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing, software
-# distributed under the License is distributed on an "AS IS" BASIS,
-# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-# See the License for the specific language governing permissions and
-# limitations under the License.
-#
-
-from __future__ import absolute_import
-import unittest
-
-from yardstick.network_services.traffic_profile.base import TrafficProfile
-from yardstick.network_services.traffic_profile.http import \
- TrafficProfileGenericHTTP
-
-
-class TestTrafficProfileGenericHTTP(unittest.TestCase):
- def test___init__(self):
- traffic_profile_generic_htt_p = \
- TrafficProfileGenericHTTP(TrafficProfile)
- self.assertIsNotNone(traffic_profile_generic_htt_p)
-
- def test_execute(self):
- traffic_profile_generic_htt_p = \
- TrafficProfileGenericHTTP(TrafficProfile)
- traffic_generator = {}
- self.assertIsNone(
- traffic_profile_generic_htt_p.execute(traffic_generator))
-
- def test__send_http_request(self):
- traffic_profile_generic_htt_p = \
- TrafficProfileGenericHTTP(TrafficProfile)
- self.assertIsNone(traffic_profile_generic_htt_p._send_http_request(
- "10.1.1.1", "250", "/req"))
diff --git a/tests/unit/network_services/traffic_profile/test_http_ixload.py b/tests/unit/network_services/traffic_profile/test_http_ixload.py
deleted file mode 100644
index 5110439fd..000000000
--- a/tests/unit/network_services/traffic_profile/test_http_ixload.py
+++ /dev/null
@@ -1,272 +0,0 @@
-# Copyright (c) 2017 Intel Corporation
-#
-# Licensed under the Apache License, Version 2.0 (the "License");
-# you may not use this file except in compliance with the License.
-# You may obtain a copy of the License at
-#
-# http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing, software
-# distributed under the License is distributed on an "AS IS" BASIS,
-# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-# See the License for the specific language governing permissions and
-# limitations under the License.
-
-
-from __future__ import absolute_import
-import unittest
-import mock
-
-from oslo_serialization import jsonutils
-
-from yardstick.network_services.traffic_profile import http_ixload
-from yardstick.network_services.traffic_profile.http_ixload import \
- join_non_strings, validate_non_string_sequence
-
-
-class TestJoinNonStrings(unittest.TestCase):
-
- def test_validate_non_string_sequence(self):
- self.assertEqual(validate_non_string_sequence([1, 2, 3]), [1, 2, 3])
- self.assertIsNone(validate_non_string_sequence('123'))
- self.assertIsNone(validate_non_string_sequence(1))
-
- self.assertEqual(validate_non_string_sequence(1, 2), 2)
- self.assertEqual(validate_non_string_sequence(1, default=2), 2)
-
- with self.assertRaises(RuntimeError):
- validate_non_string_sequence(1, raise_exc=RuntimeError)
-
- def test_join_non_strings(self):
- self.assertEqual(join_non_strings(':'), '')
- self.assertEqual(join_non_strings(':', 'a'), 'a')
- self.assertEqual(join_non_strings(':', 'a', 2, 'c'), 'a:2:c')
- self.assertEqual(join_non_strings(':', ['a', 2, 'c']), 'a:2:c')
- self.assertEqual(join_non_strings(':', 'abc'), 'abc')
-
-
-class TestIxLoadTrafficGen(unittest.TestCase):
-
- def test_parse_run_test(self):
- ports = [1, 2, 3]
- test_input = {
- "remote_server": "REMOTE_SERVER",
- "ixload_cfg": "IXLOAD_CFG",
- "result_dir": "RESULT_DIR",
- "ixia_chassis": "IXIA_CHASSIS",
- "IXIA": {
- "card": "CARD",
- "ports": ports,
- },
- }
- j = jsonutils.dump_as_bytes(test_input)
- ixload = http_ixload.IXLOADHttpTest(j)
- self.assertDictEqual(ixload.test_input, test_input)
- self.assertIsNone(ixload.parse_run_test())
- self.assertEqual(ixload.ports_to_reassign, [
- ["IXIA_CHASSIS", "CARD", 1],
- ["IXIA_CHASSIS", "CARD", 2],
- ["IXIA_CHASSIS", "CARD", 3],
- ])
-
- def test_format_ports_for_reassignment(self):
- ports = [
- ["IXIA_CHASSIS", "CARD", 1],
- ["IXIA_CHASSIS", "CARD", 2],
- ["IXIA_CHASSIS", "CARD", 3],
- ]
- formatted = http_ixload.IXLOADHttpTest.format_ports_for_reassignment(ports)
- self.assertEqual(formatted, [
- "IXIA_CHASSIS;CARD;1",
- "IXIA_CHASSIS;CARD;2",
- "IXIA_CHASSIS;CARD;3",
- ])
-
- def test_reassign_ports(self):
- ports = [1, 2, 3]
- test_input = {
- "remote_server": "REMOTE_SERVER",
- "ixload_cfg": "IXLOAD_CFG",
- "result_dir": "RESULT_DIR",
- "ixia_chassis": "IXIA_CHASSIS",
- "IXIA": {
- "card": "CARD",
- "ports": ports,
- },
- }
- j = jsonutils.dump_as_bytes(test_input)
- ixload = http_ixload.IXLOADHttpTest(j)
- repository = mock.Mock()
- test = mock.MagicMock()
- test.setPorts = mock.Mock()
- ports_to_reassign = [(1, 2, 3), (1, 2, 4)]
- ixload.format_ports_for_reassignment = mock.Mock(return_value=["1;2;3"])
- self.assertIsNone(ixload.reassign_ports(test, repository, ports_to_reassign))
-
- def test_reassign_ports_error(self):
- ports = [1, 2, 3]
- test_input = {
- "remote_server": "REMOTE_SERVER",
- "ixload_cfg": "IXLOAD_CFG",
- "result_dir": "RESULT_DIR",
- "ixia_chassis": "IXIA_CHASSIS",
- "IXIA": {
- "card": "CARD",
- "ports": ports,
- },
- }
- j = jsonutils.dump_as_bytes(test_input)
- ixload = http_ixload.IXLOADHttpTest(j)
- repository = mock.Mock()
- test = "test"
- ports_to_reassign = [(1, 2, 3), (1, 2, 4)]
- ixload.format_ports_for_reassignment = mock.Mock(return_value=["1;2;3"])
- ixload.ix_load = mock.MagicMock()
- ixload.ix_load.delete = mock.Mock()
- ixload.ix_load.disconnect = mock.Mock()
- with self.assertRaises(Exception):
- ixload.reassign_ports(test, repository, ports_to_reassign)
-
- def test_stat_collector(self):
- args = [0, 1]
- self.assertIsNone(
- http_ixload.IXLOADHttpTest.stat_collector(*args))
-
- def test_IxL_StatCollectorCommand(self):
- args = [[0, 1, 2, 3], [0, 1, 2, 3]]
- self.assertIsNone(
- http_ixload.IXLOADHttpTest.IxL_StatCollectorCommand(*args))
-
- def test_set_results_dir(self):
- test_stat_collector = mock.MagicMock()
- test_stat_collector.setResultDir = mock.Mock()
- results_on_windows = "c:/Results"
- self.assertIsNone(
- http_ixload.IXLOADHttpTest.set_results_dir(test_stat_collector,
- results_on_windows))
-
- def test_set_results_dir_error(self):
- test_stat_collector = ""
- results_on_windows = "c:/Results"
- with self.assertRaises(Exception):
- http_ixload.IXLOADHttpTest.set_results_dir(test_stat_collector, results_on_windows)
-
- def test_load_config_file(self):
- ports = [1, 2, 3]
- test_input = {
- "remote_server": "REMOTE_SERVER",
- "ixload_cfg": "IXLOAD_CFG",
- "result_dir": "RESULT_DIR",
- "ixia_chassis": "IXIA_CHASSIS",
- "IXIA": {
- "card": "CARD",
- "ports": ports,
- },
- }
- j = jsonutils.dump_as_bytes(test_input)
- ixload = http_ixload.IXLOADHttpTest(j)
- ixload.ix_load = mock.MagicMock()
- ixload.ix_load.new = mock.Mock(return_value="")
- self.assertIsNotNone(ixload.load_config_file("ixload.cfg"))
-
- def test_load_config_file_error(self):
- ports = [1, 2, 3]
- test_input = {
- "remote_server": "REMOTE_SERVER",
- "ixload_cfg": "IXLOAD_CFG",
- "result_dir": "RESULT_DIR",
- "ixia_chassis": "IXIA_CHASSIS",
- "IXIA": {
- "card": "CARD",
- "ports": ports,
- },
- }
- j = jsonutils.dump_as_bytes(test_input)
- ixload = http_ixload.IXLOADHttpTest(j)
- ixload.ix_load = "test"
- with self.assertRaises(Exception):
- ixload.load_config_file("ixload.cfg")
-
- @mock.patch('yardstick.network_services.traffic_profile.http_ixload.IxLoad')
- @mock.patch('yardstick.network_services.traffic_profile.http_ixload.StatCollectorUtils')
- def test_start_http_test_connect_error(self, mock_collector_type, mock_ixload_type):
- ports = [1, 2, 3]
- test_input = {
- "remote_server": "REMOTE_SERVER",
- "ixload_cfg": "IXLOAD_CFG",
- "result_dir": "RESULT_DIR",
- "ixia_chassis": "IXIA_CHASSIS",
- "IXIA": {
- "card": "CARD",
- "ports": ports,
- },
- }
-
- j = jsonutils.dump_as_bytes(test_input)
-
- mock_ixload = mock_ixload_type()
- mock_ixload.connect.side_effect = RuntimeError
-
- ixload = http_ixload.IXLOADHttpTest(j)
- ixload.results_on_windows = 'windows_result_dir'
- ixload.result_dir = 'my_result_dir'
-
- with self.assertRaises(RuntimeError):
- ixload.start_http_test()
-
- @mock.patch('yardstick.network_services.traffic_profile.http_ixload.IxLoad')
- @mock.patch('yardstick.network_services.traffic_profile.http_ixload.StatCollectorUtils')
- def test_start_http_test(self, mock_collector_type, mock_ixload_type):
- ports = [1, 2, 3]
- test_input = {
- "remote_server": "REMOTE_SERVER",
- "ixload_cfg": "IXLOAD_CFG",
- "result_dir": "RESULT_DIR",
- "ixia_chassis": "IXIA_CHASSIS",
- "IXIA": {
- "card": "CARD",
- "ports": ports,
- },
- }
-
- j = jsonutils.dump_as_bytes(test_input)
-
- ixload = http_ixload.IXLOADHttpTest(j)
- ixload.results_on_windows = 'windows_result_dir'
- ixload.result_dir = 'my_result_dir'
- ixload.load_config_file = mock.MagicMock()
-
- self.assertIsNone(ixload.start_http_test())
-
- @mock.patch('yardstick.network_services.traffic_profile.http_ixload.IxLoad')
- @mock.patch('yardstick.network_services.traffic_profile.http_ixload.StatCollectorUtils')
- def test_start_http_test_reassign_error(self, mock_collector_type, mock_ixload_type):
- ports = [1, 2, 3]
- test_input = {
- "remote_server": "REMOTE_SERVER",
- "ixload_cfg": "IXLOAD_CFG",
- "result_dir": "RESULT_DIR",
- "ixia_chassis": "IXIA_CHASSIS",
- "IXIA": {
- "card": "CARD",
- "ports": ports,
- },
- }
-
- j = jsonutils.dump_as_bytes(test_input)
-
- ixload = http_ixload.IXLOADHttpTest(j)
- ixload.load_config_file = mock.MagicMock()
-
- reassign_ports = mock.Mock(side_effect=RuntimeError)
- ixload.reassign_ports = reassign_ports
- ixload.results_on_windows = 'windows_result_dir'
- ixload.result_dir = 'my_result_dir'
-
- ixload.start_http_test()
- self.assertEqual(reassign_ports.call_count, 1)
-
- @mock.patch("yardstick.network_services.traffic_profile.http_ixload.IXLOADHttpTest")
- def test_main(self, IXLOADHttpTest):
- args = ["1", "2", "3"]
- http_ixload.main(args)
diff --git a/tests/unit/network_services/traffic_profile/test_ixia_rfc2544.py b/tests/unit/network_services/traffic_profile/test_ixia_rfc2544.py
deleted file mode 100644
index e8910d62b..000000000
--- a/tests/unit/network_services/traffic_profile/test_ixia_rfc2544.py
+++ /dev/null
@@ -1,611 +0,0 @@
-# Copyright (c) 2016-2017 Intel Corporation
-#
-# Licensed under the Apache License, Version 2.0 (the "License");
-# you may not use this file except in compliance with the License.
-# You may obtain a copy of the License at
-#
-# http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing, software
-# distributed under the License is distributed on an "AS IS" BASIS,
-# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-# See the License for the specific language governing permissions and
-# limitations under the License.
-#
-
-from __future__ import absolute_import
-from __future__ import division
-import unittest
-import mock
-
-from copy import deepcopy
-
-from tests.unit import STL_MOCKS
-
-STLClient = mock.MagicMock()
-stl_patch = mock.patch.dict("sys.modules", STL_MOCKS)
-stl_patch.start()
-
-if stl_patch:
- from yardstick.network_services.traffic_profile.trex_traffic_profile \
- import TrexProfile
- from yardstick.network_services.traffic_profile.ixia_rfc2544 import \
- IXIARFC2544Profile
- from yardstick.network_services.traffic_profile import ixia_rfc2544
-
-
-class TestIXIARFC2544Profile(unittest.TestCase):
-
- TRAFFIC_PROFILE = {
- "schema": "isb:traffic_profile:0.1",
- "name": "fixed",
- "description": "Fixed traffic profile to run UDP traffic",
- "traffic_profile": {
- "traffic_type": "FixedTraffic",
- "frame_rate": 100, # pps
- "flow_number": 10,
- "frame_size": 64,
- },
- }
-
- PROFILE = {'description': 'Traffic profile to run RFC2544 latency',
- 'name': 'rfc2544',
- 'traffic_profile': {'traffic_type': 'IXIARFC2544Profile',
- 'frame_rate': 100},
- IXIARFC2544Profile.DOWNLINK: {'ipv4':
- {'outer_l2': {'framesize':
- {'64B': '100', '1518B': '0',
- '128B': '0', '1400B': '0',
- '256B': '0', '373b': '0',
- '570B': '0'}},
- 'outer_l3v4': {'dstip4': '1.1.1.1-1.15.255.255',
- 'proto': 'udp', 'count': '1',
- 'srcip4': '90.90.1.1-90.105.255.255',
- 'dscp': 0, 'ttl': 32},
- 'outer_l4': {'srcport': '2001',
- 'dsrport': '1234'}}},
- IXIARFC2544Profile.UPLINK: {'ipv4':
- {'outer_l2': {'framesize':
- {'64B': '100', '1518B': '0',
- '128B': '0', '1400B': '0',
- '256B': '0', '373b': '0',
- '570B': '0'}},
- 'outer_l3v4': {'dstip4': '9.9.1.1-90.105.255.255',
- 'proto': 'udp', 'count': '1',
- 'srcip4': '1.1.1.1-1.15.255.255',
- 'dscp': 0, 'ttl': 32},
- 'outer_l4': {'dstport': '2001',
- 'srcport': '1234'}}},
- 'schema': 'isb:traffic_profile:0.1'}
-
- def test_get_ixia_traffic_profile_error(self):
- traffic_generator = mock.Mock(autospec=TrexProfile)
- traffic_generator.my_ports = [0, 1]
- traffic_generator.uplink_ports = [-1]
- traffic_generator.downlink_ports = [1]
- traffic_generator.client = \
- mock.Mock(return_value=True)
- STATIC_TRAFFIC = {
- IXIARFC2544Profile.UPLINK: {
- "id": 1,
- "bidir": "False",
- "duration": 60,
- "iload": "100",
- "outer_l2": {
- "dstmac": "00:00:00:00:00:03",
- "framesPerSecond": True,
- "framesize": 64,
- "srcmac": "00:00:00:00:00:01"
- },
- "outer_l3": {
- "dscp": 0,
- "dstip4": "152.16.40.20",
- "proto": "udp",
- "srcip4": "152.16.100.20",
- "ttl": 32
- },
- "outer_l3v4": {
- "dscp": 0,
- "dstip4": "152.16.40.20",
- "proto": "udp",
- "srcip4": "152.16.100.20",
- "ttl": 32
- },
- "outer_l3v6": {
- "count": 1024,
- "dscp": 0,
- "dstip4": "152.16.100.20",
- "proto": "udp",
- "srcip4": "152.16.40.20",
- "ttl": 32
- },
- "outer_l4": {
- "dstport": "2001",
- "srcport": "1234"
- },
- "traffic_type": "continuous"
- },
- IXIARFC2544Profile.DOWNLINK: {
- "id": 2,
- "bidir": "False",
- "duration": 60,
- "iload": "100",
- "outer_l2": {
- "dstmac": "00:00:00:00:00:04",
- "framesPerSecond": True,
- "framesize": 64,
- "srcmac": "00:00:00:00:00:01"
- },
- "outer_l3": {
- "count": 1024,
- "dscp": 0,
- "dstip4": "152.16.100.20",
- "proto": "udp",
- "srcip4": "152.16.40.20",
- "ttl": 32
- },
- "outer_l3v4": {
- "count": 1024,
- "dscp": 0,
- "dstip4": "152.16.100.20",
- "proto": "udp",
- "srcip4": "152.16.40.20",
- "ttl": 32
- },
- "outer_l3v6": {
- "count": 1024,
- "dscp": 0,
- "dstip4": "152.16.100.20",
- "proto": "udp",
- "srcip4": "152.16.40.20",
- "ttl": 32
- },
- "outer_l4": {
- "dstport": "1234",
- "srcport": "2001"
- },
- "traffic_type": "continuous"
- }
- }
- ixia_rfc2544.STATIC_TRAFFIC = STATIC_TRAFFIC
-
- r_f_c2544_profile = IXIARFC2544Profile(self.TRAFFIC_PROFILE)
- r_f_c2544_profile.rate = 100
- mac = {"src_mac_0": "00:00:00:00:00:01",
- "src_mac_1": "00:00:00:00:00:02",
- "src_mac_2": "00:00:00:00:00:02",
- "dst_mac_0": "00:00:00:00:00:03",
- "dst_mac_1": "00:00:00:00:00:04",
- "dst_mac_2": "00:00:00:00:00:04"}
- result = r_f_c2544_profile._get_ixia_traffic_profile(self.PROFILE, mac)
- self.assertIsNotNone(result)
-
- def test_get_ixia_traffic_profile(self):
- traffic_generator = mock.Mock(autospec=TrexProfile)
- traffic_generator.my_ports = [0, 1]
- traffic_generator.uplink_ports = [-1]
- traffic_generator.downlink_ports = [1]
- traffic_generator.client = \
- mock.Mock(return_value=True)
- STATIC_TRAFFIC = {
- IXIARFC2544Profile.UPLINK: {
- "id": 1,
- "bidir": "False",
- "duration": 60,
- "iload": "100",
- "outer_l2": {
- "dstmac": "00:00:00:00:00:03",
- "framesPerSecond": True,
- "framesize": 64,
- "srcmac": "00:00:00:00:00:01"
- },
- "outer_l3": {
- "dscp": 0,
- "dstip4": "152.16.40.20",
- "proto": "udp",
- "srcip4": "152.16.100.20",
- "ttl": 32
- },
- "outer_l3v4": {
- "dscp": 0,
- "dstip4": "152.16.40.20",
- "proto": "udp",
- "srcip4": "152.16.100.20",
- "ttl": 32,
- "count": "1"
- },
- "outer_l3v6": {
- "count": 1024,
- "dscp": 0,
- "dstip4": "152.16.100.20",
- "proto": "udp",
- "srcip4": "152.16.40.20",
- "ttl": 32,
- },
- "outer_l4": {
- "dstport": "2001",
- "srcport": "1234",
- "count": "1"
- },
- "traffic_type": "continuous"
- },
- IXIARFC2544Profile.DOWNLINK: {
- "id": 2,
- "bidir": "False",
- "duration": 60,
- "iload": "100",
- "outer_l2": {
- "dstmac": "00:00:00:00:00:04",
- "framesPerSecond": True,
- "framesize": 64,
- "srcmac": "00:00:00:00:00:01"
- },
- "outer_l3": {
- "count": 1024,
- "dscp": 0,
- "dstip4": "152.16.100.20",
- "proto": "udp",
- "srcip4": "152.16.40.20",
- "ttl": 32
- },
- "outer_l3v4": {
- "count": 1024,
- "dscp": 0,
- "dstip4": "152.16.100.20",
- "proto": "udp",
- "srcip4": "152.16.40.20",
- "ttl": 32,
- },
- "outer_l3v6": {
- "count": 1024,
- "dscp": 0,
- "dstip4": "152.16.100.20",
- "proto": "udp",
- "srcip4": "152.16.40.20",
- "ttl": 32,
- },
- "outer_l4": {
- "dstport": "1234",
- "srcport": "2001",
- "count": "1"
- },
- "traffic_type": "continuous"
- }
- }
- ixia_rfc2544.STATIC_TRAFFIC = STATIC_TRAFFIC
-
- r_f_c2544_profile = IXIARFC2544Profile(self.TRAFFIC_PROFILE)
- r_f_c2544_profile.rate = 100
- mac = {"src_mac_0": "00:00:00:00:00:01",
- "src_mac_1": "00:00:00:00:00:02",
- "src_mac_2": "00:00:00:00:00:02",
- "dst_mac_0": "00:00:00:00:00:03",
- "dst_mac_1": "00:00:00:00:00:04",
- "dst_mac_2": "00:00:00:00:00:04"}
- result = r_f_c2544_profile._get_ixia_traffic_profile(self.PROFILE, mac)
- self.assertIsNotNone(result)
-
- @mock.patch("yardstick.network_services.traffic_profile.ixia_rfc2544.open")
- def test_get_ixia_traffic_profile_v6(self, *args):
- traffic_generator = mock.Mock(autospec=TrexProfile)
- traffic_generator.my_ports = [0, 1]
- traffic_generator.uplink_ports = [-1]
- traffic_generator.downlink_ports = [1]
- traffic_generator.client = \
- mock.Mock(return_value=True)
- STATIC_TRAFFIC = {
- IXIARFC2544Profile.UPLINK: {
- "id": 1,
- "bidir": "False",
- "duration": 60,
- "iload": "100",
- "outer_l2": {
- "dstmac": "00:00:00:00:00:03",
- "framesPerSecond": True,
- "framesize": 64,
- "srcmac": "00:00:00:00:00:01"
- },
- "outer_l3": {
- "dscp": 0,
- "dstip4": "152.16.40.20",
- "proto": "udp",
- "srcip4": "152.16.100.20",
- "ttl": 32
- },
- "outer_l3v4": {
- "dscp": 0,
- "dstip4": "152.16.40.20",
- "proto": "udp",
- "srcip4": "152.16.100.20",
- "ttl": 32
- },
- "outer_l3v6": {
- "count": 1024,
- "dscp": 0,
- "dstip4": "152.16.100.20",
- "proto": "udp",
- "srcip4": "152.16.40.20",
- "ttl": 32
- },
- "outer_l4": {
- "dstport": "2001",
- "srcport": "1234"
- },
- "traffic_type": "continuous"
- },
- IXIARFC2544Profile.DOWNLINK: {
- "id": 2,
- "bidir": "False",
- "duration": 60,
- "iload": "100",
- "outer_l2": {
- "dstmac": "00:00:00:00:00:04",
- "framesPerSecond": True,
- "framesize": 64,
- "srcmac": "00:00:00:00:00:01"
- },
- "outer_l3": {
- "count": 1024,
- "dscp": 0,
- "dstip4": "152.16.100.20",
- "proto": "udp",
- "srcip4": "152.16.40.20",
- "ttl": 32
- },
- "outer_l3v4": {
- "count": 1024,
- "dscp": 0,
- "dstip4": "152.16.100.20",
- "proto": "udp",
- "srcip4": "152.16.40.20",
- "ttl": 32
- },
- "outer_l3v6": {
- "count": 1024,
- "dscp": 0,
- "dstip4": "152.16.100.20",
- "proto": "udp",
- "srcip4": "152.16.40.20",
- "ttl": 32
- },
- "outer_l4": {
- "dstport": "1234",
- "srcport": "2001"
- },
- "traffic_type": "continuous"
- }
- }
- ixia_rfc2544.STATIC_TRAFFIC = STATIC_TRAFFIC
-
- r_f_c2544_profile = IXIARFC2544Profile(self.TRAFFIC_PROFILE)
- r_f_c2544_profile.rate = 100
- mac = {"src_mac_0": "00:00:00:00:00:01",
- "src_mac_1": "00:00:00:00:00:02",
- "src_mac_2": "00:00:00:00:00:02",
- "dst_mac_0": "00:00:00:00:00:03",
- "dst_mac_1": "00:00:00:00:00:04",
- "dst_mac_2": "00:00:00:00:00:04"}
- profile_data = {'description': 'Traffic profile to run RFC2544',
- 'name': 'rfc2544',
- 'traffic_profile':
- {'traffic_type': 'IXIARFC2544Profile',
- 'frame_rate': 100},
- IXIARFC2544Profile.DOWNLINK:
- {'ipv4':
- {'outer_l2': {'framesize':
- {'64B': '100', '1518B': '0',
- '128B': '0', '1400B': '0',
- '256B': '0', '373b': '0',
- '570B': '0'}},
- 'outer_l3v4': {'dstip4': '1.1.1.1-1.15.255.255',
- 'proto': 'udp', 'count': '1',
- 'srcip4': '90.90.1.1-90.105.255.255',
- 'dscp': 0, 'ttl': 32},
- 'outer_l3v6': {'dstip6': '1.1.1.1-1.15.255.255',
- 'proto': 'udp', 'count': '1',
- 'srcip6': '90.90.1.1-90.105.255.255',
- 'dscp': 0, 'ttl': 32},
- 'outer_l4': {'srcport': '2001',
- 'dsrport': '1234'}}},
- IXIARFC2544Profile.UPLINK: {'ipv4':
- {'outer_l2': {'framesize':
- {'64B': '100', '1518B': '0',
- '128B': '0', '1400B': '0',
- '256B': '0', '373b': '0',
- '570B': '0'}},
- 'outer_l3v4':
- {'dstip4': '9.9.1.1-90.105.255.255',
- 'proto': 'udp', 'count': '1',
- 'srcip4': '1.1.1.1-1.15.255.255',
- 'dscp': 0, 'ttl': 32},
- 'outer_l3v6':
- {'dstip6': '9.9.1.1-90.105.255.255',
- 'proto': 'udp', 'count': '1',
- 'srcip6': '1.1.1.1-1.15.255.255',
- 'dscp': 0, 'ttl': 32},
-
- 'outer_l4': {'dstport': '2001',
- 'srcport': '1234'}}},
- 'schema': 'isb:traffic_profile:0.1'}
- result = r_f_c2544_profile._get_ixia_traffic_profile(profile_data, mac)
- self.assertIsNotNone(result)
-
- def test__get_ixia_traffic_profile_default_args(self):
- r_f_c2544_profile = IXIARFC2544Profile(self.TRAFFIC_PROFILE)
-
- expected = {}
- result = r_f_c2544_profile._get_ixia_traffic_profile({})
- self.assertDictEqual(result, expected)
-
- def test__ixia_traffic_generate(self):
- traffic_generator = mock.Mock(autospec=TrexProfile)
- traffic_generator.networks = {
- "uplink_0": ["xe0"],
- "downlink_0": ["xe1"],
- }
- traffic_generator.client = \
- mock.Mock(return_value=True)
- traffic = {IXIARFC2544Profile.DOWNLINK: {'iload': 10},
- IXIARFC2544Profile.UPLINK: {'iload': 10}}
- ixia_obj = mock.MagicMock()
- r_f_c2544_profile = IXIARFC2544Profile(self.TRAFFIC_PROFILE)
- r_f_c2544_profile.rate = 100
- result = r_f_c2544_profile._ixia_traffic_generate(traffic, ixia_obj)
- self.assertIsNone(result)
-
- def test_execute(self):
- traffic_generator = mock.Mock(autospec=TrexProfile)
- traffic_generator.networks = {
- "uplink_0": ["xe0"],
- "downlink_0": ["xe1"],
- }
- traffic_generator.client = \
- mock.Mock(return_value=True)
- r_f_c2544_profile = IXIARFC2544Profile(self.TRAFFIC_PROFILE)
- r_f_c2544_profile.first_run = True
- r_f_c2544_profile.params = {IXIARFC2544Profile.DOWNLINK: {'iload': 10},
- IXIARFC2544Profile.UPLINK: {'iload': 10}}
-
- r_f_c2544_profile.get_streams = mock.Mock()
- r_f_c2544_profile.full_profile = {}
- r_f_c2544_profile._get_ixia_traffic_profile = mock.Mock()
- r_f_c2544_profile.get_multiplier = mock.Mock()
- r_f_c2544_profile._ixia_traffic_generate = mock.Mock()
- ixia_obj = mock.MagicMock()
- self.assertIsNone(r_f_c2544_profile.execute_traffic(traffic_generator, ixia_obj))
-
- def test_update_traffic_profile(self):
- traffic_generator = mock.Mock(autospec=TrexProfile)
- traffic_generator.networks = {
- "uplink_0": ["xe0"], # private, one value for intfs
- "downlink_0": ["xe1", "xe2"], # public, two values for intfs
- "downlink_1": ["xe3"], # not in TRAFFIC PROFILE
- "tenant_0": ["xe4"], # not public or private
- }
-
- ports_expected = [8, 3, 5]
- traffic_generator.vnfd_helper.port_num.side_effect = ports_expected
- traffic_generator.client.return_value = True
-
- traffic_profile = deepcopy(self.TRAFFIC_PROFILE)
- traffic_profile.update({
- "uplink_0": ["xe0"],
- "downlink_0": ["xe1", "xe2"],
- })
-
- r_f_c2544_profile = IXIARFC2544Profile(traffic_profile)
- r_f_c2544_profile.full_profile = {}
- r_f_c2544_profile.get_streams = mock.Mock()
-
- self.assertIsNone(r_f_c2544_profile.update_traffic_profile(traffic_generator))
- self.assertEqual(r_f_c2544_profile.ports, ports_expected)
-
- def test_get_drop_percentage(self):
- r_f_c2544_profile = IXIARFC2544Profile(self.TRAFFIC_PROFILE)
- r_f_c2544_profile.params = self.PROFILE
- ixia_obj = mock.MagicMock()
- r_f_c2544_profile.execute = mock.Mock()
- r_f_c2544_profile._get_ixia_traffic_profile = mock.Mock()
- r_f_c2544_profile._ixia_traffic_generate = mock.Mock()
- r_f_c2544_profile.get_multiplier = mock.Mock()
- r_f_c2544_profile.tmp_throughput = 0
- r_f_c2544_profile.tmp_drop = 0
- r_f_c2544_profile.full_profile = {}
- samples = {}
- for ifname in range(1):
- name = "xe{}".format(ifname)
- samples[name] = {"rx_throughput_fps": 20,
- "tx_throughput_fps": 20,
- "rx_throughput_mbps": 10,
- "tx_throughput_mbps": 10,
- "RxThroughput": 10,
- "TxThroughput": 10,
- "in_packets": 1000,
- "out_packets": 1000}
- tol_min = 100.0
- tolerance = 0.0
- self.assertIsNotNone(
- r_f_c2544_profile.get_drop_percentage(samples, tol_min, tolerance,
- ixia_obj))
-
- def test_get_drop_percentage_update(self):
- r_f_c2544_profile = IXIARFC2544Profile(self.TRAFFIC_PROFILE)
- r_f_c2544_profile.params = self.PROFILE
- ixia_obj = mock.MagicMock()
- r_f_c2544_profile.execute = mock.Mock()
- r_f_c2544_profile._get_ixia_traffic_profile = mock.Mock()
- r_f_c2544_profile._ixia_traffic_generate = mock.Mock()
- r_f_c2544_profile.get_multiplier = mock.Mock()
- r_f_c2544_profile.tmp_throughput = 0
- r_f_c2544_profile.tmp_drop = 0
- r_f_c2544_profile.full_profile = {}
- samples = {}
- for ifname in range(1):
- name = "xe{}".format(ifname)
- samples[name] = {"rx_throughput_fps": 20,
- "tx_throughput_fps": 20,
- "rx_throughput_mbps": 10,
- "tx_throughput_mbps": 10,
- "RxThroughput": 10,
- "TxThroughput": 10,
- "in_packets": 1000,
- "out_packets": 1002}
- tol_min = 0.0
- tolerance = 1.0
- self.assertIsNotNone(
- r_f_c2544_profile.get_drop_percentage(samples, tol_min, tolerance,
- ixia_obj))
-
- def test_get_drop_percentage_div_zero(self):
- r_f_c2544_profile = IXIARFC2544Profile(self.TRAFFIC_PROFILE)
- r_f_c2544_profile.params = self.PROFILE
- ixia_obj = mock.MagicMock()
- r_f_c2544_profile.execute = mock.Mock()
- r_f_c2544_profile._get_ixia_traffic_profile = mock.Mock()
- r_f_c2544_profile._ixia_traffic_generate = mock.Mock()
- r_f_c2544_profile.get_multiplier = mock.Mock()
- r_f_c2544_profile.tmp_throughput = 0
- r_f_c2544_profile.tmp_drop = 0
- r_f_c2544_profile.full_profile = {}
- samples = {}
- for ifname in range(1):
- name = "xe{}".format(ifname)
- samples[name] = {"rx_throughput_fps": 20,
- "tx_throughput_fps": 20,
- "rx_throughput_mbps": 10,
- "tx_throughput_mbps": 10,
- "RxThroughput": 10,
- "TxThroughput": 10,
- "in_packets": 1000,
- "out_packets": 0}
- tol_min = 0.0
- tolerance = 0.0
- r_f_c2544_profile.tmp_throughput = 0
- self.assertIsNotNone(
- r_f_c2544_profile.get_drop_percentage(samples, tol_min, tolerance,
- ixia_obj))
-
- def test_get_multiplier(self):
- r_f_c2544_profile = IXIARFC2544Profile(self.TRAFFIC_PROFILE)
- r_f_c2544_profile.max_rate = 100
- r_f_c2544_profile.min_rate = 100
- self.assertEqual("1.0", r_f_c2544_profile.get_multiplier())
-
- def test_start_ixia_latency(self):
- traffic_generator = mock.Mock(autospec=TrexProfile)
- traffic_generator.networks = {
- "uplink_0": ["xe0"],
- "downlink_0": ["xe1"],
- }
- traffic_generator.client = \
- mock.Mock(return_value=True)
- r_f_c2544_profile = IXIARFC2544Profile(self.TRAFFIC_PROFILE)
- r_f_c2544_profile.max_rate = 100
- r_f_c2544_profile.min_rate = 100
- ixia_obj = mock.MagicMock()
- r_f_c2544_profile._get_ixia_traffic_profile = \
- mock.Mock(return_value={})
- r_f_c2544_profile.full_profile = {}
- r_f_c2544_profile._ixia_traffic_generate = mock.Mock()
- self.assertIsNone(
- r_f_c2544_profile.start_ixia_latency(traffic_generator, ixia_obj))
diff --git a/tests/unit/network_services/traffic_profile/test_prox_acl.py b/tests/unit/network_services/traffic_profile/test_prox_acl.py
deleted file mode 100644
index ef5bac0d5..000000000
--- a/tests/unit/network_services/traffic_profile/test_prox_acl.py
+++ /dev/null
@@ -1,78 +0,0 @@
-# Copyright (c) 2017 Intel Corporation
-#
-# Licensed under the Apache License, Version 2.0 (the "License");
-# you may not use this file except in compliance with the License.
-# You may obtain a copy of the License at
-#
-# http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing, software
-# distributed under the License is distributed on an "AS IS" BASIS,
-# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-# See the License for the specific language governing permissions and
-# limitations under the License.
-#
-
-from __future__ import absolute_import
-
-import unittest
-import mock
-
-from tests.unit import STL_MOCKS
-
-STLClient = mock.MagicMock()
-stl_patch = mock.patch.dict("sys.modules", STL_MOCKS)
-stl_patch.start()
-
-if stl_patch:
- from yardstick.network_services.traffic_profile.prox_ACL import ProxACLProfile
- from yardstick.network_services.vnf_generic.vnf.prox_helpers import ProxTestDataTuple
-
-
-class TestProxACLProfile(unittest.TestCase):
-
- def test_run_test_with_pkt_size(self):
- def target(*args, **kwargs):
- runs.append(args[2])
- if args[2] < 0 or args[2] > 100:
- raise RuntimeError(' '.join([str(args), str(runs)]))
- if args[2] > 75.0:
- return fail_tuple, {}
- return success_tuple, {}
-
- def get_mock_samples(*args, **kwargs):
- if args[2] < 0:
- raise RuntimeError(' '.join([str(args), str(runs)]))
- return success_tuple
-
- tp_config = {
- 'traffic_profile': {
- 'upper_bound': 100.0,
- 'lower_bound': 0.0,
- 'tolerated_loss': 50.0,
- 'attempts': 20
- },
- }
-
- runs = []
- success_tuple = ProxTestDataTuple(10.0, 1, 2, 3, 4, [5.1, 5.2, 5.3], 995, 1000, 123.4)
- fail_tuple = ProxTestDataTuple(10.0, 1, 2, 3, 4, [5.6, 5.7, 5.8], 850, 1000, 123.4)
-
- traffic_gen = mock.MagicMock()
-
- profile_helper = mock.MagicMock()
- profile_helper.run_test = target
-
- profile = ProxACLProfile(tp_config)
- profile.init(mock.MagicMock())
-
- profile.prox_config["attempts"] = 20
- profile.queue = mock.MagicMock()
- profile.tolerated_loss = 50.0
- profile.pkt_size = 128
- profile.duration = 30
- profile.test_value = 100.0
- profile.tolerated_loss = 100.0
- profile._profile_helper = profile_helper
-
- profile.run_test_with_pkt_size(traffic_gen, profile.pkt_size, profile.duration)
diff --git a/tests/unit/network_services/traffic_profile/test_prox_binsearch.py b/tests/unit/network_services/traffic_profile/test_prox_binsearch.py
deleted file mode 100644
index 1b4189b48..000000000
--- a/tests/unit/network_services/traffic_profile/test_prox_binsearch.py
+++ /dev/null
@@ -1,185 +0,0 @@
-# Copyright (c) 2017 Intel Corporation
-#
-# Licensed under the Apache License, Version 2.0 (the "License");
-# you may not use this file except in compliance with the License.
-# You may obtain a copy of the License at
-#
-# http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing, software
-# distributed under the License is distributed on an "AS IS" BASIS,
-# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-# See the License for the specific language governing permissions and
-# limitations under the License.
-#
-
-from __future__ import absolute_import
-
-import unittest
-import mock
-
-from tests.unit import STL_MOCKS
-
-STLClient = mock.MagicMock()
-stl_patch = mock.patch.dict("sys.modules", STL_MOCKS)
-stl_patch.start()
-
-if stl_patch:
- from yardstick.network_services.vnf_generic.vnf.prox_helpers import ProxTestDataTuple
- from yardstick.network_services.traffic_profile.prox_binsearch import ProxBinSearchProfile
-
-
-class TestProxBinSearchProfile(unittest.TestCase):
-
- def test_execute_1(self):
- def target(*args, **_):
- runs.append(args[2])
- if args[2] < 0 or args[2] > 100:
- raise RuntimeError(' '.join([str(args), str(runs)]))
- if args[2] > 75.0:
- return fail_tuple, {}
- return success_tuple, {}
-
- tp_config = {
- 'traffic_profile': {
- 'packet_sizes': [200],
- 'test_precision': 2.0,
- 'tolerated_loss': 0.001,
- },
- }
-
- runs = []
- success_tuple = ProxTestDataTuple(10.0, 1, 2, 3, 4, [5.1, 5.2, 5.3], 995, 1000, 123.4)
- fail_tuple = ProxTestDataTuple(10.0, 1, 2, 3, 4, [5.6, 5.7, 5.8], 850, 1000, 123.4)
-
- traffic_generator = mock.MagicMock()
-
- profile_helper = mock.MagicMock()
- profile_helper.run_test = target
-
- profile = ProxBinSearchProfile(tp_config)
- profile.init(mock.MagicMock())
- profile._profile_helper = profile_helper
-
- profile.execute_traffic(traffic_generator)
- self.assertEqual(round(profile.current_lower, 2), 74.69)
- self.assertEqual(round(profile.current_upper, 2), 76.09)
- self.assertEqual(len(runs), 7)
-
- # Result Samples inc theor_max
- result_tuple = {"Result_Actual_throughput": 7.5e-07,
- "Result_theor_max_throughput": 0.00012340000000000002,
- "Result_pktSize": 200}
- profile.queue.put.assert_called_with(result_tuple)
-
- success_result_tuple = {"Success_CurrentDropPackets": 0.5,
- "Success_DropPackets": 0.5,
- "Success_LatencyAvg": 5.3,
- "Success_LatencyMax": 5.2,
- "Success_LatencyMin": 5.1,
- "Success_PktSize": 200,
- "Success_RxThroughput": 7.5e-07,
- "Success_Throughput": 7.5e-07,
- "Success_TxThroughput": 0.00012340000000000002}
-
- calls = profile.queue.put(success_result_tuple)
- profile.queue.put.assert_has_calls(calls)
-
- success_result_tuple2 = {"Success_CurrentDropPackets": 0.5,
- "Success_DropPackets": 0.5,
- "Success_LatencyAvg": 5.3,
- "Success_LatencyMax": 5.2,
- "Success_LatencyMin": 5.1,
- "Success_PktSize": 200,
- "Success_RxThroughput": 7.5e-07,
- "Success_Throughput": 7.5e-07,
- "Success_TxThroughput": 123.4,
- "Success_can_be_lost": 409600,
- "Success_drop_total": 20480,
- "Success_rx_total": 4075520,
- "Success_tx_total": 4096000}
-
- calls = profile.queue.put(success_result_tuple2)
- profile.queue.put.assert_has_calls(calls)
-
- def test_execute_2(self):
- def target(*args, **_):
- runs.append(args[2])
- if args[2] < 0 or args[2] > 100:
- raise RuntimeError(' '.join([str(args), str(runs)]))
- if args[2] > 25.0:
- return fail_tuple, {}
- return success_tuple, {}
-
- tp_config = {
- 'traffic_profile': {
- 'packet_sizes': [200],
- 'test_precision': 2.0,
- 'tolerated_loss': 0.001,
- },
- }
-
- runs = []
- success_tuple = ProxTestDataTuple(10.0, 1, 2, 3, 4, [5.1, 5.2, 5.3], 995, 1000, 123.4)
- fail_tuple = ProxTestDataTuple(10.0, 1, 2, 3, 4, [5.6, 5.7, 5.8], 850, 1000, 123.4)
-
- traffic_generator = mock.MagicMock()
-
- profile_helper = mock.MagicMock()
- profile_helper.run_test = target
-
- profile = ProxBinSearchProfile(tp_config)
- profile.init(mock.MagicMock())
- profile._profile_helper = profile_helper
-
- profile.execute_traffic(traffic_generator)
- self.assertEqual(round(profile.current_lower, 2), 24.06)
- self.assertEqual(round(profile.current_upper, 2), 25.47)
- self.assertEqual(len(runs), 7)
-
- def test_execute_3(self):
- def target(*args, **_):
- runs.append(args[2])
- if args[2] < 0 or args[2] > 100:
- raise RuntimeError(' '.join([str(args), str(runs)]))
- if args[2] > 75.0:
- return fail_tuple, {}
- return success_tuple, {}
-
- tp_config = {
- 'traffic_profile': {
- 'packet_sizes': [200],
- 'test_precision': 2.0,
- 'tolerated_loss': 0.001,
- },
- }
-
- runs = []
- success_tuple = ProxTestDataTuple(10.0, 1, 2, 3, 4, [5.1, 5.2, 5.3], 995, 1000, 123.4)
- fail_tuple = ProxTestDataTuple(10.0, 1, 2, 3, 4, [5.6, 5.7, 5.8], 850, 1000, 123.4)
-
- traffic_generator = mock.MagicMock()
-
- profile_helper = mock.MagicMock()
- profile_helper.run_test = target
-
- profile = ProxBinSearchProfile(tp_config)
- profile.init(mock.MagicMock())
- profile._profile_helper = profile_helper
-
- profile.upper_bound = 100.0
- profile.lower_bound = 99.0
- profile.execute_traffic(traffic_generator)
-
-
- # Result Samples
- result_tuple = {"Result_theor_max_throughput": 0, "Result_pktSize": 200}
- profile.queue.put.assert_called_with(result_tuple)
-
- # Check for success_ tuple (None expected)
- calls = profile.queue.put.mock_calls
- for call in calls:
- for call_detail in call[1]:
- for k in call_detail:
- if "Success_" in k:
- self.assertRaises(AttributeError)
diff --git a/tests/unit/network_services/traffic_profile/test_prox_profile.py b/tests/unit/network_services/traffic_profile/test_prox_profile.py
deleted file mode 100644
index e5b36096f..000000000
--- a/tests/unit/network_services/traffic_profile/test_prox_profile.py
+++ /dev/null
@@ -1,130 +0,0 @@
-# Copyright (c) 2017 Intel Corporation
-#
-# Licensed under the Apache License, Version 2.0 (the "License");
-# you may not use this file except in compliance with the License.
-# You may obtain a copy of the License at
-#
-# http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing, software
-# distributed under the License is distributed on an "AS IS" BASIS,
-# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-# See the License for the specific language governing permissions and
-# limitations under the License.
-#
-
-from __future__ import absolute_import
-
-import unittest
-import mock
-
-from tests.unit import STL_MOCKS
-
-STLClient = mock.MagicMock()
-stl_patch = mock.patch.dict("sys.modules", STL_MOCKS)
-stl_patch.start()
-
-if stl_patch:
- from yardstick.network_services.traffic_profile.prox_profile import ProxProfile
- from yardstick.network_services.vnf_generic.vnf.prox_helpers import ProxResourceHelper
-
-
-class TestProxProfile(unittest.TestCase):
-
- def test_sort_vpci(self):
- traffic_generator = mock.Mock()
- interface_1 = {'virtual-interface': {'vpci': 'id1'}, 'name': 'name1'}
- interface_2 = {'virtual-interface': {'vpci': 'id2'}, 'name': 'name2'}
- interface_3 = {'virtual-interface': {'vpci': 'id3'}, 'name': 'name3'}
- interfaces = [interface_2, interface_3, interface_1]
- traffic_generator.vnfd_helper = {
- 'vdu': [{'external-interface': interfaces}]}
- output = ProxProfile.sort_vpci(traffic_generator)
- self.assertEqual([interface_1, interface_2, interface_3], output)
-
- def test_fill_samples(self):
- samples = {}
-
- traffic_generator = mock.MagicMock()
- interfaces = [
- ['id1', 'name1'],
- ['id2', 'name2']
- ]
- traffic_generator.resource_helper.sut.port_stats.side_effect = [
- list(range(12)),
- list(range(10, 22)),
- ]
-
- expected = {
- 'name1': {
- 'in_packets': 6,
- 'out_packets': 7,
- },
- 'name2': {
- 'in_packets': 16,
- 'out_packets': 17,
- },
- }
- with mock.patch.object(ProxProfile, 'sort_vpci', return_value=interfaces):
- ProxProfile.fill_samples(samples, traffic_generator)
-
- self.assertDictEqual(samples, expected)
-
- def test_init(self):
- tp_config = {
- 'traffic_profile': {},
- }
-
- profile = ProxProfile(tp_config)
- queue = mock.Mock()
- profile.init(queue)
- self.assertIs(profile.queue, queue)
-
- def test_execute_traffic(self):
- packet_sizes = [
- 10,
- 100,
- 1000,
- ]
- tp_config = {
- 'traffic_profile': {
- 'packet_sizes': packet_sizes,
- },
- }
-
- traffic_generator = mock.MagicMock()
-
- setup_helper = traffic_generator.setup_helper
- setup_helper.find_in_section.return_value = None
-
- prox_resource_helper = ProxResourceHelper(setup_helper)
- traffic_generator.resource_helper = prox_resource_helper
-
- profile = ProxProfile(tp_config)
-
- self.assertFalse(profile.done)
- for _ in packet_sizes:
- with self.assertRaises(NotImplementedError):
- profile.execute_traffic(traffic_generator)
-
- self.assertIsNone(profile.execute_traffic(traffic_generator))
- self.assertTrue(profile.done)
-
- def test_bounds_iterator(self):
- tp_config = {
- 'traffic_profile': {},
- }
-
- profile = ProxProfile(tp_config)
- value = 0.0
- for value in profile.bounds_iterator():
- pass
-
- self.assertEqual(value, 100.0)
-
- mock_logger = mock.MagicMock()
- for _ in profile.bounds_iterator(mock_logger):
- pass
-
- self.assertEqual(mock_logger.debug.call_count, 1)
- self.assertEqual(mock_logger.info.call_count, 10)
diff --git a/tests/unit/network_services/traffic_profile/test_prox_ramp.py b/tests/unit/network_services/traffic_profile/test_prox_ramp.py
deleted file mode 100644
index 1acec2f68..000000000
--- a/tests/unit/network_services/traffic_profile/test_prox_ramp.py
+++ /dev/null
@@ -1,97 +0,0 @@
-# Copyright (c) 2017 Intel Corporation
-#
-# Licensed under the Apache License, Version 2.0 (the "License");
-# you may not use this file except in compliance with the License.
-# You may obtain a copy of the License at
-#
-# http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing, software
-# distributed under the License is distributed on an "AS IS" BASIS,
-# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-# See the License for the specific language governing permissions and
-# limitations under the License.
-#
-
-from __future__ import absolute_import
-
-import unittest
-import mock
-
-from tests.unit import STL_MOCKS
-
-STLClient = mock.MagicMock()
-stl_patch = mock.patch.dict("sys.modules", STL_MOCKS)
-stl_patch.start()
-
-if stl_patch:
- from yardstick.network_services.traffic_profile.prox_ramp import ProxRampProfile
- from yardstick.network_services.vnf_generic.vnf.prox_helpers import ProxProfileHelper
- from yardstick.network_services.vnf_generic.vnf.prox_helpers import ProxTestDataTuple
-
-
-class TestProxRampProfile(unittest.TestCase):
-
- def test_run_test_with_pkt_size(self):
- tp_config = {
- 'traffic_profile': {
- 'lower_bound': 10.0,
- 'upper_bound': 100.0,
- 'step_value': 10.0,
- },
- }
-
- success_tuple = ProxTestDataTuple(10.0, 1, 2, 3, 4, [5.1, 5.2, 5.3], 995, 1000, 123.4)
-
- traffic_gen = mock.MagicMock()
- traffic_gen._test_type = 'Generic'
-
- profile_helper = ProxProfileHelper(traffic_gen.resource_helper)
- profile_helper.run_test = run_test = mock.MagicMock(return_value=success_tuple)
-
- profile = ProxRampProfile(tp_config)
- profile.fill_samples = fill_samples = mock.MagicMock()
- profile.queue = mock.MagicMock()
- profile._profile_helper = profile_helper
-
- profile.run_test_with_pkt_size(traffic_gen, 128, 30)
- self.assertEqual(run_test.call_count, 10)
- self.assertEqual(fill_samples.call_count, 10)
-
- def test_run_test_with_pkt_size_with_fail(self):
- tp_config = {
- 'traffic_profile': {
- 'lower_bound': 10.0,
- 'upper_bound': 100.0,
- 'step_value': 10.0,
- },
- }
-
- success_tuple = ProxTestDataTuple(10.0, 1, 2, 3, 4, [5.1, 5.2, 5.3], 995, 1000, 123.4)
- fail_tuple = ProxTestDataTuple(10.0, 1, 2, 3, 4, [5.6, 5.7, 5.8], 850, 1000, 123.4)
-
- result_list = [
- success_tuple,
- success_tuple,
- success_tuple,
- fail_tuple,
- success_tuple,
- fail_tuple,
- fail_tuple,
- fail_tuple,
- ]
-
- traffic_gen = mock.MagicMock()
- traffic_gen._test_type = 'Generic'
-
- profile_helper = ProxProfileHelper(traffic_gen.resource_helper)
- profile_helper.run_test = run_test = mock.MagicMock(side_effect=result_list)
-
- profile = ProxRampProfile(tp_config)
- profile.fill_samples = fill_samples = mock.MagicMock()
- profile.queue = mock.MagicMock()
- profile._profile_helper = profile_helper
-
- profile.run_test_with_pkt_size(traffic_gen, 128, 30)
- self.assertEqual(run_test.call_count, 4)
- self.assertEqual(fill_samples.call_count, 3)
diff --git a/tests/unit/network_services/traffic_profile/test_rfc2544.py b/tests/unit/network_services/traffic_profile/test_rfc2544.py
deleted file mode 100644
index cb3a547ee..000000000
--- a/tests/unit/network_services/traffic_profile/test_rfc2544.py
+++ /dev/null
@@ -1,276 +0,0 @@
-# Copyright (c) 2016-2017 Intel Corporation
-#
-# Licensed under the Apache License, Version 2.0 (the "License");
-# you may not use this file except in compliance with the License.
-# You may obtain a copy of the License at
-#
-# http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing, software
-# distributed under the License is distributed on an "AS IS" BASIS,
-# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-# See the License for the specific language governing permissions and
-# limitations under the License.
-#
-
-import unittest
-import mock
-
-from tests.unit import STL_MOCKS
-
-
-STLClient = mock.MagicMock()
-stl_patch = mock.patch.dict("sys.modules", STL_MOCKS)
-stl_patch.start()
-
-if stl_patch:
- from yardstick.network_services.traffic_profile.trex_traffic_profile \
- import TrexProfile
- from yardstick.network_services.traffic_profile.rfc2544 import \
- RFC2544Profile
-
-
-class TestRFC2544Profile(unittest.TestCase):
- TRAFFIC_PROFILE = {
- "schema": "isb:traffic_profile:0.1",
- "name": "fixed",
- "description": "Fixed traffic profile to run UDP traffic",
- "traffic_profile": {
- "traffic_type": "FixedTraffic",
- "frame_rate": 100, # pps
- "flow_number": 10,
- "frame_size": 64}}
-
- PROFILE = {'description': 'Traffic profile to run RFC2544 latency',
- 'name': 'rfc2544',
- 'traffic_profile': {'traffic_type': 'RFC2544Profile',
- 'frame_rate': 100},
- 'downlink_0': {'ipv4':
- {'outer_l2': {'framesize':
- {'64B': '100', '1518B': '0',
- '128B': '0', '1400B': '0',
- '256B': '0', '373b': '0',
- '570B': '0'}},
- 'outer_l3v4': {'dstip4': '1.1.1.1-1.15.255.255',
- 'proto': 'udp',
- 'srcip4': '90.90.1.1-90.105.255.255',
- 'dscp': 0, 'ttl': 32, 'count': 1},
- 'outer_l4': {'srcport': '2001',
- 'dsrport': '1234', 'count': 1}}},
- 'uplink_0': {'ipv4':
- {'outer_l2': {'framesize':
- {'64B': '100', '1518B': '0',
- '128B': '0', '1400B': '0',
- '256B': '0', '373b': '0',
- '570B': '0'}},
- 'outer_l3v4': {'dstip4': '9.9.1.1-90.105.255.255',
- 'proto': 'udp',
- 'srcip4': '1.1.1.1-1.15.255.255',
- 'dscp': 0, 'ttl': 32, 'count': 1},
- 'outer_l4': {'dstport': '2001',
- 'srcport': '1234', 'count': 1}}},
- 'schema': 'isb:traffic_profile:0.1'}
-
- def test___init__(self):
- r_f_c2544_profile = RFC2544Profile(self.TRAFFIC_PROFILE)
- self.assertIsNotNone(r_f_c2544_profile.rate)
-
- def test_execute(self):
- traffic_generator = mock.Mock(autospec=TrexProfile)
- traffic_generator.networks = {
- "uplink_0": ["xe0"],
- "downlink_0": ["xe1"],
- }
- traffic_generator.client.return_value = True
- r_f_c2544_profile = RFC2544Profile(self.TRAFFIC_PROFILE)
- r_f_c2544_profile.params = self.PROFILE
- r_f_c2544_profile.first_run = True
- self.assertIsNone(r_f_c2544_profile.execute_traffic(traffic_generator))
-
- def test_get_drop_percentage(self):
- traffic_generator = mock.Mock(autospec=TrexProfile)
- traffic_generator.networks = {
- "uplink_0": ["xe0"],
- "downlink_0": ["xe1"],
- }
- traffic_generator.client.return_value = True
-
- r_f_c2544_profile = RFC2544Profile(self.TRAFFIC_PROFILE)
- r_f_c2544_profile.params = self.PROFILE
- r_f_c2544_profile.register_generator(traffic_generator)
- self.assertIsNone(r_f_c2544_profile.execute_traffic(traffic_generator))
-
- samples = {}
- for ifname in range(1):
- name = "xe{}".format(ifname)
- samples[name] = {
- "rx_throughput_fps": 20,
- "tx_throughput_fps": 20,
- "rx_throughput_mbps": 10,
- "tx_throughput_mbps": 10,
- "in_packets": 1000,
- "out_packets": 1000,
- }
-
- expected = {
- 'DropPercentage': 0.0,
- 'RxThroughput': 100 / 3.0,
- 'TxThroughput': 100 / 3.0,
- 'CurrentDropPercentage': 0.0,
- 'Throughput': 66.66666666666667,
- 'xe0': {
- 'tx_throughput_fps': 20,
- 'in_packets': 1000,
- 'out_packets': 1000,
- 'rx_throughput_mbps': 10,
- 'tx_throughput_mbps': 10,
- 'rx_throughput_fps': 20,
- },
- }
- traffic_generator.generate_samples.return_value = samples
- traffic_generator.RUN_DURATION = 30
- traffic_generator.rfc2544_helper.tolerance_low = 0.0001
- traffic_generator.rfc2544_helper.tolerance_high = 0.0001
- result = r_f_c2544_profile.get_drop_percentage(traffic_generator)
- self.assertDictEqual(result, expected)
-
- def test_get_drop_percentage_update(self):
- traffic_generator = mock.Mock(autospec=RFC2544Profile)
- traffic_generator.networks = {
- "uplink_0": ["xe0"],
- "downlink_0": ["xe1"],
- }
- traffic_generator.client = mock.Mock(return_value=True)
-
- r_f_c2544_profile = RFC2544Profile(self.TRAFFIC_PROFILE)
- r_f_c2544_profile.params = self.PROFILE
- r_f_c2544_profile.register_generator(traffic_generator)
- self.assertIsNone(r_f_c2544_profile.execute_traffic())
-
- samples = {}
- for ifname in range(1):
- name = "xe{}".format(ifname)
- samples[name] = {
- "rx_throughput_fps": 20,
- "tx_throughput_fps": 20,
- "rx_throughput_mbps": 10,
- "tx_throughput_mbps": 10,
- "in_packets": 1000,
- "out_packets": 1002,
- }
- expected = {
- 'DropPercentage': 0.1996,
- 'RxThroughput': 33.333333333333336,
- 'TxThroughput': 33.4,
- 'CurrentDropPercentage': 0.1996,
- 'Throughput': 66.66666666666667,
- 'xe0': {
- 'tx_throughput_fps': 20,
- 'in_packets': 1000,
- 'out_packets': 1002,
- 'rx_throughput_mbps': 10,
- 'tx_throughput_mbps': 10,
- 'rx_throughput_fps': 20,
- },
- }
- traffic_generator.generate_samples = mock.MagicMock(
- return_value=samples)
- traffic_generator.RUN_DURATION = 30
- traffic_generator.rfc2544_helper.tolerance_low = 0.0001
- traffic_generator.rfc2544_helper.tolerance_high = 0.0001
- result = r_f_c2544_profile.get_drop_percentage(traffic_generator)
- self.assertDictEqual(expected, result)
-
- def test_get_drop_percentage_div_zero(self):
- traffic_generator = mock.Mock(autospec=TrexProfile)
- traffic_generator.networks = {
- "uplink_0": ["xe0"],
- "downlink_0": ["xe1"],
- }
- traffic_generator.client = \
- mock.Mock(return_value=True)
- r_f_c2544_profile = RFC2544Profile(self.TRAFFIC_PROFILE)
- r_f_c2544_profile.params = self.PROFILE
- self.assertIsNone(r_f_c2544_profile.execute_traffic(traffic_generator))
- samples = {}
- for ifname in range(1):
- name = "xe{}".format(ifname)
- samples[name] = {"rx_throughput_fps": 20,
- "tx_throughput_fps": 20,
- "rx_throughput_mbps": 10,
- "tx_throughput_mbps": 10,
- "in_packets": 1000,
- "out_packets": 0}
- r_f_c2544_profile.throughput_max = 0
- expected = {
- 'DropPercentage': 100.0, 'RxThroughput': 100 / 3.0,
- 'TxThroughput': 0.0, 'CurrentDropPercentage': 100.0,
- 'Throughput': 66.66666666666667,
- 'xe0': {
- 'tx_throughput_fps': 20, 'in_packets': 1000,
- 'out_packets': 0, 'rx_throughput_mbps': 10,
- 'tx_throughput_mbps': 10, 'rx_throughput_fps': 20
- }
- }
- traffic_generator.generate_samples = mock.Mock(return_value=samples)
- traffic_generator.RUN_DURATION = 30
- traffic_generator.rfc2544_helper.tolerance_low = 0.0001
- traffic_generator.rfc2544_helper.tolerance_high = 0.0001
- self.assertDictEqual(expected,
- r_f_c2544_profile.get_drop_percentage(traffic_generator))
-
- def test_get_multiplier(self):
- r_f_c2544_profile = RFC2544Profile(self.TRAFFIC_PROFILE)
- r_f_c2544_profile.max_rate = 100
- r_f_c2544_profile.min_rate = 100
- self.assertEqual("1.0", r_f_c2544_profile.get_multiplier())
-
- def test_calculate_pps(self):
- r_f_c2544_profile = RFC2544Profile(self.TRAFFIC_PROFILE)
- r_f_c2544_profile.rate = 100
- r_f_c2544_profile.pps = 100
- samples = {'Throughput': 4549093.33}
- self.assertEqual((2274546.67, 1.0),
- r_f_c2544_profile.calculate_pps(samples))
-
- def test_create_single_stream(self):
- r_f_c2544_profile = RFC2544Profile(self.TRAFFIC_PROFILE)
- r_f_c2544_profile._create_single_packet = mock.MagicMock()
- r_f_c2544_profile.pg_id = 1
- self.assertIsNotNone(
- r_f_c2544_profile.create_single_stream(64, 2274546.67))
-
- def test_create_single_stream_no_pg_id(self):
- r_f_c2544_profile = RFC2544Profile(self.TRAFFIC_PROFILE)
- r_f_c2544_profile._create_single_packet = mock.MagicMock()
- r_f_c2544_profile.pg_id = 0
- self.assertIsNotNone(
- r_f_c2544_profile.create_single_stream(64, 2274546.67))
-
- def test_execute_latency(self):
- traffic_generator = mock.Mock(autospec=TrexProfile)
- traffic_generator.networks = {
- "private_0": ["xe0"],
- "public_0": ["xe1"],
- }
- traffic_generator.client = \
- mock.Mock(return_value=True)
- r_f_c2544_profile = RFC2544Profile(self.TRAFFIC_PROFILE)
- r_f_c2544_profile.params = self.PROFILE
- r_f_c2544_profile.first_run = True
- samples = {}
- for ifname in range(1):
- name = "xe{}".format(ifname)
- samples[name] = {"rx_throughput_fps": 20,
- "tx_throughput_fps": 20,
- "rx_throughput_mbps": 10,
- "tx_throughput_mbps": 10,
- "in_packets": 1000,
- "out_packets": 0}
-
- samples['Throughput'] = 4549093.33
- r_f_c2544_profile.calculate_pps = mock.Mock(return_value=[2274546.67,
- 1.0])
-
- self.assertIsNone(r_f_c2544_profile.execute_latency(traffic_generator,
- samples))
diff --git a/tests/unit/network_services/traffic_profile/test_trex_traffic_profile.py b/tests/unit/network_services/traffic_profile/test_trex_traffic_profile.py
deleted file mode 100644
index d1009a5e8..000000000
--- a/tests/unit/network_services/traffic_profile/test_trex_traffic_profile.py
+++ /dev/null
@@ -1,309 +0,0 @@
-# Copyright (c) 2016-2017 Intel Corporation
-#
-# Licensed under the Apache License, Version 2.0 (the "License");
-# you may not use this file except in compliance with the License.
-# You may obtain a copy of the License at
-#
-# http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing, software
-# distributed under the License is distributed on an "AS IS" BASIS,
-# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-# See the License for the specific language governing permissions and
-# limitations under the License.
-
-import ipaddress
-
-import mock
-import six
-import unittest
-
-from tests.unit import STL_MOCKS
-from yardstick.common import exceptions as y_exc
-
-STLClient = mock.MagicMock()
-stl_patch = mock.patch.dict("sys.modules", STL_MOCKS)
-stl_patch.start()
-
-if stl_patch:
- from yardstick.network_services.traffic_profile.base import TrafficProfile
- from yardstick.network_services.traffic_profile.trex_traffic_profile import TrexProfile
- from yardstick.network_services.traffic_profile.trex_traffic_profile import SRC
- from yardstick.network_services.traffic_profile.trex_traffic_profile import DST
- from yardstick.network_services.traffic_profile.trex_traffic_profile import ETHERNET
- from yardstick.network_services.traffic_profile.trex_traffic_profile import IP
- from yardstick.network_services.traffic_profile.trex_traffic_profile import IPv6
- from yardstick.network_services.traffic_profile.trex_traffic_profile import UDP
- from yardstick.network_services.traffic_profile.trex_traffic_profile import SRC_PORT
- from yardstick.network_services.traffic_profile.trex_traffic_profile import DST_PORT
- from yardstick.network_services.traffic_profile.trex_traffic_profile import TYPE_OF_SERVICE
-
-
-class TestTrexProfile(unittest.TestCase):
- TRAFFIC_PROFILE = {
- "schema": "isb:traffic_profile:0.1",
- "name": "fixed",
- "description": "Fixed traffic profile to run UDP traffic",
- "traffic_profile": {
- "traffic_type": "FixedTraffic",
- "frame_rate": 100, # pps
- "flow_number": 10,
- "frame_size": 64}}
-
- EXAMPLE_ETHERNET_ADDR = "00:00:00:00:00:01"
- EXAMPLE_IP_ADDR = "10.0.0.1"
- EXAMPLE_IPv6_ADDR = "0064:ff9b:0:0:0:0:9810:6414"
-
- PROFILE = {
- 'description': 'Traffic profile to run RFC2544 latency',
- 'name': 'rfc2544',
- 'traffic_profile': {'traffic_type': 'RFC2544Profile',
- 'frame_rate': 100},
- TrafficProfile.DOWNLINK: {
- 'ipv4': {'outer_l2': {'framesize': {'64B': '100',
- '1518B': '0',
- '128B': '0',
- '1400B': '0',
- '256B': '0',
- '373b': '0',
- '570B': '0'},
- "srcmac": "00:00:00:00:00:02",
- "dstmac": "00:00:00:00:00:01"},
- 'outer_l3v4': {'dstip4': '1.1.1.1-1.1.2.2',
- 'proto': 'udp',
- 'srcip4': '9.9.1.1-90.1.2.2',
- 'dscp': 0, 'ttl': 32,
- 'count': 1},
- 'outer_l4': {'srcport': '2001',
- 'dsrport': '1234',
- 'count': 1}}},
- TrafficProfile.UPLINK: {
- 'ipv4':
- {'outer_l2': {'framesize':
- {'64B': '100', '1518B': '0',
- '128B': '0', '1400B': '0',
- '256B': '0', '373b': '0',
- '570B': '0'},
- "srcmac": "00:00:00:00:00:01",
- "dstmac": "00:00:00:00:00:02"},
- 'outer_l3v4': {'dstip4': '9.9.1.1-90.105.255.255',
- 'proto': 'udp',
- 'srcip4': '1.1.1.1-1.15.255.255',
- 'dscp': 0, 'ttl': 32, 'count': 1},
- 'outer_l4': {'dstport': '2001',
- 'srcport': '1234',
- 'count': 1}}},
- 'schema': 'isb:traffic_profile:0.1'}
- PROFILE_v6 = {
- 'description': 'Traffic profile to run RFC2544 latency',
- 'name': 'rfc2544',
- 'traffic_profile': {'traffic_type': 'RFC2544Profile',
- 'frame_rate': 100},
- TrafficProfile.DOWNLINK: {
- 'ipv6': {'outer_l2': {'framesize':
- {'64B': '100', '1518B': '0',
- '128B': '0', '1400B': '0',
- '256B': '0', '373b': '0',
- '570B': '0'},
- "srcmac": "00:00:00:00:00:02",
- "dstmac": "00:00:00:00:00:01"},
- 'outer_l3v4': {
- 'dstip6':
- '0064:ff9b:0:0:0:0:9810:6414-0064:ff9b:0:0:0:0:9810:6420',
- 'proto': 'udp',
- 'srcip6':
- '0064:ff9b:0:0:0:0:9810:2814-0064:ff9b:0:0:0:0:9810:2820',
- 'dscp': 0, 'ttl': 32,
- 'count': 1},
- 'outer_l4': {'srcport': '2001',
- 'dsrport': '1234',
- 'count': 1}}},
- TrafficProfile.UPLINK: {
- 'ipv6': {'outer_l2': {'framesize':
- {'64B': '100', '1518B': '0',
- '128B': '0', '1400B': '0',
- '256B': '0', '373b': '0',
- '570B': '0'},
- "srcmac": "00:00:00:00:00:01",
- "dstmac": "00:00:00:00:00:02"},
- 'outer_l3v4': {
- 'dstip6':
- '0064:ff9b:0:0:0:0:9810:2814-0064:ff9b:0:0:0:0:9810:2820',
- 'proto': 'udp',
- 'srcip6':
- '0064:ff9b:0:0:0:0:9810:6414-0064:ff9b:0:0:0:0:9810:6420',
- 'dscp': 0, 'ttl': 32,
- 'count': 1},
- 'outer_l4': {'dstport': '2001',
- 'srcport': '1234',
- 'count': 1}}},
- 'schema': 'isb:traffic_profile:0.1'}
-
- def test___init__(self):
- TrafficProfile.params = self.PROFILE
- trex_profile = \
- TrexProfile(TrafficProfile)
- self.assertEqual(trex_profile.pps, 100)
-
- def test_qinq(self):
- qinq = {"S-VLAN": {"id": 128, "priority": 0, "cfi": 0},
- "C-VLAN": {"id": 512, "priority": 0, "cfi": 0}}
-
- trex_profile = \
- TrexProfile(TrafficProfile)
- self.assertIsNone(trex_profile.set_qinq(qinq))
-
- qinq = {"S-VLAN": {"id": "128-130", "priority": 0, "cfi": 0},
- "C-VLAN": {"id": "512-515", "priority": 0, "cfi": 0}}
- self.assertIsNone(trex_profile.set_qinq(qinq))
-
- def test__set_outer_l2_fields(self):
- trex_profile = \
- TrexProfile(TrafficProfile)
- qinq = {"S-VLAN": {"id": 128, "priority": 0, "cfi": 0},
- "C-VLAN": {"id": 512, "priority": 0, "cfi": 0}}
- outer_l2 = self.PROFILE[TrafficProfile.UPLINK]['ipv4']['outer_l2']
- outer_l2['QinQ'] = qinq
- self.assertIsNone(trex_profile._set_outer_l2_fields(outer_l2))
-
- def test__set_outer_l3v4_fields(self):
- trex_profile = \
- TrexProfile(TrafficProfile)
- outer_l3v4 = self.PROFILE[TrafficProfile.UPLINK]['ipv4']['outer_l3v4']
- outer_l3v4['proto'] = 'tcp'
- self.assertIsNone(trex_profile._set_outer_l3v4_fields(outer_l3v4))
-
- def test__set_outer_l3v6_fields(self):
- trex_profile = \
- TrexProfile(TrafficProfile)
- outer_l3v6 = self.PROFILE_v6[TrafficProfile.UPLINK]['ipv6']['outer_l3v4']
- outer_l3v6['proto'] = 'tcp'
- outer_l3v6['tc'] = 1
- outer_l3v6['hlim'] = 10
- self.assertIsNone(trex_profile._set_outer_l3v6_fields(outer_l3v6))
-
- def test__set_outer_l4_fields(self):
- trex_profile = \
- TrexProfile(TrafficProfile)
- outer_l4 = self.PROFILE[TrafficProfile.UPLINK]['ipv4']['outer_l4']
- self.assertIsNone(trex_profile._set_outer_l4_fields(outer_l4))
-
- def test_get_streams(self):
- trex_profile = \
- TrexProfile(TrafficProfile)
- trex_profile.params = self.PROFILE
- profile_data = self.PROFILE[TrafficProfile.UPLINK]
- self.assertIsNotNone(trex_profile.get_streams(profile_data))
- trex_profile.pg_id = 1
- self.assertIsNotNone(trex_profile.get_streams(profile_data))
- trex_profile.params = self.PROFILE_v6
- trex_profile.profile_data = self.PROFILE_v6[TrafficProfile.UPLINK]
- self.assertIsNotNone(trex_profile.get_streams(profile_data))
- trex_profile.pg_id = 1
- self.assertIsNotNone(trex_profile.get_streams(profile_data))
-
- def test_generate_packets(self):
- trex_profile = \
- TrexProfile(TrafficProfile)
- trex_profile.fsize = 10
- trex_profile.base_pkt = [10]
- self.assertIsNone(trex_profile.generate_packets())
-
- def test_generate_imix_data_error(self):
- trex_profile = \
- TrexProfile(TrafficProfile)
- self.assertEqual({}, trex_profile.generate_imix_data(False))
-
- def test__count_ip_ipv4(self):
- start, end, count = TrexProfile._count_ip('1.1.1.1', '1.2.3.4')
- self.assertEqual('1.1.1.1', str(start))
- self.assertEqual('1.2.3.4', str(end))
- diff = (int(ipaddress.IPv4Address(six.u('1.2.3.4'))) -
- int(ipaddress.IPv4Address(six.u('1.1.1.1'))))
- self.assertEqual(diff, count)
-
- def test__count_ip_ipv6(self):
- start_ip = '0064:ff9b:0:0:0:0:9810:6414'
- end_ip = '0064:ff9b:0:0:0:0:9810:6420'
- start, end, count = TrexProfile._count_ip(start_ip, end_ip)
- self.assertEqual(0x98106414, start)
- self.assertEqual(0x98106420, end)
- self.assertEqual(0x98106420 - 0x98106414, count)
-
- def test__count_ip_ipv6_exception(self):
- start_ip = '0064:ff9b:0:0:0:0:9810:6420'
- end_ip = '0064:ff9b:0:0:0:0:9810:6414'
- with self.assertRaises(y_exc.IPv6RangeError):
- TrexProfile._count_ip(start_ip, end_ip)
-
- def test__dscp_range_action_partial_actual_count_zero(self):
- traffic_profile = TrexProfile(TrafficProfile)
- dscp_partial = traffic_profile._dscp_range_action_partial()
-
- flow_vars_initial_length = len(traffic_profile.vm_flow_vars)
- dscp_partial('1', '1', 'unneeded')
- self.assertEqual(len(traffic_profile.vm_flow_vars), flow_vars_initial_length + 2)
-
- def test__dscp_range_action_partial_count_greater_than_actual(self):
- traffic_profile = TrexProfile(TrafficProfile)
- dscp_partial = traffic_profile._dscp_range_action_partial()
-
- flow_vars_initial_length = len(traffic_profile.vm_flow_vars)
- dscp_partial('1', '10', '100')
- self.assertEqual(len(traffic_profile.vm_flow_vars), flow_vars_initial_length + 2)
-
- def test__udp_range_action_partial_actual_count_zero(self):
- traffic_profile = TrexProfile(TrafficProfile)
- traffic_profile.udp['field1'] = 'value1'
- udp_partial = traffic_profile._udp_range_action_partial('field1')
-
- flow_vars_initial_length = len(traffic_profile.vm_flow_vars)
- udp_partial('1', '1', 'unneeded')
- self.assertEqual(len(traffic_profile.vm_flow_vars), flow_vars_initial_length + 2)
-
- def test__udp_range_action_partial_count_greater_than_actual(self):
- traffic_profile = TrexProfile(TrafficProfile)
- traffic_profile.udp['field1'] = 'value1'
- udp_partial = traffic_profile._udp_range_action_partial('field1', 'not_used_count')
-
- flow_vars_initial_length = len(traffic_profile.vm_flow_vars)
- udp_partial('1', '10', '100')
- self.assertEqual(len(traffic_profile.vm_flow_vars), flow_vars_initial_length + 2)
-
- def test__general_single_action_partial(self):
- trex_profile = TrexProfile(TrafficProfile)
-
- trex_profile._general_single_action_partial(ETHERNET)(SRC)(
- self.EXAMPLE_ETHERNET_ADDR)
- self.assertEqual(self.EXAMPLE_ETHERNET_ADDR,
- trex_profile.ether_packet.src)
-
- trex_profile._general_single_action_partial(IP)(DST)(
- self.EXAMPLE_IP_ADDR)
- self.assertEqual(self.EXAMPLE_IP_ADDR, trex_profile.ip_packet.dst)
-
- trex_profile._general_single_action_partial(IPv6)(DST)(
- self.EXAMPLE_IPv6_ADDR)
- self.assertEqual(self.EXAMPLE_IPv6_ADDR, trex_profile.ip6_packet.dst)
-
- trex_profile._general_single_action_partial(UDP)(SRC_PORT)(5060)
- self.assertEqual(5060, trex_profile.udp_packet.sport)
-
- trex_profile._general_single_action_partial(IP)(TYPE_OF_SERVICE)(0)
- self.assertEqual(0, trex_profile.ip_packet.tos)
-
- def test__set_proto_addr(self):
- trex_profile = TrexProfile(TrafficProfile)
-
- ether_range = "00:00:00:00:00:01-00:00:00:00:00:02"
- ip_range = "1.1.1.2-1.1.1.10"
- ipv6_range = '0064:ff9b:0:0:0:0:9810:6414-0064:ff9b:0:0:0:0:9810:6420'
-
- trex_profile._set_proto_addr(ETHERNET, SRC, ether_range)
- trex_profile._set_proto_addr(ETHERNET, DST, ether_range)
- trex_profile._set_proto_addr(IP, SRC, ip_range)
- trex_profile._set_proto_addr(IP, DST, ip_range)
- trex_profile._set_proto_addr(IPv6, SRC, ipv6_range)
- trex_profile._set_proto_addr(IPv6, DST, ipv6_range)
- trex_profile._set_proto_addr(UDP, SRC_PORT, "5060-5090")
- trex_profile._set_proto_addr(UDP, DST_PORT, "5060")
diff --git a/tests/unit/network_services/vnf_generic/vnf/test_acl_vnf.py b/tests/unit/network_services/vnf_generic/vnf/test_acl_vnf.py
index 2971b55eb..efde669d2 100644
--- a/tests/unit/network_services/vnf_generic/vnf/test_acl_vnf.py
+++ b/tests/unit/network_services/vnf_generic/vnf/test_acl_vnf.py
@@ -29,6 +29,7 @@ stl_patch.start()
if stl_patch:
from yardstick.network_services.vnf_generic.vnf.acl_vnf import AclApproxVnf
from yardstick.network_services.nfvi.resource import ResourceProfile
+ from yardstick.network_services.vnf_generic.vnf.acl_vnf import AclApproxSetupEnvSetupEnvHelper
TEST_FILE_YAML = 'nsb_test_case.yaml'
@@ -345,3 +346,27 @@ class TestAclApproxVnf(unittest.TestCase):
acl_approx_vnf.dpdk_devbind = "dpdk-devbind.py"
acl_approx_vnf._resource_collect_stop = mock.Mock()
self.assertIsNone(acl_approx_vnf.terminate())
+
+
+class TestAclApproxSetupEnvSetupEnvHelper(unittest.TestCase):
+
+ @mock.patch('yardstick.network_services.vnf_generic.vnf.sample_vnf.open')
+ @mock.patch.object(utils, 'find_relative_file')
+ @mock.patch('yardstick.network_services.vnf_generic.vnf.sample_vnf.MultiPortConfig')
+ @mock.patch.object(utils, 'open_relative_file')
+ def test_build_config(self, *args):
+ vnfd_helper = mock.Mock()
+ ssh_helper = mock.Mock()
+ scenario_helper = mock.Mock()
+ scenario_helper.vnf_cfg = {'lb_config': 'HW'}
+ scenario_helper.all_options = {}
+
+ acl_approx_setup_helper = AclApproxSetupEnvSetupEnvHelper(vnfd_helper,
+ ssh_helper,
+ scenario_helper)
+
+ acl_approx_setup_helper.ssh_helper.provision_tool = mock.Mock(return_value='tool_path')
+ acl_approx_setup_helper.ssh_helper.all_ports = mock.Mock()
+ acl_approx_setup_helper.vnfd_helper.port_nums = mock.Mock(return_value=[0, 1])
+ expected = 'sudo tool_path -p 0x3 -f /tmp/acl_config -s /tmp/acl_script --hwlb 3'
+ self.assertEqual(acl_approx_setup_helper.build_config(), expected)
diff --git a/tests/unit/network_services/vnf_generic/vnf/test_base.py b/tests/unit/network_services/vnf_generic/vnf/test_base.py
index 664373f8f..9ef6473f0 100644
--- a/tests/unit/network_services/vnf_generic/vnf/test_base.py
+++ b/tests/unit/network_services/vnf_generic/vnf/test_base.py
@@ -215,9 +215,11 @@ class TestGenericVNF(unittest.TestCase):
with self.assertRaises(TypeError) as exc:
# pylint: disable=abstract-class-instantiated
base.GenericVNF('vnf1', VNFD['vnfd:vnfd-catalog']['vnfd'][0])
- msg = ("Can't instantiate abstract class GenericVNF with abstract "
- "methods collect_kpi, instantiate, scale, terminate, "
- "wait_for_instantiate")
+
+ msg = ("Can't instantiate abstract class GenericVNF with abstract methods "
+ "collect_kpi, instantiate, scale, start_collect, "
+ "stop_collect, terminate, wait_for_instantiate")
+
self.assertEqual(msg, str(exc.exception))
diff --git a/tests/unit/network_services/vnf_generic/vnf/test_cgnapt_vnf.py b/tests/unit/network_services/vnf_generic/vnf/test_cgnapt_vnf.py
index edaa0ad7e..b7731b649 100644
--- a/tests/unit/network_services/vnf_generic/vnf/test_cgnapt_vnf.py
+++ b/tests/unit/network_services/vnf_generic/vnf/test_cgnapt_vnf.py
@@ -20,7 +20,7 @@ import mock
from tests.unit import STL_MOCKS
from tests.unit.network_services.vnf_generic.vnf.test_base import mock_ssh
-
+from yardstick.common import utils
STLClient = mock.MagicMock()
stl_patch = mock.patch.dict("sys.modules", STL_MOCKS)
@@ -79,6 +79,27 @@ link 1 up
with self.assertRaises(NotImplementedError):
helper.scale()
+ @mock.patch('yardstick.network_services.vnf_generic.vnf.sample_vnf.open')
+ @mock.patch.object(utils, 'find_relative_file')
+ @mock.patch('yardstick.network_services.vnf_generic.vnf.sample_vnf.MultiPortConfig')
+ @mock.patch.object(utils, 'open_relative_file')
+ def test_build_config(self, *args):
+ vnfd_helper = mock.Mock()
+ ssh_helper = mock.Mock()
+ scenario_helper = mock.Mock()
+ scenario_helper.vnf_cfg = {'lb_config': 'HW'}
+ scenario_helper.all_options = {}
+
+ cgnat_approx_setup_helper = CgnaptApproxSetupEnvHelper(vnfd_helper,
+ ssh_helper,
+ scenario_helper)
+
+ cgnat_approx_setup_helper.ssh_helper.provision_tool = mock.Mock(return_value='tool_path')
+ cgnat_approx_setup_helper.ssh_helper.all_ports = mock.Mock()
+ cgnat_approx_setup_helper.vnfd_helper.port_nums = mock.Mock(return_value=[0, 1])
+ expected = 'sudo tool_path -p 0x3 -f /tmp/cgnapt_config -s /tmp/cgnapt_script --hwlb 3'
+ self.assertEqual(cgnat_approx_setup_helper.build_config(), expected)
+
@mock.patch("yardstick.network_services.vnf_generic.vnf.sample_vnf.Process")
class TestCgnaptApproxVnf(unittest.TestCase):
diff --git a/tests/unit/network_services/vnf_generic/vnf/test_sample_vnf.py b/tests/unit/network_services/vnf_generic/vnf/test_sample_vnf.py
index 26bd1dadd..ff71bed9d 100644
--- a/tests/unit/network_services/vnf_generic/vnf/test_sample_vnf.py
+++ b/tests/unit/network_services/vnf_generic/vnf/test_sample_vnf.py
@@ -607,12 +607,15 @@ class TestDpdkVnfSetupEnvHelper(unittest.TestCase):
dpdk_setup_helper.CFG_SCRIPT = 'script'
dpdk_setup_helper.pipeline_kwargs = {}
dpdk_setup_helper.all_ports = [0, 1, 2]
+ dpdk_setup_helper.scenario_helper.vnf_cfg = {'lb_config': 'HW',
+ 'worker_threads': 1}
expected = {
'cfg_file': 'config',
'script': 'script',
'port_mask_hex': '0x3',
'tool_path': 'tool_path',
+ 'hwlb': ' --hwlb 1',
}
dpdk_setup_helper._build_pipeline_kwargs()
self.assertDictEqual(dpdk_setup_helper.pipeline_kwargs, expected)
@@ -1661,42 +1664,6 @@ class TestSampleVnf(unittest.TestCase):
# test the default resource helper is MyResourceHelper, not subclass
self.assertEqual(type(sample_vnf.resource_helper), MyResourceHelper)
- def test__get_port0localip6(self):
- sample_vnf = SampleVNF('vnf1', self.VNFD_0)
- expected = '0064:ff9b:0:0:0:0:9810:6414'
- result = sample_vnf._get_port0localip6()
- self.assertEqual(result, expected)
-
- def test__get_port1localip6(self):
- sample_vnf = SampleVNF('vnf1', self.VNFD_0)
- expected = '0064:ff9b:0:0:0:0:9810:2814'
- result = sample_vnf._get_port1localip6()
- self.assertEqual(result, expected)
-
- def test__get_port0prefixip6(self):
- sample_vnf = SampleVNF('vnf1', self.VNFD_0)
- expected = '112'
- result = sample_vnf._get_port0prefixlen6()
- self.assertEqual(result, expected)
-
- def test__get_port1prefixip6(self):
- sample_vnf = SampleVNF('vnf1', self.VNFD_0)
- expected = '112'
- result = sample_vnf._get_port1prefixlen6()
- self.assertEqual(result, expected)
-
- def test__get_port0gateway6(self):
- sample_vnf = SampleVNF('vnf1', self.VNFD_0)
- expected = '0064:ff9b:0:0:0:0:9810:6414'
- result = sample_vnf._get_port0gateway6()
- self.assertEqual(result, expected)
-
- def test__get_port1gateway6(self):
- sample_vnf = SampleVNF('vnf1', self.VNFD_0)
- expected = '0064:ff9b:0:0:0:0:9810:2814'
- result = sample_vnf._get_port1gateway6()
- self.assertEqual(result, expected)
-
@mock.patch('yardstick.network_services.vnf_generic.vnf.sample_vnf.Process')
def test__start_vnf(self, *args):
vnfd = self.VNFD['vnfd:vnfd-catalog']['vnfd'][0]
@@ -1750,6 +1717,64 @@ class TestSampleVnf(unittest.TestCase):
self.assertIsNone(sample_vnf.instantiate(scenario_cfg, {}))
self.assertEqual(sample_vnf.nfvi_context, context2)
+ def test__update_collectd_options(self):
+ scenario_cfg = {'options':
+ {'collectd':
+ {'interval': 3,
+ 'plugins':
+ {'plugin3': {'param': 3}}},
+ 'vnf__0':
+ {'collectd':
+ {'interval': 2,
+ 'plugins':
+ {'plugin3': {'param': 2},
+ 'plugin2': {'param': 2}}}}}}
+ context_cfg = {'nodes':
+ {'vnf__0':
+ {'collectd':
+ {'interval': 1,
+ 'plugins':
+ {'plugin3': {'param': 1},
+ 'plugin2': {'param': 1},
+ 'plugin1': {'param': 1}}}}}}
+ expected = {'interval': 1,
+ 'plugins':
+ {'plugin3': {'param': 1},
+ 'plugin2': {'param': 1},
+ 'plugin1': {'param': 1}}}
+
+ vnfd = self.VNFD['vnfd:vnfd-catalog']['vnfd'][0]
+ sample_vnf = SampleVNF('vnf__0', vnfd)
+ sample_vnf._update_collectd_options(scenario_cfg, context_cfg)
+ self.assertEqual(sample_vnf.setup_helper.collectd_options, expected)
+
+ def test__update_options(self):
+ options1 = {'interval': 1,
+ 'param1': 'value1',
+ 'plugins':
+ {'plugin3': {'param': 3},
+ 'plugin2': {'param': 1},
+ 'plugin1': {'param': 1}}}
+ options2 = {'interval': 2,
+ 'param2': 'value2',
+ 'plugins':
+ {'plugin4': {'param': 4},
+ 'plugin2': {'param': 2},
+ 'plugin1': {'param': 2}}}
+ expected = {'interval': 1,
+ 'param1': 'value1',
+ 'param2': 'value2',
+ 'plugins':
+ {'plugin4': {'param': 4},
+ 'plugin3': {'param': 3},
+ 'plugin2': {'param': 1},
+ 'plugin1': {'param': 1}}}
+
+ vnfd = self.VNFD['vnfd:vnfd-catalog']['vnfd'][0]
+ sample_vnf = SampleVNF('vnf1', vnfd)
+ sample_vnf._update_options(options2, options1)
+ self.assertEqual(options2, expected)
+
@mock.patch("yardstick.network_services.vnf_generic.vnf.sample_vnf.time")
@mock.patch("yardstick.ssh.SSH")
def test_wait_for_instantiate_empty_queue(self, ssh, *args):
@@ -1785,16 +1810,6 @@ class TestSampleVnf(unittest.TestCase):
self.assertEqual(sample_vnf.wait_for_instantiate(), 0)
- def test__build_ports(self):
- vnfd = self.VNFD['vnfd:vnfd-catalog']['vnfd'][0]
- sample_vnf = SampleVNF('vnf1', vnfd)
-
- self.assertIsNone(sample_vnf._build_ports())
- self.assertIsNotNone(sample_vnf.networks)
- self.assertIsNotNone(sample_vnf.uplink_ports)
- self.assertIsNotNone(sample_vnf.downlink_ports)
- self.assertIsNotNone(sample_vnf.my_ports)
-
@mock.patch("yardstick.network_services.vnf_generic.vnf.sample_vnf.time")
def test_vnf_execute_with_queue_data(self, *args):
queue_size_list = [
diff --git a/tests/unit/network_services/vnf_generic/vnf/test_vfw_vnf.py b/tests/unit/network_services/vnf_generic/vnf/test_vfw_vnf.py
index 48fc87ed4..aaad66381 100644
--- a/tests/unit/network_services/vnf_generic/vnf/test_vfw_vnf.py
+++ b/tests/unit/network_services/vnf_generic/vnf/test_vfw_vnf.py
@@ -30,6 +30,7 @@ stl_patch.start()
if stl_patch:
from yardstick.network_services.vnf_generic.vnf.vfw_vnf import FWApproxVnf
from yardstick.network_services.nfvi.resource import ResourceProfile
+ from yardstick.network_services.vnf_generic.vnf.vfw_vnf import FWApproxSetupEnvHelper
TEST_FILE_YAML = 'nsb_test_case.yaml'
@@ -349,3 +350,25 @@ pipeline>
'rules': ""}}
self.scenario_cfg.update({"nodes": {"vnf__1": ""}})
self.assertIsNone(vfw_approx_vnf.instantiate(self.scenario_cfg, self.context_cfg))
+
+
+class TestFWApproxSetupEnvHelper(unittest.TestCase):
+
+ @mock.patch('yardstick.network_services.vnf_generic.vnf.sample_vnf.open')
+ @mock.patch.object(utils, 'find_relative_file')
+ @mock.patch('yardstick.network_services.vnf_generic.vnf.sample_vnf.MultiPortConfig')
+ @mock.patch.object(utils, 'open_relative_file')
+ def test_build_config(self, *args):
+ vnfd_helper = mock.Mock()
+ ssh_helper = mock.Mock()
+ scenario_helper = mock.Mock()
+ scenario_helper.vnf_cfg = {'lb_config': 'HW'}
+ scenario_helper.all_options = {}
+
+ vfw_approx_setup_helper = FWApproxSetupEnvHelper(vnfd_helper, ssh_helper, scenario_helper)
+
+ vfw_approx_setup_helper.ssh_helper.provision_tool = mock.Mock(return_value='tool_path')
+ vfw_approx_setup_helper.ssh_helper.all_ports = mock.Mock()
+ vfw_approx_setup_helper.vnfd_helper.port_nums = mock.Mock(return_value=[0, 1])
+ expected = 'sudo tool_path -p 0x3 -f /tmp/vfw_config -s /tmp/vfw_script --hwlb 3'
+ self.assertEqual(vfw_approx_setup_helper.build_config(), expected)
diff --git a/tests/unit/network_services/vnf_generic/vnf/test_vpe_vnf.py b/tests/unit/network_services/vnf_generic/vnf/test_vpe_vnf.py
index 8c45d973e..9857e95b6 100644
--- a/tests/unit/network_services/vnf_generic/vnf/test_vpe_vnf.py
+++ b/tests/unit/network_services/vnf_generic/vnf/test_vpe_vnf.py
@@ -18,7 +18,7 @@ import os
import time
import mock
-import six.moves.configparser as configparser
+from six.moves import configparser
import unittest
from tests.unit import STL_MOCKS
@@ -663,7 +663,14 @@ class TestVpeApproxVnf(unittest.TestCase):
vpe_approx_vnf.ssh_helper.bin_path = mock.Mock()
vpe_approx_vnf.ssh_helper.upload_config_file = mock.MagicMock()
self.assertIsNone(vpe_approx_vnf._build_vnf_ports())
- self.assertIsNotNone(vpe_approx_vnf.build_config())
+
+ vpe_approx_vnf.ssh_helper.provision_tool = mock.Mock(return_value='tool_path')
+ vpe_approx_vnf.ssh_helper.all_ports = mock.Mock()
+ vpe_approx_vnf.vnfd_helper.port_nums = mock.Mock(return_value=[0, 1])
+ vpe_approx_vnf.scenario_helper.vnf_cfg = {'lb_config': 'HW'}
+
+ expected = 'sudo tool_path -p 0x3 -f /tmp/vpe_config -s /tmp/vpe_script --hwlb 3'
+ self.assertEqual(vpe_approx_vnf.build_config(), expected)
@mock.patch(SSH_HELPER)
def test_wait_for_instantiate(self, ssh):