统一用例风格

Signed-off-by: liyanlin02 <liyanlin15@huawei.com>
This commit is contained in:
liyanlin02
2024-06-21 15:06:30 +08:00
parent 2359b79d48
commit 3326d775f3
32 changed files with 1677 additions and 32 deletions
@@ -1,4 +1,5 @@
# -*- coding:utf-8 -*-
import traceback
import uuid
import sys
import subprocess
@@ -19,6 +20,7 @@ from aw.Common.Constant import CONSTANT
from aw.Common.Common import getFileName
from aw.ExtractFile.ExtractFile import *
from aw.Common.Common import getHostIp, copyFile, copyDirectory
total_time = ""
lock_suffix = CONSTANT.File.LOCK_SUFFIX
suc_file = CONSTANT.File.SUC_FILE
@@ -26,7 +28,6 @@ failed_file = CONSTANT.File.FAILED_FILE
REBOOT_TIMEOUT = 20000000
class liteOsUpgrade_RK3568(BaseApp):
'''
@author: cwx1076044
@@ -76,7 +77,6 @@ class liteOsUpgrade_RK3568(BaseApp):
except Exception as e:
logger.error(e)
raise e
@dec_stepmsg("upgrade")
@timeout(3600)
@@ -91,7 +91,7 @@ class liteOsUpgrade_RK3568(BaseApp):
# @return: True or Flase
#===================================================================================
'''
global local_image_path, loader_tool_path, sn, LocationID ,test_num, system_type
global local_image_path, loader_tool_path, sn, LocationID, test_num, system_type
system_type = platform.system()
hostname = socket.gethostname()
ipaddress = socket.gethostbyname(hostname)
@@ -192,12 +192,14 @@ class liteOsUpgrade_RK3568(BaseApp):
# os.system("hdc_std -t %s shell hilog -w start -t kmsg" % sn)
if upgrade_test_type == "null":
return True
screenshot_path = os.path.join(local_image_path, "screenshot")
# 临时安装
self.install_third_packages()
resource_path = os.path.join(screenshot_path, 'resource')
logger.info(resource_path)
py_path = os.path.join(resource_path, "capturescreentest.py")
# py_path = "main.py"
screenshot_path = os.path.join(local_image_path, "screenshot")
script_path = os.path.join(screenshot_path, 'new_script')
logger.info(script_path)
# py_path = os.path.join(script_path, "main.py")
py_file = "main.py"
new_report_path = os.path.join(report_path, "result")
logger.info(new_report_path)
time_sleep = random.randint(3, 7)
@@ -216,11 +218,11 @@ class liteOsUpgrade_RK3568(BaseApp):
if not upgrade_test_type or upgrade_test_type == "smoke_test":
# 进到工程目录
# cur_path = os.getcwd()
# os.chdir(resource_path)
test_return = cmd_test(resource_path, py_path, new_report_path, resource_path, sn, test_num, pr_url)
cur_path = os.getcwd()
os.chdir(script_path)
test_return = cmd_test(script_path, py_file, sn, test_num, new_report_path)
# 执行完回到原来的目录
# os.chdir(cur_path)
os.chdir(cur_path)
if test_return == 1:
return True
if test_return == 98:
@@ -230,6 +232,20 @@ class liteOsUpgrade_RK3568(BaseApp):
else:
return False
def install_third_packages(self):
try:
pip_list = os.system('python -m pip list')
if 'pytest ' not in pip_list:
os.system('python -m pip install pytest==8.2.2')
if 'pytest-html' not in pip_list:
os.system('python -m pip install pytest-html==4.1.1')
if 'pytest-metadata' not in pip_list:
os.system('python -m pip install pytest-metadata==3.1.1')
if 'iniconfig' not in pip_list:
os.system('python -m pip install iniconfig==2.0.0')
except:
logger.error(traceback.format_exc())
@timeout(1000)
def flash_version(self):
partList = ["boot_linux", "system", "vendor", "userdata", "resource", "ramdisk", "chipset", "sys-prod", "chip-prod"]
@@ -265,12 +281,12 @@ class liteOsUpgrade_RK3568(BaseApp):
check_mode_cmd = "%s LD" % loader_tool_path
g = sendCmd(check_mode_cmd)
logger.info(g)
#time.sleep(40)
# time.sleep(40)
if "LocationID=%s Mode=Loader" % LocationID in g:
logger.info("3568 board has entered the Loader mode successfully!")
return True
else:
#if test_num != "2/2":
# if test_num != "2/2":
# hdc_kill()
os.system("hdc -t %s shell reboot loader" % sn)
time.sleep(5)
@@ -460,7 +476,7 @@ def start_cmd(sn):
@timeout(900)
def cmd_test(screenshot_path, py_path, new_report_path, resource_path, sn, test_num, pr_url):
def cmd_test(screenshot_path, py_file, device_num, test_num, new_report_path):
global total_time
save_screenshot_path = os.path.join(new_report_path, "screenshot_result")
logger.info(save_screenshot_path)
@@ -477,11 +493,10 @@ def cmd_test(screenshot_path, py_path, new_report_path, resource_path, sn, test_
except Exception as e:
logger.error(e)
return 98
config_path = os.path.join(screenshot_path, "app_capture_screen_test_config.json")
py_cmd = "python %s --config %s --anwser_path %s --save_path %s --device_num %s --test_num %s --tools_path %s --pr_url %s" \
% (py_path, config_path, resource_path, save_screenshot_path, sn, test_num, screenshot_path, pr_url)
# config_path = os.path.join(screenshot_path, "app_capture_screen_test_config.json")
py_cmd = "python {} --device_num {} --test_num {} --save_path {}".format(py_file, device_num, test_num, save_screenshot_path)
time1 = time.time()
result = outCmd(py_cmd, save_screenshot_path, base_screenshot_path, resource_path)
result = outCmd(py_cmd, save_screenshot_path, base_screenshot_path, screenshot_path)
time2 = time.time()
total_time = int(time2 - time1)
logger.info("total_time: %s" % total_time)
@@ -496,7 +511,7 @@ def cmd_test(screenshot_path, py_path, new_report_path, resource_path, sn, test_
@timeout(900)
def outCmd(cmd, save_screenshot_path, base_screenshot_path, resource_path):
def outCmd(cmd, save_screenshot_path, base_screenshot_path, script_path):
logger.info("cmd is: %s" % cmd)
# if system_type == "Windows":
# shell = False
@@ -506,16 +521,16 @@ def outCmd(cmd, save_screenshot_path, base_screenshot_path, resource_path):
# encoding = "utf-8"
p = subprocess.Popen(cmd, shell=True, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, encoding='utf-8', errors='ignore', universal_newlines=True)
curline = p.stdout.readline()
list_png_name = []
# list_png_name = []
try:
while "End of check" not in curline:
curline = p.stdout.readline()
logger.info(curline)
if 'SmokeTest: End of check, test succeeded!' in curline:
return True
if "abnarmal" in curline:
png_name = curline.split(" ")[3].split(".")[0]
list_png_name.append(png_name)
# if "abnarmal" in curline:
# png_name = curline.split(" ")[3].split(".")[0]
# list_png_name.append(png_name)
if "SmokeTest find some fatal problems" in curline:
logger.error("SmokeTest find some fatal problems!")
return 99
@@ -523,14 +538,16 @@ def outCmd(cmd, save_screenshot_path, base_screenshot_path, resource_path):
logger.error(e)
logger.error("execute smoke_test.py failed!")
return 99
l = list(set(list_png_name))
if l:
logger.error(l)
# l = list(set(list_png_name))
# if l:
# logger.error(l)
try:
for i in l:
result = os.path.join(resource_path, "%s.jpeg" % i)
base = os.path.join(base_screenshot_path, "%s.jpeg" % i)
shutil.copy(result, base)
resource_path = os.path.join(script_path, 'resource')
for jpeg_file in os.listdir(resource_path):
if jpeg_file.endswith('jpeg'):
result = os.path.join(resource_path, jpeg_file)
base = os.path.join(base_screenshot_path, jpeg_file)
shutil.copy(result, base)
except Exception as t:
logger.info(t)
p.wait()
@@ -562,4 +579,4 @@ def exec_cmd(mini_path, sn, save_path, archive_path):
logger.info("mini_system_test is ok!")
return True
logger.error("mini_system_test failed!")
return 98
return 98
@@ -0,0 +1,44 @@
import logging
import os.path
import pytest
from utils.device import Device
BASE_DIR = os.path.dirname(__file__)
def pytest_addoption(parser):
parser.addoption('--sn', default='')
@pytest.fixture(scope='session', autouse=True)
def device(request):
sn = request.config.option.sn
return Device(sn)
@pytest.fixture(scope='module')
def setup_teardown(request, device):
logging.info('前置操作')
current_case = os.path.basename(request.path)[:-3]
# 日志截图等保存路径
device.report_path = os.path.join(os.path.dirname(request.config.option.htmlpath), current_case)
logging.info('设置当前用例的报告路径为{}'.format(device.report_path))
device.resource_path = os.path.join(os.path.dirname(__file__), 'resource')
os.makedirs(device.report_path, exist_ok=True)
device.rm_faultlog()
device.start_hilog()
device.wakeup()
device.set_screen_timeout()
device.unlock()
device.go_home()
yield
logging.info('后置操作')
logging.info('点击右下角多任务键清理所有任务')
device.click(514, 1244)
device.click(360, 1170)
device.clean_app_data(request.param)
device.stop_and_collect_hilog()
@@ -0,0 +1,54 @@
import argparse
import json
import os.path
import re
import sys
import traceback
import pytest
BASE_DIR = os.path.dirname(__file__)
parser = argparse.ArgumentParser(description='manual to this scription')
parser.add_argument('--test_num', type=str, default='1/1')
parser.add_argument('--save_path', type=str, default='./report')
parser.add_argument('--device_num', type=str, default='')
args = parser.parse_args()
run_params = ['-vs']
try:
sn = args.device_num
run_params.append('--sn={}'.format(sn))
# 报告保存路径
report = os.path.join(args.save_path, sn, 'report.html')
run_params.append('--html={}'.format(report))
with open(os.path.join(BASE_DIR, 'testcases.json'), 'r', encoding='utf-8') as f:
test_cases_list = [case.get('case_file') for case in json.load(f)]
test_num = args.test_num
if test_num == '1/2':
selected_case = test_cases_list[0:1] + test_cases_list[8:]
elif test_num == '2/2':
selected_case = test_cases_list[0:8]
else:
selected_case = test_cases_list
run_params.extend(selected_case)
print(run_params)
pytest.main(run_params)
print('获取测试结果')
with open(report, 'r', encoding='utf-8') as f:
text = f.read()
passed = int(re.findall(r'<span class="passed">(\d+) Passed,</span>', text)[0])
if passed == len(selected_case):
print('SmokeTest: End of check, test succeeded!')
sys.exit(0)
else:
print('SmokeTest: End of check, SmokeTest find some fatal problems! passed {}/{}'.format(passed, len(selected_case)))
sys.exit(98)
except:
print('SmokeTest: End of check, SmokeTest find some fatal problems! {}'.format(traceback.format_exc()))
sys.exit(98)
@@ -0,0 +1,5 @@
[pytest]
log_cli = true
log_cli_level = DEBUG
log_cli_format = %(asctime)s [%(levelname)s]|%(filename)s:%(lineno)s|%(message)s
log_cli_date_format = %Y-%m-%d %H:%M:%S
@@ -0,0 +1,247 @@
[
{
"processName": "hiview",
"acls": [
"ohos.permission.DUMP",
"ohos.permission.GET_WIFI_PEERS_MAC"
]
},
{
"processName": "privacy_service",
"acls": [
"ohos.permission.MANAGE_DISPOSED_APP_STATUS",
"ohos.permission.MICROPHONE_CONTROL"
]
},
{
"processName": "inputmethod_service",
"acls": [
"ohos.permission.INPUT_MONITORING"
]
},
{
"processName": "memmgrservice",
"acls": [
"ohos.permission.INTERACT_ACROSS_LOCAL_ACCOUNTS_EXTENSION"
]
},
{
"processName": "locationhub",
"acls": [
"ohos.permission.GET_SENSITIVE_PERMISSIONS"
]
},
{
"processName": "useriam",
"acls": [
"ohos.permission.ACCESS_AUTH_RESPOOL",
"ohos.permission.INTERACT_ACROSS_LOCAL_ACCOUNTS_EXTENSION"
]
},
{
"processName": "pinauth",
"acls": [
"ohos.permission.ACCESS_AUTH_RESPOOL"
]
},
{
"processName": "foundation",
"acls": [
"ohos.permission.PUBLISH_SYSTEM_COMMON_EVENT",
"ohos.permission.PERMISSION_START_ABILITIES_FROM_BACKGROUND",
"ohos.permission.GRANT_SENSITIVE_PERMISSIONS",
"ohos.permission.REVOKE_SENSITIVE_PERMISSIONS",
"ohos.permission.MANAGE_HAP_TOKENID",
"ohos.permission.START_INVISIBLE_ABILITY",
"ohos.permission.INPUT_MONITORING",
"ohos.permission.INSTALL_SANDBOX_BUNDLE",
"ohos.permission.INTERACT_ACROSS_LOCAL_ACCOUNTS_EXTENSION",
"ohos.permission.MANAGE_USER_ACCOUNT_INFO",
"ohos.permission.MONITOR_DEVICE_NETWORK_STATE",
"ohos.permission.CAPTURE_SCREEN"
]
},
{
"processName": "dscreen",
"acls": [
"ohos.permission.CAPTURE_SCREEN"
]
},
{
"processName": "sensors",
"acls": [
"ohos.permission.GET_SENSITIVE_PERMISSIONS"
]
},
{
"processName": "camera_service",
"acls": [
"ohos.permission.GET_SENSITIVE_PERMISSIONS"
]
},
{
"processName": "audio_server",
"acls": [
"ohos.permission.GET_SENSITIVE_PERMISSIONS",
"ohos.permission.MONITOR_DEVICE_NETWORK_STATE"
]
},
{
"processName": "msdp_sa",
"acls": [
"ohos.permission.INPUT_MONITORING",
"ohos.permission.ACCESS_DISTRIBUTED_HARDWARE",
"ohos.permission.INTERCEPT_INPUT_EVENT",
"ohos.permission.MONITOR_DEVICE_NETWORK_STATE"
]
},
{
"processName": "dslm_service",
"acls": [
"ohos.permission.ACCESS_IDS"
]
},
{
"processName": "accountmgr",
"acls": [
"ohos.permission.ENFORCE_USER_IDM",
"ohos.permission.STORAGE_MANAGER_CRYPT"
]
},
{
"processName": "hdcd",
"acls": [
"ohos.permission.GET_BUNDLE_INFO_PRIVILEGED",
"ohos.permission.INSTALL_BUNDLE",
"ohos.permission.LISTEN_BUNDLE_CHANGE",
"ohos.permission.CHANGE_ABILITY_ENABLED_STATE",
"ohos.permission.REMOVE_CACHE_FILES",
"ohos.permission.START_ABILITIES_FROM_BACKGROUND",
"ohos.permission.PERMISSION_USED_STATS",
"ohos.permission.DUMP",
"ohos.permission.NOTIFICATION_CONTROLLER",
"ohos.permission.PUBLISH_SYSTEM_COMMON_EVENT",
"ohos.permission.CLEAN_APPLICATION_DATA",
"ohos.permission.START_SYSTEM_DIALOG",
"ohos.permission.GET_RUNNING_INFO"
]
},
{
"processName": "softbus_server",
"acls": [
"ohos.permission.GET_SENSITIVE_PERMISSIONS",
"ohos.permission.ACCESS_IDS"
]
},
{
"processName": "backup_sa",
"acls": [
"ohos.permission.INSTALL_BUNDLE",
"ohos.permission.ACCESS_EXT_SYSTEM_ABILITY"
]
},
{
"processName": "media_service",
"acls": [
"ohos.permission.CAPTURE_SCREEN"
]
},
{
"processName": "security_component_service",
"acls": [
"ohos.permission.GRANT_SENSITIVE_PERMISSIONS",
"ohos.permission.REVOKE_SENSITIVE_PERMISSIONS"
]
},
{
"processName": "distributedsched",
"acls": [
"ohos.permission.INPUT_MONITORING",
"ohos.permission.MANAGE_MISSIONS",
"ohos.permission.MONITOR_DEVICE_NETWORK_STATE",
"ohos.permission.START_INVISIBLE_ABILITY"
]
},
{
"processName": "accessibility",
"acls": [
"ohos.permission.INTERCEPT_INPUT_EVENT"
]
},
{
"processName": "dlp_permission_service",
"acls": [
"ohos.permission.INSTALL_SANDBOX_BUNDLE",
"ohos.permission.UNINSTALL_SANDBOX_BUNDLE"
]
},
{
"processName": "quick_fix",
"acls": [
"ohos.permission.INSTALL_QUICK_FIX_BUNDLE",
"ohos.permission.UNINSTALL_QUICK_FIX_BUNDLE"
]
},
{
"processName": "hidumper_service",
"acls": [
"ohos.permission.DUMP"
]
},
{
"processName": "sharing_service",
"acls": [
"ohos.permission.CAPTURE_SCREEN"
]
},
{
"processName": "pasteboard_service",
"acls": [
"ohos.permission.INPUT_MONITORING",
"ohos.permission.MONITOR_DEVICE_NETWORK_STATE"
]
},
{
"processName": "token_sync_service",
"acls": [
"ohos.permission.MONITOR_DEVICE_NETWORK_STATE"
]
},
{
"processName": "deviceprofile",
"acls": [
"ohos.permission.MONITOR_DEVICE_NETWORK_STATE"
]
},
{
"processName": "distributedfiledaemon",
"acls": [
"ohos.permission.MONITOR_DEVICE_NETWORK_STATE"
]
},
{
"processName": "distributeddata",
"acls": [
"ohos.permission.MONITOR_DEVICE_NETWORK_STATE"
]
},
{
"processName": "dhardware",
"acls": [
"ohos.permission.MONITOR_DEVICE_NETWORK_STATE"
]
},
{
"processName": "netmanager",
"acls": [
"ohos.permission.MANAGE_WIFI_HOTSPOT"
]
},
{
"processName": "samgr",
"acls": [
"ohos.permission.MONITOR_DEVICE_NETWORK_STATE",
"ohos.permission.MANAGE_SYSTEM_ABILITY"
]
}
]
@@ -0,0 +1,90 @@
[
{
"bundle&processName": "com.ohos.launcher",
"apl": "2"
},
{
"bundle&processName": "com.ohos.settings",
"apl": "2"
},
{
"bundle&processName": "com.ohos.systemui",
"apl": "2"
},
{
"bundle&processName": "com.ohos.screenlock",
"apl": "2"
},
{
"bundle&processName": "com.ohos.adminprovisioning",
"apl": "2"
},
{
"bundle&processName": "edm",
"apl": "3"
},
{
"bundle&processName": "com.ohos.settings.faceauth",
"apl": "2"
},
{
"bundle&processName": "cn.openharmony.inputmethodchoosedialog",
"apl":"3"
},
{
"bundle&processName":"media_service",
"apl":"3"
},
{
"bundle&processName":"com.ohos.amsdialog",
"apl":"3"
},
{
"bundle&processName":"com.ohos.useriam.authwidget",
"apl":"2"
},
{
"bundle&processName":"com.ohos.powerdialog",
"apl":"2"
},
{
"bundle&processName":"com.ohos.filepicker",
"apl":"2"
},
{
"bundle&processName":"com.ohos.camera",
"apl":"3"
},
{
"bundle&processName":"com.ohos.smartperf",
"apl":"2"
},
{
"bundle&processName":"com.ohos.devicemanagerui",
"apl":"2"
},
{
"bundle&processName":"ohos.telephony.resources",
"apl":"3"
},
{
"bundle&processName":"com.ohos.notificationdialog",
"apl":"2"
},
{
"bundle&processName":"ohos.samples.distributedcalc",
"apl":"3"
},
{
"bundle&processName":"ohos.samples.distributedmusicplayer",
"apl":"3"
},
{
"bundle&processName":"com.ohos.note",
"apl":"3"
},
{
"bundle&processName":"com.ohos.devicetest",
"apl":"2"
}
]
Binary file not shown.

After

Width:  |  Height:  |  Size: 41 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 67 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 74 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 78 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 22 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 98 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 48 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 67 KiB

@@ -0,0 +1,54 @@
[
{
"case_file": "testcases/test_launcher.py",
"description": "开机测试"
},
{
"case_file": "testcases/test_process.py",
"description": "进程检查"
},
{
"case_file": "testcases/test_apl_check.py",
"description": "外部用例APL白名单检查"
},
{
"case_file": "testcases/test_acl_check.py",
"description": "外部用例ACL白名单检查"
},
{
"case_file": "testcases/test_settings_wifi.py",
"description": "设置和wlan开关测试"
},
{
"case_file": "testcases/test_crash_check.py",
"description": "crash检查"
},
{
"case_file": "testcases/test_photos.py",
"description": "图库测试"
},
{
"case_file": "testcases/test_contacts.py",
"description": "联系人测试"
},
{
"case_file": "testcases/test_mms.py",
"description": "短信息测试"
},
{
"case_file": "testcases/test_distributed_musicplayer.py",
"description": "音乐测试"
},
{
"case_file": "testcases/test_camera.py",
"description": "照相机测试"
},
{
"case_file": "testcases/test_controller_center.py",
"description": "控制中心测试"
},
{
"case_file": "testcases/test_note.py",
"description": "备忘录应用测试"
}
]
@@ -0,0 +1,67 @@
import json
import logging
import os
import time
import pytest
class TestACLCheck:
@pytest.mark.parametrize('setup_teardown', [None], indirect=True)
def test(self, setup_teardown, device):
return
check_list_file = os.path.join(device.resource_path, 'acl_whitelist.json')
assert os.path.exists(check_list_file), '基线文件{}不存在'.format(check_list_file)
logging.info('读取{}文件内容'.format(check_list_file))
whitelist_dict = {}
json_data = json.load(open(check_list_file, 'r'))
for item in json_data:
whitelist_dict.update({item.get('processName'): item.get('acls')})
logging.info('导出token_info')
token_file = 'token_info_{}.txt'.format(time.time_ns())
device.hdc_shell('atm dump -t > /data/{}'.format(token_file))
device.hdc_file_recv('/data/{}'.format(token_file))
local_file = os.path.join(device.report_path, token_file)
assert os.path.exists(local_file), 'token_info导出失败'
device.hdc_shell('rm -rf /data/{}'.format(token_file))
acls_in_device = self.check_and_get_native_acls(local_file)
check_rst = True
for process, permission_list in acls_in_device.items():
if process not in whitelist_dict.keys():
check_rst = False
logging.error('processName={}未配置白名单权限:{}'.format(process, permission_list))
else:
whitelist_set = set(whitelist_dict[process])
permission_set = set(permission_list)
not_applied = permission_set.difference(whitelist_set)
if not_applied:
check_rst = False
logging.error('processName={}未配置白名单权限:{}'.format(process, not_applied))
assert check_rst, 'ACL检查失败'
@staticmethod
def check_and_get_native_acls(token_file):
check_pass = True
with open(token_file, 'r') as f:
lines = f.readlines()
native_acls_dict = {}
process = ''
for line in lines:
if 'processName' in line:
process = line.split(':')[1].strip().strip('",')
elif 'invalidPermList' in line:
check_pass = False
logging.error('invalidPermList is detected in processName = {}'.format(process))
elif 'nativeAcls' in line:
permissions = line.split(':')[1].strip().strip('",')
if not permissions:
continue
native_acls_dict.update(
{
process: permissions.split(',')
}
)
assert check_pass, 'ACL检查失败'
return native_acls_dict
@@ -0,0 +1,80 @@
import json
import logging
import math
import os.path
import sqlite3
import pytest
class TestAPLCheck:
@pytest.mark.parametrize('setup_teardown', [None], indirect=True)
def test(self, setup_teardown, device):
return
check_list_file = os.path.join(device.resource_path, 'apl_check_list.json')
logging.info('{}文件获取预置的check list'.format(check_list_file))
json_data = json.load(open(check_list_file, 'r'))
whitelist_dict = {}
for data in json_data:
whitelist_dict.update({data.get('bundle&processName'): int(data.get('apl'))})
assert whitelist_dict, 'check list为空'
logging.info('导出access_token.db')
device.hdc_file_recv('/data/service/el1/public/access_token/access_token.db')
db_file = os.path.join(device.report_path, 'access_token.db')
assert os.path.exists(db_file), '{}不存在'.format(db_file)
logging.info('查询hap_token_info_table')
hap_apl_result = self.query_records(db_file, 'select bundle_name,apl from hap_token_info_table')
assert hap_apl_result, 'hap_token_info_table为空'
logging.info('查询native_token_info_table')
native_apl_result = self.query_records(db_file, 'select process_name,apl from native_token_info_table')
assert native_apl_result, 'native_token_info_table为空'
logging.info('hap apl检查')
hap_check_rst = self.compare_db_with_whitelist(hap_apl_result, whitelist_dict, 1)
logging.info('native apl检查')
native_check_rst = self.compare_db_with_whitelist(native_apl_result, whitelist_dict, 2)
assert hap_check_rst, 'hap apl检查失败'
assert native_check_rst, 'native apl检查失败'
@staticmethod
def query_records(db_file, sql):
conn = sqlite3.connect(db_file)
assert conn, 'sqlit数据库连接失败'
cursor = conn.cursor()
cursor.execute(sql)
results = cursor.fetchall()
conn.close()
if not results:
return
result_dict = {}
for line in results:
key = line[0]
value = 0 if math.isnan(line[1]) else line[1]
result_dict.update({key: value})
return result_dict
@staticmethod
def compare_db_with_whitelist(db_data: dict, whitelist_dict: dict, basic_value):
"""
数据库和白名单对比
:param db_data:
:param whitelist_dict:
:param basic_value:
:return:
"""
check_rst = True
for key, apl in db_data.items():
if key not in whitelist_dict.keys():
continue
if apl <= basic_value:
continue
is_pass = whitelist_dict[key] == apl
if not is_pass:
logging.error('bundleName/processName = {} apl = {} | 校验未通过'.format(key, apl))
check_rst = False
else:
logging.info('bundleName/processName = {} apl = {} | 校验通过'.format(key, apl))
return check_rst
@@ -0,0 +1,46 @@
import logging
import os.path
import time
import pytest
class TestCamera:
camera_ability_name = 'com.ohos.camera.MainAbility'
camera_bundle_name = 'com.ohos.camera'
photo_ability_name = 'com.ohos.photos.MainAbility'
photo_bundle_name = 'com.ohos.photos'
@pytest.mark.parametrize('setup_teardown', [camera_bundle_name], indirect=True)
def test(self, setup_teardown, device):
logging.info('启动相机应用')
device.start_ability(self.camera_bundle_name, self.camera_ability_name)
time.sleep(2)
device.save_snapshot_to_local('step1.jpeg')
logging.info('点击拍照')
device.click(360, 1095)
time.sleep(2)
device.save_snapshot_to_local('step2.jpeg')
logging.info('切到录像模式')
device.click(430, 980)
time.sleep(2)
device.save_snapshot_to_local('step3.jpeg')
logging.info('点击录制')
device.click(360, 1095)
time.sleep(5)
device.save_snapshot_to_local('step4.jpeg')
logging.info('停止录制')
device.click(320, 1095)
time.sleep(2)
device.save_snapshot_to_local('step5.jpeg')
logging.info('点击左下角切到相册')
device.click(200, 1095)
time.sleep(5)
device.save_snapshot_to_local('step6.jpeg')
device.assert_process_running(self.photo_bundle_name)
@@ -0,0 +1,36 @@
import logging
import os.path
import time
import pytest
from utils.images import compare_image_similarity, crop_picture
class TestContacts:
ability_name = 'com.ohos.contacts.MainAbility'
bundle_name = 'com.ohos.contacts'
@pytest.mark.parametrize('setup_teardown', [bundle_name], indirect=True)
def test(self, setup_teardown, device):
logging.info('启动联系人应用')
device.start_ability(self.bundle_name, self.ability_name)
time.sleep(2)
logging.info('联系人界面截图对比')
standard_pic = os.path.join(device.resource_path, 'contacts.jpeg')
contacts_page_pic = device.save_snapshot_to_local('contacts.jpeg')
crop_picture(contacts_page_pic)
similarity = compare_image_similarity(contacts_page_pic, standard_pic)
assert similarity > 0.5, '截图对比失败'
logging.info('联系人界面控件检查')
current_layout = device.generate_layout_object('contacts.json')
current_layout.assert_text_exist('电话')
current_layout.assert_text_exist('联系人')
current_layout.assert_text_exist('收藏')
current_layout.assert_text_exist('1')
current_layout.assert_text_exist('3')
current_layout.assert_text_exist('5')
current_layout.assert_text_exist('7')
current_layout.assert_text_exist('9')
@@ -0,0 +1,24 @@
import logging
import os.path
import time
import pytest
from utils.images import compare_image_similarity, crop_picture
class TestControllerCenter:
@pytest.mark.parametrize('setup_teardown', [None], indirect=True)
def test(self, setup_teardown, device):
logging.info('下拉控制中心')
device.swipe(from_x=500, from_y=1, to_x=500, to_y=120)
time.sleep(2)
logging.info('控制中心界面截图对比')
standard_pic = os.path.join(device.resource_path, 'controller_center.jpeg')
controller_page_pic = device.save_snapshot_to_local('controller_center.jpeg')
crop_picture(controller_page_pic)
similarity = compare_image_similarity(controller_page_pic, standard_pic)
assert similarity > 0.5, '截图对比失败'
@@ -0,0 +1,11 @@
import pytest
class TestCrashCheck:
@pytest.mark.parametrize('setup_teardown', [None], indirect=True)
def test(self, setup_teardown, device):
crashes = device.hdc_shell('cd /data/log/faultlog/temp && grep "Process name" -rnw ./')
assert 'foundation' not in crashes, '检查到有 foundation crash'
assert 'render_service' not in crashes, '检查到有 render_service crash'
assert 'appspawn' not in crashes, '检查到有 appspawn crash'
@@ -0,0 +1,33 @@
import logging
import os.path
import time
import pytest
from utils.images import compare_image_similarity, crop_picture
class TestDistributedMusicplayer:
ability_name = 'ohos.samples.distributedmusicplayer.MainAbility'
bundle_name = 'ohos.samples.distributedmusicplayer'
@pytest.mark.parametrize('setup_teardown', [bundle_name], indirect=True)
def test(self, setup_teardown, device):
logging.info('启动音乐应用')
device.start_ability(self.bundle_name, self.ability_name)
time.sleep(2)
# 弹窗
device.stop_permission()
logging.info('音乐界面截图对比')
standard_pic = os.path.join(device.resource_path, 'distributedmusicplayer.jpeg')
music_page_pic = device.save_snapshot_to_local('distributedmusicplayer.jpeg')
crop_picture(music_page_pic)
similarity = compare_image_similarity(music_page_pic, standard_pic)
assert similarity > 0.5, '截图对比失败'
logging.info('音乐界面控件检查')
current_layout = device.generate_layout_object('music.json')
current_layout.assert_key_exist('image1')
current_layout.assert_key_exist('image2')
current_layout.assert_key_exist('image3')
current_layout.assert_key_exist('image4')
@@ -0,0 +1,25 @@
import logging
import os
import pytest
from utils.images import compare_image_similarity
class TestLauncher:
@pytest.mark.parametrize('setup_teardown', [None], indirect=True)
def test(self, setup_teardown, device):
logging.info('桌面截图对比')
# usb弹窗
device.click(360, 715)
device.dirc_fling(1)
device.dirc_fling(1)
standard_pic = os.path.join(device.resource_path, 'launcher.jpeg')
launcher_pic = device.save_snapshot_to_local('launcher.jpeg')
similarity = compare_image_similarity(launcher_pic, standard_pic)
assert similarity > 0.5, '截图对比失败'
logging.info('检查桌面图标控件是否存在')
current_layout = device.generate_layout_object('desktop.json')
current_layout.assert_type_exist('Badge')
@@ -0,0 +1,29 @@
import logging
import os.path
import time
import pytest
from utils.images import compare_image_similarity, crop_picture
class TestMms:
ability_name = 'com.ohos.mms.MainAbility'
bundle_name = 'com.ohos.mms'
@pytest.mark.parametrize('setup_teardown', [bundle_name], indirect=True)
def test(self, setup_teardown, device):
logging.info('启动信息应用')
device.start_ability(self.bundle_name, self.ability_name)
time.sleep(2)
logging.info('信息界面截图对比')
standard_pic = os.path.join(device.resource_path, 'mms.jpeg')
mms_page_pic = device.save_snapshot_to_local('mms.jpeg')
crop_picture(mms_page_pic)
similarity = compare_image_similarity(mms_page_pic, standard_pic)
assert similarity > 0.5, '截图对比失败'
logging.info('信息界面控件检查')
current_layout = device.generate_layout_object('mms.json')
current_layout.assert_text_exist('信息')
@@ -0,0 +1,43 @@
import logging
import os.path
import time
import pytest
from utils.images import compare_image_similarity, crop_picture
class TestNote:
ability_name = 'MainAbility'
bundle_name = 'com.ohos.note'
@pytest.mark.parametrize('setup_teardown', [bundle_name], indirect=True)
def test(self, setup_teardown, device):
logging.info('启动备忘录应用')
device.start_ability(self.bundle_name, self.ability_name)
time.sleep(2)
logging.info('主界面截图')
device.save_snapshot_to_local('step1.jpeg')
logging.info('导出界面布局')
note_layout = device.generate_layout_object('note.json')
logging.info('点击数据公式')
article = note_layout.get_element_by_text('数学公式')
device.click(*note_layout.center_of_element(article))
time.sleep(2)
device.save_snapshot_to_local('step2.jpeg')
logging.info('点击笔记内容区域')
device.click(360, 300)
time.sleep(3)
standard_pic = os.path.join(device.resource_path, 'note.jpeg')
note_pic = device.save_snapshot_to_local('step3.jpeg')
crop_picture(note_pic)
similarity = compare_image_similarity(note_pic, standard_pic)
assert similarity > 0.5, '截图对比失败'
logging.info('导出笔记界面布局')
note_layout2 = device.generate_layout_object('note_content.json')
logging.info('控件检查')
note_layout2.assert_text_exist('好好学习,天天向上')
note_layout2.assert_text_exist('space')
@@ -0,0 +1,40 @@
import logging
import os.path
import time
import pytest
from utils.images import compare_image_similarity, crop_picture
class TestPhotos:
ability_name = 'com.ohos.photos.MainAbility'
bundle_name = 'com.ohos.photos'
@pytest.mark.parametrize('setup_teardown', [bundle_name], indirect=True)
def test(self, setup_teardown, device):
logging.info('启动图库应用')
device.start_ability(self.bundle_name, self.ability_name)
time.sleep(2)
logging.info('图库界面截图对比')
standard_pic = os.path.join(device.resource_path, 'photos.jpeg')
photos_page_pic = device.save_snapshot_to_local('photos.jpeg')
crop_picture(photos_page_pic)
similarity = compare_image_similarity(photos_page_pic, standard_pic)
assert similarity > 0.5, '截图对比失败'
logging.info('图库界面控件检查')
current_layout = device.generate_layout_object('photos.json')
current_layout.assert_text_exist('照片')
current_layout.assert_text_exist('相册')
logging.info('medialibrarydata进程检查')
process = 'com.ohos.medialibrary.medialibrarydata'
device.assert_process_running(process)
time.sleep(1)
logging.info('sandbox path检查')
pid = device.get_pid(process)
sanboxf = device.hdc_shell('echo "ls /storage/media/local/"|nsenter -t {} -m sh'.format(pid))
assert 'files' in sanboxf, '{}中未检测到files'.format(sanboxf)
@@ -0,0 +1,78 @@
import logging
import re
import pytest
class TestProcess:
pid_list = [
'com.ohos.launcher',
'render_service',
]
ps_list = [
'hdf_devmgr',
'param_watcher',
'storage_manager',
'appspawn',
'hilogd',
'samgr',
'storage_daemon',
'uinput_inject',
'multimodalinput',
'huks_service',
'memmgrservice',
'bluetooth_servi',
'resource_schedu',
'bgtaskmgr_servi',
'audio_server',
'deviceauth_service',
'softbus_server',
'wifi_hal_service',
'faultloggerd',
'accountmgr',
'time_service',
'distributeddata',
'useriam',
'inputmethod_ser',
'ui_service',
'netmanager',
'sensors',
'media_service',
'wifi_manager_se',
'installs',
'hiview',
'telephony',
'camera_service',
'foundation',
'hdcd',
'light_host',
'vibrator_host',
'sensor_host',
'input_user_host',
'camera_host',
'audio_host',
'wifi_host',
'usb_host',
'blue_host',
'wifi_hal_service',
'com.ohos.systemui',
'power_host',
]
@pytest.mark.parametrize('setup_teardown', [None], indirect=True)
def test(self, setup_teardown, device):
lost_process = []
logging.info('检查进程号是否存在')
for process in self.pid_list:
pid = device.get_pid(process)
if not re.search(r'\d+', pid):
lost_process.append(process)
ps_elf_rst = device.hdc_shell('ps -elf')
for pname in self.ps_list:
if pname not in ps_elf_rst:
lost_process.append(pname)
logging.info('丢失的进程有{}'.format(lost_process))
assert not lost_process
@@ -0,0 +1,47 @@
import logging
import os.path
import time
import pytest
from utils.images import compare_image_similarity, crop_picture
class TestSettingsWifi:
ability_name = 'com.ohos.settings.MainAbility'
bundle_name = 'com.ohos.settings'
@pytest.mark.parametrize('setup_teardown', [bundle_name], indirect=True)
def test(self, setup_teardown, device):
logging.info('启动设置应用')
device.start_ability(self.bundle_name, self.ability_name)
time.sleep(2)
logging.info('设置界面截图对比')
standard_pic = os.path.join(device.resource_path, 'settings.jpeg')
settings_page_pic = device.save_snapshot_to_local('settings.jpeg')
crop_picture(settings_page_pic)
similarity = compare_image_similarity(settings_page_pic, standard_pic)
assert similarity > 0.5, '截图对比失败'
logging.info('设置界面控件检查')
settings_layout = device.generate_layout_object('settings.json')
settings_layout.assert_text_exist('设置')
settings_layout.assert_text_exist('WLAN')
logging.info('进入wlan页面')
wlan_element = settings_layout.get_element_by_text('WLAN')
device.click(*settings_layout.center_of_element(wlan_element))
time.sleep(2)
device.save_snapshot_to_local('wlan_snapshot1.jpeg')
wlan_page_layout = device.generate_layout_object('wlan_before_clicked.json')
wlan_switch = wlan_page_layout.get_element_by_type('Toggle')
before_click = device.get_wifi_status().get('active')
logging.info('打开/关闭 wlan开关')
device.click(*wlan_page_layout.center_of_element(wlan_switch))
time.sleep(5)
device.save_snapshot_to_local('wlan_snapshot2.jpeg')
after_click = device.get_wifi_status().get('active')
logging.info('wlan开关状态变化:{} => {}'.format(before_click, after_click))
assert before_click != after_click, 'wlan开关切换失败'
@@ -0,0 +1,442 @@
import json
import logging
import os.path
import re
import subprocess
import time
from utils.layout import Layout
class Device:
@classmethod
def _execute_cmd(cls, cmd):
logging.info(f'[In]{cmd}')
rst = subprocess.run(cmd, capture_output=True, shell=True, encoding='utf-8', timeout=30)
out_put = rst.stdout or rst.stderr
time.sleep(0.5)
logging.info(f'[Out]{out_put}')
return out_put
def __init__(self, sn):
self.sn = sn
self.report_path = ''
self.resource_path = ''
self.width = 720
self.height = 1280
self.velocity_range = (200, 40000)
self.get_render_size()
def hdc_shell(self, cmd):
return self._execute_cmd(f'hdc -t {self.sn} shell "{cmd}"')
def hdc(self, cmd):
return self._execute_cmd(f'hdc -t {self.sn} {cmd}')
def hdc_version(self):
return self._execute_cmd('hdc -v')
def hdc_list_targets(self):
devices = self._execute_cmd('hdc list targets')
if 'Empty' in devices:
return []
return devices.splitlines()
def install_hap(self, hap_path, replace=True):
"""
暗账应用
:param hap_path: hap包的路径
:param replace: 是否覆盖安装,true覆盖,否则不覆盖
:return:
"""
logging.debug('安装hap包')
if replace:
cmd = f'app install -r {hap_path}'
else:
cmd = f'app install {hap_path}'
return self.hdc(cmd)
def install_multi_hap(self, hap_list: list):
"""
安装更新, 多hap可以指定多个文件路径
:param hap_list:
:return:
"""
logging.debug('安装hap包')
haps = ' '.join(hap_list)
cmd = 'install {}'.format(haps)
return self.hdc(cmd)
def bm_install(self, hap_list: list):
"""
bm工具安装更新多个hap
:param hap_list:
:return:
"""
logging.debug('安装hap包')
haps = ' '.join(hap_list)
cmd = 'bm install -p {}'.format(haps)
return self.hdc_shell(cmd)
def uninstall_hap(self, bundle_name):
# 两个命令都可以
logging.debug(f'卸载{bundle_name}应用')
# cmd = 'hdc app uninstall {}'.format(bundle_name)
return self.hdc(f'uninstall {bundle_name}')
def bm_uninstall(self, bundle_name):
logging.debug(f'卸载{bundle_name}应用')
return self.hdc_shell(f'bm uninstall -n {bundle_name}')
def hdc_file_send(self, local, remote):
logging.debug('发送文件')
return self.hdc(f'file send "{local}" "{remote}"')
def hdc_file_recv(self, remote, local=''):
logging.debug('接收文件')
local = local or self.report_path
return self.hdc(f'file recv "{remote}" "{local}"')
def hilog(self):
logging.debug('hilog')
return self.hdc('hilog')
def get_udid(self):
logging.debug('获取udid')
return self.hdc('bm get --udid')
def kill(self):
logging.debug('杀掉hdc进程')
return self._execute_cmd('hdc kill')
def restart(self):
logging.debug('重启hdc进程')
return self._execute_cmd('hdc start -r')
def reboot(self):
logging.debug('重启设备')
return self.hdc_shell('reboot')
def start_ability(self, bundle_name, ability_name):
logging.debug(f'打开{bundle_name}应用')
return self.hdc_shell(f'aa start -b {bundle_name} -a {ability_name}')
def force_stop(self, bundle_name):
logging.debug(f'停掉{bundle_name}应用')
return self.hdc_shell(f'aa force-stop {bundle_name}')
def clean_app_data(self, bundle_name):
"""
清除应用缓存和数据, -c缓存, -d应用数据
:param bundle_name:
:return:
"""
if not bundle_name:
return
logging.debug(f'清理{bundle_name}应用数据和缓存')
return self.hdc_shell(f'bm clean -n {bundle_name} -c -d')
def disable_app(self, bundle_name):
"""
禁止应用,应用在桌面消失
:param bundle_name:
:return:
"""
logging.debug(f'禁止{bundle_name}应用,应用在桌面消失')
return self.hdc_shell(f'bm disable -n {bundle_name}')
def enable_app(self, bundle_name):
"""
允许应用,应用显示在桌面上
:param bundle_name:
:return:
"""
logging.debug(f'允许{bundle_name}应用,应用显示在桌面上')
return self.hdc_shell(f'bm enable -n {bundle_name}')
def dump_hap_configuration(self, bundle_name):
"""
查看应用配置信息
:param bundle_name:
:return:
"""
logging.debug(f'查看{bundle_name}应用配置信息')
return self.hdc_shell(f'bm dump -n {bundle_name}')
def click(self, x: int, y: int):
"""
模拟触摸按下
:param x:
:param y:
:return:
"""
x = self._auto_fix(x, 1, self.width)
y = self._auto_fix(y, 1, self.height)
logging.debug(f'点击({x},{y})坐标')
return self.hdc_shell(f'uitest uiInput click {x} {y}')
def stop_permission(self):
logging.debug(f'消掉权限请求的弹窗')
return self.click(516, 688)
# return self.force_stop('com.ohos.permissionmanager')
def double_click(self, x, y):
x = self._auto_fix(x, 1, self.width)
y = self._auto_fix(y, 1, self.height)
logging.debug(f'双击({x},{y})坐标')
return self.hdc_shell(f'uitest uiInput doubleClick {x} {y}')
def long_click(self, x, y):
x = self._auto_fix(x, 1, self.width)
y = self._auto_fix(y, 1, self.height)
logging.debug(f'长按({x},{y})坐标')
return self.hdc_shell(f'uitest uiInput longClick {x} {y}')
def dirc_fling(self, direct=0):
"""
模拟指定方向滑动
:param direct:direction (可选参数,滑动方向,可选值: [0,1,2,3], 滑动方向: [左,右,上,下],默认值: 0)
swipeVelocityPps_ (可选参数,滑动速度,取值范围: 200-40000, 默认值: 600, 单位: px/s)
stepLength(可选参数,滑动步长,默认值:滑动距离/50, 单位: px)
:return:
"""
direct_map = {
0: '',
1: '',
2: '',
3: '',
}
if direct not in direct_map.keys():
direct = 0
logging.debug(f'{direct_map.get(direct)} 滑动')
return self.hdc_shell(f'uitest uiInput dircFling {direct}')
def swipe(self, from_x, from_y, to_x, to_y, velocity=600):
"""
模拟慢滑操作
:param from_x:(必选参数,滑动起点x坐标)
:param from_y:(必选参数,滑动起点y坐标)
:param to_x:(必选参数,滑动终点x坐标)
:param to_y:(必选参数,滑动终点y坐标)
:param velocity: (可选参数,滑动速度,取值范围: 200-40000, 默认值: 600, 单位: px/s)
:return:
"""
# 保证数据取值范围合理
from_x = self._auto_fix(from_x, 1, self.width)
from_y = self._auto_fix(from_y, 1, self.height)
to_x = self._auto_fix(to_x, 1, self.width)
to_y = self._auto_fix(to_y, 1, self.height)
velocity = self._auto_fix(velocity, *self.velocity_range)
logging.debug(f'从({from_x},{from_y})滑动到({to_x},{to_y}),滑动速度:{velocity}px/s')
return self.hdc_shell(f'uitest uiInput swipe {from_x} {from_y} {to_x} {to_y} {velocity}')
def fling(self, from_x, from_y, to_x, to_y, velocity=600):
"""
模拟快滑操作
:param from_x:(必选参数,滑动起点x坐标)
:param from_y:(必选参数,滑动起点y坐标)
:param to_x:(必选参数,滑动终点x坐标)
:param to_y:(必选参数,滑动终点y坐标)
:param velocity: (可选参数,滑动速度,取值范围: 200-40000, 默认值: 600, 单位: px/s)
:return:
"""
# 保证数据取值范围合理
# 保证数据取值范围合理
from_x = self._auto_fix(from_x, 1, self.width)
from_y = self._auto_fix(from_y, 1, self.height)
to_x = self._auto_fix(to_x, 1, self.width)
to_y = self._auto_fix(to_y, 1, self.height)
# velocity = self._auto_fix(velocity, *self.velocity_range)
logging.debug(f'从({from_x},{from_y})快速滑动到({to_x},{to_y})')
return self.hdc_shell(f'uitest uiInput fling {from_x} {from_y} {to_x} {to_y}')
def drag(self, from_x, from_y, to_x, to_y, velocity=600):
"""
拖拽,从(x1, y1)拖拽到(x2, y2
:param from_x:(必选参数,滑动起点x坐标)
:param from_y:(必选参数,滑动起点y坐标)
:param to_x:(必选参数,滑动终点x坐标)
:param to_y:(必选参数,滑动终点y坐标)
:param velocity: (可选参数,滑动速度,取值范围: 200-40000, 默认值: 600, 单位: px/s)
"""
from_x = self._auto_fix(from_x, 1, self.width)
from_y = self._auto_fix(from_y, 1, self.height)
to_x = self._auto_fix(to_x, 1, self.width)
to_y = self._auto_fix(to_y, 1, self.height)
velocity = self._auto_fix(velocity, *self.velocity_range)
logging.debug(f'从({from_x},{from_y})拖到({to_x},{to_y}),拖动速度:{velocity}px/s')
return self.hdc_shell(f'uitest uiInput drag {from_x} {from_y} {to_x} {to_y} {velocity}')
def key_event(self, key_code):
logging.debug(f'按下{key_code}')
return self.hdc_shell(f'uitest uiInput keyEvent {key_code}')
def go_home(self):
return self.key_event('Home')
def go_back(self):
return self.key_event('Back')
def press_power_key(self):
return self.key_event('Power')
def input_text(self, x, y, text=''):
x = self._auto_fix(x, 1, self.width)
y = self._auto_fix(y, 1, self.height)
logging.debug(f'向({x, y})坐标处输入“{text}')
return self.hdc_shell(f'uitest uiInput inputText {x} {y} {text}')
def wakeup(self):
logging.debug('点亮屏幕')
return self.hdc_shell('power-shell wakeup')
def suspend(self):
logging.debug('熄灭屏幕')
return self.hdc_shell('power-shell suspend')
def set_power_mode(self, mode='602'):
"""
设置电源模式
:param mode:600 normal mode 正常模式
601 power save mode省电模式
602 performance mode性能模式,屏幕会常亮
603 extreme power save mode极端省电模式
:return:
"""
power_map = {
'600': '正常模式',
'601': '省电模式',
'602': '性能模式',
'603': '极端省电模式',
}
if mode not in power_map.keys():
mode = '602'
logging.debug(f'设置电源为{power_map.get(mode)}')
return self.hdc_shell(f'power-shell setmode {mode}')
def display_power_state(self):
logging.debug('获取屏幕点亮状态')
return self.hdc_shell('hidumper -s 3308')
def get_render_size(self):
logging.debug('获取屏幕分辨率')
rst = self.hdc_shell('hidumper -s RenderService -a screen')
xy = re.findall(r'render size: (\d+)x(\d+)', rst)[0]
self.width, self.height = [int(i) for i in xy]
def dump_layout(self, file_name=''):
logging.debug('获取当前页面布局')
self.hdc_shell('rm -rf /data/local/tmp/*.json')
if file_name:
file_path = '/data/local/tmp/' + file_name
dump_rst = self.hdc_shell(f'uitest dumpLayout -p {file_path}')
else:
dump_rst = self.hdc_shell(f'uitest dumpLayout')
layout_file = dump_rst.split(':')[1].strip()
return layout_file
def generate_layout_object(self, file_name=''):
"""
:param file_name: 文件名不包含路径
:return:
"""
tmp_file = self.dump_layout(file_name)
self.hdc_file_recv(tmp_file, self.report_path)
local_file = os.path.join(self.report_path, tmp_file.split('/')[-1])
if not os.path.exists(local_file):
logging.error(f'没有找到{local_file}文件')
return
json_data = json.load(open(local_file, 'r', encoding='utf-8'))
return Layout(json_data)
def snapshot_display(self, jpeg=''):
"""jpeg必须在/data/local/tmp目录"""
logging.debug('获取当前页面截图')
self.hdc_shell('rm -rf /data/local/tmp/*.jpeg')
if jpeg:
jpeg = '/data/local/tmp/' + jpeg
shot_rst = self.hdc_shell(f'snapshot_display -f {jpeg}')
else:
shot_rst = self.hdc_shell('snapshot_display')
if 'success' not in shot_rst:
return ''
return re.findall(r'write to(.*)as jpeg', shot_rst)[0].strip()
def save_snapshot_to_local(self, file_name=''):
tmp_file = self.snapshot_display(file_name)
save_file = os.path.join(self.report_path, tmp_file.split('/')[-1])
self.hdc_file_recv(tmp_file, self.report_path)
return save_file
def set_screen_timeout(self, timeout=600):
"""
设置屏幕超时时间
:param timeout: 单位s
:return:
"""
logging.debug(f'设置屏幕{timeout}s无操作后休眠')
return self.hdc_shell(f'power-shell timeout -o {timeout * 1000}')
def rm_faultlog(self):
logging.debug('删除fault log')
self.hdc_shell('rm -f /data/log/faultlog/SERVICE_BLOCK*')
self.hdc_shell('rm -f /data/log/faultlog/appfreeze*')
self.hdc_shell('rm -f /data/log/faultlog/temp/cppcrash*')
self.hdc_shell('rm -f /data/log/faultlog/faultlogger/jscrash*')
self.hdc_shell('rm -f /data/log/faultlog/faultlogger/appfreeze*')
self.hdc_shell('rm -f /data/log/faultlog/faultlogger/cppcrash*')
def start_hilog(self):
logging.debug('开启hilog')
self.hdc_shell('hilog -w stop;hilog -w clear')
self.hdc_shell('hilog -r;hilog -b INFO;hilog -w start -l 10M -n 1000')
def stop_and_collect_hilog(self, local_dir=''):
logging.debug('停止并收集hilog')
self.hdc_shell('hilog -w stop')
self.hdc_file_recv('/data/log/hilog/', local_dir)
def unlock(self):
"""
滑动解锁
:return:
"""
logging.debug('解锁屏幕')
return self.dirc_fling(2)
def assert_process_running(self, process):
logging.debug(f'检查{process}进程是否存在')
rst = self.hdc_shell(f'ps -ef | grep -w {process} | grep -v grep')
assert process in rst, f'进程{process}不存在'
def get_pid(self, process):
logging.debug(f'获取{process}进程PID')
return self.hdc_shell(f'pidof {process}').strip()
def get_wifi_status(self):
# hidumper -ls查看所有hidumper服务
logging.debug('获取wifi状态')
status = self.hdc_shell('hidumper -s WifiDevice')
active_state = re.findall(r'WiFi active state: (.*)', status)[0].strip()
connection_status = re.findall(r'WiFi connection status: (.*)', status)[0].strip()
scan_status = self.hdc_shell('hidumper -s WifiScan')
is_scan_running = re.findall(r'Is scan service running: (.*)', scan_status)[0].strip()
return {
'active': active_state,
'connected': connection_status,
'scanning': is_scan_running
}
@staticmethod
def _auto_fix(p: int, min_num: int, max_num: int):
"""
自动修正数据以保证数据的准确性
:param p:
:param min_num:
:param max_num:
:return:
"""
return min_num if p < min_num else max_num if p > max_num else p
@@ -0,0 +1,44 @@
import logging
import os.path
import cv2
def crop_picture(picture, x1=0, y1=72, x2=720, y2=1208):
img = cv2.imread(picture)
img = img[y1:y2, x1:x2]
cv2.imwrite(picture, img)
def compare_image_similarity(image1, image2):
logging.info('{}是否存在?[{}]'.format(image1, os.path.exists(image1)))
logging.info('{}是否存在?[{}]'.format(image2, os.path.exists(image2)))
if not os.path.exists(image1) or not os.path.exists(image2):
logging.warning('文件缺失,相似度为0%')
return 0
image1 = cv2.imread(image1, 0)
image2 = cv2.imread(image2, 0)
# 初始化特征点检测器和描述符
orb = cv2.ORB_create(edgeThreshold=5, patchSize=30)
keypoints1, descriptors1 = orb.detectAndCompute(image1, None)
keypoints2, descriptors2 = orb.detectAndCompute(image2, None)
# 初始化匹配器
bf = cv2.BFMatcher(cv2.NORM_HAMMING, crossCheck=True)
# 匹配特征点
matches = bf.match(descriptors1, descriptors2)
if not matches:
logging.warning('没有匹配到特征点,相似度为0%')
return 0
if not keypoints1:
if not keypoints2:
logging.warning('相似度为100%')
return 1
logging.warning('相似度为0%')
return 0
# 计算相似度
similarity = len(matches) / len(keypoints1)
logging.info('相似度为:{}%'.format(similarity * 100))
return similarity
@@ -0,0 +1,89 @@
import logging
import re
class Layout:
def __init__(self, json_data):
self.element_list = self.parse_attribute_nodes(json_data)
def parse_attribute_nodes(self, json_obj, attr_list=None):
if attr_list is None:
attr_list = []
if isinstance(json_obj, dict):
for key, value in json_obj.items():
if key == 'attributes' and isinstance(value, dict):
attr_list.append(value)
elif isinstance(value, (dict, list)):
self.parse_attribute_nodes(value, attr_list)
elif isinstance(json_obj, list):
for item in json_obj:
self.parse_attribute_nodes(item, attr_list)
return attr_list
def get_elements_by_text(self, text):
ems = []
for e in self.element_list:
if e.get('text') == text:
ems.append(e)
return ems
def get_element_by_text(self, text, index=0):
ems = self.get_elements_by_text(text)
if not ems:
return
return ems[index]
def assert_text_exist(self, text):
element = self.get_elements_by_text(text)
rst = '' if element else ''
logging.info('检查[文本]="{}"是否存在?[{}]'.format(text, rst))
assert element
def get_elements_by_type(self, _type):
ems = []
for e in self.element_list:
if e.get('type') == _type:
ems.append(e)
return ems
def get_element_by_type(self, _type, index=0):
ems = self.get_elements_by_type(_type)
if not ems:
return
return ems[index]
def assert_type_exist(self, _type):
element = self.get_elements_by_type(_type)
rst = '' if element else ''
logging.info('检查[type]="{}"是否存在?[{}]'.format(_type, rst))
assert element
def get_elements_by_key(self, key):
ems = []
for e in self.element_list:
if e.get('key') == key:
ems.append(e)
return ems
def get_element_by_key(self, key, index=0):
ems = self.get_elements_by_key(key)
if not ems:
return
return ems[index]
def assert_key_exist(self, key):
element = self.get_elements_by_key(key)
rst = '' if element else ''
logging.info('检查[key]="{}"是否存在?[{}]'.format(key, rst))
assert element
@staticmethod
def center_of_element(e):
assert e, '控件不存在'
bounds = e.get('bounds')
x1, y1, x2, y2 = [int(i) for i in re.findall(r'\d+', bounds)]
x = (x1 + x2) // 2
y = (y1 + y2) // 2
return x, y