############################################################################## # Copyright (c) 2016 Huawei Technologies Co.,Ltd and others. # # All rights reserved. This program and the accompanying materials # are made available under the terms of the Apache License, Version 2.0 # which accompanies this distribution, and is available at # http://www.apache.org/licenses/LICENSE-2.0 ############################################################################## import os import re import ConfigParser import logging import fileinput import consts.files as files import consts.parameters as parameters # ------------------------------------------------------ # List of common variables # ------------------------------------------------------ LOG = None CONF_FILE = None DEPLOYMENT_UNIT = None ITERATIONS = None BASE_DIR = None TEMPLATE_DIR = None TEMPLATE_NAME = None TEMPLATE_EXTENSION = None # ------------------------------------------------------ # Initialization and Input 'heat_templates/'validation # ------------------------------------------------------ def init(api=False): global BASE_DIR # BASE_DIR = os.getcwd() BASE_DIR = os.path.dirname(os.path.abspath(__file__)) BASE_DIR = BASE_DIR.replace('/heat', '') BASE_DIR = InputValidation.validate_directory_exist_and_format( BASE_DIR, "Error 000001") conf_file_init(api) log_init() general_vars_init(api) def conf_file_init(api=False): global CONF_FILE if api: CONF_FILE = ConfigurationFile(files.get_sections_api(), '/tmp/bottlenecks.conf') else: CONF_FILE = ConfigurationFile(cf.get_sections(), '/tmp/bottlenecks.conf') def general_vars_init(api=False): global TEMPLATE_EXTENSION global TEMPLATE_NAME global TEMPLATE_DIR global ITERATIONS TEMPLATE_EXTENSION = '.yaml' # Check Section in Configuration File InputValidation.validate_configuration_file_section( files.GENERAL, "Section " + files.GENERAL + "is not present in configuration file") InputValidation.validate_configuration_file_section( files.OPENSTACK, "Section " + files.OPENSTACK + "is not present in configuration file") TEMPLATE_DIR = '/tmp/heat_templates/' if not api: # Validate template name InputValidation.validate_configuration_file_parameter( files.GENERAL, files.TEMPLATE_NAME, "Parameter " + files.TEMPLATE_NAME + "is not present in configuration file") TEMPLATE_NAME = CONF_FILE.get_variable(files.GENERAL, files.TEMPLATE_NAME) InputValidation.validate_file_exist( TEMPLATE_DIR + TEMPLATE_NAME, "The provided template file does not exist") # Validate and assign Iterations if files.ITERATIONS in CONF_FILE.get_variable_list(files.GENERAL): ITERATIONS = int(CONF_FILE.get_variable(files.GENERAL, files.ITERATIONS)) else: ITERATIONS = 1 def log_init(): global LOG LOG = logging.getLogger() LOG.setLevel(level=logging.DEBUG) log_formatter = logging.Formatter("%(asctime)s --- %(message)s") file_handler = logging.FileHandler("{0}/{1}.log".format("./", "benchmark")) file_handler.setFormatter(log_formatter) file_handler.setLevel(logging.DEBUG) LOG.addHandler(file_handler) # ------------------------------------------------------ # Configuration file access # ------------------------------------------------------ class ConfigurationFile: """ Used to extract data from the configuration file """ def __init__(self, sections, config_file='conf.cfg'): """ Reads configuration file sections :param sections: list of strings representing the sections to be loaded :param config_file: name of the configuration file (string) :return: None """ InputValidation.validate_string( config_file, "The configuration file name must be a string") InputValidation.validate_file_exist( config_file, 'The provided configuration file does not exist') self.config = ConfigParser.ConfigParser() self.config.read(config_file) for section in sections: setattr( self, section, ConfigurationFile. _config_section_map(section, self.config)) @staticmethod def _config_section_map(section, config_file): """ Returns a dictionary with the configuration values for the specific section :param section: section to be loaded (string) :param config_file: name of the configuration file (string) :return: dict """ dict1 = dict() options = config_file.options(section) for option in options: dict1[option] = config_file.get(section, option) return dict1 def get_variable(self, section, variable_name): """ Returns the value correspondent to a variable :param section: section to be loaded (string) :param variable_name: name of the variable (string) :return: string """ message = "The variable name must be a string" InputValidation.validate_string(variable_name, message) if variable_name in self.get_variable_list(section): sect = getattr(self, section) return sect[variable_name] else: 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): """ Returns the list of the available variables in a section :param section: section to be loaded (string) :return: list """ try: return getattr(self, section) except: msg = 'Section {} not found in the configuration file'.\ format(section) raise ValueError(msg) # ------------------------------------------------------ # Manage files # ------------------------------------------------------ def get_heat_template_params(): """ Returns the list of deployment parameters from the configuration file for the heat template :return: dict """ heat_parameters_list = CONF_FILE.get_variable_list( files.DEPLOYMENT_PARAMETERS) testcase_parameters = dict() for param in heat_parameters_list: testcase_parameters[param] = CONF_FILE.get_variable( files.DEPLOYMENT_PARAMETERS, param) return testcase_parameters def get_testcase_params(): """ Returns the list of testcase parameters from the configuration file :return: dict """ testcase_parameters = dict() parameters = CONF_FILE.get_variable_list(files.TESTCASE_PARAMETERS) for param in parameters: testcase_parameters[param] = CONF_FILE.get_variable( files.TESTCASE_PARAMETERS, param) return testcase_parameters def get_file_first_line(file_name): """ Returns the first line of a file :param file_name: name of the file to be read (str) :return: str """ message = "name of the file must be a string" InputValidation.validate_string(file_name, message) message = 'file {} does not exist'.format(file_name) InputValidation.validate_file_exist(file_name, message) res = open(file_name, 'r') return res.readline() def replace_in_file(file, text_to_search, text_to_replace): """ Replaces a string within a file :param file: name of the file (str) :param text_to_search: text to be replaced :param text_to_replace: new text that will replace the previous :return: None """ message = 'text to be replaced in the file must be a string' InputValidation.validate_string(text_to_search, message) message = 'text to replace in the file must be a string' InputValidation.validate_string(text_to_replace, message) message = "name of the file must be a string" InputValidation.validate_string(file, message) message = "The file does not exist" InputValidation.validate_file_exist(file, message) for line in fileinput.input(file, inplace=True): print(line.replace(text_to_search, text_to_replace).rstrip()) # ------------------------------------------------------ # Shell interaction # ------------------------------------------------------ def run_command(command): LOG.info("Running command: {}".format(command)) return os.system(command) # ------------------------------------------------------ # Expose variables to other modules # ------------------------------------------------------ def get_base_dir(): return BASE_DIR def get_template_dir(): return TEMPLATE_DIR # ------------------------------------------------------ # Configuration Variables from Config File # ------------------------------------------------------ def get_deployment_configuration_variables_from_conf_file(): variables = dict() types = dict() all_variables = CONF_FILE.get_variable_list(files.EXPERIMENT_VNF) for var in all_variables: v = CONF_FILE.get_variable(files.EXPERIMENT_VNF, var) type = re.findall(r'@\w*', v) values = re.findall(r'\"(.+?)\"', v) variables[var] = values try: types[var] = type[0][1:] except IndexError: LOG.debug("No type has been specified for variable " + var) return variables # ------------------------------------------------------ # benchmarks from Config File # ------------------------------------------------------ def get_benchmarks_from_conf_file(): requested_benchmarks = list() benchmarks = CONF_FILE.get_variable( files.GENERAL, files.BENCHMARKS).split(', ') for benchmark in benchmarks: requested_benchmarks.append(benchmark) return requested_benchmarks class InputValidation(object): @staticmethod def validate_string(param, message): if not isinstance(param, str): raise ValueError(message) return True @staticmethod def validate_integer(param, message): if not isinstance(param, int): raise ValueError(message) return True @staticmethod def validate_dictionary(param, message): if not isinstance(param, dict): raise ValueError(message) return True @staticmethod def validate_file_exist(file_name, message): if not os.path.isfile(file_name): raise ValueError(message + ' ' + file_name) return True @staticmethod def validate_directory_exist_and_format(directory, message): if not os.path.isdir(directory): raise ValueError(message) if not directory.endswith('/'): return directory + '/' return directory @staticmethod def validate_configuration_file_parameter(section, parameter, message): params = CONF_FILE.get_variable_list(section) if parameter not in params: raise ValueError(message) return True @staticmethod def validate_configuration_file_section(section, message): if section not in files.get_sections(): raise ValueError(message) return True @staticmethod def validate_boolean(boolean, message): if isinstance(boolean, bool): return boolean if isinstance(boolean, str): if boolean == 'True': return True if boolean == 'False': return False raise ValueError(message) @staticmethod def validate_os_credentials(credentials): if not isinstance(credentials, dict): raise ValueError( 'The provided openstack_credentials ' 'variable must be in dictionary format') credential_keys = ['user', 'password', 'ip_controller', 'heat_url', 'auth_uri', 'project'] missing = [ credential_key for credential_key in credential_keys if credential_key not in credentials.keys() ] if len(missing) == 0: return True msg = 'OpenStack Credentials Error! ' \ 'The following parameters are missing: {}'.\ format(", ".join(missing)) raise ValueError(msg)