aboutsummaryrefslogtreecommitdiffstats
path: root/functest/opnfv_tests/vnf/router/vnf_controller
diff options
context:
space:
mode:
authorCédric Ollivier <cedric.ollivier@orange.com>2018-02-27 14:25:49 +0100
committerCédric Ollivier <cedric.ollivier@orange.com>2018-02-27 14:27:57 +0100
commitbaa8f2d5f67d45e5761f92cb93fe22050f08d0fe (patch)
tree05ddb33dc893cad35369b3286db944eac79ffe4d /functest/opnfv_tests/vnf/router/vnf_controller
parent53cd7f8176c996014decb7311d9f546f6b8f2497 (diff)
Clean all OpenStack related modules
Xtesting is only focused on the framework and entry points. Change-Id: I1a4146ed8519438b13810a20ddf1140c35bb6ecd Signed-off-by: Cédric Ollivier <cedric.ollivier@orange.com>
Diffstat (limited to 'functest/opnfv_tests/vnf/router/vnf_controller')
-rw-r--r--functest/opnfv_tests/vnf/router/vnf_controller/__init__.py0
-rw-r--r--functest/opnfv_tests/vnf/router/vnf_controller/checker.py63
-rw-r--r--functest/opnfv_tests/vnf/router/vnf_controller/command_generator.py38
-rw-r--r--functest/opnfv_tests/vnf/router/vnf_controller/ssh_client.py135
-rw-r--r--functest/opnfv_tests/vnf/router/vnf_controller/vm_controller.py146
-rw-r--r--functest/opnfv_tests/vnf/router/vnf_controller/vnf_controller.py141
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)