functest/opnfv_tests/vnf/router/test_controller/function_test_exec.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 function test execution module"""
  11
  12 import logging
  13 import time
  14 import yaml
  15
  16 from functest.opnfv_tests.vnf.router.utilvnf import Utilvnf
  17 from functest.opnfv_tests.vnf.router.vnf_controller.vnf_controller import (
  18     VnfController)
  19
  20
  21 class FunctionTestExec(object):
  22     """vrouter function test execution class"""
  23
  24     logger = logging.getLogger(__name__)
  25
  26     def __init__(self, util_info):
  27         self.logger.debug("init test exec")
  28         self.util = Utilvnf()
  29         credentials = util_info["credentials"]
  30         self.vnf_ctrl = VnfController(util_info)
  31
  32         test_cmd_map_file = open(self.util.vnf_data_dir +
  33                                  self.util.opnfv_vnf_data_dir +
  34                                  self.util.command_template_dir +
  35                                  self.util.test_cmd_map_yaml_file,
  36                                  'r')
  37         self.test_cmd_map_yaml = yaml.safe_load(test_cmd_map_file)
  38         test_cmd_map_file.close()
  39
  40         self.util.set_credentials(credentials["username"],
  41                                   credentials["password"],
  42                                   credentials["auth_url"],
  43                                   credentials["tenant_name"],
  44                                   credentials["region_name"])
  45
  46         with open(self.util.test_env_config_yaml) as file_fd:
  47             test_env_config_yaml = yaml.safe_load(file_fd)
  48         file_fd.close()
  49
  50         self.protocol_stable_wait = test_env_config_yaml.get("general").get(
  51             "protocol_stable_wait")
  52
  53     def config_target_vnf(self, target_vnf, reference_vnf, test_kind):
  54         self.logger.debug("Configuration to target vnf")
  55         test_info = self.test_cmd_map_yaml[target_vnf["os_type"]]
  56         test_cmd_file_path = test_info[test_kind]["pre_command_target"]
  57         target_parameter_file_path = test_info[test_kind]["parameter_target"]
  58         prompt_file_path = test_info["prompt"]
  59
  60         return self.vnf_ctrl.config_vnf(target_vnf,
  61                                         reference_vnf,
  62                                         test_cmd_file_path,
  63                                         target_parameter_file_path,
  64                                         prompt_file_path)
  65
  66     def config_reference_vnf(self, target_vnf, reference_vnf, test_kind):
  67         self.logger.debug("Configuration to reference vnf")
  68         test_info = self.test_cmd_map_yaml[reference_vnf["os_type"]]
  69         test_cmd_file_path = test_info[test_kind]["pre_command_reference"]
  70         reference_parameter_file_path = test_info[test_kind][
  71             "parameter_reference"]
  72         prompt_file_path = test_info["prompt"]
  73
  74         return self.vnf_ctrl.config_vnf(reference_vnf,
  75                                         target_vnf,
  76                                         test_cmd_file_path,
  77                                         reference_parameter_file_path,
  78                                         prompt_file_path)
  79
  80     def result_check(self, target_vnf, reference_vnf, test_kind, test_list):
  81         test_info = self.test_cmd_map_yaml[target_vnf["os_type"]]
  82         target_parameter_file_path = test_info[test_kind]["parameter_target"]
  83         prompt_file_path = test_info["prompt"]
  84         check_rule_file_path_list = []
  85
  86         for test in test_list:
  87             check_rule_file_path_list.append(test_info[test_kind][test])
  88
  89         return self.vnf_ctrl.result_check(target_vnf,
  90                                           reference_vnf,
  91                                           check_rule_file_path_list,
  92                                           target_parameter_file_path,
  93                                           prompt_file_path)
  94
  95     def run(self, target_vnf, reference_vnf_list, test_info, test_list):
  96         test_result_data = {}
  97         test_kind = test_info["protocol"]
  98         for reference_vnf in reference_vnf_list:
  99             self.logger.debug("Start config command " +
 100                               target_vnf["vnf_name"] + " and " +
 101                               reference_vnf["vnf_name"])
 102
 103             result = self.config_target_vnf(target_vnf,
 104                                             reference_vnf,
 105                                             test_kind)
 106             if not result:
 107                 return False, test_result_data
 108
 109             result = self.config_reference_vnf(target_vnf,
 110                                                reference_vnf,
 111                                                test_kind)
 112             if not result:
 113                 return False, test_result_data
 114
 115             self.logger.debug("Finish config command.")
 116
 117             self.logger.debug("Waiting for protocol stable.")
 118             time.sleep(self.protocol_stable_wait)
 119
 120             self.logger.debug("Start check method")
 121
 122             (result, res_dict_data_list) = self.result_check(target_vnf,
 123                                                              reference_vnf,
 124                                                              test_kind,
 125                                                              test_list)
 126
 127             test_result_data = {"test_kind": test_info["test_kind"],
 128                                 "protocol": test_info["protocol"],
 129                                 "result": res_dict_data_list}
 130
 131             if not result:
 132                 self.logger.debug("Error check method.")
 133                 return False, test_result_data
 134
 135             self.logger.debug("Finish check method.")
 136
 137         return True, test_result_data