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_ftrace.txt -o /data/local/tmp/test_ftrace.htrace -t 10 -s -k"') 32 33 34def task_event(): 35 subprocess.check_output(f'hdc shell "hiprofiler_cmd -c /data/local/tmp/config_ftrace_event.txt -o /data/local/tmp/test_ftrace_event.htrace -t 10 -s -k"') 36 37 38def task_freq(): 39 subprocess.check_output(f'hdc shell "hiprofiler_cmd -c /data/local/tmp/config_ftrace_freq.txt -o /data/local/tmp/test_ftrace_freq.htrace -t 30 -s -k"') 40 41 42class TestHiprofilerFtracePlugin: 43 @pytest.mark.L0 44 def test_ftraceplugin(self): 45 subprocess.check_output(r'hdc file send ..\\inputfiles\\ftrace\\config_ftrace.txt /data/local/tmp/', 46 text=True, encoding="utf-8") 47 task_thread = threading.Thread(target=task, args=()) 48 task_thread.start() 49 task_thread.join() 50 51 subprocess.check_output(f"hdc file recv /data/local/tmp/test_ftrace.htrace ../outputfiles/ ", 52 text=True, encoding="utf-8") 53 # 检查文件大小 54 file_size = get_file_size(f"../outputfiles/test_ftrace.htrace") 55 assert (file_size > 1024) 56 subprocess.check_output( 57 r"../inputfiles/trace_streamer_db.exe ../outputfiles/test_ftrace.htrace -e ../outputfiles/test_ftrace.db") 58 # 连接数据库文件 59 conn = sqlite3.connect(r'../outputfiles/test_ftrace.db') 60 # 创建游标对象 61 cursor = conn.cursor() 62 # 执行SQL查询 63 # 检查binder 64 cursor.execute("select * from callstack where cat ='binder' limit 0,10") 65 result = cursor.fetchall() 66 for row in result: 67 assert(row[5] == 'binder transaction' or row[5] == 'binder reply' or row[5] == 'binder transaction async' or row[5] == 'binder async rcv') 68 cursor.close() 69 conn.close() 70 71 @pytest.mark.L0 72 def test_ftrace_events(self): 73 subprocess.check_output(f"hdc file send ..\\inputfiles\\ftrace\\config_ftrace_event.txt /data/local/tmp/", 74 text=True, encoding="utf-8") 75 task_thread = threading.Thread(target=task_event, args=()) 76 task_thread.start() 77 task_thread.join() 78 subprocess.check_output(f"hdc file recv /data/local/tmp/test_ftrace_event.htrace ../outputfiles/ ", 79 text=True, encoding="utf-8") 80 # 检查文件大小 81 file_size = get_file_size(f"../outputfiles/test_ftrace_event.htrace") 82 assert (file_size > 1024) 83 assert (file_size < 1024 * 1024 * 1024) 84 subprocess.check_output( 85 r"../inputfiles/trace_streamer_db.exe ../outputfiles/test_ftrace_event.htrace -e ../outputfiles/test_ftrace_event.db") 86 # 连接数据库文件 87 conn = sqlite3.connect(r'../outputfiles/test_ftrace_event.db') 88 # 创建游标对象 89 cursor = conn.cursor() 90 # 执行SQL查询 91 check_wake = False 92 check_newtask = False 93 check_exit = False 94 table_list = [a for a in cursor.execute('SELECT name FROM sqlite_master WHERE type = "table"')] 95 for table in table_list: 96 cursor.execute('SELECT * FROM ' + table[0]) 97 result = cursor.fetchall() 98 for row in result: 99 if 'sched_wakeup' in row: 100 check_wake = True 101 if 'task_newtask' in row: 102 check_newtask = True 103 if 'sched_process_exit' in row: 104 check_exit = True 105 # 检查 wakeup 和waking 事件 106 cursor.execute("select * from instant where name = 'sched_wakeup' limit 10") 107 result = cursor.fetchall() 108 for row in result: 109 assert(row[2] > 0) 110 assert(row[3] > 0) 111 cursor.execute("select * from instant where name = 'sched_waking' limit 10") 112 result = cursor.fetchall() 113 for row in result: 114 assert(row[2] > 0) 115 assert(row[3] > 0) 116 cursor.close() 117 conn.close() 118 assert check_wake 119 assert check_newtask 120 assert check_exit 121 122 @pytest.mark.L0 123 def test_ftrace_freq(self): 124 subprocess.check_output(f"hdc file send ..\\inputfiles\\ftrace\\config_ftrace_freq.txt /data/local/tmp/", 125 text=True, encoding="utf-8") 126 task_thread = threading.Thread(target=task_freq, args=()) 127 task_thread.start() 128 task_thread.join() 129 subprocess.check_output(f"hdc file recv /data/local/tmp/test_ftrace_freq.htrace ../outputfiles/ ", 130 text=True, encoding="utf-8") 131 # 检查文件大小 132 file_size = get_file_size(f"../outputfiles/test_ftrace_freq.htrace") 133 assert (file_size > 1024) 134 subprocess.check_output( 135 r"../inputfiles/trace_streamer_db.exe ../outputfiles/test_ftrace_freq.htrace -e ../outputfiles/test_ftrace_freq.db") 136 137 # 连接数据库文件 138 conn = sqlite3.connect(r'../outputfiles/test_ftrace_freq.db') 139 # 创建游标对象 140 cursor = conn.cursor() 141 # 执行SQL查询 142 cursor.execute("select end_ts - start_ts as time from trace_range") 143 result = cursor.fetchall() 144 for row in result: 145 assert(row[0] > 10 * 1000 * 1000 * 1000) 146 #检查cpu 频率 147 cursor.execute("select count(0) from cpu_measure_filter where name = 'cpu_frequency'") 148 result = cursor.fetchall() 149 for row in result: 150 assert(row[0] == 12) 151 152 cursor.execute("select * from measure,cpu_measure_filter where filter_id = id and name ='cpu_frequency' limit 10") 153 result = cursor.fetchall() 154 for row in result: 155 assert(row[3] > 0) 156 cursor.close() 157 conn.close()