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 os 20import random 21import re 22import string 23import subprocess 24import shutil 25import platform 26import glob 27import time 28import sys 29from xdevice import Plugin 30from xdevice import platform_logger 31from xdevice import DeviceAllocationState 32from xdevice import ParamError 33from xdevice import ITestKit 34from xdevice import get_config_value 35from xdevice import get_file_absolute_path 36from xdevice import get_local_ip 37from xdevice import FilePermission 38from xdevice import DeviceTestType 39from xdevice import DeviceLabelType 40from xdevice import JsonParser 41from ohos.constants import ComType 42from ohos.constants import CKit 43from ohos.constants import DeviceLiteKernel 44from ohos.drivers.constants import get_nfs_server 45from ohos.error import ErrorMessage 46from ohos.exception import LiteDeviceConnectError 47from ohos.exception import LiteDeviceError 48from ohos.exception import LiteDeviceMountError 49from ohos.utils import parse_strings_key_value 50 51__all__ = ["DeployKit", "MountKit", "RootFsKit", "QueryKit", "LiteShellKit", 52 "LiteAppInstallKit", "DeployToolKit"] 53LOG = platform_logger("KitLite") 54RESET_CMD = "0xEF, 0xBE, 0xAD, 0xDE, 0x0C, 0x00, 0x87, 0x78, 0x00, 0x00, 0x61, 0x94" 55 56 57def copy_file_to_nfs(nfs_config, src): 58 ip = nfs_config.get("ip") 59 port = nfs_config.get("port") 60 nfs_dir = nfs_config.get("dir") 61 username = nfs_config.get("username") 62 password = nfs_config.get("password") 63 is_remote = nfs_config.get("remote", "false") 64 65 dst_file = os.path.join(nfs_dir, os.path.basename(src)) 66 if is_remote.lower() == "true": 67 # copy to remote 68 client = None 69 try: 70 LOG.info(f"Trying to copy the file from {src} to nfs server") 71 import paramiko 72 client = paramiko.Transport(ip, int(port)) 73 client.connect(username=username, password=password) 74 sftp = paramiko.SFTPClient.from_transport(client) 75 sftp.put(localpath=src, remotepath=dst_file) 76 except OSError as e: 77 LOG.warning(f"Copy file to nfs server failed, {e}") 78 finally: 79 if client is not None: 80 client.close() 81 else: 82 # copy to local 83 for count in range(1, 4): 84 LOG.info(f"Trying to copy the file from {src} to nfs server") 85 try: 86 os.remove(dst_file) 87 except (FileNotFoundError, IOError, PermissionError): 88 pass 89 90 try: 91 shutil.copy(src, nfs_dir) 92 except (FileNotFoundError, IOError, PermissionError) as e: 93 LOG.warning(f"Copy file to nfs server failed, {e}") 94 95 if check_server_file(src, nfs_dir): 96 break 97 LOG.info(f"Trying to copy the file from {src} to nfs server {count} times") 98 if count == 3: 99 LOG.error("Copy file to nfs server failed after retry") 100 LOG.debug("Nfs server: {}".format(glob.glob(os.path.join(nfs_dir, '*.*')))) 101 102 103def get_file_from_nfs(nfs_config, src, dst): 104 ip = nfs_config.get("ip") 105 port = nfs_config.get("port") 106 username = nfs_config.get("username") 107 password = nfs_config.get("password") 108 is_remote = nfs_config.get("remote", "false") 109 110 try: 111 if is_remote.lower() == "true": 112 import paramiko 113 client = None 114 try: 115 client = paramiko.Transport(ip, int(port)) 116 client.connect(username=username, password=password) 117 sftp = paramiko.SFTPClient.from_transport(client) 118 sftp.get(remotepath=src, localpath=dst) 119 finally: 120 if client is not None: 121 client.close() 122 else: 123 if os.path.exists(dst): 124 os.remove(dst) 125 if os.path.exists(src): 126 shutil.copy(src, dst) 127 except (FileNotFoundError, IOError) as e: 128 LOG.error(f"Get file from nfs server failed, {e}") 129 130 131def execute_query(device, query, request): 132 if not query: 133 LOG.debug("query bin is none") 134 return 135 if device.device_props: 136 LOG.debug("query bin has been executed") 137 return 138 LOG.debug("execute query bin begins") 139 commands = ["cd /"] 140 if device.__get_device_kernel__() == DeviceLiteKernel.linux_kernel: 141 # query_pth, /storage/test_root/tools/querySmall.bin 142 query_pth = f"/storage{query}" 143 query_dir = os.path.dirname(query_pth) 144 query_bin = os.path.basename(query_pth) 145 commands.append(f"chmod +x {query_pth}") 146 commands.append(f"cd {query_dir}") 147 commands.append("ls") 148 commands.append(f"./{query_bin}") 149 commands.append("ls") 150 else: 151 # query, /test_root/tools/querySmall.bin 152 query_dir = os.path.dirname(query) 153 query_bin = os.path.basename(query) 154 commands.append(f"cd {query_dir}") 155 commands.append("ls") 156 commands.append(f"./{query_bin}") 157 commands.append("ls") 158 159 content = "" 160 for command in commands: 161 output, _, _ = device.execute_command_with_timeout(command=command, timeout=10) 162 LOG.debug(output) 163 # 取bin文件的执行输出 164 if query_bin in command: 165 content = output 166 167 nfs_config = get_nfs_server(request) 168 src = os.path.join(nfs_config.get("dir"), "querySmall.txt") 169 dst = os.path.join(request.config.report_path, "log", "querySmall.txt") 170 get_file_from_nfs(nfs_config, src, dst) 171 if os.path.exists(dst): 172 LOG.debug("querySmall.txt exists") 173 with open(dst, encoding="utf-8") as query_result_f: 174 content = query_result_f.read() 175 os.remove(dst) 176 params = parse_strings_key_value(content) 177 device.update_device_props(params) 178 LOG.debug("execute query bin ends") 179 180 181@Plugin(type=Plugin.TEST_KIT, id=CKit.deploy) 182class DeployKit(ITestKit): 183 def __init__(self): 184 self.burn_file = "" 185 self.burn_command = "" 186 self.timeout = "" 187 self.paths = "" 188 189 def __check_config__(self, config): 190 self.timeout = str(int(get_config_value( 191 'timeout', config, is_list=False, default=0)) // 1000) 192 self.burn_file = get_config_value('burn_file', config, is_list=False) 193 burn_command = get_config_value('burn_command', config, is_list=False, 194 default=RESET_CMD) 195 self.burn_command = burn_command.replace(" ", "").split(",") 196 self.paths = get_config_value('paths', config) 197 if self.timeout == "0" or not self.burn_file: 198 raise ParamError(ErrorMessage.Config.Code_0302007.format(self.timeout, self.burn_file)) 199 200 def _reset(self, device): 201 cmd_com = device.device.com_dict.get(ComType.cmd_com) 202 try: 203 cmd_com.connect() 204 cmd_com.execute_command( 205 command='AT+RST={}'.format(self.timeout)) 206 cmd_com.close() 207 except (LiteDeviceConnectError, IOError) as error: 208 device.device_allocation_state = DeviceAllocationState.unusable 209 LOG.error( 210 "The exception {} happened in deploy kit running".format( 211 error), error_no=getattr(error, "error_no", 212 "00000")) 213 raise LiteDeviceError(ErrorMessage.Device.Code_0303019.format(cmd_com.serial_port)) 214 finally: 215 if cmd_com: 216 cmd_com.close() 217 218 def _send_file(self, device): 219 burn_tool_name = "HiBurn.exe" if os.name == "nt" else "HiBurn" 220 burn_tool_path = get_file_absolute_path( 221 os.path.join("tools", burn_tool_name), self.paths) 222 patch_file = get_file_absolute_path(self.burn_file, self.paths) 223 deploy_serial_port = device.device.com_dict.get( 224 ComType.deploy_com).serial_port 225 deploy_baudrate = device.device.com_dict. \ 226 get(ComType.deploy_com).baud_rate 227 port_number = re.findall(r'\d+$', deploy_serial_port) 228 if not port_number: 229 raise LiteDeviceError(ErrorMessage.Config.Code_0302036.format(deploy_serial_port)) 230 new_temp_tool_path = copy_file_as_temp(burn_tool_path, 10) 231 cmd = '{} -com:{} -bin:{} -signalbaud:{}' \ 232 .format(new_temp_tool_path, port_number[0], patch_file, 233 deploy_baudrate) 234 LOG.info('The running cmd is {}'.format(cmd)) 235 LOG.info('The burn tool is running, please wait..') 236 return_code, out = subprocess.getstatusoutput(cmd) 237 LOG.info( 238 'Deploy kit to execute burn tool finished with return code: {} ' 239 'output: {}'.format(return_code, out)) 240 os.remove(new_temp_tool_path) 241 if 0 != return_code: 242 device.device_allocation_state = DeviceAllocationState.unusable 243 raise LiteDeviceError(ErrorMessage.Device.Code_0303019.format(deploy_serial_port)) 244 245 def __setup__(self, device, **kwargs): 246 """ 247 Execute reset command on the device by cmd serial port and then upload 248 patch file by deploy tool. 249 Parameters: 250 device: the instance of LocalController with one or more 251 ComController 252 """ 253 del kwargs 254 self._reset(device) 255 self._send_file(device) 256 257 def __teardown__(self, device): 258 pass 259 260 261@Plugin(type=Plugin.TEST_KIT, id=CKit.mount) 262class MountKit(ITestKit): 263 def __init__(self): 264 self.remote = None 265 self.paths = "" 266 self.mount_list = [] 267 self.mounted_dir = set() 268 self.server = "" 269 self.file_name_list = [] 270 self.remote_info = None 271 self.hcptest = False 272 self.type_kernel = None 273 274 def __check_config__(self, config): 275 self.remote = get_config_value('server', config, is_list=False) 276 self.paths = get_config_value('paths', config) 277 self.mount_list = get_config_value('mount', config, is_list=True) 278 self.server = get_config_value('server', config, is_list=False, 279 default="NfsServer") 280 self.type_kernel = get_config_value('type_kernel', config, is_list=False) 281 if not self.mount_list: 282 msg = "The config for mount kit is invalid with mount:{}" \ 283 .format(self.mount_list) 284 LOG.error(msg, error_no="00108") 285 raise TypeError(ErrorMessage.Config.Code_0302017) 286 287 def mount_on_board(self, device=None, remote_info=None, case_type="", request=None): 288 """ 289 Init the environment on the device server, e.g. mount the testcases to 290 server 291 292 Parameters: 293 device: DeviceLite, device lite on local or remote 294 remote_info: dict, includes 295 linux_host: str, nfs_server ip 296 linux_directory: str, the directory on the linux 297 is_remote: str, server is remote or not 298 case_type: str, CppTestLite or CTestLite, default value is 299 DeviceTestType.cpp_test_lite 300 301 Returns: 302 True or False, represent init Failed or success 303 """ 304 if not remote_info: 305 raise ParamError(ErrorMessage.Device.Code_0303011) 306 307 linux_host = remote_info.get("ip", "") 308 linux_directory = remote_info.get("dir", "") 309 is_remote = remote_info.get("remote", "false") 310 liteos_commands = ["cd /", "umount device_directory", 311 "mount nfs_ip:nfs_directory device_directory nfs"] 312 linux_commands = ["cd /{}".format("storage"), 313 "umount -f /{}/{}".format("storage", "device_directory"), 314 "toybox mount -t nfs -o nolock,addr=nfs_ip nfs_ip:nfs_directory " 315 "/{}/{}".format("storage", "device_directory"), 316 "chmod 755 -R /{}/{}".format( 317 "storage", "device_directory")] 318 if not linux_host or not linux_directory: 319 raise LiteDeviceMountError(ErrorMessage.Config.Code_0302023) 320 321 commands = [] 322 if device.label == "ipcamera": 323 if self.type_kernel and \ 324 self.type_kernel in [DeviceLiteKernel.linux_kernel, DeviceLiteKernel.lite_kernel]: 325 commands = linux_commands 326 device.__set_device_kernel__(self.type_kernel) 327 else: 328 env_result, status, _ = device.execute_command_with_timeout( 329 command="uname", timeout=1, retry=2) 330 if status: 331 if env_result.find(DeviceLiteKernel.linux_kernel) != -1 or \ 332 env_result.find("Linux") != -1: 333 commands = linux_commands 334 device.__set_device_kernel__(DeviceLiteKernel.linux_kernel) 335 else: 336 commands = liteos_commands 337 device.__set_device_kernel__(DeviceLiteKernel.lite_kernel) 338 else: 339 raise LiteDeviceMountError(ErrorMessage.Config.Code_0302037) 340 341 for mount_file in self.mount_list: 342 target = mount_file.get("target", "/test_root") 343 if target in self.mounted_dir: 344 LOG.debug("%s is mounted" % target) 345 continue 346 if target == "/test_root/tools" and device.device_props: 347 LOG.debug("query bin has been executed, '/test_root/tools' no need to mount again") 348 continue 349 mkdir_on_board(device, target) 350 self.mounted_dir.add(target) 351 352 temp_linux_directory = linux_directory 353 if is_remote.lower() == "false": 354 temp_linux_directory = get_mount_dir(linux_directory) 355 for command in commands: 356 command = command.replace("nfs_ip", linux_host). \ 357 replace("nfs_directory", temp_linux_directory).replace( 358 "device_directory", target).replace("//", "/") 359 timeout = 15 if command.startswith("mount") else 1 360 if command.startswith("mount"): 361 for mount_time in range(1, 4): 362 result, status, _ = device. \ 363 execute_command_with_timeout(command=command, 364 case_type=case_type, 365 timeout=timeout) 366 if status: 367 break 368 if "already mounted" in result: 369 LOG.info("{} is mounted".format(target)) 370 break 371 LOG.info("Mount failed,try " 372 "again {} time".format(mount_time)) 373 if mount_time == 3: 374 raise LiteDeviceMountError(ErrorMessage.Device.Code_0303020) 375 else: 376 result, status, _ = device.execute_command_with_timeout( 377 command=command, case_type=case_type, timeout=timeout) 378 LOG.info('Prepare environment success') 379 execute_query(device, '/test_root/tools/querySmall.bin', request) 380 381 def __setup__(self, device, **kwargs): 382 """ 383 Mount the file to the board by the nfs server. 384 """ 385 LOG.debug("Start mount kit setup") 386 387 request = kwargs.get("request", None) 388 if not request: 389 raise ParamError(ErrorMessage.Config.Code_0302004) 390 self.check_hcp_mode(request) 391 device.connect() 392 393 remote_info = get_nfs_server(request) 394 copy_list = self.copy_to_server(remote_info, request.config.testcases_path) 395 self.mount_on_board(device=device, remote_info=remote_info, 396 case_type=DeviceTestType.cpp_test_lite, request=request) 397 return copy_list 398 399 def copy_to_server(self, remote_info, testcases_dir): 400 file_local_paths = [] 401 # find querySmall.bin 402 query_small_src = "resource/tools/querySmall.bin" 403 try: 404 file_path = get_file_absolute_path(query_small_src, self.paths) 405 file_local_paths.append(file_path) 406 self.mount_list.append({"source": query_small_src, "target": "/test_root/tools"}) 407 except ParamError: 408 LOG.debug("query bin is not found") 409 410 is_query_small_copy_before = False 411 for mount_file in self.mount_list: 412 source = mount_file.get("source") 413 if not source: 414 raise TypeError(ErrorMessage.Config.Code_0302005) 415 if source == query_small_src: 416 is_query_small_copy_before = True 417 if is_query_small_copy_before: 418 LOG.debug("query bin has been copy to nfs server") 419 continue 420 source = source.replace("$testcases/", "").replace("$resources/", "") 421 file_path = get_file_absolute_path(source, self.paths) 422 if os.path.isdir(file_path): 423 for root, _, files in os.walk(file_path): 424 for _file in files: 425 if _file.endswith(".json"): 426 continue 427 file_local_paths.append(os.path.join(root, _file)) 428 else: 429 file_local_paths.append(file_path) 430 431 ip = linux_host = remote_info.get("ip", "") 432 port = remote_info.get("port", "") 433 remote_dir = linux_directory = remote_info.get("dir", "") 434 435 if (str(get_local_ip()) == linux_host) and ( 436 linux_directory == ("/data%s" % testcases_dir)): 437 return [] 438 if not ip or not port or not remote_dir: 439 LOG.warning("Nfs server's ip or port or dir is empty") 440 return [] 441 for _file in file_local_paths: 442 copy_file_to_nfs(remote_info, _file) 443 self.file_name_list.append(os.path.basename(_file)) 444 445 return self.file_name_list 446 447 def __teardown__(self, device): 448 if device.__get_device_kernel__() == DeviceLiteKernel.linux_kernel: 449 device.execute_command_with_timeout(command="cd /storage", 450 timeout=1) 451 for mounted_dir in self.mounted_dir: 452 device.execute_command_with_timeout(command="umount -f " 453 "/storage{}". 454 format(mounted_dir), 455 timeout=2) 456 device.execute_command_with_timeout(command="rm -r /storage{}". 457 format(mounted_dir), 458 timeout=1) 459 else: 460 device.execute_command_with_timeout(command="cd /", timeout=1) 461 for mounted_dir in self.mounted_dir: 462 for mount_time in range(1, 3): 463 result, status, _ = device.execute_command_with_timeout( 464 command="umount {}".format(mounted_dir), 465 timeout=2) 466 if result.find("Resource busy") == -1: 467 device.execute_command_with_timeout(command="rm -r {}".format(mounted_dir), timeout=1) 468 if status: 469 break 470 LOG.info("Umount failed,try " 471 "again {} time".format(mount_time)) 472 time.sleep(1) 473 474 def check_hcp_mode(self, request): 475 config_file = request.root.source.config_file 476 json_config = JsonParser(config_file) 477 role = get_config_value('role', json_config.get_driver(), False) 478 if role: 479 self.hcptest = True 480 481 482def copy_file_as_temp(original_file, str_length): 483 """ 484 To obtain a random string with specified length 485 Parameters: 486 original_file : the original file path 487 str_length: the length of random string 488 """ 489 if os.path.isfile(original_file): 490 random_str = random.sample(string.ascii_letters + string.digits, 491 str_length) 492 new_temp_tool_path = '{}_{}{}'.format( 493 os.path.splitext(original_file)[0], "".join(random_str), 494 os.path.splitext(original_file)[1]) 495 return shutil.copyfile(original_file, new_temp_tool_path) 496 return None 497 498 499def mkdir_on_board(device, dir_path): 500 """ 501 Liteos L1 board don't support mkdir -p 502 Parameters: 503 device : the L1 board 504 dir_path: the dir path to make 505 """ 506 if device.__get_device_kernel__() == DeviceLiteKernel.linux_kernel: 507 device.execute_command_with_timeout(command="cd /storage", timeout=1) 508 else: 509 device.execute_command_with_timeout(command="cd /", timeout=1) 510 for sub_dir in dir_path.split("/"): 511 if sub_dir in ["", "/"]: 512 continue 513 device.execute_command_with_timeout(command="mkdir {}".format(sub_dir), 514 timeout=1) 515 device.execute_command_with_timeout(command="cd {}".format(sub_dir), 516 timeout=1) 517 if device.__get_device_kernel__() == DeviceLiteKernel.linux_kernel: 518 device.execute_command_with_timeout(command="cd /storage", timeout=1) 519 else: 520 device.execute_command_with_timeout(command="cd /", timeout=1) 521 522 523def get_mount_dir(mount_dir): 524 """ 525 Use windows path to mount directly when the system is windows 526 Parameters: 527 mount_dir : the dir to mount that config in user_config.xml 528 such as: the mount_dir is: D:\\mount\\root 529 the mount command should be: mount ip:/d/mount/root 530 """ 531 if platform.system() == "Windows": 532 mount_dir = mount_dir.replace(":", "").replace("\\", "/") 533 _list = mount_dir.split("/") 534 if mount_dir.startswith("/"): 535 _list[1] = _list[1].lower() 536 else: 537 _list[0] = _list[0].lower() 538 mount_dir = "/".join(_list) 539 mount_dir = "/%s" % mount_dir 540 return mount_dir 541 542 543def check_server_file(local_file, target_path): 544 for file_list in glob.glob(os.path.join(target_path, '*.*')): 545 if os.path.basename(local_file) in file_list: 546 return True 547 return False 548 549 550@Plugin(type=Plugin.TEST_KIT, id=CKit.rootfs) 551class RootFsKit(ITestKit): 552 def __init__(self): 553 self.checksum_command = None 554 self.hash_file_name = None 555 self.device_label = None 556 557 def __check_config__(self, config): 558 self.checksum_command = get_config_value("command", config, 559 is_list=False) 560 self.hash_file_name = get_config_value("hash_file_name", config, 561 is_list=False) 562 self.device_label = get_config_value("device_label", config, 563 is_list=False) 564 if not self.checksum_command or not self.hash_file_name or \ 565 not self.device_label: 566 err_msg = ErrorMessage.Config.Code_0302008.format( 567 self.checksum_command, self.hash_file_name, self.device_label) 568 LOG.error(err_msg) 569 raise TypeError(err_msg) 570 571 def __setup__(self, device, **kwargs): 572 del kwargs 573 574 # check device label 575 if not device.label == self.device_label: 576 LOG.error("Device label is not match '%s '" % "demo label", 577 error_no="00108") 578 return False 579 else: 580 report_path = self._get_report_dir() 581 if report_path and os.path.exists(report_path): 582 583 # execute command of checksum 584 device.connect() 585 device.execute_command_with_timeout( 586 command="cd /", case_type=DeviceTestType.cpp_test_lite) 587 result, _, _ = device.execute_command_with_timeout( 588 command=self.checksum_command, 589 case_type=DeviceTestType.cpp_test_lite) 590 device.close() 591 # get serial from device and then join new file name 592 pos = self.hash_file_name.rfind(".") 593 serial = "_%s" % device.__get_serial__() 594 if pos > 0: 595 hash_file_name = "".join((self.hash_file_name[:pos], 596 serial, 597 self.hash_file_name[pos:])) 598 else: 599 hash_file_name = "".join((self.hash_file_name, serial)) 600 hash_file_path = os.path.join(report_path, hash_file_name) 601 # write result to file 602 hash_file_path_open = os.open(hash_file_path, os.O_WRONLY | 603 os.O_CREAT | os.O_APPEND, 604 FilePermission.mode_755) 605 606 with os.fdopen(hash_file_path_open, mode="w") as hash_file: 607 hash_file.write(result) 608 hash_file.flush() 609 else: 610 msg = "RootFsKit teardown, log path [%s] not exists!" \ 611 % report_path 612 LOG.error(msg, error_no="00440") 613 return False 614 return True 615 616 def __teardown__(self, device): 617 pass 618 619 @staticmethod 620 def _get_report_dir(): 621 from xdevice import Variables 622 report_path = os.path.join(Variables.exec_dir, 623 Variables.report_vars.report_dir, 624 Variables.task_name) 625 return report_path 626 627 628@Plugin(type=Plugin.TEST_KIT, id=CKit.query) 629class QueryKit(ITestKit): 630 def __init__(self): 631 self.mount_kit = MountKit() 632 self.query = "" 633 self.properties = "" 634 635 def __check_config__(self, config): 636 setattr(self.mount_kit, "mount_list", 637 get_config_value('mount', config)) 638 setattr(self.mount_kit, "server", get_config_value( 639 'server', config, is_list=False, default="NfsServer")) 640 self.query = get_config_value('query', config, is_list=False) 641 self.properties = get_config_value('properties', config, is_list=False) 642 643 if not self.query: 644 err_msg = ErrorMessage.Config.Code_0302009.format(self.query) 645 LOG.error(err_msg) 646 raise TypeError(err_msg) 647 648 def __setup__(self, device, **kwargs): 649 LOG.debug("Start query kit setup") 650 if device.label != DeviceLabelType.ipcamera: 651 return 652 request = kwargs.get("request", None) 653 if not request: 654 raise ParamError(ErrorMessage.Config.Code_0302010) 655 self.mount_kit.__setup__(device, request=request) 656 execute_query(device, self.query, request) 657 658 def __teardown__(self, device): 659 if device.label != DeviceLabelType.ipcamera: 660 return 661 device.connect() 662 self.mount_kit.__teardown__(device) 663 device.close() 664 665 666@Plugin(type=Plugin.TEST_KIT, id=CKit.liteshell) 667class LiteShellKit(ITestKit): 668 def __init__(self): 669 self.command_list = [] 670 self.tear_down_command = [] 671 self.paths = None 672 673 def __check_config__(self, config): 674 self.command_list = get_config_value('run-command', config) 675 self.tear_down_command = get_config_value('teardown-command', config) 676 677 def __setup__(self, device, **kwargs): 678 del kwargs 679 device.connect() 680 LOG.debug("LiteShellKit setup, device:{}".format(device.device_sn)) 681 if len(self.command_list) == 0: 682 LOG.info("No setup command to run, skipping!") 683 return 684 for command in self.command_list: 685 run_command(device, command) 686 687 def __teardown__(self, device): 688 LOG.debug("LiteShellKit teardown: device:{}".format(device.device_sn)) 689 if len(self.tear_down_command) == 0: 690 LOG.info("No teardown command to run, skipping!") 691 return 692 for command in self.tear_down_command: 693 run_command(device, command) 694 695 696def run_command(device, command): 697 LOG.debug("The command:{} is running".format(command)) 698 if command.strip() == "reset": 699 device.reboot() 700 else: 701 if command.startswith("chr"): 702 command = eval(command) 703 device.execute_command_with_timeout(command, timeout=1) 704 705 706@Plugin(type=Plugin.TEST_KIT, id=CKit.liteinstall) 707class LiteAppInstallKit(ITestKit): 708 def __init__(self): 709 self.app_list = "" 710 self.is_clean = "" 711 self.alt_dir = "" 712 self.bundle_name = None 713 self.paths = "" 714 self.signature = False 715 716 def __check_config__(self, options): 717 self.app_list = get_config_value('test-file-name', options) 718 self.is_clean = get_config_value('cleanup-apps', options, False) 719 self.signature = get_config_value('signature', options, False) 720 self.alt_dir = get_config_value('alt-dir', options, False) 721 if self.alt_dir and self.alt_dir.startswith("resource/"): 722 self.alt_dir = self.alt_dir[len("resource/"):] 723 self.paths = get_config_value('paths', options) 724 725 def __setup__(self, device, **kwargs): 726 del kwargs 727 LOG.debug("LiteAppInstallKit setup, device:{}". 728 format(device.device_sn)) 729 if len(self.app_list) == 0: 730 LOG.info("No app to install, skipping!") 731 return 732 733 for app in self.app_list: 734 if app.endswith(".hap"): 735 device.execute_command_with_timeout("cd /", timeout=1) 736 if self.signature: 737 device.execute_command_with_timeout( 738 command="./bin/bm set -d enable", timeout=10) 739 else: 740 device.execute_command_with_timeout( 741 command="./bin/bm set -s disable", timeout=10) 742 743 device.execute_command_with_timeout( 744 "./bin/bm install -p %s" % app, timeout=60) 745 746 def __teardown__(self, device): 747 LOG.debug("LiteAppInstallKit teardown: device:{}".format( 748 device.device_sn)) 749 if self.is_clean and str(self.is_clean).lower() == "true" \ 750 and self.bundle_name: 751 device.execute_command_with_timeout( 752 "./bin/bm uninstall -n %s" % self.bundle_name, timeout=90) 753 754 755@Plugin(type=Plugin.TEST_KIT, id=CKit.deploytool) 756class DeployToolKit(ITestKit): 757 def __init__(self): 758 self.config = None 759 self.auto_deploy = None 760 self.device_label = None 761 self.time_out = None 762 self.burn_file = None 763 self.paths = None 764 self.upgrade_file_path = None 765 self.burn_tools = None 766 767 def __check_config__(self, config): 768 self.config = config 769 self.paths = get_config_value('paths', config) 770 self.burn_file = get_config_value('burn_file', config, is_list=False) 771 self.auto_deploy = get_config_value('auto_deploy', 772 config, is_list=False) 773 self.device_label = get_config_value("device_label", config, 774 is_list=False) 775 self.time_out = get_config_value("timeout", config, 776 is_list=False) 777 self.upgrade_file_path = get_config_value("upgrade_file_path", config, 778 is_list=False) 779 self.burn_tools = get_config_value('burn_tools', config, is_list=False) 780 781 if not self.auto_deploy or not self.time_out: 782 err_msg = ErrorMessage.Config.Code_0302011.format(self.device_label, self.time_out) 783 LOG.error(err_msg) 784 raise TypeError(err_msg) 785 786 def __setup__(self, device, **kwargs): 787 LOG.info("upgrade_file_path:{}".format(self.upgrade_file_path)) 788 upgrade_file_name = os.path.basename(self.upgrade_file_path) 789 if self.upgrade_file_path.startswith("resource"): 790 self.upgrade_file_path = get_file_absolute_path( 791 os.path.join("tools", upgrade_file_name), self.paths) 792 sys.path.insert(0, os.path.dirname(self.upgrade_file_path)) 793 serial_port = device.device.com_dict.get( 794 ComType.deploy_com).serial_port 795 LOG.debug("serial_port:{}".format(serial_port)) 796 usb_port = device.device.com_dict.get(ComType.cmd_com).usb_port 797 patch_file = get_file_absolute_path(self.burn_file, self.paths) 798 upgrade_name = upgrade_file_name.split(".py")[0] 799 import_cmd_str = "from {} import {} as upgrade_device".format( 800 upgrade_name, upgrade_name) 801 scope = {} 802 exec(import_cmd_str, scope) 803 upgrade_device = scope.get("upgrade_device", 'none') 804 upgrade = upgrade_device(serial_port=serial_port, baund_rate=115200, 805 patch_file=patch_file, usb_port=usb_port) 806 upgrade_result = upgrade.burn() 807 if upgrade_result: 808 return upgrade.reset_device() 809 return None 810 811 def __teardown__(self, device): 812 pass 813