path: root/yardstick/vTC/apexlake/tests/instantiation_validation_bench_test.py

# Copyright (c) 2015 Intel Research and Development Ireland Ltd.
#
# 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
import os
import experimental_framework.constants.conf_file_sections as cfs
import experimental_framework.common as common
import experimental_framework.benchmarks.\
    instantiation_validation_benchmark as iv_module
from experimental_framework.benchmarks.\
    instantiation_validation_benchmark import InstantiationValidationBenchmark


kill_counter = [0, 0]
command_counter = [0, 0, 0, 0, 0]
replace_counter = [0, 0, 0]


def dummy_os_kill(pid, signal, get_counters=None):
    if get_counters:
        return kill_counter
    if pid == 1234:
        kill_counter[0] += 1
        return
    if pid == 4321:
        kill_counter[1] += 1
        return
    raise Exception(pid)


def dummy_run_command(command, get_counters=None):
    if get_counters:
        return command_counter
    if command == 'sudo smcroute -k':
        command_counter[0] += 1
        return
    elif command == 'sudo ip link delete interface.100':
        command_counter[1] += 1
        return
    elif command == 'sudo kill 1234':
        kill_counter[0] += 1
        return
    elif command == 'sudo kill 4321':
        kill_counter[1] += 1
        return
    raise Exception(command)


def dummy_run_command_2(command, get_counters=None):
    if get_counters:
        return command_counter
    if command == 'sudo ip link add link interface name interface.' \
                  '100 type vlan id 100':
        command_counter[0] += 1
        return
    elif command == 'sudo ifconfig interface.100 10.254.254.254 up' \
                    ' netmask 255.255.255.248':
        command_counter[1] += 1
        return
    elif command == "sudo echo 'mgroup from interface.100 group" \
                    " 224.192.16.1' > /etc/smcroute.conf":
        command_counter[2] += 1
        return
    elif command == "sudo smcroute -d":
        command_counter[3] += 1
        return
    elif command == "sudo test_sniff interface.100 128 &":
        command_counter[4] += 1
        return


def dummy_replace_in_file(file, str_from, str_to, get_couters=None):
    if get_couters:
        return replace_counter
    if file == 'file':
        if str_from == 'local out_file = "result_file"':
            if str_to == 'local out_file = ""':
                replace_counter[0] += 1
                return
        if str_from == 'local traffic_rate = 100':
            if str_to == 'local traffic_rate = 0':
                replace_counter[1] += 1
                return
        if str_from == 'local traffic_delay = 60':
            if str_to == 'local traffic_delay = 0':
                replace_counter[2] += 1
                return
        if str_from == 'local out_file = ""':
            if str_to == 'local out_file = "result_file"':
                replace_counter[3] += 1
                return
        if str_from == 'local traffic_rate = 0':
            if str_to == 'local traffic_rate = 100':
                replace_counter[4] += 1
                return
        if str_from == 'local traffic_delay = 0':
            if str_to == 'local traffic_delay = 60':
                replace_counter[5] += 1
                return
    raise Exception(file + ' ' + str_from + ' ' + str_to)


class DummyDpdkPacketGenerator():

    counter = 0

    def __init__(self):
        DummyDpdkPacketGenerator.counter = [0, 0]

    def init_dpdk_pktgen(self, dpdk_interfaces, lua_script, pcap_file_0,
                         pcap_file_1, vlan_0, vlan_1):
        if dpdk_interfaces == 1:
            if lua_script == 'constant_traffic.lua':
                if pcap_file_0 == 'packet_512.pcap':
                    if pcap_file_1 == 'igmp.pcap':
                        if vlan_0 == '-1':
                            if vlan_1 == '-1':
                                DummyDpdkPacketGenerator.counter[0] += 1

    def send_traffic(self):
        DummyDpdkPacketGenerator.counter[1] += 1
        return


class DummyInstantiaionValidationBenchmark(InstantiationValidationBenchmark):

    counter = [0, 0, 0, 0, 0]

    def _configure_lua_file(self, traffic_rate_percentage, traffic_time):
        DummyInstantiaionValidationBenchmark.counter[0] += 1

    def _init_packet_checker(self):
        DummyInstantiaionValidationBenchmark.counter[1] += 1

    def _finalize_packet_checker(self):
        DummyInstantiaionValidationBenchmark.counter[2] += 1

    def _reset_lua_file(self, traffic_rate_percentage, traffic_time):
        if traffic_rate_percentage == '1' and traffic_time == '10':
            DummyInstantiaionValidationBenchmark.counter[3] += 1

    def _get_results(self):
        DummyInstantiaionValidationBenchmark.counter[4] += 1
        res = {'test': 'result'}
        return res


class InstantiationValidationInitTest(unittest.TestCase):

    def setUp(self):
        common.BASE_DIR = os.getcwd()
        self.iv = InstantiationValidationBenchmark('InstantiationValidation',
                                                   dict())

    def tearDown(self):
        common.BASE_DIR = None

    @mock.patch('experimental_framework.common.get_base_dir')
    def test___init___for_success(self, mock_base_dir):
        mock_base_dir.return_value = 'base_dir/'
        iv = InstantiationValidationBenchmark('InstantiationValidation',
                                              dict())
        self.assertEqual(iv.base_dir,
                         'base_dir/experimental_framework/'
                         'packet_generators/dpdk_pktgen/')
        self.assertEqual(iv.results_file,
                         'base_dir/experimental_framework/'
                         'packet_generators/dpdk_pktgen/packets.res')
        self.assertEqual(iv.lua_file,
                         'base_dir/experimental_framework/'
                         'packet_generators/dpdk_pktgen/constant_traffic.lua')
        self.assertEqual(iv.pkt_checker_command,
                         'base_dir/experimental_framework/'
                         'libraries/packet_checker/test_sniff ')
        self.assertEqual(iv.res_dir, '')
        self.assertEqual(iv.interface_name, '')

    def test_init_for_success(self):
        self.iv.init()

    def test_finalize_for_success(self):
        self.iv.finalize()

    def test_get_features_for_success(self):

        expected = dict()
        expected['description'] = 'Instantiation Validation Benchmark'
        expected['parameters'] = [
            iv_module.THROUGHPUT,
            iv_module.VLAN_SENDER,
            iv_module.VLAN_RECEIVER
        ]
        expected['allowed_values'] = dict()
        expected['allowed_values'][iv_module.THROUGHPUT] = \
            map(str, range(0, 100))
        expected['allowed_values'][iv_module.VLAN_SENDER] = \
            map(str, range(-1, 4096))
        expected['allowed_values'][iv_module.VLAN_RECEIVER] = \
            map(str, range(-1, 4096))
        expected['default_values'] = dict()
        expected['default_values'][iv_module.THROUGHPUT] = '1'
        expected['default_values'][iv_module.VLAN_SENDER] = '-1'
        expected['default_values'][iv_module.VLAN_RECEIVER] = '-1'
        output = self.iv.get_features()
        self.assertEqual(expected, output)

    @mock.patch('commands.getoutput')
    def test__get_pids_for_success(self, mock_getoutput):
        expected = [1234]
        mock_getoutput.return_value = '1234'
        output = self.iv._get_pids()
        self.assertEqual(expected, output)

        expected = [1234, 4321]
        mock_getoutput.return_value = '1234\n4321'
        output = self.iv._get_pids()
        self.assertEqual(expected, output)

        expected = []
        mock_getoutput.return_value = None
        output = self.iv._get_pids()
        self.assertEqual(expected, output)

    @mock.patch('experimental_framework.common.run_command',
                side_effect=dummy_run_command)
    @mock.patch('os.kill', side_effect=dummy_os_kill)
    @mock.patch('experimental_framework.benchmarks.'
                'instantiation_validation_benchmark.'
                'InstantiationValidationBenchmark._get_pids')
    def test__finalize_packet_checker_for_success(self,
                                                  mock_pids,
                                                  mock_os_kill,
                                                  mock_run_command):
        global command_counter
        global kill_counter
        command_counter = [0, 0, 0, 0, 0]
        kill_counter = [0, 0]
        mock_pids.return_value = [1234, 4321]
        self.iv.interface_name = 'interface'
        self.iv.params[iv_module.VLAN_RECEIVER] = '100'
        self.iv._finalize_packet_checker()
        self.assertEqual(dummy_os_kill('', '', True), [1, 1])
        self.assertEqual(dummy_run_command('', True), [1, 1, 0, 0, 0])

    @mock.patch('os.chdir')
    @mock.patch('experimental_framework.common.run_command',
                side_effect=dummy_run_command_2)
    @mock.patch('experimental_framework.benchmarks.'
                'instantiation_validation_benchmark.'
                'InstantiationValidationBenchmark._get_pids')
    @mock.patch('os.kill', side_effect=dummy_os_kill)
    def test__init_packet_checker_for_success(self, mock_kill, mock_pids,
                                              mock_run_command, mock_chdir):
        global command_counter
        command_counter = [0, 0, 0, 0, 0]
        mock_pids.return_value = [1234, 4321]
        self.iv.pkt_checker_command = 'test_sniff '
        self.iv.interface_name = 'interface'
        self.iv.params[iv_module.VLAN_RECEIVER] = '100'
        self.iv._init_packet_checker()
        self.assertEqual(dummy_run_command('', True), [1, 1, 1, 1, 1])

    @mock.patch('experimental_framework.common.get_file_first_line')
    def test__get_results_for_success(self, mock_get_file):
        self.iv.res_dir = 'directory'
        mock_get_file.side_effect = ['100', '50']
        expected = {'failure': '0'}
        output = self.iv._get_results()
        self.assertEqual(expected, output)

        mock_get_file.side_effect = ['10', '50']
        expected = {'failure': '1'}
        output = self.iv._get_results()
        self.assertEqual(expected, output)

    @mock.patch('experimental_framework.common.replace_in_file',
                side_effect=dummy_replace_in_file)
    def test__reset_lua_file_for_success(self, mock_replace):
        global replace_counter
        replace_counter = [0, 0, 0, 0, 0, 0]
        traffic_rate_percentage = '100'
        traffic_time = '60'
        self.iv.lua_file = 'file'
        self.iv.results_file = 'result_file'
        self.iv._reset_lua_file(traffic_rate_percentage, traffic_time)
        self.assertEqual(dummy_replace_in_file('', '', '', True),
                         [1, 1, 1, 0, 0, 0])

    @mock.patch('experimental_framework.common.replace_in_file',
                side_effect=dummy_replace_in_file)
    def test__configure_lua_file_for_success(self, mock_replace):
        global replace_counter
        replace_counter = [0, 0, 0, 0, 0, 0]
        traffic_rate_percentage = '100'
        traffic_time = '60'
        self.iv.lua_file = 'file'
        self.iv.results_file = 'result_file'
        self.iv._configure_lua_file(traffic_rate_percentage, traffic_time)
        self.assertEqual(dummy_replace_in_file('', '', '', True),
                         [0, 0, 0, 1, 1, 1])

    @mock.patch('experimental_framework.common.LOG')
    @mock.patch('experimental_framework.packet_generators.'
                'dpdk_packet_generator.DpdkPacketGenerator',
                side_effect=DummyDpdkPacketGenerator)
    @mock.patch('experimental_framework.common.get_dpdk_pktgen_vars')
    def test_run_for_success(self, mock_common_get_vars, mock_pktgen,
                             mock_log):
        rval = dict()
        rval[cfs.CFSP_DPDK_BUS_SLOT_NIC_2] = 'bus_2'
        rval[cfs.CFSP_DPDK_NAME_IF_2] = 'if_2'
        mock_common_get_vars.return_value = rval
        expected = {'test': 'result'}
        iv = DummyInstantiaionValidationBenchmark('InstantiationValidation',
                                                  dict())
        iv.params[iv_module.THROUGHPUT] = '1'
        output = iv.run()
        self.assertEqual(expected, output)
        self.assertEqual(DummyDpdkPacketGenerator.counter,
                         [1, 1])
        self.assertEqual(DummyInstantiaionValidationBenchmark.counter,
                         [1, 1, 1, 1, 1])