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 """vrouter controll module"""
  11
  12 import logging
  13 import os
  14 import prettytable
  15 import time
  16 import yaml
  17
  18 from functest.opnfv_tests.vnf.router.utilvnf import Utilvnf
  19 from functest.opnfv_tests.vnf.router.vnf_controller.checker import Checker
  20 from functest.opnfv_tests.vnf.router.vnf_controller.ssh_client import (
  21     SshClient)
  22 from functest.opnfv_tests.vnf.router.vnf_controller.vm_controller import (
  23     VmController)
  24
  25
  26 class VnfController(object):
  27     """vrouter controll class"""
  28
  29     logger = logging.getLogger(__name__)
  30
  31     def __init__(self, util_info):
  32         self.logger.debug("init vnf controller")
  33         self.util = Utilvnf()
  34         self.vm_controller = VmController(util_info)
  35
  36         with open(self.util.test_env_config_yaml) as file_fd:
  37             test_env_config_yaml = yaml.safe_load(file_fd)
  38         file_fd.close()
  39
  40         self.cmd_wait = test_env_config_yaml.get("general").get("command_wait")
  41         self.ssh_connect_timeout = test_env_config_yaml.get("general").get(
  42             "ssh_connect_timeout")
  43         self.ssh_connect_retry_count = test_env_config_yaml.get("general").get(
  44             "ssh_connect_retry_count")
  45
  46     def config_vnf(self, source_vnf, destination_vnf, test_cmd_file_path,
  47                    parameter_file_path, prompt_file_path):
  48         parameter_file = open(parameter_file_path,
  49                               'r')
  50         cmd_input_param = yaml.safe_load(parameter_file)
  51         parameter_file.close()
  52
  53         cmd_input_param["macaddress"] = source_vnf["data_plane_network_mac"]
  54         cmd_input_param["source_ip"] = source_vnf["data_plane_network_ip"]
  55         cmd_input_param["destination_ip"] = destination_vnf[
  56             "data_plane_network_ip"]
  57
  58         return self.vm_controller.config_vm(source_vnf,
  59                                             test_cmd_file_path,
  60                                             cmd_input_param,
  61                                             prompt_file_path)
  62
  63     def result_check(self, target_vnf, reference_vnf,
  64                      check_rule_file_path_list, parameter_file_path,
  65                      prompt_file_path):
  66
  67         res_dict_data_list = []
  68
  69         parameter_file = open(parameter_file_path,
  70                               'r')
  71         cmd_input_param = yaml.safe_load(parameter_file)
  72         parameter_file.close()
  73
  74         cmd_input_param["source_ip"] = target_vnf["data_plane_network_ip"]
  75         cmd_input_param["destination_ip"] = reference_vnf[
  76             "data_plane_network_ip"]
  77
  78         prompt_file = open(prompt_file_path,
  79                            'r')
  80         prompt = yaml.safe_load(prompt_file)
  81         prompt_file.close()
  82         terminal_mode_prompt = prompt["terminal_mode"]
  83
  84         ssh = SshClient(target_vnf["floating_ip"],
  85                         target_vnf["user"],
  86                         target_vnf["pass"])
  87
  88         result = ssh.connect(self.ssh_connect_timeout,
  89                              self.ssh_connect_retry_count)
  90         if not result:
  91             return False, res_dict_data_list
  92
  93         checker = Checker()
  94
  95         res_table = prettytable.PrettyTable(
  96                         header_style='upper', padding_width=5,
  97                         field_names=['test item', 'result'])
  98
  99         status = True
 100         res_data_list = []
 101         for check_rule_file_path in check_rule_file_path_list:
 102             (check_rule_dir, check_rule_file) = os.path.split(
 103                 check_rule_file_path)
 104             check_rules = checker.load_check_rule(check_rule_dir,
 105                                                   check_rule_file,
 106                                                   cmd_input_param)
 107             (res, res_data) = self.vm_controller.command_execute(
 108                 ssh,
 109                 check_rules["command"],
 110                 terminal_mode_prompt)
 111             res_data_list.append(res_data)
 112             if not res:
 113                 status = False
 114                 break
 115
 116             (res, res_dict_data) = checker.regexp_information(res_data,
 117                                                               check_rules)
 118             res_dict_data_list.append(res_dict_data)
 119             res_table.add_row([res_dict_data["test_name"],
 120                                res_dict_data["result"]])
 121             if not res:
 122                 status = False
 123
 124             time.sleep(self.cmd_wait)
 125
 126         ssh.close()
 127
 128         self.logger.info("Test result:\n\n%s\n", res_table.get_string())
 129
 130         self.output_check_result_detail_data(res_data_list)
 131
 132         return status, res_dict_data_list
 133
 134     def output_check_result_detail_data(self, res_data_list):
 135         for res_data in res_data_list:
 136             self.logger.debug(res_data)