:param test_cases: Test cases to be ran on the workload
(dict() of dict())
- Each string represents a test case and it is one
- of the strings provided by the
- "get_available_test_cases()" function output.
+
+ Example:
+ test_case = dict()
+ test_case['name'] = 'module.Class'
+ test_case['params'] = dict()
+ test_case['params']['throughput'] = '1'
+ test_case['params']['vlan_sender'] = '1007'
+ test_case['params']['vlan_receiver'] = '1006'
+ test_cases = [test_case]
:param iterations: Number of cycles to be executed (int)
correspond to the place holders (#parameter_name)
specified in the heat template.
- :return: None
+ :return: dict() Containing results
"""
+ common.init(api=True)
# Input Validation
common.InputValidation.validate_os_credentials(openstack_credentials)
credentials = openstack_credentials
+
msg = 'The provided heat_template does not exist'
template = "{}{}".format(common.get_template_dir(), heat_template)
common.InputValidation.validate_file_exist(template, msg)
+
msg = 'The provided iterations variable must be an integer value'
common.InputValidation.validate_integer(iterations, msg)
+
msg = 'The provided heat_template_parameters variable must be a ' \
'dictionary'
common.InputValidation.validate_dictionary(heat_template_parameters,
common.LOG.info("Benchmarking Unit initialization")
benchmarking_unit.initialize()
common.LOG.info("Benchmarking Unit Running")
- benchmarking_unit.run_benchmarks()
+ results = benchmarking_unit.run_benchmarks()
finally:
common.LOG.info("Benchmarking Unit Finalization")
benchmarking_unit.finalize()
+ return results
def get_name(self):
return self.name
+ def get_params(self):
+ return self.params
+
def get_features(self):
features = dict()
features['description'] = 'Please implement the method ' \
def __init__(self, name, params):
base.BenchmarkBaseClass.__init__(self, name, params)
- self.base_dir = common.get_base_dir() + \
- fp.EXPERIMENTAL_FRAMEWORK_DIR + fp.DPDK_PKTGEN_DIR
+ self.base_dir = "{}{}{}".format(
+ common.get_base_dir(), fp.EXPERIMENTAL_FRAMEWORK_DIR,
+ fp.DPDK_PKTGEN_DIR)
self.results_file = self.base_dir + PACKETS_FILE_NAME
self.lua_file = self.base_dir + 'constant_traffic.lua'
self.res_dir = ''
if self.res_dir:
packet_checker_res = \
int(common.get_file_first_line(self.res_dir +
- '/packet_checker.res'))
+ 'packet_checker.res'))
pkt_gen_res = int(common.get_file_first_line(self.results_file))
if pkt_gen_res <= packet_checker_res or \
(float(pkt_gen_res - packet_checker_res) / pkt_gen_res) <= 0.1:
:return:
"""
# Kill any other process running from previous failed execution
- self.res_dir = os.getcwd()
+ self.res_dir = common.get_result_dir()
pids = self._get_pids()
for pid in pids:
os.kill(pid, signal.SIGTERM)
common.run_command(command)
# Start the packet checker
+ # TODO: Compile "make" the packet sniffer
+ command = "chmod +x {}".format(self.pkt_checker_command)
+ common.run_command(command)
command = self.pkt_checker_command
command += self.interface_name + '.' + self.params[VLAN_RECEIVER]
command += ' 128'
features['allowed_values'][NUMBER_OF_CORES] = \
['0', '1', '2', '3', '4', '5', '6', '7', '8', '9', '10']
features['allowed_values'][AMOUNT_OF_RAM] = \
- ['250M', '1G', '2G', '3G', '4G', '5G', '6G', '7G', '8G', '9G',
+ ['256M', '1G', '2G', '3G', '4G', '5G', '6G', '7G', '8G', '9G',
'10G']
features['default_values'][NUM_OF_NEIGHBORS] = '1'
features['default_values'][NUMBER_OF_CORES] = '1'
- features['default_values'][AMOUNT_OF_RAM] = '250M'
+ features['default_values'][AMOUNT_OF_RAM] = '256M'
return features
def init(self):
features['allowed_values']['number_of_cores'] = \
['0', '1', '2', '3', '4', '5', '6', '7', '8', '9', '10']
features['allowed_values']['amount_of_ram'] = \
- ['250M', '1G', '2G', '3G', '4G', '5G', '6G', '7G', '8G', '9G',
+ ['256M', '1G', '2G', '3G', '4G', '5G', '6G', '7G', '8G', '9G',
'10G']
features['default_values']['num_of_neighbours'] = '1'
features['default_values']['number_of_cores'] = '1'
- features['default_values']['amount_of_ram'] = '250M'
+ features['default_values']['amount_of_ram'] = '256M'
return features
def init(self):
+
# Copyright (c) 2015 Intel Research and Development Ireland Ltd.
#
# Licensed under the Apache License, Version 2.0 (the "License");
to measure the throughput of the workload
:return: Results of the testcase (type: dict)
"""
- ret_val = dict()
packet_size = self._extract_packet_size_from_params()
- ret_val[PACKET_SIZE] = packet_size
# Packetgen management
packetgen = dpdk.DpdkPacketGenerator()
packetgen.send_traffic()
common.LOG.debug('Stop the packet generator')
- # Result Collection
- results = self._get_results()
- for metric_name in results.keys():
- ret_val[metric_name] = results[metric_name]
- self._reset_lua_file()
- return ret_val
+ return self._get_results()
def _extract_packet_size_from_params(self):
"""
# See the License for the specific language governing permissions and
# limitations under the License.
+import time
from experimental_framework.benchmarks import benchmark_base_class as base
return features
def run(self):
+ time.sleep(10)
return dict()
PKTGEN_MEMCHANNEL = None
PKTGEN_BUS_SLOT_NIC_1 = None
PKTGEN_BUS_SLOT_NIC_2 = None
+PKTGEN_NAME_NIC_1 = None
+PKTGEN_NAME_NIC_2 = None
+# TODO: remove Influx
INFLUXDB_IP = None
INFLUXDB_PORT = None
INFLUXDB_DB_NAME = None
def init(api=False):
global BASE_DIR
- BASE_DIR = os.getcwd()
+ # BASE_DIR = os.getcwd()
+ BASE_DIR = os.path.dirname(os.path.abspath(__file__))
BASE_DIR = BASE_DIR.replace('/experimental_framework', '')
BASE_DIR = InputValidation.validate_directory_exist_and_format(
BASE_DIR, "Error 000001")
init_conf_file(api)
- init_general_vars()
init_log()
+ init_general_vars(api)
if len(CONF_FILE.get_variable_list(cf.CFS_PKTGEN)) > 0:
init_pktgen()
def init_conf_file(api=False):
global CONF_FILE
if api:
- CONF_FILE = ConfigurationFile(cf.get_sections_api())
+ CONF_FILE = ConfigurationFile(cf.get_sections_api(),
+ '/etc/apexlake/apexlake.conf')
else:
- CONF_FILE = ConfigurationFile(cf.get_sections())
+ CONF_FILE = ConfigurationFile(cf.get_sections(),
+ '/etc/apexlake/apexlake.conf')
-def init_general_vars():
+def init_general_vars(api=False):
global TEMPLATE_FILE_EXTENSION
global TEMPLATE_NAME
global TEMPLATE_DIR
"Section " + cf.CFS_GENERAL +
"is not present in configuration file")
- TEMPLATE_DIR = BASE_DIR + 'heat_templates/'
-
- # Validate template name
- InputValidation.\
- validate_configuration_file_parameter(
- cf.CFS_GENERAL,
- cf.CFSG_TEMPLATE_NAME,
- "Parameter " + cf.CFSG_TEMPLATE_NAME +
- "is not present in configuration file")
+ TEMPLATE_DIR = '/tmp/apexlake/heat_templates/'
+ if not os.path.exists(TEMPLATE_DIR):
+ os.makedirs(TEMPLATE_DIR)
+ cmd = "cp /etc/apexlake/heat_templates/*.yaml {}".format(TEMPLATE_DIR)
+ run_command(cmd)
+
+ if not api:
+ # Validate template name
+ InputValidation.\
+ validate_configuration_file_parameter(
+ cf.CFS_GENERAL,
+ cf.CFSG_TEMPLATE_NAME,
+ "Parameter " + cf.CFSG_TEMPLATE_NAME +
+ "is not present in configuration file")
+ TEMPLATE_NAME = CONF_FILE.get_variable(cf.CFS_GENERAL,
+ cf.CFSG_TEMPLATE_NAME)
+ InputValidation.validate_file_exist(
+ TEMPLATE_DIR + TEMPLATE_NAME,
+ "The provided template file does not exist")
- TEMPLATE_NAME = CONF_FILE.get_variable(cf.CFS_GENERAL,
- cf.CFSG_TEMPLATE_NAME)
- InputValidation.validate_file_exist(
- TEMPLATE_DIR + TEMPLATE_NAME,
- "The provided template file does not exist")
- RESULT_DIR = BASE_DIR + 'results/'
+ RESULT_DIR = "/tmp/apexlake/results/"
# Validate and assign Iterations
if cf.CFSG_ITERATIONS in CONF_FILE.get_variable_list(cf.CFS_GENERAL):
def init_log():
global LOG
- if cf.CFSG_DEBUG in CONF_FILE.get_variable_list(cf.CFS_GENERAL) and \
- CONF_FILE.get_variable(cf.CFS_GENERAL, cf.CFSG_DEBUG):
- logging.basicConfig(level=logging.DEBUG)
- else:
- logging.basicConfig(level=logging.INFO)
-
LOG = logging.getLogger()
+ LOG.setLevel(level=logging.INFO)
log_formatter = logging.Formatter("%(asctime)s --- %(message)s")
file_handler = logging.FileHandler("{0}/{1}.log".format("./", "benchmark"))
file_handler.setFormatter(log_formatter)
global PKTGEN_BUS_SLOT_NIC_1
global PKTGEN_BUS_SLOT_NIC_2
global PKTGEN_DPDK_DIRECTORY
+ global PKTGEN_NAME_NIC_1
+ global PKTGEN_NAME_NIC_2
msg = "Section {} is not present in the configuration file".\
format(cf.CFS_PKTGEN)
PKTGEN_BUS_SLOT_NIC_2 = CONF_FILE.get_variable(
cf.CFS_PKTGEN, cf.CFSP_DPDK_BUS_SLOT_NIC_2)
+ # Validation of the DPDK NIC 1
+ msg = "Parameter {} is not present in section {}".format(
+ cf.CFSP_DPDK_NAME_IF_1, cf.CFS_PKTGEN)
+ InputValidation.validate_configuration_file_parameter(
+ cf.CFS_PKTGEN, cf.CFSP_DPDK_NAME_IF_1, msg)
+ PKTGEN_NAME_NIC_1 = CONF_FILE.get_variable(
+ cf.CFS_PKTGEN, cf.CFSP_DPDK_NAME_IF_1)
+
+ # Validation of the DPDK NIC 2
+ msg = "Parameter {} is not present in section {}".format(
+ cf.CFSP_DPDK_NAME_IF_2, cf.CFS_PKTGEN)
+ InputValidation.validate_configuration_file_parameter(
+ cf.CFS_PKTGEN, cf.CFSP_DPDK_NAME_IF_2, msg)
+ PKTGEN_NAME_NIC_2 = CONF_FILE.get_variable(
+ cf.CFS_PKTGEN, cf.CFSP_DPDK_NAME_IF_2)
+
# Validation of DPDK directory parameter
msg = "Parameter {} is not present in section {}".format(
cf.CFSP_DPDK_DPDK_DIRECTORY, cf.CFS_PKTGEN)
"""
InputValidation.validate_string(
config_file, "The configuration file name must be a string")
- config_file = BASE_DIR + config_file
+ # config_file = BASE_DIR + config_file
InputValidation.validate_file_exist(
config_file, 'The provided configuration file does not exist')
self.config = ConfigParser.ConfigParser()
sect = getattr(self, section)
return sect[variable_name]
else:
- exc_msg = 'Parameter {} is not in the {} section of the conf file'
- exc_msg.format(variable_name, section)
+ exc_msg = 'Parameter {} is not in the {} section of the ' \
+ 'conf file'.format(variable_name, section)
raise ValueError(exc_msg)
def get_variable_list(self, section):
# Shell interaction
# ------------------------------------------------------
def run_command(command):
- LOG.info("Running command: " + command)
+ LOG.info("Running command: {}".format(command))
return os.system(command)
return TEMPLATE_DIR
+def get_result_dir():
+ return RESULT_DIR
+
+
def get_dpdk_pktgen_vars():
if not (PKTGEN == 'dpdk_pktgen'):
return dict()
ret_val = dict()
ret_val[cf.CFSP_DPDK_PKTGEN_DIRECTORY] = PKTGEN_DIR
+ ret_val[cf.CFSP_DPDK_DPDK_DIRECTORY] = PKTGEN_DPDK_DIRECTORY
ret_val[cf.CFSP_DPDK_PROGRAM_NAME] = PKTGEN_PROGRAM
ret_val[cf.CFSP_DPDK_COREMASK] = PKTGEN_COREMASK
ret_val[cf.CFSP_DPDK_MEMORY_CHANNEL] = PKTGEN_MEMCHANNEL
ret_val[cf.CFSP_DPDK_BUS_SLOT_NIC_1] = PKTGEN_BUS_SLOT_NIC_1
ret_val[cf.CFSP_DPDK_BUS_SLOT_NIC_2] = PKTGEN_BUS_SLOT_NIC_2
- ret_val[cf.CFSP_DPDK_DPDK_DIRECTORY] = PKTGEN_DPDK_DIRECTORY
+ ret_val[cf.CFSP_DPDK_NAME_IF_1] = PKTGEN_NAME_NIC_1
+ ret_val[cf.CFSP_DPDK_NAME_IF_2] = PKTGEN_NAME_NIC_2
return ret_val
from heatclient import client as heatClient
from heatclient.common import template_utils
-from experimental_framework import common
+import experimental_framework.common as common
class HeatManager:
self.project_id = credentials['project']
self.heat = None
- # TODO: verify that init_heat is useless in the constructor
- # self.init_heat()
-
def init_heat(self):
keystone = keystoneClient.Client(username=self.user,
password=self.password,
tenant_name=self.project_id,
auth_url=self.auth_uri)
auth_token = keystone.auth_token
+ self.heat_url = keystone.service_catalog.url_for(
+ service_type='orchestration')
self.heat = heatClient.Client('1', endpoint=self.heat_url,
token=auth_token)
'''
-This file contains the code to Generate the heat templates from the base
-template
+Generation of the heat templates from the base template
'''
import json
# Delete the templates eventually generated in previous running of the
# framework
common.LOG.info("Removing the heat templates previously generated")
- os.system("rm " + template_dir + template_name + "_*")
+ command = "rm {}{}_*".format(template_dir, template_name)
+ os.system(command)
# Creation of the tree with all the new configurations
common.LOG.info("Creation of the tree with all the new configurations")
base_template = template_base_name
else:
base_template = template_dir + template_base_name
- if os.path.isabs(template_name):
- new_template = template_name
- else:
- new_template = template_dir + template_name
+ new_template = template_dir + template_name
new_template += "_" + str(counter) + template_file_extension
shutil.copy(base_template, new_template)
int write_file()
{
- FILE *f = fopen("packet_checker.res", "w");
+ FILE *f = fopen("/tmp/apexlake/results/packet_checker.res", "w");
if (f == NULL)
{
printf("Error opening file!\n");
import os
import base_packet_generator
import experimental_framework.common as common
+import time
from experimental_framework.constants import conf_file_sections as conf_file
from experimental_framework.constants import framework_parameters as fp
def _change_vlan(pcap_directory, pcap_file, vlan):
common.LOG.info("Changing VLAN Tag on Packet: " + pcap_file +
". New VLAN Tag is " + vlan)
+ command = "chmod +x {}{}".format(pcap_directory, 'vlan_tag.sh')
+ common.run_command(command)
command = pcap_directory + 'vlan_tag.sh '
command += pcap_directory + pcap_file + ' ' + vlan
common.run_command(command)
common.run_command(dpdk_vars[conf_file.CFSP_DPDK_DPDK_DIRECTORY] +
'tools/dpdk_nic_bind.py --unbind ' +
dpdk_vars[conf_file.CFSP_DPDK_BUS_SLOT_NIC_1])
+ time.sleep(5)
common.run_command(dpdk_vars[conf_file.CFSP_DPDK_DPDK_DIRECTORY] +
'tools/dpdk_nic_bind.py --bind=ixgbe ' +
dpdk_vars[conf_file.CFSP_DPDK_BUS_SLOT_NIC_1])
common.run_command(dpdk_vars[conf_file.CFSP_DPDK_DPDK_DIRECTORY] +
'tools/dpdk_nic_bind.py --unbind ' +
dpdk_vars[conf_file.CFSP_DPDK_BUS_SLOT_NIC_2])
+ time.sleep(5)
common.run_command(dpdk_vars[conf_file.CFSP_DPDK_DPDK_DIRECTORY] +
'tools/dpdk_nic_bind.py --bind=ixgbe ' +
dpdk_vars[conf_file.CFSP_DPDK_BUS_SLOT_NIC_2])
FrameworkApi.init()
mock_init.assert_called_once_with(api=True)
- @mock.patch('experimental_framework.benchmarking_unit.BenchmarkingUnit.'
- 'get_available_test_cases',
- side_effect=DummyBenchmarkingUnit.get_available_test_cases)
- def test_get_available_test_cases_for_success(self, mock_bench):
- expected = ['BenchA', 'BenchB']
- output = FrameworkApi.get_available_test_cases()
- self.assertEqual(expected, output)
-
- @mock.patch('experimental_framework.benchmarking_unit.BenchmarkingUnit.'
- 'get_required_benchmarks',
- side_effect=DummyBenchmarkingUnit.get_required_benchmarks)
- def test_get_test_case_features_for_success(self, mock_get_req_bench):
-
- expected = dict()
- expected['description'] = 'Instantiation Validation Benchmark'
- expected['parameters'] = [
- iv.THROUGHPUT,
- iv.VLAN_SENDER,
- iv.VLAN_RECEIVER]
- expected['allowed_values'] = dict()
- expected['allowed_values'][iv.THROUGHPUT] = \
- map(str, range(0, 100))
- expected['allowed_values'][iv.VLAN_SENDER] = \
- map(str, range(-1, 4096))
- expected['allowed_values'][iv.VLAN_RECEIVER] = \
- map(str, range(-1, 4096))
- expected['default_values'] = dict()
- expected['default_values'][iv.THROUGHPUT] = '1'
- expected['default_values'][iv.VLAN_SENDER] = '-1'
- expected['default_values'][iv.VLAN_RECEIVER] = '-1'
-
- test_case = 'instantiation_validation_benchmark.' \
- 'InstantiationValidationBenchmark'
- output = FrameworkApi.get_test_case_features(test_case)
- self.assertEqual(expected, output)
+ # @mock.patch('experimental_framework.benchmarking_unit.BenchmarkingUnit.'
+ # 'get_available_test_cases',
+ # side_effect=DummyBenchmarkingUnit.get_available_test_cases)
+ # def test_get_available_test_cases_for_success(self, mock_bench):
+ # expected = ['BenchA', 'BenchB']
+ # output = FrameworkApi.get_available_test_cases()
+ # self.assertEqual(expected, output)
+
+ # @mock.patch('experimental_framework.benchmarking_unit.BenchmarkingUnit.'
+ # 'get_required_benchmarks',
+ # side_effect=DummyBenchmarkingUnit.get_required_benchmarks)
+ # def test_get_test_case_features_for_success(self, mock_get_req_bench):
+ #
+ # expected = dict()
+ # expected['description'] = 'Instantiation Validation Benchmark'
+ # expected['parameters'] = [
+ # iv.THROUGHPUT,
+ # iv.VLAN_SENDER,
+ # iv.VLAN_RECEIVER]
+ # expected['allowed_values'] = dict()
+ # expected['allowed_values'][iv.THROUGHPUT] = \
+ # map(str, range(0, 100))
+ # expected['allowed_values'][iv.VLAN_SENDER] = \
+ # map(str, range(-1, 4096))
+ # expected['allowed_values'][iv.VLAN_RECEIVER] = \
+ # map(str, range(-1, 4096))
+ # expected['default_values'] = dict()
+ # expected['default_values'][iv.THROUGHPUT] = '1'
+ # expected['default_values'][iv.VLAN_SENDER] = '-1'
+ # expected['default_values'][iv.VLAN_RECEIVER] = '-1'
+ #
+ # test_case = 'instantiation_validation_benchmark.' \
+ # 'InstantiationValidationBenchmark'
+ # output = FrameworkApi.get_test_case_features(test_case)
+ # self.assertEqual(expected, output)
def test____for_failure(self):
self.assertRaises(
ValueError, FrameworkApi.get_test_case_features, 111)
+ @mock.patch('experimental_framework.common.init')
@mock.patch('experimental_framework.common.LOG')
@mock.patch('experimental_framework.common.get_credentials')
@mock.patch('experimental_framework.heat_template_generation.'
@mock.patch('experimental_framework.benchmarking_unit.BenchmarkingUnit',
side_effect=DummyBenchmarkingUnit2)
def test_execute_framework_for_success(self, mock_b_unit, mock_heat,
- mock_credentials, mock_log):
+ mock_credentials, mock_log,
+ mock_common_init):
common.TEMPLATE_DIR = "{}/{}/".format(
os.getcwd(), 'tests/data/generated_templates'
)
class DummyConfigurationFile(common.ConfigurationFile):
- def __init__(self, sections):
+ def __init__(self, sections, conf_file=''):
pass
def get_variable(self, section, variable_name):
if variable_name == cf.CFSG_TEMPLATE_NAME:
return 'vTC.yaml'
if variable_name == cf.CFSG_ITERATIONS:
- return 2
+ return '2'
if variable_name == cf.CFSG_DEBUG:
- return True
+ return 'True'
if variable_name == cf.CFSP_PACKET_GENERATOR:
if self.pktgen_counter == 1:
return 'non_supported'
expected = self.dir.split('experimental_framework/')[0]
self.assertEqual(common.BASE_DIR, expected)
- def test_init_general_vars_for_success(self):
+ @mock.patch('os.path.exists')
+ @mock.patch('os.makedirs')
+ @mock.patch('experimental_framework.common.LOG')
+ def test_init_general_vars_for_success(self, mock_log, mock_makedirs,
+ mock_path_exists):
common.BASE_DIR = "{}/".format(os.getcwd())
+ mock_path_exists.return_value = False
common.init_general_vars()
self.assertEqual(common.TEMPLATE_FILE_EXTENSION, '.yaml')
- heat_dir = self.dir.split('experimental_framework/')[0]
- self.assertEqual(common.TEMPLATE_DIR,
- '{}{}'.format(heat_dir, 'heat_templates/'))
+ self.assertEqual(common.TEMPLATE_DIR, '/tmp/apexlake/heat_templates/')
self.assertEqual(common.TEMPLATE_NAME, 'vTC.yaml')
- self.assertEqual(common.RESULT_DIR,
- '{}{}'.format(heat_dir, 'results/'))
+ self.assertEqual(common.RESULT_DIR, '/tmp/apexlake/results/')
self.assertEqual(common.ITERATIONS, 1)
+ mock_makedirs.assert_called_once_with('/tmp/apexlake/heat_templates/')
class TestCommonInit2(unittest.TestCase):
reset_common()
common.CONF_FILE = None
- def test_init_general_vars_2_for_success(self):
+ @mock.patch('experimental_framework.common.LOG')
+ def test_init_general_vars_2_for_success(self, mock_log):
common.BASE_DIR = "{}/".format(os.getcwd())
common.init_general_vars()
self.assertEqual(common.TEMPLATE_FILE_EXTENSION, '.yaml')
- heat_dir = self.dir.split('experimental_framework/')[0]
- self.assertEqual(common.TEMPLATE_DIR,
- '{}{}'.format(heat_dir, 'heat_templates/'))
+ self.assertEqual(common.TEMPLATE_DIR, '/tmp/apexlake/heat_templates/')
self.assertEqual(common.TEMPLATE_NAME, 'vTC.yaml')
- self.assertEqual(common.RESULT_DIR,
- '{}{}'.format(heat_dir, 'results/'))
+ self.assertEqual(common.RESULT_DIR, '/tmp/apexlake/results/')
self.assertEqual(common.ITERATIONS, 2)
def test_init_log_2_for_success(self):
'Deployment-parameters',
'Testcase-parameters'
]
- c_file = '/tests/data/common/conf.cfg'
+ c_file = './tests/data/common/conf.cfg'
common.BASE_DIR = os.getcwd()
self.conf_file = common.ConfigurationFile(self.sections, c_file)
sections = ['General', 'OpenStack', 'Experiment-VNF', 'PacketGen',
'Deployment-parameters', 'Testcase-parameters']
c = DummyConfigurationFile3(
- sections, config_file='/tests/data/common/conf.cfg')
+ sections, config_file='./tests/data/common/conf.cfg')
self.assertEqual(
DummyConfigurationFile3._config_section_map('', '', True),
6)
'Deployment-parameters',
'Testcase-parameters'
]
- config_file = '/tests/data/common/conf.cfg'
+ config_file = './tests/data/common/conf.cfg'
common.BASE_DIR = os.getcwd()
common.CONF_FILE = DummyConfigurationFile4(self.sections, config_file)
common.PKTGEN_MEMCHANNEL = 'var'
common.PKTGEN_BUS_SLOT_NIC_1 = 'var'
common.PKTGEN_BUS_SLOT_NIC_2 = 'var'
+ common.PKTGEN_NAME_NIC_1 = 'var'
+ common.PKTGEN_NAME_NIC_2 = 'var'
common.PKTGEN_DPDK_DIRECTORY = 'var'
expected = {
'bus_slot_nic_1': 'var',
'bus_slot_nic_2': 'var',
+ 'name_if_1': 'var',
+ 'name_if_2': 'var',
'coremask': 'var',
'dpdk_directory': 'var',
'memory_channels': 'var',
parameters, 0)
self.assertEqual(output, True)
+ @mock.patch('experimental_framework.common.LOG')
@mock.patch('experimental_framework.heat_manager.HeatManager',
side_effect=DummyHeatManagerFailed)
@mock.patch('os.path.isfile')
def test_deploy_heat_template_for_success_2(self, mock_os_is_file,
- mock_heat_manager):
+ mock_heat_manager, mock_log):
mock_os_is_file.return_value = True
du = DummyDeploymentUnit(dict())
template_file = ''
def test__change_vlan_for_success(self, mock_run_command, mock_log):
mut.DpdkPacketGenerator._change_vlan('/directory/', 'pcap_file', '10')
expected_param = '/directory/vlan_tag.sh /directory/pcap_file 10'
- mock_run_command.assert_called_once_with(expected_param)
+ mock_run_command.assert_called_with(expected_param)
# See the License for the specific language governing permissions and
# limitations under the License.
+__author__ = 'gpetralx'
+
+
import unittest
import experimental_framework.heat_template_generation as heat_gen
import mock
self.assertFalse(self.heat_manager.delete_stack('stack_1'))
+class ServiceCatalog():
+ def url_for(self, service_type):
+ return 'http://heat_url'
+
+
class KeystoneMock(object):
@property
def auth_token(self):
return 'token'
+ service_catalog = ServiceCatalog()
+
class TestHeatInit(unittest.TestCase):
def setUp(self):
elif command == "test_sniff interface.100 128 &":
command_counter[4] += 1
return
- raise Exception(command)
def dummy_replace_in_file(file, str_from, str_to, get_couters=None):
# See the License for the specific language governing permissions and
# limitations under the License.
+__author__ = 'gpetralx'
+
import unittest
import mock
def test_run_for_success(self, mock_dpdk, mock_get_results,
mock_extract_size, conf_lua_file_mock,
reset_lua_file_mock, mock_common_log):
- expected = {'results': 0, 'packet_size': '1'}
+ expected = {'results': 0}
mock_extract_size.return_value = '1'
mock_get_results.return_value = {'results': 0}
output = self.benchmark.run()
Code Review / yardstick.git / commitdiff