openharmony/testfwk_xdevice
1
0
Fork 0
mirror of https://gitee.com/openharmony/testfwk_xdevice synced 2024-11-30 11:00:37 +00:00
Code Issues Actions Packages Projects Releases Wiki Activity

!203 优化设备日志抓取逻辑

Browse Source 
Merge pull request !203 from liguangjie/master
This commit is contained in:
openharmony_ci 2023-01-19 02:29:11 +00:00 committed by Gitee
commit c4c23897a4
No known key found for this signature in database
GPG Key ID: 173E9B9CA92EEF8F
6 changed files with 278 additions and 2028 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
plugins/ohos/setup.py
View File

@ -46,8 +46,6 @@ def main():
'driver': [
'drivers=ohos.drivers.drivers',
'drivers_lite=ohos.drivers.drivers_lite',
'homevision=ohos.drivers.homevision',
'kunpeng=ohos.drivers.kunpeng',
'openharmony=ohos.drivers.openharmony'
],
'listener': [

1648
plugins/ohos/src/ohos/drivers/drivers.py
View File

File diff suppressed because it is too large Load Diff

134
plugins/ohos/src/ohos/drivers/homevision.py
View File

@ -1,134 +0,0 @@
#!/usr/bin/env python3
# coding=utf-8
#
# Copyright (c) 2022 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 os
import platform
import subprocess
import sys
from xdevice import IDriver
from xdevice import Plugin
from xdevice import platform_logger
from xdevice import get_device_log_file
from xdevice import JsonParser
from xdevice import get_config_value
from xdevice import FilePermission
from xdevice import get_decode
HOMEVISION_TEST = "HomeVision"
LOG = platform_logger(HOMEVISION_TEST)
@Plugin(type=Plugin.DRIVER, id=HOMEVISION_TEST)
class HomeVisionTest(IDriver):
"""
HomeVisionTest is a Test that runs a driver test on given tv devices.
"""
# test driver config
config = None
result = ""
def __check_environment__(self, device_options):
pass
def __check_config__(self, config):
pass
def __execute__(self, request):
self.config = request.config
self.config.device = request.get_devices()[0]
self.config.devices = request.get_devices()
# get config file
config_file = request.get_config_file()
if not config_file:
LOG.error("Config file not exists")
return
LOG.debug("HomeVisionTest config file Path: %s" % config_file)
device_log_pipes = []
try:
for device in self.config.devices:
device_name = device.get("name", "")
device_log = get_device_log_file(
request.config.report_path, device.__get_serial__(),
"device_log", device_name)
hilog = get_device_log_file(
request.config.report_path, device.__get_serial__(),
"device_hilog", device_name)
device_log_open = os.open(device_log, os.O_WRONLY |
os.O_CREAT | os.O_APPEND,
FilePermission.mode_755)
hilog_open = os.open(hilog, os.O_WRONLY | os.O_CREAT |
os.O_APPEND, FilePermission.mode_755)
device_log_pipe = os.fdopen(device_log_open, "a")
hilog_pipe = os.fdopen(hilog_open, "a")
device.start_catch_device_log(device_log_pipe, hilog_pipe)
device_log_pipes.extend([device_log_pipe, hilog_pipe])
self._run_homevision_test(config_file)
finally:
for device_log_pipe in device_log_pipes:
device_log_pipe.flush()
device_log_pipe.close()
for device in self.config.devices:
device.stop_catch_device_log()
@classmethod
def _run_homevision_test(cls, config_file):
from xdevice import Variables
# insert RegressionTest path for loading homevision module
homevision_test_module = os.path.join(Variables.exec_dir,
"RegressionTest")
sys.path.insert(1, homevision_test_module)
json_config = JsonParser(config_file)
device_ip = get_config_value("device-ip", json_config.get_driver(),
False)
job_id = get_config_value("job-id", json_config.get_driver(), False)
home_vision_app_name = get_config_value(
"home-vision-app-name", json_config.get_driver(), False)
cmd_parts = []
if platform.system() == "Windows":
cmd_parts.append("python")
else:
cmd_parts.append("python3")
relative_path = "startAutoTest.py"
cmd_parts.append(os.path.abspath(os.path.join(homevision_test_module,
relative_path)))
cmd_parts.append(device_ip)
cmd_parts.append(job_id)
cmd_parts.append(home_vision_app_name)
cmd = " ".join(cmd_parts)
LOG.info("Start HomeVision test with cmd: %s" % cmd)
try:
proc = subprocess.Popen(cmd_parts, shell=False,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE)
(out, _) = proc.communicate()
out = get_decode(out).strip()
for line in out.split("\n"):
LOG.info(line)
except (subprocess.CalledProcessError, FileNotFoundError,
Exception) as error:
LOG.error("HomeVision test error: %s" % error)
def __result__(self):
return self.result if os.path.exists(self.result) else ""

118
plugins/ohos/src/ohos/drivers/kunpeng.py
View File

@ -1,118 +0,0 @@
#!/usr/bin/env python3
# coding=utf-8
#
# Copyright (c) 2022 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 os
import platform
import subprocess
from subprocess import Popen, PIPE, STDOUT
import sys
import uuid
from xdevice import platform_logger
from xdevice import IDriver
from xdevice import Plugin
from xdevice import get_config_value
from xdevice import ConfigConst
KUNPENG_TEST = "KunpengTest"
LOG = platform_logger(KUNPENG_TEST)
@Plugin(type=Plugin.DRIVER, id=KUNPENG_TEST)
class KunpengTest(IDriver):
"""
KunpengTest is a Test that runs a host-driven test on given kunpeng
servers.
"""
# test driver config
config = None
result = ""
def __check_environment__(self, device_options):
pass
def __check_config__(self, config):
pass
def __execute__(self, request):
self.config = request.config
result_dir = os.path.join(request.config.report_path, "result")
# status.db dir
log_dir = os.path.join(request.config.report_path, "log")
if not os.path.exists(result_dir):
os.makedirs(result_dir)
if not os.path.exists(log_dir):
os.makedirs(log_dir)
self.result = os.path.join(result_dir, 'result.xml')
testargs = request.get(ConfigConst.testargs)
mainconfig_file = testargs.get("main_config_path", [None])[0]
project_src_path = testargs.get("main_config_path", [None])[1]
if not mainconfig_file:
LOG.info('can not find mainconfig_file in '
'testargs!!, use default mainconfig_file')
return
tmp_id = str(uuid.uuid1())
tmp_folder = os.path.join(self.config.report_path, "temp")
self.config.tmp_sub_folder = os.path.join(tmp_folder, "task_" + tmp_id)
os.makedirs(self.config.tmp_sub_folder, exist_ok=True)
# 3.test execution
# mainconfig_file
self._start_kunpengtest_with_cmd(mainconfig_file, project_src_path,
log_dir)
return
def _get_driver_config(self, json_config):
self.config.main_config = get_config_value(
'main_config', json_config.get_driver(), False)
self.config.test_bed = get_config_value(
'test_bed', json_config.get_driver(), False)
self.config.test_set = get_config_value(
'test_set', json_config.get_driver(), False)
def _start_kunpengtest_with_cmd(self, mainconfig_file, project_src_path,
log_dir): # , mainconfig_file
sys.path.append(project_src_path)
cmd_parts = []
if platform.system() == "Windows":
cmd_parts.append("python")
else:
cmd_parts.append("python3")
start_script_path = os.path.join(project_src_path, 'bin', 'kprun.py')
cmd_parts.append(start_script_path)
cmd_parts.append("-c")
cmd_parts.append(mainconfig_file)
cmd_parts.append("-rp")
cmd_parts.append(self.result)
cmd_parts.append("-le")
cmd_parts.append(log_dir)
cmd = " ".join(cmd_parts)
LOG.info("start kunpengtest with cmd: %s" % cmd)
try:
with Popen(cmd, shell=False, stdout=PIPE, stderr=STDOUT) as p:
for line in p.stdout:
if line:
LOG.info(line.strip().decode('utf-8'))
else:
break
except (subprocess.CalledProcessError, FileNotFoundError) as error:
LOG.error("kunpeng test error: %s" % error)
def __result__(self):
return self.result if os.path.exists(self.result) else ""

43
plugins/ohos/src/ohos/drivers/openharmony.py
View File

@ -61,6 +61,11 @@ class OHKernelTestDriver(IDriver):
self.kits = []
self.config = None
self.runner = None
# log
self.device_log = None
self.hilog = None
self.log_proc = None
self.hilog_proc = None
def __check_environment__(self, device_options):
pass
@ -80,24 +85,25 @@ class OHKernelTestDriver(IDriver):
self.result = "%s.xml" % \
os.path.join(request.config.report_path,
"result", request.get_module_name())
device_log = get_device_log_file(
self.device_log = get_device_log_file(
request.config.report_path,
request.config.device.__get_serial__(),
"device_log")
hilog = get_device_log_file(
self.hilog = get_device_log_file(
request.config.report_path,
request.config.device.__get_serial__(),
"device_hilog")
device_log_open = os.open(device_log, os.O_WRONLY | os.O_CREAT |
device_log_open = os.open(self.device_log, os.O_WRONLY | os.O_CREAT |
os.O_APPEND, FilePermission.mode_755)
hilog_open = os.open(hilog, os.O_WRONLY | os.O_CREAT | os.O_APPEND,
hilog_open = os.open(self.hilog, os.O_WRONLY | os.O_CREAT | os.O_APPEND,
FilePermission.mode_755)
self.config.device.device_log_collector.add_log_address(self.device_log, self.hilog)
with os.fdopen(device_log_open, "a") as log_file_pipe, \
os.fdopen(hilog_open, "a") as hilog_file_pipe:
self.config.device.start_catch_device_log(log_file_pipe,
hilog_file_pipe)
self.log_proc, self.hilog_proc = self.config.device.device_log_collector.\
start_catch_device_log(log_file_pipe, hilog_file_pipe)
self._run_oh_kernel(config_file, request.listeners, request)
log_file_pipe.flush()
hilog_file_pipe.flush()
@ -109,7 +115,9 @@ class OHKernelTestDriver(IDriver):
raise exception
finally:
do_module_kit_teardown(request)
self.config.device.stop_catch_device_log()
self.config.device.device_log_collector.remove_log_address(self.device_log, self.hilog)
self.config.device.device_log_collector.stop_catch_device_log(self.log_proc)
self.config.device.device_log_collector.stop_catch_device_log(self.hilog_proc)
self.result = check_result_report(
request.config.report_path, self.result, self.error_message)
@ -217,6 +225,11 @@ class OHJSUnitTestDriver(IDriver):
self.runner = None
self.rerun = True
self.rerun_all = True
# log
self.device_log = None
self.hilog = None
self.log_proc = None
self.hilog_proc = None
def __check_environment__(self, device_options):
pass
@ -242,20 +255,22 @@ class OHJSUnitTestDriver(IDriver):
request.root.source.source_string, error_no="00110")
LOG.debug("Test case file path: %s" % suite_file)
self.config.device.set_device_report_path(request.config.report_path)
hilog = get_device_log_file(request.config.report_path,
self.hilog = get_device_log_file(request.config.report_path,
request.config.device.__get_serial__() + "_" + request.
get_module_name(),
"device_hilog")
hilog_open = os.open(hilog, os.O_WRONLY | os.O_CREAT | os.O_APPEND,
hilog_open = os.open(self.hilog, os.O_WRONLY | os.O_CREAT | os.O_APPEND,
0o755)
self.config.device.device_log_collector.add_log_address(self.device_log, self.hilog)
self.config.device.execute_shell_command(command="hilog -r")
with os.fdopen(hilog_open, "a") as hilog_file_pipe:
if hasattr(self.config, "device_log") \
and self.config.device_log == ConfigConst.device_log_on \
and hasattr(self.config.device, "clear_crash_log"):
self.config.device.clear_crash_log()
self.config.device.start_catch_device_log(hilog_file_pipe=hilog_file_pipe)
self.config.device.device_log_collector.clear_crash_log()
self.log_proc, self.hilog_proc = self.config.device.device_log_collector.\
start_catch_device_log(hilog_file_pipe=hilog_file_pipe)
self._run_oh_jsunit(config_file, request)
except Exception as exception:
self.error_message = exception
@ -270,8 +285,10 @@ class OHJSUnitTestDriver(IDriver):
if hasattr(self.config, "device_log") and \
self.config.device_log == ConfigConst.device_log_on \
and hasattr(self.config.device, "start_get_crash_log"):
self.config.device.start_get_crash_log(log_tar_file_name)
self.config.device.stop_catch_device_log()
self.config.device.device_log_collector.start_get_crash_log(log_tar_file_name)
self.config.device.device_log_collector.remove_log_address(self.device_log, self.hilog)
self.config.device.device_log_collector.stop_catch_device_log(self.log_proc)
self.config.device.device_log_collector.stop_catch_device_log(self.hilog_proc)
self.result = check_result_report(
request.config.report_path, self.result, self.error_message)

361
plugins/ohos/src/ohos/environment/device.py
View File

@ -149,8 +149,7 @@ class Device(IDevice):
log = platform_logger("Device")
device_state_monitor = None
reboot_timeout = 2 * 60 * 1000
log_file_pipe = None
hilog_file_pipe = None
_device_log_collector = None
_proxy = None
_is_harmony = None
@ -166,6 +165,7 @@ class Device(IDevice):
module_package = None
module_ablity_name = None
_device_log_path = ""
_device_report_path = ""
model_dict = {
'default': ProductForm.phone,
@ -285,7 +285,7 @@ class Device(IDevice):
self.device_state_monitor.wait_for_device_available(BACKGROUND_TIME)
cmd = "target mount"
self.connector_command(cmd)
self.start_catch_device_log()
self.device_log_collector.restart_catch_device_log()
return status
def wait_for_device_not_available(self, wait_time):
@ -308,7 +308,7 @@ class Device(IDevice):
self.device_state_monitor.wait_for_device_available(
self.reboot_timeout)
self.enable_hdc_root()
self.start_catch_device_log()
self.device_log_collector.restart_catch_device_log()
@perform_device_action
def install_package(self, package_path, command=""):
@ -385,152 +385,6 @@ class Device(IDevice):
return True
return False
def start_catch_device_log(self, log_file_pipe=None,
hilog_file_pipe=None):
"""
Starts hdc log for each device in separate subprocesses and save
the logs in files.
"""
self._sync_device_time()
if log_file_pipe:
self.log_file_pipe = log_file_pipe
if hilog_file_pipe:
self.hilog_file_pipe = hilog_file_pipe
self._start_catch_device_log()
def stop_catch_device_log(self):
"""
Stops all hdc log subprocesses.
"""
self._stop_catch_device_log()
def _start_catch_device_log(self):
if self.hilog_file_pipe:
command = "hilog"
if self.host != "127.0.0.1":
cmd = [HdcHelper.CONNECTOR_NAME, "-s", "{}:{}".format(self.host, self.port),
"-t", self.device_sn, "shell", command]
else:
cmd = [HdcHelper.CONNECTOR_NAME, "-t", self.device_sn, "shell", command]
LOG.info("execute command: %s" % " ".join(cmd).replace(
self.device_sn, convert_serial(self.device_sn)))
self.device_hilog_proc = start_standing_subprocess(
cmd, self.hilog_file_pipe)
def _stop_catch_device_log(self):
if self.device_hilog_proc:
stop_standing_subprocess(self.device_hilog_proc)
self.device_hilog_proc = None
self.hilog_file_pipe = None
def start_hilog_task(self, log_size="50M"):
self._sync_device_time()
self.clear_crash_log()
# 先停止一下
cmd = "hilog -w stop"
out = self.execute_shell_command(cmd)
# 清空日志
cmd = "hilog -r"
out = self.execute_shell_command(cmd)
cmd = "rm -rf /data/log/hilog/*"
out = self.execute_shell_command(cmd)
# 开始日志任务 设置落盘文件个数最大值1000, 单个文件20M，链接https://gitee.com/openharmony/hiviewdfx_hilog
cmd = "hilog -w start -l {} -n 1000".format(log_size)
out = self.execute_shell_command(cmd)
LOG.info("Execute command: {}, result is {}".format(cmd, out))
def stop_hilog_task(self, log_name):
cmd = "hilog -w stop"
out = self.execute_shell_command(cmd)
self.pull_file("/data/log/hilog/", "{}/log/".format(self._device_log_path))
try:
os.rename("{}/log/hilog".format(self._device_log_path),
"{}/log/{}_hilog".format(self._device_log_path, log_name))
except Exception as e:
self.log.error("Rename hilog folder {}_hilog failed. error: {}".format(log_name, e))
# 把hilog文件夹下所有文件拉出来 由于hdc不支持整个文件夹拉出只能采用先压缩再拉取文件
cmd = "cd /data/log/hilog && tar -zcvf /data/log/{}_hilog.tar.gz *".format(log_name)
out = self.execute_shell_command(cmd)
LOG.info("Execute command: {}, result is {}".format(cmd, out))
if "No space left on device" not in out:
self.pull_file("/data/log/{}_hilog.tar.gz".format(log_name), "{}/log/".format(self._device_log_path))
cmd = "rm -rf /data/log/{}_hilog.tar.gz".format(log_name)
out = self.execute_shell_command(cmd)
# 获取crash日志
self.start_get_crash_log(log_name)
def _get_log(self, log_cmd, *params):
def filter_by_name(log_name, args):
for starts_name in args:
if log_name.startswith(starts_name):
return True
return False
data_list = list()
log_name_array = list()
log_result = self.execute_shell_command(log_cmd)
if log_result is not None and len(log_result) != 0:
log_name_array = log_result.strip().replace("\r", "").split("\n")
for log_name in log_name_array:
log_name = log_name.strip()
if len(params) == 0 or \
filter_by_name(log_name, params):
data_list.append(log_name)
return data_list
def get_cur_crash_log(self, crash_path, log_name):
log_name_map = {'cppcrash': NATIVE_CRASH_PATH,
"jscrash": JS_CRASH_PATH,
"SERVICE_BLOCK": ROOT_PATH,
"appfreeze": ROOT_PATH}
if not os.path.exists(crash_path):
os.makedirs(crash_path)
if "Not support std mode" in log_name:
return
def get_log_path(logname):
name_array = logname.split("-")
if len(name_array) <= 1:
return ROOT_PATH
return log_name_map.get(name_array[0])
log_path = get_log_path(log_name)
temp_path = "%s/%s" % (log_path, log_name)
self.pull_file(temp_path, crash_path)
LOG.debug("Finish pull file: %s" % log_name)
def start_get_crash_log(self, task_name):
log_array = list()
native_crash_cmd = "ls {}".format(NATIVE_CRASH_PATH)
js_crash_cmd = '"ls {} | grep jscrash"'.format(JS_CRASH_PATH)
block_crash_cmd = '"ls {}"'.format(ROOT_PATH)
# 获取crash日志文件
log_array.extend(self._get_log(native_crash_cmd, "cppcrash"))
log_array.extend(self._get_log(js_crash_cmd, "jscrash"))
log_array.extend(self._get_log(block_crash_cmd, "SERVICE_BLOCK", "appfreeze"))
LOG.debug("crash log file {}, length is {}".format(str(log_array), str(len(log_array))))
crash_path = "{}/log/{}_crash_log/".format(self._device_log_path, task_name)
for log_name in log_array:
log_name = log_name.strip()
self.get_cur_crash_log(crash_path, log_name)
def clear_crash_log(self):
clear_block_crash_cmd = "rm -f {}/*".format(ROOT_PATH)
clear_native_crash_cmd = "rm -f {}/*".format(NATIVE_CRASH_PATH)
clear_debug_crash_cmd = "rm -f {}/debug/*".format(ROOT_PATH)
clear_js_crash_cmd = "rm -f {}/*".format(JS_CRASH_PATH)
self.execute_shell_command(clear_block_crash_cmd)
self.execute_shell_command(clear_native_crash_cmd)
self.execute_shell_command(clear_debug_crash_cmd)
self.execute_shell_command(clear_js_crash_cmd)
def _sync_device_time(self):
# 先同步PC和设备的时间
iso_time_format = '%Y-%m-%d %H:%M:%S'
cur_time = get_cst_time().strftime(iso_time_format)
self.execute_shell_command("date '{}'".format(cur_time))
self.execute_shell_command("hwclock --systohc")
def get_recover_result(self, retry=RETRY_ATTEMPTS):
command = "param get bootevent.boot.completed"
stdout = self.execute_shell_command(command, timeout=5 * 1000,
@ -562,6 +416,7 @@ class Device(IDevice):
self._proxy = None
self.remove_ports()
self.stop_harmony_rpc()
self.device_log_collector.stop_restart_catch_device_log()
@property
def proxy(self):
@ -796,17 +651,21 @@ class Device(IDevice):
path = ""
try:
temp_path = os.path.join(self._device_log_path, "temp")
if not os.path.exists(temp_path):
os.makedirs(temp_path)
path = os.path.join(temp_path, name)
self.execute_shell_command(
"snapshot_display -f /data/local/tmp/screen.png")
self.pull_file("/data/local/tmp/screen.png", path)
picture_name = os.path.basename(name)
out = self.execute_shell_command(
"snapshot_display -f /data/local/tmp/{}".format(picture_name))
self.log.debug("result: {}".format(out))
if "error" in out and "success" not in out:
return False
else:
self.pull_file("/data/local/tmp/{}".format(picture_name), path)
except Exception as error:
self.log.error("devicetest take_picture: {}".format(str(error)))
return path
def set_device_report_path(self, path):
self._device_log_path = path
def execute_shell_in_daemon(self, command):
if self.host != "127.0.0.1":
cmd = [HdcHelper.CONNECTOR_NAME, "-s", "{}:{}".format(
@ -827,3 +686,193 @@ class Device(IDevice):
else os.setsid,
close_fds=True)
return process
@property
def device_log_collector(self):
if self._device_log_collector is None:
self._device_log_collector = DeviceLogCollector(self)
return self._device_log_collector
def set_device_report_path(self, path):
self._device_log_path = path
def get_device_report_path(self):
return self._device_log_path
class DeviceLogCollector:
hilog_file_address = []
log_file_address = []
device = None
restart_proc = []
def __init__(self, device):
self.device = device
def restart_catch_device_log(self):
from xdevice import FilePermission
for _, path in enumerate(self.hilog_file_address):
hilog_open = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_APPEND,
FilePermission.mode_755)
with os.fdopen(hilog_open, "a") as hilog_file_pipe:
_, proc = self.start_catch_device_log(hilog_file_pipe=hilog_file_pipe)
self.restart_proc.append(proc)
def stop_restart_catch_device_log(self):
# when device free stop restart log proc
for _, proc in enumerate(self.restart_proc):
self.stop_catch_device_log(proc)
self.restart_proc.clear()
self.hilog_file_address.clear()
self.log_file_address.clear()
def start_catch_device_log(self, log_file_pipe=None,
hilog_file_pipe=None):
"""
Starts hdc log for each device in separate subprocesses and save
the logs in files.
"""
self._sync_device_time()
device_hilog_proc = None
if hilog_file_pipe:
command = "hilog"
if self.device.host != "127.0.0.1":
cmd = [HdcHelper.CONNECTOR_NAME, "-s", "{}:{}".format(self.device.host, self.device.port),
"-t", self.device.device_sn, "shell", command]
else:
cmd = [HdcHelper.CONNECTOR_NAME, "-t", self.device.device_sn, "shell", command]
LOG.info("execute command: %s" % " ".join(cmd).replace(
self.device.device_sn, convert_serial(self.device.device_sn)))
device_hilog_proc = start_standing_subprocess(
cmd, hilog_file_pipe)
return None, device_hilog_proc
def stop_catch_device_log(self, proc):
"""
Stops all hdc log subprocesses.
"""
if proc:
stop_standing_subprocess(proc)
self.device.log.debug("Stop catch device hilog.")
def start_hilog_task(self, log_size="50M"):
self._sync_device_time()
self.clear_crash_log()
# 先停止一下
cmd = "hilog -w stop"
out = self.device.execute_shell_command(cmd)
# 清空日志
cmd = "hilog -r"
out = self.device.execute_shell_command(cmd)
cmd = "rm -rf /data/log/hilog/*"
out = self.device.execute_shell_command(cmd)
# 开始日志任务 设置落盘文件个数最大值1000, 单个文件20M，链接https://gitee.com/openharmony/hiviewdfx_hilog
cmd = "hilog -w start -l {} -n 1000".format(log_size)
out = self.device.execute_shell_command(cmd)
LOG.info("Execute command: {}, result is {}".format(cmd, out))
def stop_hilog_task(self, log_name):
cmd = "hilog -w stop"
out = self.device.execute_shell_command(cmd)
self.device.pull_file("/data/log/hilog/", "{}/log/".format(self.device.get_device_report_path()))
try:
os.rename("{}/log/hilog".format(self.device.get_device_report_path()),
"{}/log/{}_hilog".format(self.device.get_device_report_path(), log_name))
except Exception as e:
self.device.log.warning("Rename hilog folder {}_hilog failed. error: {}".format(log_name, e))
# 把hilog文件夹下所有文件拉出来 由于hdc不支持整个文件夹拉出只能采用先压缩再拉取文件
cmd = "cd /data/log/hilog && tar -zcvf /data/log/{}_hilog.tar.gz *".format(log_name)
out = self.device.execute_shell_command(cmd)
LOG.info("Execute command: {}, result is {}".format(cmd, out))
if "No space left on device" not in out:
self.device.pull_file("/data/log/{}_hilog.tar.gz".format(log_name),
"{}/log/".format(self.device.get_device_report_path()))
cmd = "rm -rf /data/log/{}_hilog.tar.gz".format(log_name)
out = self.device.execute_shell_command(cmd)
# 获取crash日志
self.start_get_crash_log(log_name)
def _get_log(self, log_cmd, *params):
def filter_by_name(log_name, args):
for starts_name in args:
if log_name.startswith(starts_name):
return True
return False
data_list = list()
log_name_array = list()
log_result = self.device.execute_shell_command(log_cmd)
if log_result is not None and len(log_result) != 0:
log_name_array = log_result.strip().replace("\r", "").split("\n")
for log_name in log_name_array:
log_name = log_name.strip()
if len(params) == 0 or \
filter_by_name(log_name, params):
data_list.append(log_name)
return data_list
def get_cur_crash_log(self, crash_path, log_name):
log_name_map = {'cppcrash': NATIVE_CRASH_PATH,
"jscrash": JS_CRASH_PATH,
"SERVICE_BLOCK": ROOT_PATH,
"appfreeze": ROOT_PATH}
if not os.path.exists(crash_path):
os.makedirs(crash_path)
if "Not support std mode" in log_name:
return
def get_log_path(logname):
name_array = logname.split("-")
if len(name_array) <= 1:
return ROOT_PATH
return log_name_map.get(name_array[0])
log_path = get_log_path(log_name)
temp_path = "%s/%s" % (log_path, log_name)
self.device.pull_file(temp_path, crash_path)
LOG.debug("Finish pull file: %s" % log_name)
def start_get_crash_log(self, task_name):
log_array = list()
native_crash_cmd = "ls {}".format(NATIVE_CRASH_PATH)
js_crash_cmd = '"ls {} | grep jscrash"'.format(JS_CRASH_PATH)
block_crash_cmd = '"ls {}"'.format(ROOT_PATH)
# 获取crash日志文件
log_array.extend(self._get_log(native_crash_cmd, "cppcrash"))
log_array.extend(self._get_log(js_crash_cmd, "jscrash"))
log_array.extend(self._get_log(block_crash_cmd, "SERVICE_BLOCK", "appfreeze"))
LOG.debug("crash log file {}, length is {}".format(str(log_array), str(len(log_array))))
crash_path = "{}/log/{}_crash_log/".format(self.device.get_device_report_path(), task_name)
for log_name in log_array:
log_name = log_name.strip()
self.get_cur_crash_log(crash_path, log_name)
def clear_crash_log(self):
clear_block_crash_cmd = "rm -f {}/*".format(ROOT_PATH)
clear_native_crash_cmd = "rm -f {}/*".format(NATIVE_CRASH_PATH)
clear_debug_crash_cmd = "rm -f {}/debug/*".format(ROOT_PATH)
clear_js_crash_cmd = "rm -f {}/*".format(JS_CRASH_PATH)
self.device.execute_shell_command(clear_block_crash_cmd)
self.device.execute_shell_command(clear_native_crash_cmd)
self.device.execute_shell_command(clear_debug_crash_cmd)
self.device.execute_shell_command(clear_js_crash_cmd)
def _sync_device_time(self):
# 先同步PC和设备的时间
iso_time_format = '%Y-%m-%d %H:%M:%S'
cur_time = get_cst_time().strftime(iso_time_format)
self.device.execute_shell_command("date '{}'".format(cur_time))
self.device.execute_shell_command("hwclock --systohc")
def add_log_address(self, log_file_address, hilog_file_address):
# record to restart catch log when reboot device
if log_file_address:
self.log_file_address.append(log_file_address)
if hilog_file_address:
self.hilog_file_address.append(hilog_file_address)
def remove_log_address(self, log_file_address, hilog_file_address):
if log_file_address:
self.log_file_address.remove(log_file_address)
if hilog_file_address:
self.hilog_file_address.remove(hilog_file_address)