functest/opnfv_tests/vnf/router/vnf_controller/vnf_controller.py

   1 #!/usr/bin/env python
   2
   3 # Copyright (c) 2017 Okinawa Open Laboratory and others.
   4 #
   5 # All rights reserved. This program and the accompanying materials
   6 # are made available under the terms of the Apache License, Version 2.0
   7 # which accompanies this distribution, and is available at
   8 # http://www.apache.org/licenses/LICENSE-2.0
   9
  10 # pylint: disable=missing-docstring
  11
  12 """vrouter controll module"""
  13
  14 import logging
  15 import os
  16 import time
  17 import yaml
  18
  19 import prettytable
  20
  21 from functest.opnfv_tests.vnf.router.utilvnf import Utilvnf
  22 from functest.opnfv_tests.vnf.router.vnf_controller.checker import Checker
  23 from functest.opnfv_tests.vnf.router.vnf_controller.ssh_client import (
  24     SshClient)
  25 from functest.opnfv_tests.vnf.router.vnf_controller.vm_controller import (
  26     VmController)
  27
  28
  29 class VnfController():
  30     """vrouter controll class"""
  31
  32     logger = logging.getLogger(__name__)
  33
  34     def __init__(self, util_info):
  35         self.logger.debug("init vnf controller")
  36         self.util = Utilvnf()
  37         self.vm_controller = VmController(util_info)
  38
  39         with open(self.util.test_env_config_yaml) as file_fd:
  40             test_env_config_yaml = yaml.safe_load(file_fd)
  41         file_fd.close()
  42
  43         self.cmd_wait = test_env_config_yaml.get("general").get("command_wait")
  44         self.ssh_connect_timeout = test_env_config_yaml.get("general").get(
  45             "ssh_connect_timeout")
  46         self.ssh_connect_retry_count = test_env_config_yaml.get("general").get(
  47             "ssh_connect_retry_count")
  48
  49     def config_vnf(self, source_vnf, destination_vnf, test_cmd_file_path,
  50                    parameter_file_path, prompt_file_path):
  51         # pylint: disable=too-many-arguments
  52         parameter_file = open(parameter_file_path,
  53                               'r')
  54         cmd_input_param = yaml.safe_load(parameter_file)
  55         parameter_file.close()
  56
  57         cmd_input_param["macaddress"] = source_vnf["data_plane_network_mac"]
  58         cmd_input_param["source_ip"] = source_vnf["data_plane_network_ip"]
  59         cmd_input_param["destination_ip"] = destination_vnf[
  60             "data_plane_network_ip"]
  61
  62         return self.vm_controller.config_vm(source_vnf,
  63                                             test_cmd_file_path,
  64                                             cmd_input_param,
  65                                             prompt_file_path)
  66
  67     def result_check(self, target_vnf, reference_vnf,
  68                      check_rule_file_path_list, parameter_file_path,
  69                      prompt_file_path):
  70         # pylint: disable=too-many-arguments,too-many-locals
  71
  72         res_dict_data_list = []
  73
  74         parameter_file = open(parameter_file_path,
  75                               'r')
  76         cmd_input_param = yaml.safe_load(parameter_file)
  77         parameter_file.close()
  78
  79         cmd_input_param["source_ip"] = target_vnf["data_plane_network_ip"]
  80         cmd_input_param["destination_ip"] = reference_vnf[
  81             "data_plane_network_ip"]
  82
  83         prompt_file = open(prompt_file_path,
  84                            'r')
  85         prompt = yaml.safe_load(prompt_file)
  86         prompt_file.close()
  87         terminal_mode_prompt = prompt["terminal_mode"]
  88
  89         ssh = SshClient(target_vnf["floating_ip"],
  90                         target_vnf["user"],
  91                         target_vnf["pass"])
  92
  93         result = ssh.connect(self.ssh_connect_timeout,
  94                              self.ssh_connect_retry_count)
  95         if not result:
  96             return False, res_dict_data_list
  97
  98         checker = Checker()
  99
 100         res_table = prettytable.PrettyTable(
 101             header_style='upper', padding_width=5,
 102             field_names=['test item', 'result'])
 103
 104         status = True
 105         res_data_list = []
 106         for check_rule_file_path in check_rule_file_path_list:
 107             (check_rule_dir, check_rule_file) = os.path.split(
 108                 check_rule_file_path)
 109             check_rules = checker.load_check_rule(check_rule_dir,
 110                                                   check_rule_file,
 111                                                   cmd_input_param)
 112             (res, res_data) = self.vm_controller.command_execute(
 113                 ssh,
 114                 check_rules["command"],
 115                 terminal_mode_prompt)
 116             res_data_list.append(res_data)
 117             if not res:
 118                 status = False
 119                 break
 120
 121             (res, res_dict_data) = checker.regexp_information(res_data,
 122                                                               check_rules)
 123             res_dict_data_list.append(res_dict_data)
 124             res_table.add_row([res_dict_data["test_name"],
 125                                res_dict_data["result"]])
 126             if not res:
 127                 status = False
 128
 129             time.sleep(self.cmd_wait)
 130
 131         ssh.close()
 132
 133         self.logger.info("Test result:\n\n%s\n", res_table.get_string())
 134
 135         self.output_check_result_detail_data(res_data_list)
 136
 137         return status, res_dict_data_list
 138
 139     def output_check_result_detail_data(self, res_data_list):
 140         for res_data in res_data_list:
 141             self.logger.debug(res_data)