1#!/usr/bin/env python3 2# coding=utf-8 3 4# 5# Copyright (c) Huawei Device Co., Ltd. 2025. All right reserved. 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 os 20from typing import Dict, List, Union 21from xml.etree import ElementTree 22 23from _core.constants import Cluster, ConfigConst 24from _core.error import ErrorMessage 25from xdevice import platform_logger 26from ..models import TaskInfo 27from ..runner import Runner 28 29LOG = platform_logger("Worker") 30 31 32def new_element(tag: str, attrib: dict = None, text: str = None): 33 """ 34 tag : element's tag 35 attrib: element's attribute 36 text : element's text 37 """ 38 if attrib is None: 39 attrib = {} 40 ele = ElementTree.Element(tag, attrib) 41 ele.text = text 42 return ele 43 44 45def element_tostring(element: ElementTree.Element): 46 return ElementTree.tostring(element).decode() 47 48 49class UserConfig: 50 51 def __init__(self, devices: List[Dict[str, Union[int, str]]], testcase_dir: str, resource_dir: str, **kwargs): 52 """ 53 devices: test devices 54 testcase_dir: testcase's path 55 resource_dir: resource's path 56 """ 57 self.devices = devices 58 self.testcase_dir = testcase_dir 59 self.resource_dir = resource_dir 60 self.kwargs = kwargs 61 62 self._cfg_root = None 63 64 def _get_config(self, tag: str, clear: bool = True): 65 """获取配置节点，获取失败则新建""" 66 element = self._cfg_root.find(tag) 67 if element is not None: 68 if clear: 69 element.clear() 70 else: 71 element = new_element(tag) 72 self._cfg_root.append(element) 73 return element 74 75 def _get_devices(self): 76 77 def inner_add(_label, _device): 78 if _label not in devices.keys(): 79 devices[_label] = [] 80 devices.get(_label).append(_device) 81 82 devices = {} 83 for dev in self.devices: 84 sn = dev.get("sn") 85 if not sn: 86 continue 87 device = { 88 "ip": "127.0.0.1", 89 "port": "8710", 90 "sn": sn 91 } 92 inner_add("ohos", device) 93 return devices 94 95 @staticmethod 96 def _build_device(label, devices): 97 """build default device""" 98 usb_type = {"ohos": "usb-hdc"} 99 attrib = {"type": usb_type.get(label, ""), "label": label} 100 ele_device = new_element("device", attrib=attrib) 101 for dev in devices: 102 ele_info = new_element("info", attrib=dev) 103 ele_device.append(ele_info) 104 return ele_device 105 106 def _set_root(self): 107 """优先读取测试工程里的配置文件作为模板，否则使用这里代码设定的配置项""" 108 config_path = os.path.join(self._get_project_path(), "config", "user_config.xml") 109 if os.path.exists(config_path): 110 try: 111 root = ElementTree.parse(config_path).getroot() 112 if root.find("taskargs") is not None: 113 self._cfg_root = root 114 return 115 except ElementTree.ParseError as e: 116 LOG.error(e) 117 self._cfg_root = new_element("user_config") 118 119 def _set_environment(self): 120 environment = self._get_config("environment") 121 for label, devices in self._get_devices().items(): 122 ele_device = self._build_device(label, devices) 123 environment.append(ele_device) 124 125 def _set_testcases(self): 126 ele_dir = new_element(ConfigConst.tag_dir, text=self.testcase_dir) 127 testcases = self._get_config("testcases") 128 testcases.append(ele_dir) 129 130 def _set_resource(self): 131 ele_dir = new_element(ConfigConst.tag_dir, text=self.resource_dir) 132 resource = self._get_config("resource") 133 resource.append(ele_dir) 134 135 def _set_device_log(self): 136 device_log = self._get_config("devicelog", clear=False) 137 config = { 138 ConfigConst.tag_enable: "ON", 139 ConfigConst.tag_loglevel: "INFO", 140 ConfigConst.tag_dir: "" 141 } 142 for tag, text in config.items(): 143 if device_log.find(tag) is None: 144 device_log.append(new_element(tag, text=text)) 145 146 def _set_log_level(self): 147 log_level = self._get_config(ConfigConst.tag_loglevel) 148 log_level.text = "DEBUG" 149 150 def _get_project_path(self): 151 case_dir = self.testcase_dir 152 return os.path.dirname(case_dir) if case_dir.endswith("testcases") else case_dir 153 154 def generation(self): 155 self._set_root() 156 self._set_environment() 157 self._set_testcases() 158 self._set_resource() 159 self._set_device_log() 160 self._set_log_level() 161 return "".join([line.strip() for line in element_tostring(self._cfg_root).split("\n")]) 162 163 164class WorkerRunner(Runner): 165 166 def __init__(self, task_info: dict): 167 super().__init__(TaskInfo(**task_info)) 168 169 self.report_path = os.path.join(Cluster.report_root_path, self.task_id, self.task_info.block_id) 170 171 def run(self): 172 os.makedirs(self.report_path, exist_ok=True) 173 try: 174 devices = self.task_info.devices 175 if not devices: 176 raise Exception(ErrorMessage.Cluster.Code_0104034) 177 self.prepare_project() 178 testcase_path = os.path.join(self.project_path, "testcases") 179 if not os.path.exists(testcase_path): 180 testcase_path = self.project_path 181 resource_path = os.path.join(self.project_path, "resource") 182 if not os.path.exists(resource_path): 183 resource_path = self.project_path 184 case_list = ";".join(self.case_names) 185 env_cfg = UserConfig(devices, testcase_path, resource_path).generation() 186 command = f"run -l {case_list} -env {env_cfg} -rp {self.report_path}" 187 LOG.info(f"run command: {command}") 188 from xdevice.__main__ import main_process 189 main_process(command) 190 except Exception as e: 191 LOG.error(e) 192 self.mark_cases_error(self.case_names, str(e)) 193