1#!/usr/bin/env python3 2# -*- coding: utf-8 -*- 3# Copyright (C) 2024 Huawei Device Co., Ltd. 4# Licensed under the Apache License, Version 2.0 (the "License"); 5# you may not use this file except in compliance with the License. 6# You may obtain a copy of the License at 7# 8# http://www.apache.org/licenses/LICENSE-2.0 9# 10# Unless required by applicable law or agreed to in writing, software 11# distributed under the License is distributed on an "AS IS" BASIS, 12# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 13# See the License for the specific language governing permissions and 14# limitations under the License. 15 16import pytest 17import subprocess 18import re 19import time 20import os 21import threading 22import sqlite3 23 24 25def get_file_size(file_path): 26 size = os.path.getsize(file_path) 27 return size 28 29 30def task(): 31 subprocess.check_output(f'hdc shell "hiprofiler_cmd -c /data/local/tmp/config_xpower.txt -o /data/local/tmp/test_xpower.htrace -t 30 -s -k"') 32 33 34def task_total(): 35 subprocess.check_output(f'hdc shell "hiprofiler_cmd -c /data/local/tmp/config_xpower_total.txt -o /data/local/tmp/test_xpower_total.htrace -t 35 -s -k"') 36 37 38class TestHiprofilerXPowerPlugin: 39 @pytest.mark.L0 40 def test_xpowerplugin_app(self): 41 subprocess.check_output(f"hdc file send ..\\inputfiles\\xpower_plugin\\config_xpower.txt /data/local/tmp/", shell=False, 42 text=True, encoding="utf-8") 43 task_thread = threading.Thread(target=task, args=()) 44 task_thread.start() 45 task_thread.join() 46 subprocess.check_output(f"hdc file recv /data/local/tmp/test_xpower.htrace ../outputfiles/ ", shell=False, 47 text=True, encoding="utf-8") 48 # 检查文件大小 49 file_size = get_file_size(f"../outputfiles/test_xpower.htrace") 50 assert (file_size > 1024) 51 subprocess.check_output( 52 r"../inputfiles/trace_streamer_db.exe ../outputfiles/test_xpower.htrace -e ../outputfiles/test_xpower.db") 53 # 连接数据库文件 54 conn = sqlite3.connect(r'../outputfiles/test_xpower.db') 55 # 创建游标对象 56 cursor = conn.cursor() 57 # 执行SQL查询 58 cursor.execute("select end_ts - start_ts as time from trace_range") 59 result = cursor.fetchall() 60 #断言trace 时长27秒 61 for row in result: 62 assert(row[0] == 27 * 1000 * 1000 * 1000) 63 cursor.execute("select count(0) from xpower_measure") 64 result = cursor.fetchall() 65 for row in result: 66 assert(row[0] > 0) 67 68 cursor.execute("select * from xpower_measure where filter_id = 0 order by ts limit 0,10") 69 result = cursor.fetchall() 70 for row in result: 71 assert(row[3] > 0) 72 #电池电量 73 cursor.execute("select * from xpower_measure where filter_id = 2 order by ts limit 0,10") 74 result = cursor.fetchall() 75 for row in result: 76 assert(row[3] > 0) 77 cursor.close() 78 conn.close() 79 80 @pytest.mark.L0 81 def test_xpowerplugin_total(self): 82 subprocess.check_output(f"hdc file send ..\\inputfiles\\xpower_plugin\\config_xpower_total.txt /data/local/tmp/", shell=False, 83 text=True, encoding="utf-8") 84 task_thread = threading.Thread(target=task_total, args=()) 85 task_thread.start() 86 task_thread.join() 87 subprocess.check_output(f"hdc file recv /data/local/tmp/test_xpower_total.htrace ../outputfiles/ ", shell=False, 88 text=True, encoding="utf-8") 89 # 检查文件大小 90 file_size = get_file_size(f"../outputfiles/test_xpower_total.htrace") 91 assert (file_size > 1024) 92 subprocess.check_output( 93 r"../inputfiles/trace_streamer_db.exe ../outputfiles/test_xpower_total.htrace -e ../outputfiles/test_xpower_total.db") 94 # 连接数据库文件 95 conn = sqlite3.connect(r'../outputfiles/test_xpower_total.db') 96 # 创建游标对象 97 cursor = conn.cursor() 98 # 执行SQL查询 99 cursor.execute("select end_ts - start_ts as time from trace_range") 100 result = cursor.fetchall() 101 #断言trace 时长33秒 102 for row in result: 103 assert(row[0] == 33 * 1000 * 1000 * 1000) 104 cursor.execute("select count(0) from xpower_measure") 105 result = cursor.fetchall() 106 for row in result: 107 assert(row[0] > 0) 108 #电量百分比 109 cursor.execute("select * from xpower_measure, measure_filter where filter_id = id and name = 'Battery.Level' limit 0,10") 110 result = cursor.fetchall() 111 for row in result: 112 assert(row[3] > 0) 113 #外壳温度 114 cursor.execute("select * from xpower_measure, measure_filter where filter_id = id and name = 'ThermalReport.ShellTemp' limit 0,10") 115 result = cursor.fetchall() 116 for row in result: 117 assert(row[3] > 0) 118 #温度等级 119 cursor.execute("select * from xpower_measure, measure_filter where filter_id = id and name = 'ThermalReport.ThermalLevel' limit 0,10") 120 result = cursor.fetchall() 121 for row in result: 122 assert(row[3] >= 0) 123 cursor.close() 124 conn.close()