1#!/usr/bin/env python3 2# coding=utf-8 3 4# 5# Copyright (c) 2020-2021 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 _core.constants import HostDrivenTestType 24from _core.constants import TestExecType 25from _core.constants import ModeType 26from _core.constants import DeviceLabelType 27from _core.driver.drivers_lite import init_remote_server 28from _core.exception import DeviceError 29from _core.exception import LiteDeviceError 30from _core.exception import ParamError 31from _core.exception import ReportException 32from _core.exception import ExecuteTerminate 33from _core.interface import IDriver 34from _core.logger import platform_logger 35from _core.plugin import Plugin 36from _core.testkit.json_parser import JsonParser 37from _core.utils import get_config_value 38from _core.utils import do_module_kit_setup 39from _core.utils import do_module_kit_teardown 40from _core.utils import get_filename_extension 41from _core.utils import get_file_absolute_path 42from _core.utils import get_kit_instances 43from _core.utils import check_result_report 44from _core.utils import check_mode 45from _core.report.suite_reporter import SuiteReporter 46 47LOG = platform_logger("DeviceTest") 48PY_SUFFIX = ".py" 49PYD_SUFFIX = ".pyd" 50 51 52@Plugin(type=Plugin.DRIVER, id=HostDrivenTestType.device_test) 53class DeviceTestDriver(IDriver): 54 """ 55 DeviceTest is a Test that runs a host-driven test on given devices. 56 """ 57 # test driver config 58 config = None 59 result = "" 60 error_message = "" 61 py_file = "" 62 63 def __init__(self): 64 self.linux_host = "" 65 self.linux_directory = "" 66 67 def __check_environment__(self, device_options): 68 pass 69 70 def __check_config__(self, config=None): 71 pass 72 73 def __init_nfs_server__(self, request=None): 74 return init_remote_server(self, request) 75 76 def __execute__(self, request): 77 try: 78 # set self.config 79 self.config = request.config 80 self.config.tmp_id = str(request.uuid) 81 self.config.tmp_folder = os.path.join(self.config.report_path, 82 "temp") 83 self.config.devices = request.get_devices() 84 if request.get("exectype") == TestExecType.device_test and \ 85 not self.config.devices: 86 LOG.error("no device", error_no="00104") 87 raise ParamError("Load Error[00104]", error_no="00104") 88 89 # get source, json config and kits 90 if request.get_config_file(): 91 source = request.get_config_file() 92 LOG.debug("Test config file path: %s" % source) 93 else: 94 source = request.get_source_string() 95 LOG.debug("Test String: %s" % source) 96 97 if not source: 98 LOG.error("no config file found for '%s'" % 99 request.get_source_file(), error_no="00102") 100 raise ParamError("Load Error(00102)", error_no="00102") 101 102 json_config = JsonParser(source) 103 kits = get_kit_instances(json_config, request.config.resource_path, 104 request.config.testcases_path) 105 106 # create tmp folder 107 test_name = request.get_module_name() 108 tmp_sub_folder = self._create_tmp_folder(request) 109 self.result = "%s.xml" % os.path.join(tmp_sub_folder, test_name) 110 111 # set configs keys 112 configs = self._set_configs(json_config, kits, request, 113 tmp_sub_folder) 114 115 # get test list 116 test_list = self._get_test_list(json_config, request, source) 117 if not test_list: 118 raise ParamError("no test list to run") 119 120 self._run_devicetest(configs, test_list) 121 except (ReportException, ModuleNotFoundError, ExecuteTerminate, 122 SyntaxError, ValueError, AttributeError, TypeError, 123 KeyboardInterrupt, ParamError, DeviceError, LiteDeviceError) \ 124 as exception: 125 error_no = getattr(exception, "error_no", "00000") 126 LOG.exception(exception, exc_info=False, error_no=error_no) 127 self.error_message = exception 128 129 finally: 130 self._handle_finally(request) 131 132 def _get_test_list(self, json_config, request, source): 133 test_list = get_config_value('py_file', json_config.get_driver(), 134 is_list=True) 135 if str(request.root.source.source_file).endswith(PYD_SUFFIX) or \ 136 str(request.root.source.source_file).endswith(PY_SUFFIX): 137 test_list = [request.root.source.source_file] 138 139 if not test_list and os.path.exists(source): 140 test_list = _get_dict_test_list(os.path.dirname(source)) 141 142 # check test list 143 testcase = request.get("testcase") 144 testcase_list = [] 145 if testcase: 146 testcase_list = str(testcase).split(";") 147 148 checked_test_list = [] 149 for index, test in enumerate(test_list): 150 if not os.path.exists(test): 151 try: 152 absolute_file = get_file_absolute_path(test, [ 153 self.config.resource_path, self.config.testcases_path]) 154 except ParamError as error: 155 LOG.error(error, error_no=error.error_no) 156 continue 157 else: 158 absolute_file = test 159 160 file_name = get_filename_extension(absolute_file)[0] 161 if not testcase_list or file_name in testcase_list: 162 checked_test_list.append(absolute_file) 163 else: 164 LOG.info("test '%s' is ignored", absolute_file) 165 if checked_test_list: 166 LOG.info("test list: {}".format(checked_test_list)) 167 else: 168 LOG.error("no test list found", error_no="00109") 169 raise ParamError("Load Error(00109)", error_no="00109") 170 return checked_test_list 171 172 def _set_configs(self, json_config, kits, request, tmp_sub_folder): 173 configs = dict() 174 configs["testargs"] = self.config.testargs or {} 175 configs["testcases_path"] = self.config.testcases_path or "" 176 configs["request"] = request 177 configs["test_name"] = request.get_module_name() 178 configs["report_path"] = tmp_sub_folder 179 configs["execute"] = get_config_value( 180 'execute', json_config.get_driver(), False) 181 182 for device in self.config.devices: 183 do_module_kit_setup(request, kits) 184 if device.label == DeviceLabelType.ipcamera: 185 # add extra keys to configs for ipcamera device 186 self.__init_nfs_server__(request=request) 187 configs["linux_host"] = self.linux_host 188 configs["linux_directory"] = self.linux_directory 189 configs["kits"] = kits 190 191 return configs 192 193 def _handle_finally(self, request): 194 from xdevice import Scheduler 195 196 # do kit teardown 197 do_module_kit_teardown(request) 198 199 # close device connect 200 for device in self.config.devices: 201 if device.label == DeviceLabelType.ipcamera or device.label == \ 202 DeviceLabelType.watch_gt: 203 device.close() 204 if device.label == DeviceLabelType.phone: 205 device.close() 206 207 # check result report 208 report_name = request.root.source.test_name if \ 209 not request.root.source.test_name.startswith("{") \ 210 else "report" 211 module_name = request.get_module_name() 212 if Scheduler.mode != ModeType.decc: 213 self.result = check_result_report( 214 request.config.report_path, self.result, self.error_message, 215 report_name, module_name) 216 else: 217 tmp_list = copy.copy(SuiteReporter.get_report_result()) 218 if self.result not in [report_path for report_path, _ in tmp_list]: 219 if not self.error_message: 220 self.error_message = "Case not execute[01205]" 221 self.result = check_result_report( 222 request.config.report_path, self.result, 223 self.error_message, report_name, module_name) 224 225 def _create_tmp_folder(self, request): 226 if request.root.source.source_file.strip(): 227 folder_name = "task_%s_%s" % (self.config.tmp_id, 228 request.root.source.test_name) 229 else: 230 folder_name = "task_%s_report" % self.config.tmp_id 231 232 tmp_sub_folder = os.path.join(self.config.tmp_folder, folder_name) 233 os.makedirs(tmp_sub_folder, exist_ok=True) 234 return tmp_sub_folder 235 236 def _run_devicetest(self, configs, test_list): 237 from xdevice import Variables 238 239 # insert paths for loading _devicetest module and testcases 240 devicetest_module = os.path.join(Variables.modules_dir, "_devicetest") 241 if os.path.exists(devicetest_module): 242 sys.path.insert(1, devicetest_module) 243 if configs["testcases_path"]: 244 sys.path.insert(1, configs["testcases_path"]) 245 246 # apply data to devicetest module about resource path 247 request = configs.get('request', None) 248 if request: 249 sys.ecotest_resource_path = request.config.resource_path 250 # run devicetest 251 from _devicetest.devicetest.main import DeviceTest 252 device_test = DeviceTest(test_list=test_list, configs=configs, 253 devices=self.config.devices, log=LOG) 254 device_test.run() 255 256 def __result__(self): 257 if check_mode(ModeType.decc): 258 return self.result 259 return self.result if os.path.exists(self.result) else "" 260 261 262def _get_dict_test_list(module_path): 263 test_list = [] 264 for root, _, files in os.walk(module_path): 265 for _file in files: 266 if _file.endswith(".py") or _file.endswith(".pyd"): 267 test_list.append(os.path.join(root, _file)) 268 return test_list 269