diff options
Diffstat (limited to 'yardstick/tests/unit/network_services/vnf_generic/vnf/test_base.py')
| -rw-r--r-- | yardstick/tests/unit/network_services/vnf_generic/vnf/test_base.py | 348 |
1 files changed, 348 insertions, 0 deletions
diff --git a/yardstick/tests/unit/network_services/vnf_generic/vnf/test_base.py b/yardstick/tests/unit/network_services/vnf_generic/vnf/test_base.py new file mode 100644 index 000000000..2ea13a5e0 --- /dev/null +++ b/yardstick/tests/unit/network_services/vnf_generic/vnf/test_base.py @@ -0,0 +1,348 @@ +# 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 multiprocessing +import os +import uuid + +import mock +from oslo_config import cfg +import oslo_messaging +import unittest + +from yardstick.common import messaging +from yardstick.common.messaging import payloads +from yardstick.network_services.vnf_generic.vnf import base +from yardstick.ssh import SSH +from yardstick.tests.unit import base as ut_base + + +IP_PIPELINE_CFG_FILE_TPL = ("arp_route_tbl = ({port0_local_ip_hex}," + "{port0_netmask_hex},1,{port1_local_ip_hex}) " + "({port1_local_ip_hex},{port1_netmask_hex},0," + "{port0_local_ip_hex})") + +IP_PIPELINE_ND_CFG_FILE_TPL = """ +nd_route_tbl = ({port1_dst_ip_hex6},""" +"""{port1_dst_netmask_hex6},1,{port1_dst_ip_hex6})""" + +_LOCAL_OBJECT = object() + +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:03', + '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:04', + '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' +} + +VNFD = { + 'vnfd:vnfd-catalog': { + 'vnfd': [ + VNFD_0, + ] + } +} + + +class _DummyGenericTrafficGen(base.GenericTrafficGen): # pragma: no cover + + def run_traffic(self, *args): + pass + + def terminate(self): + pass + + def collect_kpi(self): + pass + + def instantiate(self, *args): + pass + + def scale(self, flavor=''): + pass + + +class FileAbsPath(object): + def __init__(self, module_file): + super(FileAbsPath, self).__init__() + self.module_path = os.path.dirname(os.path.abspath(module_file)) + + def get_path(self, filename): + file_path = os.path.join(self.module_path, filename) + return file_path + + +def mock_ssh(mock_ssh_type, spec=None, exec_result=_LOCAL_OBJECT, run_result=_LOCAL_OBJECT): + if spec is None: + spec = SSH + + if exec_result is _LOCAL_OBJECT: + exec_result = 0, "", "" + + if run_result is _LOCAL_OBJECT: + run_result = 0, "", "" + + mock_ssh_instance = mock.Mock(autospec=spec) + mock_ssh_instance._get_client.return_value = mock.Mock() + mock_ssh_instance.execute.return_value = exec_result + mock_ssh_instance.run.return_value = run_result + mock_ssh_type.from_node.return_value = mock_ssh_instance + return mock_ssh_instance + + +class TestQueueFileWrapper(unittest.TestCase): + def setUp(self): + self.prompt = "pipeline>" + self.q_in = multiprocessing.Queue() + self.q_out = multiprocessing.Queue() + + def test___init__(self): + queue_file_wrapper = \ + base.QueueFileWrapper(self.q_in, self.q_out, self.prompt) + self.assertEqual(queue_file_wrapper.prompt, self.prompt) + + def test_clear(self): + queue_file_wrapper = \ + base.QueueFileWrapper(self.q_in, self.q_out, self.prompt) + queue_file_wrapper.bufsize = 5 + queue_file_wrapper.write("pipeline>") + queue_file_wrapper.close() + self.assertIsNone(queue_file_wrapper.clear()) + self.assertIsNotNone(queue_file_wrapper.q_out.empty()) + + def test_close(self): + queue_file_wrapper = \ + base.QueueFileWrapper(self.q_in, self.q_out, self.prompt) + self.assertIsNone(queue_file_wrapper.close()) + + def test_read(self): + queue_file_wrapper = \ + base.QueueFileWrapper(self.q_in, self.q_out, self.prompt) + queue_file_wrapper.q_in.put("pipeline>") + self.assertEqual("pipeline>", queue_file_wrapper.read(20)) + + def test_write(self): + queue_file_wrapper = \ + base.QueueFileWrapper(self.q_in, self.q_out, self.prompt) + queue_file_wrapper.write("pipeline>") + self.assertIsNotNone(queue_file_wrapper.q_out.empty()) + + +class TestGenericVNF(ut_base.BaseUnitTestCase): + + def test_definition(self): + """Make sure that the abstract class cannot be instantiated""" + with self.assertRaises(TypeError) as exc: + # pylint: disable=abstract-class-instantiated + base.GenericVNF('vnf1', VNFD['vnfd:vnfd-catalog']['vnfd'][0], + 'task_id') + + 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)) + + +class GenericTrafficGenTestCase(ut_base.BaseUnitTestCase): + + def test_definition(self): + """Make sure that the abstract class cannot be instantiated""" + vnfd = VNFD['vnfd:vnfd-catalog']['vnfd'][0] + name = 'vnf1' + with self.assertRaises(TypeError) as exc: + # pylint: disable=abstract-class-instantiated + base.GenericTrafficGen(name, vnfd, 'task_id') + msg = ("Can't instantiate abstract class GenericTrafficGen with " + "abstract methods collect_kpi, instantiate, run_traffic, " + "scale, terminate") + self.assertEqual(msg, str(exc.exception)) + + def test_get_mq_producer_id(self): + vnfd = {'benchmark': {'kpi': mock.ANY}, + 'vdu': [{'external-interface': 'ext_int'}] + } + tg = _DummyGenericTrafficGen('name', vnfd, 'task_id') + tg._mq_producer = mock.Mock() + tg._mq_producer.id = 'fake_id' + self.assertEqual('fake_id', tg.get_mq_producer_id()) + + +class TrafficGeneratorProducerTestCase(ut_base.BaseUnitTestCase): + + @mock.patch.object(oslo_messaging, 'Target', return_value='rpc_target') + @mock.patch.object(oslo_messaging, 'RPCClient') + @mock.patch.object(oslo_messaging, 'get_rpc_transport', + return_value='rpc_transport') + @mock.patch.object(cfg, 'CONF') + def test__init(self, mock_config, mock_transport, mock_rpcclient, + mock_target): + _id = uuid.uuid1().int + tg_producer = base.TrafficGeneratorProducer(_id) + mock_transport.assert_called_once_with( + mock_config, url='rabbit://yardstick:yardstick@localhost:5672/') + mock_target.assert_called_once_with(topic=messaging.TOPIC_TG, + fanout=True, + server=messaging.SERVER) + mock_rpcclient.assert_called_once_with('rpc_transport', 'rpc_target') + self.assertEqual(_id, tg_producer._id) + self.assertEqual(messaging.TOPIC_TG, tg_producer._topic) + + @mock.patch.object(oslo_messaging, 'Target', return_value='rpc_target') + @mock.patch.object(oslo_messaging, 'RPCClient') + @mock.patch.object(oslo_messaging, 'get_rpc_transport', + return_value='rpc_transport') + @mock.patch.object(payloads, 'TrafficGeneratorPayload', + return_value='tg_pload') + def test_tg_method_started(self, mock_tg_payload, *args): + tg_producer = base.TrafficGeneratorProducer(uuid.uuid1().int) + with mock.patch.object(tg_producer, 'send_message') as mock_message: + tg_producer.tg_method_started(version=10) + + mock_message.assert_called_once_with(messaging.TG_METHOD_STARTED, + 'tg_pload') + mock_tg_payload.assert_called_once_with(version=10, iteration=0, + kpi={}) + + @mock.patch.object(oslo_messaging, 'Target', return_value='rpc_target') + @mock.patch.object(oslo_messaging, 'RPCClient') + @mock.patch.object(oslo_messaging, 'get_rpc_transport', + return_value='rpc_transport') + @mock.patch.object(payloads, 'TrafficGeneratorPayload', + return_value='tg_pload') + def test_tg_method_finished(self, mock_tg_payload, *args): + tg_producer = base.TrafficGeneratorProducer(uuid.uuid1().int) + with mock.patch.object(tg_producer, 'send_message') as mock_message: + tg_producer.tg_method_finished(version=20) + + mock_message.assert_called_once_with(messaging.TG_METHOD_FINISHED, + 'tg_pload') + mock_tg_payload.assert_called_once_with(version=20, iteration=0, + kpi={}) + + @mock.patch.object(oslo_messaging, 'Target', return_value='rpc_target') + @mock.patch.object(oslo_messaging, 'RPCClient') + @mock.patch.object(oslo_messaging, 'get_rpc_transport', + return_value='rpc_transport') + @mock.patch.object(payloads, 'TrafficGeneratorPayload', + return_value='tg_pload') + def test_tg_method_iteration(self, mock_tg_payload, *args): + tg_producer = base.TrafficGeneratorProducer(uuid.uuid1().int) + with mock.patch.object(tg_producer, 'send_message') as mock_message: + tg_producer.tg_method_iteration(100, version=30, kpi={'k': 'v'}) + + mock_message.assert_called_once_with(messaging.TG_METHOD_ITERATION, + 'tg_pload') + mock_tg_payload.assert_called_once_with(version=30, iteration=100, + kpi={'k': 'v'}) + + +class GenericVNFConsumerTestCase(ut_base.BaseUnitTestCase): + + def test__init(self): + endpoints = 'endpoint_1' + _ids = [uuid.uuid1().int] + gvnf_consumer = base.GenericVNFConsumer(_ids, endpoints) + self.assertEqual(_ids, gvnf_consumer._ids) + self.assertEqual([endpoints], gvnf_consumer._endpoints) |