1#!/usr/bin/env python3 2# coding=utf-8 3 4# 5# Copyright (c) 2020-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 os 20import re 21import random 22import string 23import shutil 24import subprocess 25 26from xdevice import platform_logger 27from xdevice import Plugin 28from xdevice import ParamError 29from xdevice import LiteDeviceError 30from xdevice import ITestKit 31from xdevice import get_config_value 32from xdevice import get_file_absolute_path 33from xdevice import DeviceAllocationState 34from ohos.constants import ComType 35from ohos.constants import CKit 36from ohos.exception import LiteDeviceConnectError 37 38__all__ = ["DeployKit"] 39LOG = platform_logger("KitLite") 40 41RESET_CMD = "0xEF, 0xBE, 0xAD, 0xDE, 0x0C, 0x00, 0x87, 0x78, 0x00, 0x00, " \ 42 "0x61, 0x94" 43 44 45@Plugin(type=Plugin.TEST_KIT, id=CKit.deploy) 46class DeployKit(ITestKit): 47 def __init__(self): 48 self.burn_file = "" 49 self.burn_command = "" 50 self.timeout = "" 51 self.paths = "" 52 53 def __check_config__(self, config): 54 self.timeout = str(int(get_config_value( 55 'timeout', config, is_list=False, default=0)) * 1000) 56 self.burn_file = get_config_value('burn_file', config, is_list=False) 57 burn_command = get_config_value('burn_command', config, is_list=False, 58 default=RESET_CMD) 59 self.burn_command = burn_command.replace(" ", "").split(",") 60 self.paths = get_config_value('paths', config) 61 if self.timeout == "0" or not self.burn_file: 62 msg = "The config for deploy kit is invalid with timeout:{}, " \ 63 "burn_file:{}".format(self.timeout, self.burn_file) 64 raise ParamError(msg, error_no="00108") 65 66 def _reset(self, device): 67 cmd_com = device.device.com_dict.get(ComType.cmd_com) 68 try: 69 cmd_com.connect() 70 cmd_com.execute_command(command='AT+RST={}'.format(self.timeout)) 71 cmd_com.close() 72 except (LiteDeviceConnectError, IOError) as error: 73 device.device_allocation_state = DeviceAllocationState.unusable 74 LOG.error( 75 "The exception {} happened in deploy kit running".format( 76 error), error_no=getattr(error, "error_no", 77 "00000")) 78 raise LiteDeviceError("%s port set_up wifiiot failed" % 79 cmd_com.serial_port, 80 error_no=getattr(error, "error_no", 81 "00000")) 82 finally: 83 if cmd_com: 84 cmd_com.close() 85 86 def _send_file(self, device, source_file): 87 burn_tool_name = "HiBurn.exe" if os.name == "nt" else "HiBurn" 88 burn_tool_path = get_file_absolute_path( 89 os.path.join("tools", burn_tool_name), self.paths) 90 91 deploy_serial_port = device.device.com_dict.get( 92 ComType.deploy_com).serial_port 93 deploy_baudrate = device.device.com_dict.\ 94 get(ComType.deploy_com).baud_rate 95 port_number = re.findall(r'\d+$', deploy_serial_port) 96 if not port_number: 97 raise LiteDeviceError("The config of serial port {} to deploy is " 98 "invalid".format(deploy_serial_port), 99 error_no="00108") 100 new_temp_tool_path = self.copy_file_as_temp(burn_tool_path, 10) 101 cmd = '{} -com:{} -bin:{} -signalbaud:{}' \ 102 .format(new_temp_tool_path, port_number[0], source_file, 103 deploy_baudrate) 104 LOG.info('The running cmd is {}'.format(cmd)) 105 LOG.info('The burn tool is running, please wait..') 106 return_code, out = subprocess.getstatusoutput(cmd) 107 LOG.info( 108 'Deploy kit to execute burn tool finished with return_code: {} ' 109 'output: {}'.format(return_code, out)) 110 os.remove(new_temp_tool_path) 111 if 0 != return_code: 112 device.device_allocation_state = DeviceAllocationState.unusable 113 raise LiteDeviceError("%s port set_up wifiiot failed" % 114 deploy_serial_port, error_no="00402") 115 116 def __setup__(self, device, **kwargs): 117 """ 118 Execute reset command on the device by cmd serial port and then upload 119 patch file by deploy tool. 120 Parameters: 121 device: the instance of LocalController with one or more 122 ComController 123 """ 124 args = kwargs 125 source_file = args.get("source_file", None) 126 self._reset(device) 127 self._send_file(device, source_file) 128 129 def __teardown__(self, device): 130 pass 131 132 def copy_file_as_temp(self, original_file, str_length): 133 """ 134 To obtain a random string with specified length 135 Parameters: 136 original_file : the original file path 137 str_length: the length of random string 138 """ 139 if os.path.isfile(original_file): 140 random_str = random.sample(string.ascii_letters + string.digits, 141 str_length) 142 new_temp_tool_path = '{}_{}{}'.format( 143 os.path.splitext(original_file)[0], "".join(random_str), 144 os.path.splitext(original_file)[1]) 145 return shutil.copyfile(original_file, new_temp_tool_path)