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_local_ip 39from xdevice import FilePermission 40from xdevice import DeviceTestType 41from xdevice import DeviceLabelType 42from ohos.exception import LiteDeviceConnectError 43from ohos.exception import LiteDeviceError 44from ohos.exception import LiteDeviceMountError 45from ohos.constants import ComType 46from ohos.constants import CKit 47from ohos.constants import DeviceLiteKernel 48from ohos.utils import parse_strings_key_value 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, e.g. 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 don't 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 device.execute_command_with_timeout(command="cd /", timeout=0.2) 582 if device.__get_device_kernel__() == DeviceLiteKernel.linux_kernel: 583 command = f"chmod +x /storage{self.query} && ./storage{self.query}" 584 output, _, _ = device.execute_command_with_timeout( 585 command=command, timeout=5) 586 else: 587 output, _, _ = device.execute_command_with_timeout( 588 command=".{}".format(self.query), timeout=5) 589 LOG.debug(output) 590 params = parse_strings_key_value(output) 591 device.update_device_props(params) 592 593 def __teardown__(self, device): 594 if device.label != DeviceLabelType.ipcamera: 595 return 596 device.connect() 597 self.mount_kit.__teardown__(device) 598 device.close() 599 600 601@Plugin(type=Plugin.TEST_KIT, id=CKit.liteshell) 602class LiteShellKit(ITestKit): 603 def __init__(self): 604 self.command_list = [] 605 self.tear_down_command = [] 606 self.paths = None 607 608 def __check_config__(self, config): 609 self.command_list = get_config_value('run-command', config) 610 self.tear_down_command = get_config_value('teardown-command', config) 611 612 def __setup__(self, device, **kwargs): 613 del kwargs 614 LOG.debug("LiteShellKit setup, device:{}".format(device.device_sn)) 615 if len(self.command_list) == 0: 616 LOG.info("No setup command to run, skipping!") 617 return 618 for command in self.command_list: 619 run_command(device, command) 620 621 def __teardown__(self, device): 622 LOG.debug("LiteShellKit teardown: device:{}".format(device.device_sn)) 623 if len(self.tear_down_command) == 0: 624 LOG.info("No teardown command to run, skipping!") 625 return 626 for command in self.tear_down_command: 627 run_command(device, command) 628 629 630def run_command(device, command): 631 LOG.debug("The command:{} is running".format(command)) 632 if command.strip() == "reset": 633 device.reboot() 634 else: 635 device.execute_shell_command(command) 636 637 638@Plugin(type=Plugin.TEST_KIT, id=CKit.liteinstall) 639class LiteAppInstallKit(ITestKit): 640 def __init__(self): 641 self.app_list = "" 642 self.is_clean = "" 643 self.alt_dir = "" 644 self.bundle_name = None 645 self.paths = "" 646 self.signature = False 647 648 def __check_config__(self, options): 649 self.app_list = get_config_value('test-file-name', options) 650 self.is_clean = get_config_value('cleanup-apps', options, False) 651 self.signature = get_config_value('signature', options, False) 652 self.alt_dir = get_config_value('alt-dir', options, False) 653 if self.alt_dir and self.alt_dir.startswith("resource/"): 654 self.alt_dir = self.alt_dir[len("resource/"):] 655 self.paths = get_config_value('paths', options) 656 657 def __setup__(self, device, **kwargs): 658 del kwargs 659 LOG.debug("LiteAppInstallKit setup, device:{}". 660 format(device.device_sn)) 661 if len(self.app_list) == 0: 662 LOG.info("No app to install, skipping!") 663 return 664 665 for app in self.app_list: 666 if app.endswith(".hap"): 667 device.execute_command_with_timeout("cd /", timeout=1) 668 if self.signature: 669 device.execute_command_with_timeout( 670 command="./bin/bm set -d enable", timeout=10) 671 else: 672 device.execute_command_with_timeout( 673 command="./bin/bm set -s disable", timeout=10) 674 675 device.execute_command_with_timeout( 676 "./bin/bm install -p %s" % app, timeout=60) 677 678 def __teardown__(self, device): 679 LOG.debug("LiteAppInstallKit teardown: device:{}".format( 680 device.device_sn)) 681 if self.is_clean and str(self.is_clean).lower() == "true" \ 682 and self.bundle_name: 683 device.execute_command_with_timeout( 684 "./bin/bm uninstall -n %s" % self.bundle_name, timeout=90) 685 686 687def process_product_info(message, product_info): 688 if "The" in message: 689 message = message[message.index("The"):] 690 items = message[len("The "):].split(" is ") 691 product_info.setdefault(items[0].strip(), 692 items[1].strip().strip("[").strip("]")) 693 694 695@Plugin(type=Plugin.TEST_KIT, id=CKit.deploytool) 696class DeployToolKit(ITestKit): 697 def __init__(self): 698 self.config = None 699 self.auto_deploy = None 700 self.device_label = None 701 self.time_out = None 702 self.paths = None 703 self.upgrade_file_path = None 704 self.burn_tools = None 705 706 def __check_config__(self, config): 707 self.config = config 708 self.paths = get_config_value("paths", config) 709 self.burn_file = get_config_value("burn_file", config, is_list=False) 710 self.auto_deploy = get_config_value('auto_deploy', 711 config, is_list=False) 712 self.device_label = get_config_value("device_label", config, 713 is_list=False) 714 self.time_out = get_config_value("timeout", config, 715 is_list=False) 716 self.upgrade_file_path = get_config_value("upgrade_file_path", config, is_list=False) 717 self.burn_tools = get_config_value("burn_tools", config, is_list=False) 718 719 if not self.auto_deploy or not self.upgrade_file_path or not self.time_out: 720 msg = "The config for deploy tool kit is" \ 721 "invalid: upgrade_file_path :{} time out:{}".format( 722 self.upgrade_file_path, self.time_out) 723 LOG.error(msg, error_no="00108") 724 return TypeError(msg) 725 726 def __setup__(self, device, **kwargs): 727 LOG.info("Upgrade file path:{}".format(self.upgrade_file_path)) 728 upgrade_file_name = os.path.basename(self.upgrade_file_path) 729 if self.upgrade_file_path.startswith("resource"): 730 self.upgrade_file_path = get_file_absolute_path( 731 os.path.join("tools", upgrade_file_name), self.paths) 732 sys.path.insert(0, os.path.dirname(self.upgrade_file_path)) 733 serial_port = device.device.com_dict.get(ComType.deploy_com).serial_port 734 LOG.debug("Serial port:{}".format(serial_port)) 735 baud_rate = device.device.com_dict.get(ComType.deploy_com).baud_rate 736 usb_port = device.device.com_dict.get(ComType.cmd_com).usb_port 737 patch_file = get_file_absolute_path(self.burn_file, self.paths) 738 upgrade_name = upgrade_file_name.split(".py")[0] 739 import_cmd_str = "from {} import {} as upgrade_device".format( 740 upgrade_name, upgrade_name) 741 scope = {} 742 exec(import_cmd_str, scope) 743 upgrade_device = scope.get("upgrade_device", "None") 744 upgrade = upgrade_device(serial_port=serial_port, baud_rate=baud_rate, 745 patch_file=patch_file, usb_port=usb_port) 746 upgrade_result = upgrade.burn() 747 if upgrade_result: 748 return upgrade.reset_device() 749 return None 750 751 def __teardown__(self, device): 752 pass 753