path: root/yardstick/tests/unit/network_services/vnf_generic/vnf/test_base.py

# 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


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(unittest.TestCase):

    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])

        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(unittest.TestCase):

    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)
        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)
        tg._mq_producer = mock.Mock()
        tg._mq_producer.id = 'fake_id'
        self.assertEqual('fake_id', tg.get_mq_producer_id())


class TrafficGeneratorProducerTestCase(unittest.TestCase):

    @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'})