1#!/usr/bin/env python3 2# coding=utf-8 3 4# 5# Copyright (c) 2020-2021 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 28 29from _core.logger import platform_logger 30from _core.plugin import Plugin 31from _core.config.config_manager import UserConfigManager 32from _core.constants import CKit 33from _core.constants import ConfigConst 34from _core.constants import ComType 35from _core.constants import DeviceLiteKernel 36from _core.constants import DeviceTestType 37from _core.exception import LiteDeviceMountError 38from _core.exception import ParamError 39from _core.exception import LiteDeviceError 40from _core.interface import ITestKit 41from _core.utils import get_config_value 42from _core.utils import get_file_absolute_path 43from _core.utils import get_local_ip 44from _core.utils import get_test_component_version 45from _core.exception import LiteDeviceConnectError 46from _core.constants import DeviceLabelType 47from _core.constants import FilePermission 48from _core.environment.manager_env import DeviceAllocationState 49 50 51__all__ = ["DeployKit", "MountKit", "RootFsKit", "QueryKit", "LiteShellKit", 52 "LiteAppInstallKit", "DeployToolKit"] 53LOG = platform_logger("KitLite") 54 55RESET_CMD = "0xEF, 0xBE, 0xAD, 0xDE, 0x0C, 0x00, 0x87, 0x78, 0x00, 0x00, " \ 56 "0x61, 0x94" 57 58 59@Plugin(type=Plugin.TEST_KIT, id=CKit.deploy) 60class DeployKit(ITestKit): 61 def __init__(self): 62 self.burn_file = "" 63 self.burn_command = "" 64 self.timeout = "" 65 self.paths = "" 66 67 def __check_config__(self, config): 68 self.timeout = str(int(get_config_value( 69 'timeout', config, is_list=False, default=0)) * 1000) 70 self.burn_file = get_config_value('burn_file', config, is_list=False) 71 burn_command = get_config_value('burn_command', config, is_list=False, 72 default=RESET_CMD) 73 self.burn_command = burn_command.replace(" ", "").split(",") 74 self.paths = get_config_value('paths', config) 75 if self.timeout == "0" or not self.burn_file: 76 msg = "The config for deploy kit is invalid with timeout:{}, " \ 77 "burn_file:{}".format(self.timeout, self.burn_file) 78 raise ParamError(msg, error_no="00108") 79 80 def _reset(self, device): 81 cmd_com = device.device.com_dict.get(ComType.cmd_com) 82 try: 83 cmd_com.connect() 84 cmd_com.execute_command( 85 command='AT+RST={}'.format(self.timeout)) 86 cmd_com.close() 87 except (LiteDeviceConnectError, IOError) as error: 88 device.device_allocation_state = DeviceAllocationState.unusable 89 LOG.error( 90 "The exception {} happened in deploy kit running".format( 91 error), error_no=getattr(error, "error_no", 92 "00000")) 93 raise LiteDeviceError("%s port set_up wifiiot failed" % 94 cmd_com.serial_port, 95 error_no=getattr(error, "error_no", 96 "00000")) 97 finally: 98 if cmd_com: 99 cmd_com.close() 100 101 def _send_file(self, device): 102 burn_tool_name = "HiBurn.exe" if os.name == "nt" else "HiBurn" 103 burn_tool_path = get_file_absolute_path( 104 os.path.join("tools", burn_tool_name), self.paths) 105 patch_file = get_file_absolute_path(self.burn_file, self.paths) 106 deploy_serial_port = device.device.com_dict.get( 107 ComType.deploy_com).serial_port 108 deploy_baudrate = device.device.com_dict.\ 109 get(ComType.deploy_com).baud_rate 110 port_number = re.findall(r'\d+$', deploy_serial_port) 111 if not port_number: 112 raise LiteDeviceError("The config of serial port {} to deploy is " 113 "invalid".format(deploy_serial_port), 114 error_no="00108") 115 new_temp_tool_path = copy_file_as_temp(burn_tool_path, 10) 116 cmd = '{} -com:{} -bin:{} -signalbaud:{}' \ 117 .format(new_temp_tool_path, port_number[0], patch_file, 118 deploy_baudrate) 119 LOG.info('The running cmd is {}'.format(cmd)) 120 LOG.info('The burn tool is running, please wait..') 121 return_code, out = subprocess.getstatusoutput(cmd) 122 LOG.info( 123 'Deploy kit to execute burn tool finished with return_code: {} ' 124 'output: {}'.format(return_code, out)) 125 os.remove(new_temp_tool_path) 126 if 0 != return_code: 127 device.device_allocation_state = DeviceAllocationState.unusable 128 raise LiteDeviceError("%s port set_up wifiiot failed" % 129 deploy_serial_port, error_no="00402") 130 131 def __setup__(self, device, **kwargs): 132 """ 133 Execute reset command on the device by cmd serial port and then upload 134 patch file by deploy tool. 135 Parameters: 136 device: the instance of LocalController with one or more 137 ComController 138 """ 139 del kwargs 140 self._reset(device) 141 self._send_file(device) 142 143 def __teardown__(self, device): 144 pass 145 146 147@Plugin(type=Plugin.TEST_KIT, id=CKit.mount) 148class MountKit(ITestKit): 149 def __init__(self): 150 self.remote = None 151 self.paths = "" 152 self.mount_list = [] 153 self.mounted_dir = set() 154 self.server = "" 155 self.file_name_list = [] 156 self.remote_info = None 157 158 def __check_config__(self, config): 159 self.remote = get_config_value('server', config, is_list=False) 160 self.paths = get_config_value('paths', config) 161 self.mount_list = get_config_value('mount', config, is_list=True) 162 self.server = get_config_value('server', config, is_list=False, 163 default="NfsServer") 164 if not self.mount_list: 165 msg = "The config for mount kit is invalid with mount:{}" \ 166 .format(self.mount_list) 167 LOG.error(msg, error_no="00108") 168 raise TypeError("Load Error[00108]") 169 170 def mount_on_board(self, device=None, remote_info=None, case_type=""): 171 """ 172 Init the environment on the device server, eg. mount the testcases to 173 server 174 175 Parameters: 176 device: DeviceLite, device lite on local or remote 177 remote_info: dict, includes 178 linux_host: str, nfs_server ip 179 linux_directory: str, the directory on the linux 180 is_remote: str, server is remote or not 181 case_type: str, CppTestLite or CTestLite, default value is 182 DeviceTestType.cpp_test_lite 183 184 Returns: 185 True or False, represent init Failed or success 186 """ 187 if not remote_info: 188 raise ParamError("failed to get server environment", 189 error_no="00108") 190 191 linux_host = remote_info.get("ip", "") 192 linux_directory = remote_info.get("dir", "") 193 is_remote = remote_info.get("remote", "false") 194 liteos_commands = ["cd /", "umount device_directory", 195 "mount nfs_ip:nfs_directory device" 196 "_directory nfs"] 197 linux_commands = ["cd /%s" % "storage", 198 "umount -f /%s/%s" % ("storage", "device_directory"), 199 "toybox mount -t nfs -o nolock,addr=nfs_ip nfs_ip:nfs_directory " 200 "/%s/%s" % ("storage", "device_directory"), 201 "chmod 755 -R /%s/%s" % ( 202 "storage", "device_directory")] 203 if not linux_host or not linux_directory: 204 raise LiteDeviceMountError( 205 "nfs server miss ip or directory[00108]", error_no="00108") 206 207 commands = [] 208 if device.label == "ipcamera": 209 env_result, status, _ = device.execute_command_with_timeout( 210 command="uname", timeout=1, retry=2) 211 if status: 212 if env_result.find(DeviceLiteKernel.linux_kernel) != -1 or \ 213 env_result.find("Linux") != -1: 214 commands = linux_commands 215 device.__set_device_kernel__(DeviceLiteKernel.linux_kernel) 216 else: 217 commands = liteos_commands 218 device.__set_device_kernel__(DeviceLiteKernel.lite_kernel) 219 else: 220 raise LiteDeviceMountError("failed to get device env[00402]", 221 error_no="00402") 222 223 for mount_file in self.mount_list: 224 target = mount_file.get("target", "/test_root") 225 if target in self.mounted_dir: 226 LOG.debug("%s is mounted" % target) 227 continue 228 mkdir_on_board(device, target) 229 230 # local nfs server need use alias of dir to mount 231 if is_remote.lower() == "false": 232 linux_directory = get_mount_dir(linux_directory) 233 for command in commands: 234 command = command.replace("nfs_ip", linux_host). \ 235 replace("nfs_directory", linux_directory).replace( 236 "device_directory", target).replace("//", "/") 237 timeout = 15 if command.startswith("mount") else 1 238 if command.startswith("mount"): 239 self.mounted_dir.add(target) 240 for mount_time in range(1, 4): 241 result, status, _ = device.\ 242 execute_command_with_timeout(command=command, 243 case_type=case_type, 244 timeout=timeout) 245 if status: 246 break 247 if "already mounted" in result: 248 LOG.info("{} is mounted".format(target)) 249 break 250 LOG.info("Mount failed,try " 251 "again {} time".format(mount_time)) 252 if mount_time == 3: 253 raise LiteDeviceMountError("Failed to mount the " 254 "device[00402]", 255 error_no="00402") 256 else: 257 result, status, _ = device.execute_command_with_timeout( 258 command=command, case_type=case_type, timeout=timeout) 259 LOG.info('prepare environment success') 260 261 def __setup__(self, device, **kwargs): 262 """ 263 Mount the file to the board by the nfs server. 264 """ 265 LOG.debug("start mount kit setup") 266 267 request = kwargs.get("request", None) 268 if not request: 269 raise ParamError("MountKit setup request is None", 270 error_no="02401") 271 device.connect() 272 273 config_manager = UserConfigManager( 274 config_file=request.get(ConfigConst.configfile, ""), 275 env=request.get(ConfigConst.test_environment, "")) 276 remote_info = config_manager.get_user_config("testcases/server", 277 filter_name=self.server) 278 279 copy_list = self.copy_to_server(remote_info.get("dir"), 280 remote_info.get("ip"), 281 request, request.config.testcases_path) 282 283 self.mount_on_board(device=device, remote_info=remote_info, 284 case_type=DeviceTestType.cpp_test_lite) 285 286 return copy_list 287 288 def copy_to_server(self, linux_directory, linux_host, request, 289 testcases_dir): 290 file_local_paths = [] 291 for mount_file in self.mount_list: 292 source = mount_file.get("source") 293 if not source: 294 raise TypeError("The source of MountKit can not be empty " 295 "in Test.json!") 296 source = source.replace("$testcases/", "").\ 297 replace("$resources/", "") 298 file_path = get_file_absolute_path(source, self.paths) 299 if os.path.isdir(file_path): 300 for root, _, files in os.walk(file_path): 301 for _file in files: 302 if _file.endswith(".json"): 303 continue 304 file_local_paths.append(os.path.join(root, _file)) 305 else: 306 file_local_paths.append(file_path) 307 308 config_manager = UserConfigManager( 309 config_file=request.get(ConfigConst.configfile, ""), 310 env=request.get(ConfigConst.test_environment, "")) 311 remote_info = config_manager.get_user_config("testcases/server", 312 filter_name=self.server) 313 self.remote_info = remote_info 314 315 if not remote_info: 316 err_msg = "The name of remote device {} does not match". \ 317 format(self.remote) 318 LOG.error(err_msg, error_no="00403") 319 raise TypeError(err_msg) 320 is_remote = remote_info.get("remote", "false") 321 if (str(get_local_ip()) == linux_host) and ( 322 linux_directory == ("/data%s" % testcases_dir)): 323 return 324 remote_ip = remote_info.get("ip", "") 325 port = remote_info.get("port", "") 326 remote_dir = remote_info.get("dir", "") 327 if not remote_ip or not port or not remote_dir: 328 LOG.warning("nfs server's ip or port or dir is empty") 329 return 330 for _file in file_local_paths: 331 # remote copy 332 LOG.info("Trying to copy the file from {} to nfs server". 333 format(_file)) 334 if not is_remote.lower() == "false": 335 try: 336 import paramiko 337 client = paramiko.Transport(remote_ip, int(port)) 338 client.connect(username=remote_info.get("username"), 339 password=remote_info.get("password")) 340 sftp = paramiko.SFTPClient.from_transport(client) 341 sftp.put(localpath=_file, remotepath=os.path.join( 342 remote_info.get("dir"), os.path.basename(_file))) 343 client.close() 344 except (OSError, Exception) as exception: 345 msg = "copy file to nfs server failed with error {}" \ 346 .format(exception) 347 LOG.error(msg, error_no="00403") 348 # local copy 349 else: 350 for count in range(1, 4): 351 try: 352 os.remove(os.path.join(remote_info.get("dir"), 353 os.path.basename(_file))) 354 except OSError as _: 355 pass 356 shutil.copy(_file, remote_info.get("dir")) 357 if check_server_file(_file, remote_info.get("dir")): 358 break 359 else: 360 LOG.info( 361 "Trying to copy the file from {} to nfs " 362 "server {} times".format(_file, count)) 363 if count == 3: 364 msg = "copy {} to nfs server " \ 365 "failed {} times".format( 366 os.path.basename(_file), count) 367 LOG.error(msg, error_no="00403") 368 LOG.debug("Nfs server:{}".format(glob.glob( 369 os.path.join(remote_info.get("dir"), '*.*')))) 370 371 self.file_name_list.append(os.path.basename(_file)) 372 373 return self.file_name_list 374 375 def __teardown__(self, device): 376 if device.__get_device_kernel__() == DeviceLiteKernel.linux_kernel: 377 device.execute_command_with_timeout(command="cd /storage", 378 timeout=1) 379 for mounted_dir in self.mounted_dir: 380 device.execute_command_with_timeout(command="umount -f " 381 "/storage{}". 382 format(mounted_dir), 383 timeout=2) 384 device.execute_command_with_timeout(command="rm -r /storage{}". 385 format(mounted_dir), 386 timeout=1) 387 else: 388 device.execute_command_with_timeout(command="cd /", timeout=1) 389 for mounted_dir in self.mounted_dir: 390 for mount_time in range(1, 3): 391 result, status, _ = device.execute_command_with_timeout( 392 command="umount {}".format(mounted_dir), 393 timeout=2) 394 if result.find("Resource busy") == -1: 395 device.execute_command_with_timeout( 396 command="rm -r {}".format(mounted_dir), 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.time_out = None 704 705 def __check_config__(self, config): 706 self.config = config 707 self.auto_deploy = get_config_value('auto_deploy', config, False) 708 self.time_out = get_config_value('timeout', config, False) 709 if self.auto_deploy or not self.time_out: 710 msg = "The config for deploytool kit is invalid " \ 711 "with auto_deploy:{},timeout:{}".format(self.auto_deploy, 712 self.time_out) 713 LOG.error(msg, error_no="00108") 714 raise ParamError(msg, error_no="00108") 715 716 def __setup__(self, device, **kwargs): 717 args = kwargs 718 request = args.get("request", None) 719 request.confing.deploy_tool_kit = self.config 720 721 def __teardown__(self, device): 722 pass 723