functest/opnfv_tests/vnf/router/vrouter_base.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 testing base class module"""
  13
  14 import datetime
  15 import json
  16 import logging
  17 import os
  18 import time
  19
  20 import pkg_resources
  21
  22 from functest.opnfv_tests.vnf.router.test_controller import function_test_exec
  23
  24 __author__ = "Shuya Nakama <shuya.nakama@okinawaopenlabs.org>"
  25
  26
  27 class VrouterOnBoardingBase(object):
  28     """vrouter testing base class"""
  29
  30     def __init__(self, util, util_info):
  31         self.logger = logging.getLogger(__name__)
  32         self.case_dir = pkg_resources.resource_filename(
  33             'functest', 'opnfv_tests/vnf/router')
  34         self.util = util
  35         self.util_info = util_info
  36         self.vnf_list = []
  37
  38     def test_vnf(self):
  39         """vrouter test execution"""
  40         result = False
  41         test_result_data_list = []
  42         test_scenario_file_path = os.path.join(self.case_dir,
  43                                                self.util.test_scenario_yaml)
  44         test_scenario_list = self.util.get_test_scenario(
  45             test_scenario_file_path)
  46         for test_scenario in test_scenario_list:
  47             if test_scenario["test_type"] == "function_test":
  48                 function_test_list = test_scenario["function_test_list"]
  49                 for function_test in function_test_list:
  50                     test_list = function_test["test_list"]
  51                     target_vnf_name = function_test["target_vnf_name"]
  52                     for test_info in test_list:
  53                         self.logger.info(
  54                             "%s %s test.", test_info["protocol"],
  55                             test_info["test_kind"])
  56                         (result, result_data) = self.function_test_vrouter(
  57                             target_vnf_name, test_info)
  58                         test_result_data_list.append(result_data)
  59                         if not result:
  60                             break
  61
  62         self.util.request_vm_delete(self.vnf_list)
  63
  64         test_result_data = json.dumps(test_result_data_list, indent=4)
  65
  66         return result, test_result_data
  67
  68     def function_test_vrouter(self, target_vnf_name, test_info):
  69         """function test execution"""
  70
  71         test_protocol = test_info["protocol"]
  72         test_list = test_info[test_protocol]
  73
  74         vnf_info_list = self.get_vnf_info_list(target_vnf_name)
  75         self.vnf_list = vnf_info_list
  76
  77         target_vnf = self.util.get_target_vnf(vnf_info_list)
  78
  79         reference_vnf_list = self.util.get_reference_vnf_list(vnf_info_list)
  80
  81         test_exec = function_test_exec.FunctionTestExec(self.util_info)
  82
  83         # start test
  84         start_time_ts = time.time()
  85         self.logger.info("vRouter test Start Time:'%s'", (
  86             datetime.datetime.fromtimestamp(start_time_ts).strftime(
  87                 '%Y-%m-%d %H:%M:%S')))
  88
  89         (result, test_result_data) = test_exec.run(target_vnf,
  90                                                    reference_vnf_list,
  91                                                    test_info,
  92                                                    test_list)
  93
  94         end_time_ts = time.time()
  95         duration = round(end_time_ts - start_time_ts, 1)
  96         self.logger.info("vRouter test duration :'%s'", duration)
  97
  98         return result, test_result_data
  99
 100     def get_vnf_info_list(self, target_vnf_name):
 101         return self.util.get_vnf_info_list(
 102             self.util_info["cfy_manager_ip"],
 103             self.util_info["deployment_name"],
 104             target_vnf_name)