| - | + | @@ -449,7 +453,7 @@ | |||
|---|---|---|---|---|---|
| Modules | -Modules Done | +Run Modules | Total Tests | Passed | Failed | diff --git a/src/xdevice/_core/testkit/__init__.py b/src/xdevice/_core/testkit/__init__.py old mode 100755 new mode 100644 diff --git a/src/xdevice/_core/testkit/json_parser.py b/src/xdevice/_core/testkit/json_parser.py index f616aa7e97c0ede63794d867c63d8fefff9e4069..921a43dc4843bc97756bd24dfbd2b67e3a15fd0b 100755 --- a/src/xdevice/_core/testkit/json_parser.py +++ b/src/xdevice/_core/testkit/json_parser.py @@ -2,7 +2,7 @@ # coding=utf-8 # -# Copyright (c) 2020 Huawei Device Co., Ltd. +# Copyright (c) 2020-2021 Huawei Device Co., Ltd. # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at @@ -18,7 +18,7 @@ import json import os - +import stat from _core.exception import ParamError from _core.logger import platform_logger from _core.plugin import Config @@ -68,11 +68,16 @@ class JsonParser: else: if not os.path.exists(path_or_content): raise ParamError("The json file {} does not exist".format( - path_or_content)) - with open(path_or_content, encoding="utf-8") as file_content: + path_or_content), error_no="00110") + + flags = os.O_RDONLY + modes = stat.S_IWUSR | stat.S_IRUSR + with os.fdopen(os.open(path_or_content, flags, modes), + "r") as file_content: json_content = json.load(file_content) except (TypeError, ValueError, AttributeError) as error: - raise ParamError("%s %s" % (path_or_content, error)) + raise ParamError("json file error: %s %s" % ( + path_or_content, error), error_no="00111") self._check_config(json_content) # set self.config diff --git a/src/xdevice/_core/testkit/kit_lite.py b/src/xdevice/_core/testkit/kit_lite.py index fc541a1650c2ff8ed99fdeea0203e583755fd80f..5ea3989de1d7867b151ea85f062bae792ecbe8c6 100755 --- a/src/xdevice/_core/testkit/kit_lite.py +++ b/src/xdevice/_core/testkit/kit_lite.py @@ -2,7 +2,7 @@ # coding=utf-8 # -# Copyright (c) 2020 Huawei Device Co., Ltd. +# Copyright (c) 2020-2021 Huawei Device Co., Ltd. # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at @@ -23,16 +23,19 @@ import string import subprocess import shutil import platform +import glob from _core.logger import platform_logger from _core.plugin import Plugin from _core.config.config_manager import UserConfigManager from _core.constants import CKit +from _core.constants import ConfigConst from _core.constants import ComType from _core.constants import DeviceLiteKernel from _core.constants import DeviceTestType from _core.exception import LiteDeviceMountError from _core.exception import ParamError +from _core.exception import LiteDeviceError from _core.interface import ITestKit from _core.utils import get_config_value from _core.utils import get_file_absolute_path @@ -40,9 +43,11 @@ from _core.utils import get_local_ip from _core.utils import get_test_component_version from _core.exception import LiteDeviceConnectError from _core.constants import DeviceLabelType +from _core.environment.manager_env import DeviceAllocationState -__all__ = ["DeployKit", "MountKit", "RootFsKit", "QueryKit"] +__all__ = ["DeployKit", "MountKit", "RootFsKit", "QueryKit", "LiteShellKit", + "LiteAppInstallKit"] LOG = platform_logger("KitLite") RESET_CMD = "0xEF, 0xBE, 0xAD, 0xDE, 0x0C, 0x00, 0x87, 0x78, 0x00, 0x00, " \ @@ -58,68 +63,53 @@ class DeployKit(ITestKit): self.paths = "" def __check_config__(self, config): - self.timeout = str(int(get_config_value('timeout', config, - is_list=False)) * 1000) + self.timeout = str(int(get_config_value( + 'timeout', config, is_list=False, default=0)) * 1000) self.burn_file = get_config_value('burn_file', config, is_list=False) burn_command = get_config_value('burn_command', config, is_list=False, default=RESET_CMD) self.burn_command = burn_command.replace(" ", "").split(",") self.paths = get_config_value('paths', config) - if not self.timeout or not self.burn_file or not self.burn_command: - msg = "The config for deploy kit is invalid with timeout:{} " \ - "burn_file:{} burn_command:{}".format(self.timeout, - self.burn_file, - self.burn_command) - LOG.error(msg) - raise TypeError(msg) + if self.timeout == "0" or not self.burn_file: + msg = "The config for deploy kit is invalid with timeout:{}, " \ + "burn_file:{}".format(self.timeout, self.burn_file) + raise ParamError(msg, error_no="00108") - def __setup__(self, device, **kwargs): - """ - Execute reset command on the device by cmd serial port and then upload - patch file by deploy tool. - Parameters: - device: the instance of LocalController with one or more - ComController - """ - del kwargs - LOG.debug( - "Deploy kit params:{}".format(self.get_plugin_config().__dict__)) - cmd_com = device.com_dict.get(ComType.cmd_com) - LOG.info("The cmd com is {}".format(cmd_com.serial_port)) + def _reset(self, device): + cmd_com = device.device.com_dict.get(ComType.cmd_com) try: cmd_com.connect() cmd_com.execute_command( command='AT+RST={}'.format(self.timeout)) cmd_com.close() except (LiteDeviceConnectError, IOError) as error: + device.device_allocation_state = DeviceAllocationState.unusable LOG.error( "The exception {} happened in deploy kit running".format( - error)) - return False + error), error_no=getattr(error, "error_no", + "00000")) + raise LiteDeviceError("%s port set_up wifiiot failed" % + cmd_com.serial_port, + error_no=getattr(error, "error_no", + "00000")) finally: if cmd_com: cmd_com.close() + + def _send_file(self, device): burn_tool_name = "HiBurn.exe" if os.name == "nt" else "HiBurn" burn_tool_path = get_file_absolute_path( os.path.join("tools", burn_tool_name), self.paths) - if not os.path.exists(burn_tool_path): - LOG.error('The burn tool {} does not exist, please check!'.format( - burn_tool_path)) - return False patch_file = get_file_absolute_path(self.burn_file, self.paths) - if not os.path.exists(patch_file): - LOG.error('The patch file {} does not exist, please check!'.format( - patch_file)) - return False - deploy_serial_port = device.com_dict.get( + deploy_serial_port = device.device.com_dict.get( ComType.deploy_com).serial_port - deploy_baudrate = device.com_dict.get(ComType.deploy_com).baund_rate + deploy_baudrate = device.device.com_dict.\ + get(ComType.deploy_com).baud_rate port_number = re.findall(r'\d+$', deploy_serial_port) if not port_number: - LOG.error( - "The config of serial port {} to deploy is invalid".format( - deploy_serial_port)) - return False + raise LiteDeviceError("The config of serial port {} to deploy is " + "invalid".format(deploy_serial_port), + error_no="00108") new_temp_tool_path = copy_file_as_temp(burn_tool_path, 10) cmd = '{} -com:{} -bin:{} -signalbaud:{}' \ .format(new_temp_tool_path, port_number[0], patch_file, @@ -131,9 +121,22 @@ class DeployKit(ITestKit): 'Deploy kit to execute burn tool finished with return_code: {} ' 'output: {}'.format(return_code, out)) os.remove(new_temp_tool_path) - if 0 == return_code: - return True - return False + if 0 != return_code: + device.device_allocation_state = DeviceAllocationState.unusable + raise LiteDeviceError("%s port set_up wifiiot failed" % + deploy_serial_port, error_no="00402") + + def __setup__(self, device, **kwargs): + """ + Execute reset command on the device by cmd serial port and then upload + patch file by deploy tool. + Parameters: + device: the instance of LocalController with one or more + ComController + """ + del kwargs + self._reset(device) + self._send_file(device) def __teardown__(self, device): pass @@ -159,8 +162,8 @@ class MountKit(ITestKit): if not self.mount_list: msg = "The config for mount kit is invalid with mount:{}" \ .format(self.mount_list) - LOG.error(msg) - raise TypeError(msg) + LOG.error(msg, error_no="00108") + raise TypeError("Load Error[00108]") def mount_on_board(self, device=None, remote_info=None, case_type=""): """ @@ -180,7 +183,8 @@ class MountKit(ITestKit): True or False, represent init Failed or success """ if not remote_info: - raise ParamError("failed to get server environment") + raise ParamError("failed to get server environment", + error_no="00108") linux_host = remote_info.get("ip", "") linux_directory = remote_info.get("dir", "") @@ -188,40 +192,47 @@ class MountKit(ITestKit): liteos_commands = ["cd /", "umount device_directory", "mount nfs_ip:nfs_directory device" "_directory nfs"] - linux_commands = ["cd /", "umount device_directory", - "mount -t nfs -o nolock -o tcp nfs_ip:nfs_directory" - "device_directory", "chmod 755 -R device_directory"] + linux_commands = ["cd /%s" % "storage", + "fuser -k /%s/%s" % ("storage", "device_directory"), + "umount -f /%s/%s" % ("storage", "device_directory"), + "mount -t nfs -o nolock -o tcp nfs_ip:nfs_directory " + "/%s/%s" % ("storage", "device_directory"), + "chmod 755 -R /%s/%s" % ( + "storage", "device_directory")] if not linux_host or not linux_directory: - raise LiteDeviceMountError("nfs server miss ip or directory") - if device.device_connect_type == "local": - device.local_device.flush_input() + raise LiteDeviceMountError( + "nfs server miss ip or directory[00108]", error_no="00108") commands = [] if device.label == "ipcamera": env_result, status, _ = device.execute_command_with_timeout( - command="uname", timeout=1) + command="uname", timeout=1, retry=2) if status: - if env_result.find(DeviceLiteKernel.linux_kernel) != -1: + if env_result.find(DeviceLiteKernel.linux_kernel) != -1 or \ + env_result.find("Linux") != -1: commands = linux_commands device.__set_device_kernel__(DeviceLiteKernel.linux_kernel) else: commands = liteos_commands device.__set_device_kernel__(DeviceLiteKernel.lite_kernel) else: - raise LiteDeviceMountError("failed to get device env") + raise LiteDeviceMountError("failed to get device env[00402]", + error_no="00402") for mount_file in self.mount_list: target = mount_file.get("target", "/test_root") if target in self.mounted_dir: + LOG.debug("%s is mounted" % target) continue mkdir_on_board(device, target) + # local nfs server need use alias of dir to mount if is_remote.lower() == "false": linux_directory = get_mount_dir(linux_directory) for command in commands: command = command.replace("nfs_ip", linux_host). \ replace("nfs_directory", linux_directory).replace( - "device_directory", target) + "device_directory", target).replace("//", "/") timeout = 15 if command.startswith("mount") else 1 result, status, _ = device.execute_command_with_timeout( command=command, case_type=case_type, timeout=timeout) @@ -233,12 +244,17 @@ class MountKit(ITestKit): """ Mount the file to the board by the nfs server. """ + LOG.debug("start mount kit setup") + request = kwargs.get("request", None) if not request: - raise ParamError("MountKit setup request is None") + raise ParamError("MountKit setup request is None", + error_no="02401") device.connect() - config_manager = UserConfigManager(env=request.config.test_environment) + config_manager = UserConfigManager( + config_file=request.get(ConfigConst.configfile, ""), + env=request.get(ConfigConst.test_environment, "")) remote_info = config_manager.get_user_config("testcases/server", filter_name=self.server) @@ -271,7 +287,9 @@ class MountKit(ITestKit): else: file_local_paths.append(file_path) - config_manager = UserConfigManager(env=request.config.test_environment) + config_manager = UserConfigManager( + config_file=request.get(ConfigConst.configfile, ""), + env=request.get(ConfigConst.test_environment, "")) remote_info = config_manager.get_user_config("testcases/server", filter_name=self.server) self.remote_info = remote_info @@ -279,7 +297,7 @@ class MountKit(ITestKit): if not remote_info: err_msg = "The name of remote device {} does not match". \ format(self.remote) - LOG.error(err_msg) + LOG.error(err_msg, error_no="00403") raise TypeError(err_msg) is_remote = remote_info.get("remote", "false") if (str(get_local_ip()) == linux_host) and ( @@ -303,47 +321,53 @@ class MountKit(ITestKit): except (OSError, Exception) as exception: msg = "copy file to nfs server failed with error {}" \ .format(exception) - LOG.error(msg) - raise LiteDeviceMountError(exception) + LOG.error(msg, error_no="00403") # local copy else: - shutil.copy(_file, remote_info.get("dir")) + for count in range(1, 4): + shutil.copy(_file, remote_info.get("dir")) + if check_server_file(_file, remote_info.get("dir")): + break + else: + LOG.info( + "Trying to copy the file from {} to nfs " + "server {} times".format(_file, count)) + if count == 3: + msg = "copy {} to nfs server " \ + "failed {} times".format( + os.path.basename(_file), count) + LOG.error(msg, error_no="00403") + LOG.debug("Nfs server:{}".format(glob.glob( + os.path.join(remote_info.get("dir"), '*.*')))) + self.file_name_list.append(os.path.basename(_file)) return self.file_name_list def __teardown__(self, device): - device.execute_command_with_timeout(command="cd /", timeout=1) - for mounted_dir in self.mounted_dir: - device.execute_command_with_timeout(command="umount {}". - format(mounted_dir), - timeout=2) - device.execute_command_with_timeout(command="rm -r {}". - format(mounted_dir), + if device.__get_device_kernel__() == DeviceLiteKernel.linux_kernel: + device.execute_command_with_timeout(command="cd /storage", timeout=1) - - is_remote = self.remote_info.get("remote", "false") - for _file in self.file_name_list: - LOG.info("Trying to delete the file from nfs server". - format(_file)) - try: - if is_remote.lower() == "true": - import paramiko - client = paramiko.Transport( - (self.remote_info.get("ip"), - int(self.remote_info.get("port")))) - client.connect( - username=self.remote_info.get("username"), - password=self.remote_info.get("password")) - sftp = paramiko.SFTPClient.from_transport(client) - file_path = "{}{}".format( - self.remote_info.get("dir"), _file) - sftp.remove(file_path) - client.close() - else: - os.remove(os.path.join(self.remote_info.get("dir"), _file)) - except FileNotFoundError: - LOG.debug("delete file %s failed" % _file) + for mounted_dir in self.mounted_dir: + device.execute_command_with_timeout(command="fuser -k {}". + format(mounted_dir), + timeout=2) + device.execute_command_with_timeout(command="umount -f " + "/storage{}". + format(mounted_dir), + timeout=2) + device.execute_command_with_timeout(command="rm -r /storage{}". + format(mounted_dir), + timeout=1) + else: + device.execute_command_with_timeout(command="cd /", timeout=1) + for mounted_dir in self.mounted_dir: + device.execute_command_with_timeout(command="umount {}". + format(mounted_dir), + timeout=2) + device.execute_command_with_timeout(command="rm -r {}". + format(mounted_dir), + timeout=1) def copy_file_as_temp(original_file, str_length): @@ -369,7 +393,10 @@ def mkdir_on_board(device, dir_path): device : the L1 board dir_path: the dir path to make """ - device.execute_command_with_timeout(command="cd /", timeout=1) + if device.__get_device_kernel__() == DeviceLiteKernel.linux_kernel: + device.execute_command_with_timeout(command="cd /storage", timeout=1) + else: + device.execute_command_with_timeout(command="cd /", timeout=1) for sub_dir in dir_path.split("/"): if sub_dir in ["", "/"]: continue @@ -377,7 +404,10 @@ def mkdir_on_board(device, dir_path): timeout=1) device.execute_command_with_timeout(command="cd {}".format(sub_dir), timeout=1) - device.execute_command_with_timeout(command="cd /", timeout=1) + if device.__get_device_kernel__() == DeviceLiteKernel.linux_kernel: + device.execute_command_with_timeout(command="cd /storage", timeout=1) + else: + device.execute_command_with_timeout(command="cd /", timeout=1) def get_mount_dir(mount_dir): @@ -400,6 +430,13 @@ def get_mount_dir(mount_dir): return mount_dir +def check_server_file(local_file, target_path): + for file_list in glob.glob(os.path.join(target_path, '*.*')): + if os.path.basename(local_file) in file_list: + return True + return False + + @Plugin(type=Plugin.TEST_KIT, id=CKit.rootfs) class RootFsKit(ITestKit): def __init__(self): @@ -420,7 +457,7 @@ class RootFsKit(ITestKit): " hash_file_name:{} device_label:{}" \ .format(self.checksum_command, self.hash_file_name, self.device_label) - LOG.error(msg) + LOG.error(msg, error_no="00108") return TypeError(msg) def __setup__(self, device, **kwargs): @@ -428,7 +465,8 @@ class RootFsKit(ITestKit): # check device label if not device.label == self.device_label: - LOG.error("device label is not match '%s '" % "demo_label") + LOG.error("device label is not match '%s '" % "demo_label", + error_no="00108") return False else: report_path = self._get_report_dir() @@ -436,6 +474,8 @@ class RootFsKit(ITestKit): # execute command of checksum device.connect() + device.execute_command_with_timeout( + command="cd /", case_type=DeviceTestType.cpp_test_lite) result, _, _ = device.execute_command_with_timeout( command=self.checksum_command, case_type=DeviceTestType.cpp_test_lite) @@ -451,13 +491,16 @@ class RootFsKit(ITestKit): hash_file_name = "".join((self.hash_file_name, serial)) hash_file_path = os.path.join(report_path, hash_file_name) # write result to file - with open(hash_file_path, mode="w", encoding="utf-8") \ - as hash_file: + hash_file_path_open = os.open(hash_file_path, os.O_WRONLY | + os.O_CREAT | os.O_APPEND, 0o755) + + with os.fdopen(hash_file_path_open, mode="w") as hash_file: hash_file.write(result) + hash_file.flush() else: msg = "RootFsKit teardown, log path [%s] not exists!" \ % report_path - LOG.error(msg) + LOG.error(msg, error_no="00440") return False return True @@ -484,28 +527,37 @@ class QueryKit(ITestKit): setattr(self.mount_kit, "server", get_config_value( 'server', config, is_list=False, default="NfsServer")) self.query = get_config_value('query', config, is_list=False) + self.properties = get_config_value('properties', config, is_list=False) if not self.query: msg = "The config for query kit is invalid with query:{}" \ .format(self.query) - LOG.error(msg) + LOG.error(msg, error_no="00108") raise TypeError(msg) def __setup__(self, device, **kwargs): + LOG.debug("start query kit setup") if device.label != DeviceLabelType.ipcamera: return request = kwargs.get("request", None) if not request: - raise ParamError("the request of queryKit is None") + raise ParamError("the request of queryKit is None", + error_no="02401") self.mount_kit.__setup__(device, request=request) - device.execute_command_with_timeout(command="cd /", timeout=0.2) - output, _, _ = device.execute_command_with_timeout( - command=".{}".format(self.query), timeout=5) - product_param = {} + if device.__get_device_kernel__() == DeviceLiteKernel.linux_kernel: + device.execute_command_with_timeout(command="cd /storage", + timeout=0.2) + output, _, _ = device.execute_command_with_timeout( + command=".{}{}".format("/storage", self.query), timeout=5) + else: + device.execute_command_with_timeout(command="cd /", timeout=0.2) + output, _, _ = device.execute_command_with_timeout( + command=".{}".format(self.query), timeout=5) + product_info = {} for line in output.split("\n"): - process_product_params(line, product_param) - product_param["version"] = get_test_component_version(request.config) - request.product_params = product_param + process_product_info(line, product_info) + product_info["version"] = get_test_component_version(request.config) + request.product_info = product_info def __teardown__(self, device): if device.label != DeviceLabelType.ipcamera: @@ -515,9 +567,95 @@ class QueryKit(ITestKit): device.close() -def process_product_params(message, product_params): +@Plugin(type=Plugin.TEST_KIT, id=CKit.liteshell) +class LiteShellKit(ITestKit): + def __init__(self): + self.command_list = [] + self.tear_down_command = [] + self.paths = None + + def __check_config__(self, config): + self.command_list = get_config_value('run-command', config) + self.tear_down_command = get_config_value('teardown-command', config) + + def __setup__(self, device, **kwargs): + del kwargs + LOG.debug("LiteShellKit setup, device:{}".format(device.device_sn)) + if len(self.command_list) == 0: + LOG.info("No setup_command to run, skipping!") + return + for command in self.command_list: + run_command(device, command) + + def __teardown__(self, device): + LOG.debug("LiteShellKit teardown: device:{}".format(device.device_sn)) + if len(self.tear_down_command) == 0: + LOG.info("No teardown_command to run, skipping!") + return + for command in self.tear_down_command: + run_command(device, command) + + +def run_command(device, command): + LOG.debug("The command:{} is running".format(command)) + if command.strip() == "reset": + device.reboot() + else: + device.execute_shell_command(command) + + +@Plugin(type=Plugin.TEST_KIT, id=CKit.liteinstall) +class LiteAppInstallKit(ITestKit): + def __init__(self): + self.app_list = "" + self.is_clean = "" + self.alt_dir = "" + self.bundle_name = None + self.paths = "" + self.signature = False + + def __check_config__(self, options): + self.app_list = get_config_value('test-file-name', options) + self.is_clean = get_config_value('cleanup-apps', options, False) + self.signature = get_config_value('signature', options, False) + self.alt_dir = get_config_value('alt-dir', options, False) + if self.alt_dir and self.alt_dir.startswith("resource/"): + self.alt_dir = self.alt_dir[len("resource/"):] + self.paths = get_config_value('paths', options) + + def __setup__(self, device, **kwargs): + del kwargs + LOG.debug("LiteAppInstallKit setup, device:{}". + format(device.device_sn)) + if len(self.app_list) == 0: + LOG.info("No app to install, skipping!") + return + + for app in self.app_list: + if app.endswith(".hap"): + device.execute_command_with_timeout("cd /", timeout=1) + if self.signature: + device.execute_command_with_timeout( + command="./bin/bm set -d enable", timeout=10) + else: + device.execute_command_with_timeout( + command="./bin/bm set -s disable", timeout=10) + + device.execute_command_with_timeout( + "./bin/bm install -p %s" % app, timeout=60) + + def __teardown__(self, device): + LOG.debug("LiteAppInstallKit teardown: device:{}".format( + device.device_sn)) + if self.is_clean and str(self.is_clean).lower() == "true" \ + and self.bundle_name: + device.execute_command_with_timeout( + "./bin/bm uninstall -n %s" % self.bundle_name, timeout=90) + + +def process_product_info(message, product_info): if "The" in message: message = message[message.index("The"):] items = message[len("The "):].split(" is ") - product_params.setdefault(items[0].strip(), - items[1].strip().strip("[").strip("]")) + product_info.setdefault(items[0].strip(), + items[1].strip().strip("[").strip("]")) diff --git a/src/xdevice/_core/testkit/uikit_lite.py b/src/xdevice/_core/testkit/uikit_lite.py deleted file mode 100755 index 82080f782fda40af1d51ec5ad921cf8a78b5d705..0000000000000000000000000000000000000000 --- a/src/xdevice/_core/testkit/uikit_lite.py +++ /dev/null @@ -1,476 +0,0 @@ -#!/usr/bin/env python3 -# coding=utf-8 - -# -# Copyright (c) 2020 Huawei Device Co., Ltd. -# Licensed under the Apache License, Version 2.0 (the "License"); -# you may not use this file except in compliance with the License. -# You may obtain a copy of the License at -# -# http://www.apache.org/licenses/LICENSE-2.0 -# -# Unless required by applicable law or agreed to in writing, software -# distributed under the License is distributed on an "AS IS" BASIS, -# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -# See the License for the specific language governing permissions and -# limitations under the License. -# - -import datetime -import platform -import signal -import os -import json -import time -import subprocess -import re -from dataclasses import dataclass - -from json import JSONDecodeError -from shutil import copyfile -from threading import Timer -from _core.constants import CKit -from _core.exception import ExecuteTerminate -from _core.exception import ParamError -from _core.logger import platform_logger -from _core.plugin import Plugin - -TIMEOUT = 90 -MAX_VALID_POSITION = 454 -LOG = platform_logger("UiKitLite") - - -def timeout_callback(proc, timeout): - try: - timeout.is_timeout = True - LOG.error("Error: execute command timeout.") - LOG.error(proc.pid) - if platform.system() != "Windows": - os.killpg(proc.pid, signal.SIGKILL) - else: - subprocess.call( - ["C:\\Windows\\System32\\taskkill", "/F", "/T", "/PID", - str(proc.pid)], - shell=False) - except (FileNotFoundError, KeyboardInterrupt, AttributeError) as error: - LOG.exception("timeout callback exception: %s" % error) - - -def get_dump_str(dump_file): - dump_str = "" - if os.path.exists(dump_file): - with open(dump_file, "r") as json_content: - dump_str = json_content.read() - return dump_str - - -def get_center_position(parsed_dict, attr_name, attr_value): - if attr_value == parsed_dict.get(attr_name, ""): - return _calculate_center_position(parsed_dict) - - child_dicts = parsed_dict.get("child", []) - for child_dict in child_dicts: - x_center, y_center = get_center_position(child_dict, attr_name, - attr_value) - if check_position(x_center, y_center): - return x_center, y_center - - return -1, -1 - - -def check_position(x_position, y_position): - return (0 <= x_position <= MAX_VALID_POSITION) and \ - (0 <= y_position <= MAX_VALID_POSITION) - - -def copy_file(destination_path, screen_file_name): - from xdevice import Variables - abs_paths = [Variables.exec_dir, Variables.top_dir, Variables.modules_dir] - for path in abs_paths: - if path: - for file_name in os.listdir(path): - if file_name.endswith(".bin") and os.path.exists(os.path.join( - path, file_name)): - dst_file = "%s%s" % (os.path.join( - destination_path, screen_file_name), ".bin") - copyfile(os.path.join(path, file_name), dst_file) - os.remove(os.path.join(path, file_name)) - break - - -def _calculate_center_position(parsed_dict): - x_value = parsed_dict.get("x", -1) - y_value = parsed_dict.get("y", -1) - if not check_position(x_value, y_value): - LOG.info("error (x, y) value, (%s, %s)" % (x_value, y_value)) - return -1, -1 - width = parsed_dict.get("width", -1) - height = parsed_dict.get("height", -1) - if not check_position(width, height): - LOG.info("error (width, height) value, (%s, %s)" % (width, height)) - return -1, -1 - return min((x_value + width / 2), 454), min((y_value + height / 2), 454) - - -def get_center(dump_str, attr_name, attr_value): - """execute hd.click method - - Parameters: - dump_str: json string - attr_name: target attribute name like 'id' or 'text' - attr_value: target attribute value - - Return: - 0: success 1: fail - """ - try: - parsed_dict = json.loads(dump_str, encoding="utf-8") - except JSONDecodeError as error: - LOG.info("format error, %s" % error.args) - return 1 - x_center, y_center = get_center_position(parsed_dict, attr_name, - attr_value) - LOG.info("(x_center, y_center) value, (%s, %s)" % (x_center, y_center)) - if check_position(x_center, y_center): - return True, (x_center, y_center) - else: - LOG.info("no valid center position (%s, %s)" % (x_center, y_center)) - return False, None - - -def get_hdc_command(command, shell=True): - if shell: - return " ".join(["hdc", "shell", command]) - else: - return " ".join(["hdc", command]) - - -def filter_json(result_str=""): - result = "no such node" - if result_str.find("{") != -1: - result_str = result_str.replace("\n", "").strip() - pattern = r"(.*)(\{.*\})(.*)" - matcher = re.match(pattern, result_str) - if matcher: - return matcher.group(2) - return result - - -@dataclass -class Timeout: - is_timeout = False - - -@Plugin(type=Plugin.TEST_KIT, id=CKit.liteuikit) -class LiteUiKit: - def __init__(self): - pass - - def __set_device__(self, device): - pass - - def __set_connect_type__(self, device): - pass - - def __check_config__(self, config): - pass - - def __setup__(self, device): - pass - - def __teardown__(self, device): - pass - - @staticmethod - def execute_hdc_cmd_with_timeout(device, command, timeout=800, - result_print=True): - """ - Executes a command on the device with timeout. - - Parameters: - device: device - command: the command to execute - timeout: time out value - result_print: - """ - from _core.environment.device_lite import get_hdc_path - cmd = [get_hdc_path(), "-p", device.serial_port.upper().replace( - "COM", "").strip()] + command.split(" ") - ret_message = "" - LOG.info("execute command: %s" % " ".join(cmd)) - - start_time = datetime.datetime.now() - LOG.info("starttime=%s with timeout=%s" % ( - start_time.strftime("%Y-%m-%d %H:%M:%S"), str(timeout))) - - proc = subprocess.Popen( - cmd, stdout=subprocess.PIPE, - stderr=subprocess.PIPE, shell=False, - preexec_fn=os.setsid if platform.system() != 'Windows' else None) - is_timeout = Timeout() - proc_timer = Timer(timeout, timeout_callback, [proc, is_timeout]) - proc_timer.start() - - try: - ret_message = LiteUiKit._result_hdc_out_process( - proc, is_timeout, result_print) - except (ExecuteTerminate, ValueError) as exception: - LOG.exception("exception: %s", str(exception)) - finally: - ret_message = "{}{}".format( - ret_message, LiteUiKit._print_hdc_stdout(proc, result_print)) - error_message = LiteUiKit._print_hdc_stderr(proc, result_print) - - proc_timer.cancel() - proc.stdout.close() - proc.stderr.close() - - LOG.info("end time=%s delta=%s" % ( - datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S"), str( - (datetime.datetime.now() - start_time).seconds))) - - if proc.returncode == 0: - LOG.info('Info: execute command success. command=%s', command) - return_code = 0 - else: - LOG.error('Error: execute command failed. command=%s', command) - return_code = 1 - return ResultValue(return_code, error_message, ret_message) - - @staticmethod - def _result_hdc_out_process(proc, timeout, result_print): - result = "" - from xdevice import Scheduler - while proc.poll() is None: - if not Scheduler.is_execute: - raise ExecuteTerminate() - line = proc.stdout.readline() - line = line.strip() - if isinstance(line, bytes): - line = line.decode('utf-8', 'ignore').strip() - if line != "": - result = "%s%s%s" % (result, line, "\n") - if result_print: - LOG.info(line) - if timeout.is_timeout: - LOG.info("timeout flag is True") - timeout.is_timeout = False - break - return result - - @classmethod - def _print_hdc_stdout(cls, proc, result_print=True): - data = proc.stdout.read() - if isinstance(data, bytes): - data = data.decode('utf-8', 'ignore') - if data != "": - if not result_print: - return data - for line in data.rstrip("\n").split("\n"): - LOG.info(line) - return data.rstrip("\n") - return "" - - @classmethod - def _print_hdc_stderr(cls, proc, result_print=True): - data = proc.stderr.read() - if isinstance(data, bytes): - data = data.decode('utf-8', 'ignore') - if data != "": - if not result_print: - return data - LOG.error("----------stderr info start------------") - for error_message in data.rstrip("\n").split("\n"): - LOG.error(error_message) - LOG.error("----------stderr info ended------------") - return data.rstrip("\n") - return "" - - @staticmethod - def click(device, x_coordinate, y_coordinate, timeout=800): - press_command = "%s%s%s%s%s" % ("uievent ", str(x_coordinate), str( - y_coordinate), " ", " PRESSDOWN") - release_command = "%s%s%s%s%s" % ("uievent ", str(x_coordinate), " ", - str(y_coordinate), " RELEASE") - press_result = LiteUiKit.execute_hdc_cmd_with_timeout( - device, get_hdc_command(press_command, True), timeout, True) - if press_result.return_code == 0: - time.sleep(timeout / 1000) - release_result = LiteUiKit.execute_hdc_cmd_with_timeout( - device, get_hdc_command(release_command, True), timeout, True) - if release_result.return_code == 0: - LOG.info("click success.") - return True - return False - - @staticmethod - def click_id(device, node_id, timeout=800): - if not node_id: - raise ParamError("miss node id") - status, point = LiteUiKit.ui_dump_id( - device, node_id=node_id, timeout=timeout) - if status: - LiteUiKit.click(device, point[0], point[1], timeout) - else: - LOG.info("click failed") - - @staticmethod - def click_text(device, text="", timeout=800): - status = False - if not text: - raise ParamError("miss node text") - dump_str = LiteUiKit.ui_dump_tree( - device, timeout=timeout) - if dump_str and text: - status, point = get_center(dump_str, "text", text) - if status: - LiteUiKit.click(point[0], point[1], timeout) - else: - LOG.info("click failed") - - @staticmethod - def screen_shot(device, screen_path, file_name, timeout=800): - result_value = LiteUiKit.execute_hdc_cmd_with_timeout( - device, get_hdc_command("screenshot"), timeout, True) - if result_value.return_code == 0: - retry_times = 2 - while retry_times > 0: - command = "rfile user/log/screenshot.bin" - if LiteUiKit.download_file(device, command, timeout=1400): - img_dir = os.path.join(screen_path, "img") - os.makedirs(img_dir, exist_ok=True) - copy_file(img_dir, file_name) - break - retry_times -= 1 - LOG.info("screen shot success.") - return True - - @staticmethod - def ui_dump_id(device, **kwargs): - args = kwargs - node_id = args.get("node_id", "") - timeout = args.get("timeout", "") - dump_node_command = "%s%s" % ("uidump node ", str(node_id)) - dump_times = 5 - while dump_times > 0: - dump_node_value = LiteUiKit.execute_hdc_cmd_with_timeout( - device, get_hdc_command(dump_node_command), timeout, True) - if dump_node_value.return_code == 0: - result = filter_json(dump_node_value.return_message) - if result != "no such node": - return get_center(result, "id", node_id) - dump_times -= 1 - LOG.error("dump fail, please check node id") - return None - - @staticmethod - def ui_dump_tree(device, **kwargs): - args = kwargs - node_id = args.get("node_id", "") - timeout = args.get("timeout", "") - if node_id: - dump_tree_command = "%s%s" % ("uidump tree ", node_id) - else: - dump_tree_command = "uidump tree" - dump_tree_value = LiteUiKit.execute_hdc_cmd_with_timeout( - device, get_hdc_command(dump_tree_command), timeout, True) - if dump_tree_value.return_code == 0: - dump_str = "" - from _core.environment.device_lite import get_hdc_path - local_path = get_hdc_path() - retry_times = 2 - while retry_times > 0: - command = "rfile user/log/dump_dom_tree.json" - if LiteUiKit.download_file(device, command, timeout=1400): - with open(local_path, "r") as file_stream: - dump_str = file_stream.read() - break - retry_times -= 1 - return dump_str - else: - LOG.error("dump failed") - raise ParamError("dump failed") - - @staticmethod - def swipe(device, start_point, end_point, timeout=800): - if not isinstance(start_point, tuple) and not isinstance( - end_point, tuple): - raise ParamError( - "The coordinates of the sliding point should be tuple") - start_x, start_y = start_point - end_x, end_y = end_point - press_first_command = "%s%s%s%s%s" % ("uievent ", str(start_x), " ", - str(start_y), " PRESSDOWN") - press_end_command = "%s%s%s%s" % ("uievent ", str(end_x), str(end_y), - " PRESSDOWN") - release_end_command = "%s%s%s%s%s" % ("uievent ", str(end_x), " ", - str(end_y), " RELEASE") - press_first_point_result = LiteUiKit.execute_hdc_cmd_with_timeout( - device, get_hdc_command(press_first_command, True), timeout) - if press_first_point_result.return_code == 0: - time.sleep(timeout / 1000) - press_medium_point_result = \ - LiteUiKit.execute_hdc_cmd_with_timeout( - device, get_hdc_command(press_end_command, True), - timeout, True) - if press_medium_point_result.return_code == 0: - time.sleep(timeout / 1000) - press_end_point_result = \ - LiteUiKit.execute_hdc_cmd_with_timeout( - device, get_hdc_command(release_end_command, True), - timeout, True) - if press_end_point_result.return_code == 0: - LOG.info("swipe success.") - return True - return False - - @staticmethod - def long_press(device, x_coordinate, y_coordinate, timeout=1200): - press_command = "%s%s%s%s%s" % ("uievent ", str(x_coordinate), " ", - str(y_coordinate), " PRESSDOWN") - release_command = "%s%s%s%s%s" % ("uievent ", str(x_coordinate), " ", - str(y_coordinate), " RELEASE") - press_result = LiteUiKit.execute_hdc_cmd_with_timeout( - device, get_hdc_command(press_command, True), timeout, True) - if press_result.return_code == 0: - time.sleep(timeout / 1000) - release_result = LiteUiKit.execute_hdc_cmd_with_timeout( - device, get_hdc_command(release_command, True), timeout, True) - if release_result.return_code == 0: - LOG.info("press success.") - return True - return False - - @staticmethod - def download_file(device, command, timeout=1200): - press_result = LiteUiKit.execute_hdc_cmd_with_timeout( - device, command, timeout, True) - if press_result.return_code == 0: - LOG.info("download file success.") - return True - return False - - -class ResultValue(object): - def __init__(self, return_code, error_message="", return_message=""): - self.return_code = return_code - self.error_message = error_message - self.return_message = return_message - - def set_return_code(self, ret_code): - self.return_code = ret_code - - def set_error_message(self, error_message): - self.error_message = error_message - - def set_return_message(self, return_message): - self.return_message = return_message - - def get_return_code(self): - return self.return_code - - def get_error_message(self): - return self.error_message - - def get_return_message(self): - return self.return_message diff --git a/src/xdevice/_core/utils.py b/src/xdevice/_core/utils.py index 205f9accb65e83f5aa4b91464cd2d6f48776ce2c..ba33461ddbcf658839d6226e2c859a6dda5d0e80 100755 --- a/src/xdevice/_core/utils.py +++ b/src/xdevice/_core/utils.py @@ -2,7 +2,7 @@ # coding=utf-8 # -# Copyright (c) 2020 Huawei Device Co., Ltd. +# Copyright (c) 2020-2021 Huawei Device Co., Ltd. # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at @@ -16,6 +16,7 @@ # limitations under the License. # +import copy import os import socket import time @@ -25,15 +26,19 @@ import subprocess import signal import uuid import json +import stat from tempfile import NamedTemporaryFile from _core.executor.listener import SuiteResult from _core.driver.parser_lite import ShellHandler from _core.exception import ParamError +from _core.exception import ExecuteTerminate from _core.logger import platform_logger from _core.report.suite_reporter import SuiteReporter from _core.plugin import get_plugin from _core.plugin import Plugin +from _core.constants import ModeType +from _core.constants import ConfigConst LOG = platform_logger("Utils") @@ -95,31 +100,27 @@ def stop_standing_subprocess(process): def get_decode(stream): - if not isinstance(stream, str) and not isinstance(stream, bytes): + if isinstance(stream, str): + return stream + + if not isinstance(stream, bytes): + return str(stream) + + try: + ret = stream.decode("utf-8", errors="ignore") + except (ValueError, AttributeError, TypeError): ret = str(stream) - else: - try: - ret = stream.decode("utf-8", errors="ignore") - except (ValueError, AttributeError, TypeError): - ret = str(stream) return ret def is_proc_running(pid, name=None): if platform.system() == "Windows": - proc_sub = subprocess.Popen(["C:\\Windows\\System32\\tasklist"], - stdout=subprocess.PIPE, - shell=False) - proc = subprocess.Popen(["C:\\Windows\\System32\\findstr", "%s" % pid], - stdin=proc_sub.stdout, - stdout=subprocess.PIPE, shell=False) + list_command = ["C:\\Windows\\System32\\tasklist"] + find_command = ["C:\\Windows\\System32\\findstr", "%s" % pid] else: - proc_sub = subprocess.Popen(["/bin/ps", "-ef"], - stdout=subprocess.PIPE, - shell=False) - proc = subprocess.Popen(["/bin/grep", "%s" % pid], - stdin=proc_sub.stdout, - stdout=subprocess.PIPE, shell=False) + list_command = ["/bin/ps", "-ef"] + find_command = ["/bin/grep", "%s" % pid] + proc = _get_find_proc(find_command, list_command) (out, _) = proc.communicate() out = get_decode(out).strip() if out == "": @@ -128,7 +129,15 @@ def is_proc_running(pid, name=None): return True if name is None else out.find(name) != -1 -def exec_cmd(cmd, timeout=5 * 60, error_print=True): +def _get_find_proc(find_command, list_command): + proc_sub = subprocess.Popen(list_command, stdout=subprocess.PIPE, + shell=False) + proc = subprocess.Popen(find_command, stdin=proc_sub.stdout, + stdout=subprocess.PIPE, shell=False) + return proc + + +def exec_cmd(cmd, timeout=5 * 60, error_print=True, join_result=False): """ Executes commands in a new shell. Directing stderr to PIPE. @@ -139,35 +148,38 @@ def exec_cmd(cmd, timeout=5 * 60, error_print=True): cmd: A sequence of commands and arguments. timeout: timeout for exe cmd. error_print: print error output or not. - + join_result: join error and out Returns: The output of the command run. """ sys_type = platform.system() - if sys_type == "Windows": - proc = subprocess.Popen(cmd, stdout=subprocess.PIPE, - stderr=subprocess.PIPE, shell=False) - elif sys_type == "Linux" or sys_type == "Darwin": + if sys_type == "Linux" or sys_type == "Darwin": proc = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, shell=False, - preexec_fn=os.setsid) # @UndefinedVariable + preexec_fn=os.setsid) else: - return + proc = subprocess.Popen(cmd, stdout=subprocess.PIPE, + stderr=subprocess.PIPE, shell=False) try: (out, err) = proc.communicate(timeout=timeout) err = get_decode(err).strip() out = get_decode(out).strip() if err and error_print: - LOG.exception(err) - return err if err else out + LOG.exception(err, exc_info=False) + if join_result: + return "%s\n %s" % (out, err) if err else out + else: + return err if err else out - except (TimeoutError, KeyboardInterrupt, AttributeError, ValueError): + except (TimeoutError, KeyboardInterrupt, AttributeError, ValueError, + EOFError, IOError): sys_type = platform.system() - if sys_type == "Windows": - os.kill(proc.pid, signal.SIGINT) - elif sys_type == "Linux" or sys_type == "Darwin": + if sys_type == "Linux" or sys_type == "Darwin": os.killpg(proc.pid, signal.SIGTERM) + else: + os.kill(proc.pid, signal.SIGINT) + raise def create_dir(path): @@ -182,15 +194,26 @@ def create_dir(path): def get_config_value(key, config_dict, is_list=True, default=None): + """get corresponding values for key in config_dict + + Args: + key: target key in config_dict + config_dict: dictionary that store values + is_list: decide return values is list type or not + default: if key not in config_dict, default value will be returned + + Returns: + corresponding values for key + """ if not isinstance(config_dict, dict): return default - value = config_dict.get(key, "") + value = config_dict.get(key, None) if isinstance(value, bool): return value - if not value: - if default: + if value is None: + if default is not None: return default return [] if is_list else "" @@ -200,28 +223,48 @@ def get_config_value(key, config_dict, is_list=True, default=None): def get_file_absolute_path(input_name, paths=None, alt_dir=None): + """find absolute path for input_name + + Args: + input_name: the target file to search + paths: path list for searching input_name + alt_dir: extra dir that appended to paths + + Returns: + absolute path for input_name + """ + input_name = str(input_name) abs_paths = set(paths) if paths else set() _update_paths(abs_paths) - for path in abs_paths: - if alt_dir: - file_path = os.path.join(path, alt_dir, input_name) + _inputs = [input_name] + if input_name.startswith("resource/"): + _inputs.append(input_name.replace("resource/", "", 1)) + elif input_name.startswith("testcases/"): + _inputs.append(input_name.replace("testcases/", "", 1)) + + for _input in _inputs: + for path in abs_paths: + if alt_dir: + file_path = os.path.join(path, alt_dir, _input) + if os.path.exists(file_path): + return os.path.abspath(file_path) + + file_path = os.path.join(path, _input) if os.path.exists(file_path): return os.path.abspath(file_path) - file_path = os.path.join(path, input_name) - if os.path.exists(file_path): - return os.path.abspath(file_path) - err_msg = "The file {} does not exist".format(input_name) - LOG.error(err_msg) + if check_mode(ModeType.decc): + LOG.error(err_msg, error_no="00109") + err_msg = "Load Error[00109]" if alt_dir: LOG.debug("alt_dir is %s" % alt_dir) LOG.debug("paths is:") for path in abs_paths: LOG.debug(path) - raise ParamError(err_msg) + raise ParamError(err_msg, error_no="00109") def _update_paths(paths): @@ -267,7 +310,9 @@ def modify_props(device, local_prop_file, target_prop_file, new_props): old_props = {} changed_prop_key = [] lines = [] - with open(local_prop_file, 'r') as old_file: + flags = os.O_RDONLY + modes = stat.S_IWUSR | stat.S_IRUSR + with os.fdopen(os.open(local_prop_file, flags, modes), "r") as old_file: lines = old_file.readlines() if lines: lines[-1] = lines[-1] + '\n' @@ -304,25 +349,42 @@ def modify_props(device, local_prop_file, target_prop_file, new_props): return is_changed -def get_device_log_file(report_path, serial=None, log_name="device_log"): +def get_device_log_file(report_path, serial=None, log_name="device_log", + device_name=""): from xdevice import Variables log_path = os.path.join(report_path, Variables.report_vars.log_dir) os.makedirs(log_path, exist_ok=True) serial = serial or time.time_ns() - device_file_name = "{}_{}.log".format(log_name, serial) + if device_name: + serial = "%s_%s" % (device_name, serial) + device_file_name = "{}_{}.log".format(log_name, str(serial).replace( + ":", "_")) device_log_file = os.path.join(log_path, device_file_name) + LOG.info("generate device log file: %s", device_log_file) return device_log_file -def check_result_report(report_path, report_file, error_message="", - report_name=""): +def check_result_report(report_root_dir, report_file, error_message="", + report_name="", module_name=""): + """ + check whether report_file exits or not. if report_file is not exist, + create empty report with error_message under report_root_dir + """ + if os.path.exists(report_file): return report_file - result_dir = os.path.join(report_path, "result") + report_dir = os.path.dirname(report_file) + if os.path.isabs(report_dir): + result_dir = report_dir + else: + result_dir = os.path.join(report_root_dir, "result", report_dir) os.makedirs(result_dir, exist_ok=True) - LOG.error("report %s not exist, create empty report under %s" % ( - report_file, result_dir)) + if check_mode(ModeType.decc): + LOG.error("report not exist, create empty report") + else: + LOG.error("report %s not exist, create empty report under %s" % ( + report_file, result_dir)) suite_name = report_name if not suite_name: @@ -330,12 +392,28 @@ def check_result_report(report_path, report_file, error_message="", suite_result = SuiteResult() suite_result.suite_name = suite_name suite_result.stacktrace = error_message + if module_name: + suite_name = module_name suite_reporter = SuiteReporter([(suite_result, [])], suite_name, - result_dir) + result_dir, modulename=module_name) suite_reporter.create_empty_report() return "%s.xml" % os.path.join(result_dir, suite_name) +def get_sub_path(test_suite_path): + pattern = "%stests%s" % (os.sep, os.sep) + file_dir = os.path.dirname(test_suite_path) + pos = file_dir.find(pattern) + if -1 == pos: + return "" + + sub_path = file_dir[pos + len(pattern):] + pos = sub_path.find(os.sep) + if -1 == pos: + return "" + return sub_path[pos + len(os.sep):] + + def is_config_str(content): return True if "{" in content and "}" in content else False @@ -346,8 +424,9 @@ def get_version(): ver_file_path = os.path.join(Variables.res_dir, 'version.txt') if not os.path.isfile(ver_file_path): return ver - - with open(ver_file_path, mode='r', encoding='UTF-8') as ver_file: + flags = os.O_RDONLY + modes = stat.S_IWUSR | stat.S_IRUSR + with os.fdopen(os.open(ver_file_path, flags, modes), "r") as ver_file: line = ver_file.readline() if '-v' in line: ver = line.strip().split('-')[1] @@ -370,17 +449,19 @@ def convert_ip(origin_ip): def convert_port(port): - if len(port) >= 2: - return "{}**{}".format(port[0], port[-1]) + _port = str(port) + if len(_port) >= 2: + return "{}{}{}".format(_port[0], "*" * (len(_port) - 2), _port[-1]) else: - return "{}**".format(port) + return "*{}".format(_port[-1]) def convert_serial(serial): if serial.startswith("local_"): - return "local_{}".format('*'*(len(serial)-6)) + return serial elif serial.startswith("remote_"): - return "remote_{}".format(convert_ip(serial.split("_")[1])) + return "remote_{}_{}".format(convert_ip(serial.split("_")[1]), + convert_port(serial.split("_")[-1])) else: length = len(serial)//3 return "{}{}{}".format( @@ -402,25 +483,45 @@ def get_shell_handler(request, parser_type): return handler -def get_kit_instances(json_config, resource_path, testcases_path): +def get_kit_instances(json_config, resource_path="", testcases_path=""): from _core.testkit.json_parser import JsonParser kit_instances = [] + + # check input param if not isinstance(json_config, JsonParser): return kit_instances + + # get kit instances for kit in json_config.config.kits: kit["paths"] = [resource_path, testcases_path] kit_type = kit.get("type", "") + device_name = kit.get("device_name", None) if get_plugin(plugin_type=Plugin.TEST_KIT, plugin_id=kit_type): test_kit = \ get_plugin(plugin_type=Plugin.TEST_KIT, plugin_id=kit_type)[0] test_kit_instance = test_kit.__class__() test_kit_instance.__check_config__(kit) + setattr(test_kit_instance, "device_name", device_name) kit_instances.append(test_kit_instance) else: - raise ParamError("kit %s not exists" % kit_type) + raise ParamError("kit %s not exists" % kit_type, error_no="00107") return kit_instances +def check_device_name(device, kit, step="setup"): + kit_device_name = getattr(kit, "device_name", None) + device_name = device.get("name") + if kit_device_name and device_name and \ + kit_device_name != device_name: + return False + if kit_device_name and device_name: + LOG.debug("do kit:%s %s for device:%s", + kit.__class__.__name__, step, device_name) + else: + LOG.debug("do kit:%s %s", kit.__class__.__name__, step) + return True + + def check_path_legal(path): if path and " " in path: return "\"%s\"" % path @@ -458,6 +559,8 @@ def get_local_ip(): else: local_ip = "127.0.0.1" return local_ip + else: + return "127.0.0.1" class SplicingAction(argparse.Action): @@ -466,16 +569,55 @@ class SplicingAction(argparse.Action): def get_test_component_version(config): + if check_mode(ModeType.decc): + return "" + try: paths = [config.resource_path, config.testcases_path] test_file = get_file_absolute_path("test_component.json", paths) - - with open(test_file, encoding="utf-8") as file_content: + flags = os.O_RDONLY + modes = stat.S_IWUSR | stat.S_IRUSR + with os.fdopen(os.open(test_file, flags, modes), "r") as file_content: json_content = json.load(file_content) version = json_content.get("version", "") return version except (ParamError, ValueError) as error: - LOG.error( - "The exception {} happened when get version".format( - error)) + LOG.error("The exception {} happened when get version".format(error)) return "" + + +def check_mode(mode): + from xdevice import Scheduler + return Scheduler.mode == mode + + +def do_module_kit_setup(request, kits): + for device in request.get_devices(): + setattr(device, ConfigConst.module_kits, []) + + from xdevice import Scheduler + for kit in kits: + run_flag = False + for device in request.get_devices(): + if not Scheduler.is_execute: + raise ExecuteTerminate() + if check_device_name(device, kit): + run_flag = True + kit_copy = copy.deepcopy(kit) + module_kits = getattr(device, ConfigConst.module_kits) + module_kits.append(kit_copy) + kit_copy.__setup__(device, request=request) + if not run_flag: + kit_device_name = getattr(kit, "device_name", None) + error_msg = "device name '%s' of '%s' not exist" % ( + kit_device_name, kit.__class__.__name__) + LOG.error(error_msg, error_no="00108") + raise ParamError(error_msg, error_no="00108") + + +def do_module_kit_teardown(request): + for device in request.get_devices(): + for kit in getattr(device, ConfigConst.module_kits, []): + if check_device_name(device, kit, step="teardown"): + kit.__teardown__(device) + setattr(device, ConfigConst.module_kits, []) diff --git a/src/xdevice/variables.py b/src/xdevice/variables.py index 6c6a5234594af3c644c6a7a73533502268d77dab..594ad67d5587cc4892b5cecbf8d1ce704b3aafd4 100755 --- a/src/xdevice/variables.py +++ b/src/xdevice/variables.py @@ -2,7 +2,7 @@ # coding=utf-8 # -# Copyright (c) 2020 Huawei Device Co., Ltd. +# Copyright (c) 2020-2021 Huawei Device Co., Ltd. # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at @@ -23,12 +23,16 @@ from dataclasses import dataclass __all__ = ["Variables"] SRC_DIR = os.path.abspath(os.path.dirname(os.path.dirname(__file__))) +SRC_ADAPTER_DIR = os.path.abspath(os.path.join(SRC_DIR, "adapter")) MODULES_DIR = os.path.abspath(os.path.dirname(__file__)) TOP_DIR = os.path.abspath( os.path.dirname(os.path.dirname(os.path.dirname(__file__)))) +TOP_ADAPTER_DIR = os.path.abspath(os.path.join(TOP_DIR, "adapter")) sys.path.insert(0, SRC_DIR) sys.path.insert(1, MODULES_DIR) sys.path.insert(2, TOP_DIR) +sys.path.insert(3, SRC_ADAPTER_DIR) +sys.path.insert(4, TOP_ADAPTER_DIR) @dataclass @@ -39,6 +43,7 @@ class ReportVariables: log_level = "" log_handler = "" pub_key_file = None + pub_key_string = "" @dataclass @@ -65,8 +70,8 @@ def _init_global_config(): # set report variables Variables.report_vars.log_dir = "log" Variables.report_vars.report_dir = "reports" - Variables.report_vars.log_format = "%(asctime)s %(name)-15s " \ - "%(levelname)-8s %(message)s" + Variables.report_vars.log_format = "[%(asctime)s] [%(name)s] " \ + "[%(levelname)s] %(message)s" Variables.report_vars.log_level = logging.INFO Variables.report_vars.log_handler = "console, file" @@ -111,7 +116,8 @@ def _init_logger(): tool_log_file = None if Variables.exec_dir and os.path.normcase( - Variables.exec_dir) == os.path.normcase(Variables.top_dir): + Variables.exec_dir) == os.path.normcase(Variables.top_dir) and \ + not hasattr(sys, "decc_mode"): host_log_path = os.path.join(Variables.exec_dir, Variables.report_vars.report_dir, Variables.report_vars.log_dir)