1#!/usr/bin/env python3 2# coding=utf-8 3 4# 5# Copyright (c) 2022 Huawei Device Co., Ltd. 6# Licensed under the Apache License, Version 2.0 (the "License"); 7# you may not use this file except in compliance with the License. 8# You may obtain a copy of the License at 9# 10# http://www.apache.org/licenses/LICENSE-2.0 11# 12# Unless required by applicable law or agreed to in writing, software 13# distributed under the License is distributed on an "AS IS" BASIS, 14# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 15# See the License for the specific language governing permissions and 16# limitations under the License. 17# 18 19import copy 20import os 21import sys 22 23from xdevice import HostDrivenTestType 24from xdevice import ModeType 25from xdevice import DeviceError 26from xdevice import LiteDeviceError 27from xdevice import ParamError 28from xdevice import ReportException 29from xdevice import ExecuteTerminate 30from xdevice import IDriver 31from xdevice import platform_logger 32from xdevice import Plugin 33from xdevice import JsonParser 34from xdevice import get_config_value 35from xdevice import do_module_kit_teardown 36from xdevice import get_filename_extension 37from xdevice import get_file_absolute_path 38from xdevice import get_kit_instances 39from xdevice import check_result_report 40from xdevice import check_mode 41from xdevice import SuiteReporter 42 43LOG = platform_logger("WindowsTest") 44PY_SUFFIX = ".py" 45PYD_SUFFIX = ".pyd" 46 47 48@Plugin(type=Plugin.DRIVER, id=HostDrivenTestType.windows_test) 49class WindowsTestDriver(IDriver): 50 """ 51 DeviceTest is a Test that runs a host-driven test on given devices. 52 """ 53 # test driver config 54 config = None 55 result = "" 56 error_message = "" 57 py_file = "" 58 59 def __init__(self): 60 self.linux_host = "" 61 self.linux_directory = "" 62 self.hilog_file_pipes = [] 63 64 def __check_environment__(self, device_options): 65 pass 66 67 def __check_config__(self, config=None): 68 pass 69 70 def __execute__(self, request): 71 try: 72 # set self.config 73 self.config = request.config 74 # get source, json config and kits 75 if request.get_config_file(): 76 source = request.get_config_file() 77 LOG.debug("Test config file path: %s" % source) 78 else: 79 source = request.get_source_string() 80 LOG.debug("Test String: %s" % source) 81 82 if not source: 83 LOG.error("No config file found for '%s'" % 84 request.get_source_file(), error_no="00102") 85 raise ParamError("Load Error(00102)", error_no="00102") 86 87 json_config = JsonParser(source) 88 kits = get_kit_instances(json_config, request.config.resource_path, 89 request.config.testcases_path) 90 91 # create tmp folder 92 test_name = request.get_module_name() 93 self.result = os.path.join(request.config.report_path, "result", "%s.xml" % test_name) 94 95 # set configs keys 96 configs = self._set_configs(json_config, kits, request) 97 98 # get test list 99 test_list = self._get_test_list(json_config, request, source) 100 if not test_list: 101 raise ParamError("no test list to run") 102 self._run_devicetest(configs, test_list) 103 except (ReportException, ModuleNotFoundError, ExecuteTerminate, 104 SyntaxError, ValueError, AttributeError, TypeError, 105 KeyboardInterrupt, ParamError, DeviceError, LiteDeviceError) \ 106 as exception: 107 error_no = getattr(exception, "error_no", "00000") 108 LOG.exception(exception, exc_info=False, error_no=error_no) 109 self.error_message = exception 110 111 finally: 112 self._handle_finally(request) 113 114 def _get_test_list(self, json_config, request, source): 115 test_list = get_config_value('py_file', json_config.get_driver(), 116 is_list=True) 117 if str(request.root.source.source_file).endswith(PYD_SUFFIX) or \ 118 str(request.root.source.source_file).endswith(PY_SUFFIX): 119 test_list = [request.root.source.source_file] 120 121 if not test_list and os.path.exists(source): 122 test_list = _get_dict_test_list(os.path.dirname(source)) 123 124 # check test list 125 testcase = request.get("testcase") 126 testcase_list = [] 127 if testcase: 128 testcase_list = str(testcase).split(";") 129 130 checked_test_list = [] 131 for _, test in enumerate(test_list): 132 if not os.path.exists(test): 133 try: 134 absolute_file = get_file_absolute_path(test, [ 135 self.config.resource_path, self.config.testcases_path]) 136 except ParamError as error: 137 LOG.error(error, error_no=error.error_no) 138 continue 139 else: 140 absolute_file = test 141 142 file_name = get_filename_extension(absolute_file)[0] 143 if not testcase_list or file_name in testcase_list: 144 checked_test_list.append(absolute_file) 145 else: 146 LOG.info("Test '%s' is ignored", absolute_file) 147 if checked_test_list: 148 LOG.info("Test list: {}".format(checked_test_list)) 149 else: 150 LOG.error("No test list found", error_no="00109") 151 raise ParamError("Load Error(00109)", error_no="00109") 152 return checked_test_list 153 154 def _set_configs(self, json_config, kits, request): 155 configs = dict() 156 configs["testargs"] = self.config.testargs or {} 157 configs["testcases_path"] = self.config.testcases_path or "" 158 configs["request"] = request 159 configs["test_name"] = request.get_module_name() 160 configs["report_path"] = request.config.report_path 161 configs["execute"] = get_config_value( 162 'execute', json_config.get_driver(), False) 163 return configs 164 165 def _handle_finally(self, request): 166 from xdevice import Scheduler 167 168 # do kit teardown 169 do_module_kit_teardown(request) 170 171 # check result report 172 report_name = request.root.source.test_name if \ 173 not request.root.source.test_name.startswith("{") \ 174 else "report" 175 module_name = request.get_module_name() 176 if Scheduler.mode != ModeType.decc: 177 self.result = check_result_report( 178 request.config.report_path, self.result, self.error_message, 179 report_name, module_name) 180 else: 181 tmp_list = copy.copy(SuiteReporter.get_report_result()) 182 if self.result not in [report_path for report_path, _ in tmp_list]: 183 if not self.error_message: 184 self.error_message = "Case not execute[01205]" 185 self.result = check_result_report( 186 request.config.report_path, self.result, 187 self.error_message, report_name, module_name) 188 189 def _run_devicetest(self, configs, test_list): 190 from xdevice import Variables 191 192 # insert paths for loading _devicetest module and testcases 193 devicetest_module = os.path.join(Variables.modules_dir, "_devicetest") 194 if os.path.exists(devicetest_module): 195 sys.path.insert(1, devicetest_module) 196 if configs["testcases_path"]: 197 sys.path.insert(1, configs["testcases_path"]) 198 199 # apply data to devicetest module about resource path 200 request = configs.get('request', None) 201 if request: 202 sys.ecotest_resource_path = request.config.resource_path 203 204 # run devicetest 205 from devicetest.main import DeviceTest 206 device_test = DeviceTest(test_list=test_list, configs=configs, 207 devices=None, log=LOG) 208 device_test.run() 209 210 def __result__(self): 211 if check_mode(ModeType.decc): 212 return self.result 213 return self.result if os.path.exists(self.result) else "" 214 215 216def _get_dict_test_list(module_path): 217 test_list = [] 218 for root, _, files in os.walk(module_path): 219 for _file in files: 220 if _file.endswith(".py") or _file.endswith(".pyd"): 221 test_list.append(os.path.join(root, _file)) 222 return test_list 223