functest/api/resources/v1/testcases.py

   1 #!/usr/bin/env python
   2
   3 # Copyright (c) 2017 Huawei Technologies Co.,Ltd 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 """
  11 Resources to handle testcase related requests
  12 """
  13
  14 import ConfigParser
  15 import logging
  16 import os
  17 import re
  18 import socket
  19 import uuid
  20
  21 from flask import jsonify
  22 from flasgger.utils import swag_from
  23 import pkg_resources
  24
  25 from functest.api.base import ApiResource
  26 from functest.api.common import api_utils, thread
  27 from functest.cli.commands.cli_testcase import Testcase
  28 from functest.api.database.v1.handlers import TasksHandler
  29 from functest.utils import config
  30 from functest.utils import env
  31 import functest.utils.functest_utils as ft_utils
  32
  33 LOGGER = logging.getLogger(__name__)
  34
  35 ADDRESS = socket.gethostbyname(socket.gethostname())
  36 ENDPOINT_TESTCASES = ('http://{}:5000/api/v1/functest/testcases'
  37                       .format(ADDRESS))
  38
  39
  40 class V1Testcases(ApiResource):
  41     """ V1Testcases Resource class"""
  42
  43     @swag_from(pkg_resources.resource_filename(
  44         'functest', 'api/swagger/testcases.yaml'))
  45     def get(self):  # pylint: disable=no-self-use
  46         """ GET all testcases """
  47         testcases_list = Testcase().list()
  48         result = {'testcases': re.split(' |\n ', testcases_list)[1:]}
  49         return jsonify(result)
  50
  51
  52 class V1Testcase(ApiResource):
  53     """ V1Testcase Resource class"""
  54
  55     @swag_from(
  56         pkg_resources.resource_filename('functest',
  57                                         'api/swagger/testcase.yaml'),
  58         endpoint='{0}/<testcase_name>'.format(ENDPOINT_TESTCASES))
  59     def get(self, testcase_name):  # pylint: disable=no-self-use
  60         """ GET the info of one testcase"""
  61         testcase = Testcase().show(testcase_name)
  62         if not testcase:
  63             return api_utils.result_handler(
  64                 status=1,
  65                 data="The test case '%s' does not exist or is not supported"
  66                 % testcase_name)
  67
  68         testcase_info = api_utils.change_obj_to_dict(testcase)
  69         dependency_dict = api_utils.change_obj_to_dict(
  70             testcase_info.get('dependency'))
  71         testcase_info.pop('name')
  72         testcase_info.pop('dependency')
  73         result = {'testcase': testcase_name}
  74         result.update(testcase_info)
  75         result.update({'dependency': dependency_dict})
  76         return jsonify(result)
  77
  78     @swag_from(
  79         pkg_resources.resource_filename('functest',
  80                                         'api/swagger/testcase_run.yaml'),
  81         endpoint='{0}/action'.format(ENDPOINT_TESTCASES))
  82     def post(self):
  83         """ Used to handle post request """
  84         return self._dispatch_post()
  85
  86     def run_test_case(self, args):
  87         """ Run a testcase """
  88         try:
  89             case_name = args['testcase']
  90         except KeyError:
  91             return api_utils.result_handler(
  92                 status=1, data='testcase name must be provided')
  93
  94         testcase = Testcase().show(case_name)
  95         if not testcase:
  96             return api_utils.result_handler(
  97                 status=1,
  98                 data="The test case '%s' does not exist or is not supported"
  99                 % case_name)
 100
 101         task_id = str(uuid.uuid4())
 102
 103         task_args = {'testcase': case_name, 'task_id': task_id}
 104
 105         task_args.update(args.get('opts', {}))
 106
 107         task_thread = thread.TaskThread(self._run, task_args, TasksHandler())
 108         task_thread.start()
 109
 110         result = {'testcase': case_name, 'task_id': task_id}
 111         return jsonify({'result': result})
 112
 113     def _run(self, args):  # pylint: disable=no-self-use
 114         """ The built_in function to run a test case """
 115
 116         case_name = args.get('testcase')
 117         self._update_logging_ini(args.get('task_id'))
 118
 119         try:
 120             cmd = "run_tests -t {}".format(case_name)
 121             runner = ft_utils.execute_command(cmd)
 122         except Exception:  # pylint: disable=broad-except
 123             result = 'FAIL'
 124             LOGGER.exception("Running test case %s failed!", case_name)
 125         if runner == os.EX_OK:
 126             result = 'PASS'
 127         else:
 128             result = 'FAIL'
 129
 130         env_info = {
 131             'installer': env.get('INSTALLER_TYPE'),
 132             'scenario': env.get('DEPLOY_SCENARIO'),
 133             'build_tag': env.get('BUILD_TAG'),
 134             'ci_loop': env.get('CI_LOOP')
 135         }
 136         result = {
 137             'task_id': args.get('task_id'),
 138             'testcase': case_name,
 139             'env_info': env_info,
 140             'result': result
 141         }
 142
 143         return {'result': result}
 144
 145     def _update_logging_ini(self, task_id):  # pylint: disable=no-self-use
 146         """ Update the log file for each task"""
 147         rconfig = ConfigParser.RawConfigParser()
 148         rconfig.read(
 149             pkg_resources.resource_filename('xtesting', 'ci/logging.ini'))
 150         log_path = os.path.join(getattr(config.CONF, 'dir_results'),
 151                                 '{}.log'.format(task_id))
 152         rconfig.set('handler_file', 'args', '("{}",)'.format(log_path))
 153
 154         with open(
 155             pkg_resources.resource_filename(
 156                 'xtesting', 'ci/logging.ini'), 'wb') as configfile:
 157             rconfig.write(configfile)