diff options
Diffstat (limited to 'functest/opnfv_tests/vnf/router/vnf_controller')
6 files changed, 0 insertions, 523 deletions
diff --git a/functest/opnfv_tests/vnf/router/vnf_controller/__init__.py b/functest/opnfv_tests/vnf/router/vnf_controller/__init__.py deleted file mode 100644 index e69de29b..00000000 --- a/functest/opnfv_tests/vnf/router/vnf_controller/__init__.py +++ /dev/null diff --git a/functest/opnfv_tests/vnf/router/vnf_controller/checker.py b/functest/opnfv_tests/vnf/router/vnf_controller/checker.py deleted file mode 100644 index a7a70f6d..00000000 --- a/functest/opnfv_tests/vnf/router/vnf_controller/checker.py +++ /dev/null @@ -1,63 +0,0 @@ -#!/usr/bin/env python - -# Copyright (c) 2017 Okinawa Open Laboratory 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 - -# pylint: disable=missing-docstring - -"""vrouter test result check module""" - -import json -import logging -import re - -from jinja2 import Environment, FileSystemLoader - - -class Checker(object): - """vrouter test result check class""" - - logger = logging.getLogger(__name__) - - def __init__(self): - self.logger.debug("init checker") - - @staticmethod - def load_check_rule(rule_file_dir, rule_file_name, parameter): - loader = FileSystemLoader(rule_file_dir, - encoding='utf8') - env = Environment(loader=loader) - check_rule_template = env.get_template(rule_file_name) - check_rule = check_rule_template.render(parameter) - check_rule_data = json.loads(check_rule) - return check_rule_data - - @staticmethod - def regexp_information(response, rules): - status = False - result_data = {} - - for rule in rules["rules"]: - result_data = { - "test_name": rule["description"], - "result": "NG" - } - - match = re.search(rule["regexp"], - response) - rule["response"] = response - if match is None: - status = False - break - - if not match.group(1) == rule["result"]: - status = False - else: - result_data["result"] = "OK" - status = True - - return status, result_data diff --git a/functest/opnfv_tests/vnf/router/vnf_controller/command_generator.py b/functest/opnfv_tests/vnf/router/vnf_controller/command_generator.py deleted file mode 100644 index 7d9116bc..00000000 --- a/functest/opnfv_tests/vnf/router/vnf_controller/command_generator.py +++ /dev/null @@ -1,38 +0,0 @@ -#!/usr/bin/env python - -# Copyright (c) 2017 Okinawa Open Laboratory 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 - -# pylint: disable=missing-docstring - -"""command generator module for vrouter testing""" - -import logging -from jinja2 import Environment, FileSystemLoader - - -class CommandGenerator(object): - """command generator class for vrouter testing""" - - logger = logging.getLogger(__name__) - - def __init__(self): - self.logger.debug("init command generator") - - @staticmethod - def load_template(template_dir, template): - # pylint disable=missing-docstring - loader = FileSystemLoader(template_dir, - encoding='utf8') - env = Environment(loader=loader) - return env.get_template(template) - - @staticmethod - def command_create(template, parameter): - # pylint disable=missing-docstring - commands = template.render(parameter) - return commands.split('\n') diff --git a/functest/opnfv_tests/vnf/router/vnf_controller/ssh_client.py b/functest/opnfv_tests/vnf/router/vnf_controller/ssh_client.py deleted file mode 100644 index 628afd30..00000000 --- a/functest/opnfv_tests/vnf/router/vnf_controller/ssh_client.py +++ /dev/null @@ -1,135 +0,0 @@ -#!/usr/bin/env python - -# Copyright (c) 2017 Okinawa Open Laboratory 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 - -"""ssh client module for vrouter testing""" - -import logging -import time -import yaml - -import paramiko - -from functest.opnfv_tests.vnf.router.utilvnf import Utilvnf - -RECEIVE_ROOP_WAIT = 1 - -DEFAULT_CONNECT_TIMEOUT = 10 -DEFAULT_CONNECT_RETRY_COUNT = 10 -DEFAULT_SEND_TIMEOUT = 10 - - -class SshClient(object): # pylint: disable=too-many-instance-attributes - """ssh client class for vrouter testing""" - - logger = logging.getLogger(__name__) - - def __init__(self, ip_address, user, password=None, key_filename=None): - self.ip_address = ip_address - self.user = user - self.password = password - self.key_filename = key_filename - self.connected = False - self.shell = None - - self.logger.setLevel(logging.INFO) - - self.ssh = paramiko.SSHClient() - self.ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy()) - - self.util = Utilvnf() - with open(self.util.test_env_config_yaml) as file_fd: - test_env_config_yaml = yaml.safe_load(file_fd) - file_fd.close() - - self.ssh_revieve_buff = test_env_config_yaml.get("general").get( - "ssh_receive_buffer") - - def connect(self, time_out=DEFAULT_CONNECT_TIMEOUT, - retrycount=DEFAULT_CONNECT_RETRY_COUNT): - # pylint: disable=missing-docstring - while retrycount > 0: - try: - self.logger.info("SSH connect to %s.", self.ip_address) - self.ssh.connect(self.ip_address, - username=self.user, - password=self.password, - key_filename=self.key_filename, - timeout=time_out, - look_for_keys=False, - allow_agent=False) - - self.logger.info("SSH connection established to %s.", - self.ip_address) - - self.shell = self.ssh.invoke_shell() - - while not self.shell.recv_ready(): - time.sleep(RECEIVE_ROOP_WAIT) - - self.shell.recv(self.ssh_revieve_buff) - break - except Exception: # pylint: disable=broad-except - self.logger.info("SSH timeout for %s...", self.ip_address) - time.sleep(time_out) - retrycount -= 1 - - if retrycount == 0: - self.logger.error("Cannot establish connection to IP '%s'. " + - "Aborting", - self.ip_address) - self.connected = False - return self.connected - - self.connected = True - return self.connected - - def send(self, cmd, prompt, timeout=DEFAULT_SEND_TIMEOUT): - # pylint: disable=missing-docstring - if self.connected is True: - self.shell.settimeout(timeout) - self.logger.debug("Commandset : '%s'", cmd) - - try: - self.shell.send(cmd + '\n') - except Exception: # pylint: disable=broad-except - self.logger.error("ssh send timeout : Command : '%s'", cmd) - return None - - res_buff = '' - while not res_buff.endswith(prompt): - time.sleep(RECEIVE_ROOP_WAIT) - try: - res = self.shell.recv(self.ssh_revieve_buff) - except Exception: # pylint: disable=broad-except - self.logger.error("ssh receive timeout : Command : '%s'", - cmd) - break - - res_buff += res - - self.logger.debug("Response : '%s'", res_buff) - return res_buff - self.logger.error("Cannot connected to IP '%s'.", self.ip_address) - return None - - def close(self): - # pylint: disable=missing-docstring - if self.connected is True: - self.ssh.close() - - @staticmethod - def error_check(response, err_strs=None): - # pylint: disable=missing-docstring - if err_strs is None: - err_strs = ["error", "warn", "unknown command", "already exist"] - for err in err_strs: - if err in response: - return False - - return True diff --git a/functest/opnfv_tests/vnf/router/vnf_controller/vm_controller.py b/functest/opnfv_tests/vnf/router/vnf_controller/vm_controller.py deleted file mode 100644 index 10e48645..00000000 --- a/functest/opnfv_tests/vnf/router/vnf_controller/vm_controller.py +++ /dev/null @@ -1,146 +0,0 @@ -#!/usr/bin/env python - -# Copyright (c) 2017 Okinawa Open Laboratory 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 - -# pylint: disable=missing-docstring - -"""vm controll module""" - -import logging -import os -import time -import yaml - -from functest.opnfv_tests.vnf.router.utilvnf import Utilvnf -from functest.opnfv_tests.vnf.router.vnf_controller.command_generator import ( - CommandGenerator) -from functest.opnfv_tests.vnf.router.vnf_controller.ssh_client import ( - SshClient) - - -class VmController(object): - """vm controll class""" - - logger = logging.getLogger(__name__) - - def __init__(self, util_info): - self.logger.debug("initialize vm controller") - self.command_gen = CommandGenerator() - credentials = util_info["credentials"] - - self.util = Utilvnf() - self.util.set_credentials(credentials["snaps_creds"]) - - with open(self.util.test_env_config_yaml) as file_fd: - test_env_config_yaml = yaml.safe_load(file_fd) - file_fd.close() - - self.reboot_wait = test_env_config_yaml.get("general").get( - "reboot_wait") - self.command_wait = test_env_config_yaml.get("general").get( - "command_wait") - self.ssh_connect_timeout = test_env_config_yaml.get("general").get( - "ssh_connect_timeout") - self.ssh_connect_retry_count = test_env_config_yaml.get("general").get( - "ssh_connect_retry_count") - - def command_gen_from_template(self, command_file_path, cmd_input_param): - (command_file_dir, command_file_name) = os.path.split( - command_file_path) - template = self.command_gen.load_template(command_file_dir, - command_file_name) - return self.command_gen.command_create(template, - cmd_input_param) - - def config_vm(self, vm_info, test_cmd_file_path, - cmd_input_param, prompt_file_path): - ssh = self.connect_ssh_and_config_vm(vm_info, - test_cmd_file_path, - cmd_input_param, - prompt_file_path) - if ssh is None: - return False - - ssh.close() - - return True - - def connect_ssh_and_config_vm(self, vm_info, test_cmd_file_path, - cmd_input_param, prompt_file_path): - - key_filename = None - if "key_path" in vm_info: - key_filename = vm_info["key_path"] - - ssh = SshClient(ip_address=vm_info["floating_ip"], - user=vm_info["user"], - password=vm_info["pass"], - key_filename=key_filename) - - result = ssh.connect(self.ssh_connect_timeout, - self.ssh_connect_retry_count) - if not result: - self.logger.debug("try to vm reboot.") - self.util.reboot_vm(vm_info["vnf_name"]) - time.sleep(self.reboot_wait) - result = ssh.connect(self.ssh_connect_timeout, - self.ssh_connect_retry_count) - if not result: - return None - - (result, _) = self.command_create_and_execute( - ssh, - test_cmd_file_path, - cmd_input_param, - prompt_file_path) - if not result: - ssh.close() - return None - - return ssh - - def command_create_and_execute(self, ssh, test_cmd_file_path, - cmd_input_param, prompt_file_path): - prompt_file = open(prompt_file_path, - 'r') - prompt = yaml.safe_load(prompt_file) - prompt_file.close() - config_mode_prompt = prompt["config_mode"] - - commands = self.command_gen_from_template(test_cmd_file_path, - cmd_input_param) - return self.command_list_execute(ssh, - commands, - config_mode_prompt) - - def command_list_execute(self, ssh, command_list, prompt): - res_data_list = [] - for command in command_list: - self.logger.debug("Command : " + command) - (res, res_data) = self.command_execute(ssh, - command, - prompt) - self.logger.debug("Response : " + res_data) - res_data_list.append(res_data) - if not res: - return res, res_data_list - - time.sleep(self.command_wait) - - return True, res_data_list - - def command_execute(self, ssh, command, prompt): - res_data = ssh.send(command, prompt) - if res_data is None: - self.logger.info("retry send command : " + command) - res_data = ssh.send(command, - prompt) - if not ssh.error_check(res_data): - return False, res_data - - return True, res_data diff --git a/functest/opnfv_tests/vnf/router/vnf_controller/vnf_controller.py b/functest/opnfv_tests/vnf/router/vnf_controller/vnf_controller.py deleted file mode 100644 index a5b1ad85..00000000 --- a/functest/opnfv_tests/vnf/router/vnf_controller/vnf_controller.py +++ /dev/null @@ -1,141 +0,0 @@ -#!/usr/bin/env python - -# Copyright (c) 2017 Okinawa Open Laboratory 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 - -# pylint: disable=missing-docstring - -"""vrouter controll module""" - -import logging -import os -import time -import yaml - -import prettytable - -from functest.opnfv_tests.vnf.router.utilvnf import Utilvnf -from functest.opnfv_tests.vnf.router.vnf_controller.checker import Checker -from functest.opnfv_tests.vnf.router.vnf_controller.ssh_client import ( - SshClient) -from functest.opnfv_tests.vnf.router.vnf_controller.vm_controller import ( - VmController) - - -class VnfController(object): - """vrouter controll class""" - - logger = logging.getLogger(__name__) - - def __init__(self, util_info): - self.logger.debug("init vnf controller") - self.util = Utilvnf() - self.vm_controller = VmController(util_info) - - with open(self.util.test_env_config_yaml) as file_fd: - test_env_config_yaml = yaml.safe_load(file_fd) - file_fd.close() - - self.cmd_wait = test_env_config_yaml.get("general").get("command_wait") - self.ssh_connect_timeout = test_env_config_yaml.get("general").get( - "ssh_connect_timeout") - self.ssh_connect_retry_count = test_env_config_yaml.get("general").get( - "ssh_connect_retry_count") - - def config_vnf(self, source_vnf, destination_vnf, test_cmd_file_path, - parameter_file_path, prompt_file_path): - # pylint: disable=too-many-arguments - parameter_file = open(parameter_file_path, - 'r') - cmd_input_param = yaml.safe_load(parameter_file) - parameter_file.close() - - cmd_input_param["macaddress"] = source_vnf["data_plane_network_mac"] - cmd_input_param["source_ip"] = source_vnf["data_plane_network_ip"] - cmd_input_param["destination_ip"] = destination_vnf[ - "data_plane_network_ip"] - - return self.vm_controller.config_vm(source_vnf, - test_cmd_file_path, - cmd_input_param, - prompt_file_path) - - def result_check(self, target_vnf, reference_vnf, - check_rule_file_path_list, parameter_file_path, - prompt_file_path): - # pylint: disable=too-many-arguments,too-many-locals - - res_dict_data_list = [] - - parameter_file = open(parameter_file_path, - 'r') - cmd_input_param = yaml.safe_load(parameter_file) - parameter_file.close() - - cmd_input_param["source_ip"] = target_vnf["data_plane_network_ip"] - cmd_input_param["destination_ip"] = reference_vnf[ - "data_plane_network_ip"] - - prompt_file = open(prompt_file_path, - 'r') - prompt = yaml.safe_load(prompt_file) - prompt_file.close() - terminal_mode_prompt = prompt["terminal_mode"] - - ssh = SshClient(target_vnf["floating_ip"], - target_vnf["user"], - target_vnf["pass"]) - - result = ssh.connect(self.ssh_connect_timeout, - self.ssh_connect_retry_count) - if not result: - return False, res_dict_data_list - - checker = Checker() - - res_table = prettytable.PrettyTable( - header_style='upper', padding_width=5, - field_names=['test item', 'result']) - - status = True - res_data_list = [] - for check_rule_file_path in check_rule_file_path_list: - (check_rule_dir, check_rule_file) = os.path.split( - check_rule_file_path) - check_rules = checker.load_check_rule(check_rule_dir, - check_rule_file, - cmd_input_param) - (res, res_data) = self.vm_controller.command_execute( - ssh, - check_rules["command"], - terminal_mode_prompt) - res_data_list.append(res_data) - if not res: - status = False - break - - (res, res_dict_data) = checker.regexp_information(res_data, - check_rules) - res_dict_data_list.append(res_dict_data) - res_table.add_row([res_dict_data["test_name"], - res_dict_data["result"]]) - if not res: - status = False - - time.sleep(self.cmd_wait) - - ssh.close() - - self.logger.info("Test result:\n\n%s\n", res_table.get_string()) - - self.output_check_result_detail_data(res_data_list) - - return status, res_dict_data_list - - def output_check_result_detail_data(self, res_data_list): - for res_data in res_data_list: - self.logger.debug(res_data) |