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