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 os 17import pytest 18import sqlite3 19import subprocess 20import sys 21import threading 22sys.path.append("..") 23from tools.utils import SMALL_FILE_EXPECTED_SIZE 24 25 26def task(): 27 subprocess.check_output(f'hdc shell "hiprofiler_cmd -c /data/local/tmp/config_hisysevent.txt -o /data/local/tmp/test_hisysevent.htrace -t 20 -s -k"') 28 29 30def task2(): 31 subprocess.check_output(f'hdc shell "echo 11 > /sys/devices/platform/modem_power/state"') 32 subprocess.check_output(f'hdc shell "hiprofiler_cmd -c /data/local/tmp/config_hisysevent2.txt -o /data/local/tmp/test_hisysevent2.htrace -t 15 -s -k"') 33 34 35def task3(): 36 subprocess.check_output(f'hdc shell "hiprofiler_cmd -c /data/local/tmp/config_hisysevent3.txt -o /data/local/tmp/test_hisysevent3.htrace -t 30 -s -k"') 37 38 39class TestHiprofilerHisyseventPlugin: 40 41 @pytest.mark.L0 42 def test_hisysevent_plugin(self): 43 subprocess.run(r'hdc file send ..\inputfiles\hisysevent_plugin\config_hisysevent.txt /data/local/tmp', 44 text=True, encoding="utf-8") 45 46 task_thread = threading.Thread(target=task, args=()) 47 task_thread.start() 48 task_thread.join() 49 subprocess.run(r'hdc file recv /data/local/tmp/test_hisysevent.htrace ..\outputfiles', 50 text=True, encoding="utf-8") 51 subprocess.run( 52 r'..\inputfiles\trace_streamer_db.exe ..\outputfiles\test_hisysevent.htrace -e ..\outputfiles\test_hisysevent.db') 53 54 # 连接数据库文件 55 conn = sqlite3.connect(r'..\outputfiles\test_hisysevent.db') 56 # # 创建游标对象 57 cursor = conn.cursor() 58 59 # # 执行SQL查询 60 check = False 61 db_relative_path = "..\\outputfiles\\test_hisysevent.htrace" 62 absolute_path = os.path.abspath(db_relative_path) 63 db_size = os.path.getsize(absolute_path) 64 # # 判断数据库大小 65 assert db_size > SMALL_FILE_EXPECTED_SIZE 66 table_list = [a for a in cursor.execute('SELECT name FROM sqlite_master WHERE type = "table"')] 67 cursor.execute('SELECT * FROM hisys_all_event') 68 result = cursor.fetchall() 69 if len(result) == 0: 70 check = True 71 assert check 72 73 @pytest.mark.L0 74 def test_hisysevent_plugin2(self): 75 76 # # 抓特定进程1 77 subprocess.run(r'hdc file send ..\inputfiles\hisysevent_plugin\config_hisysevent2.txt /data/local/tmp', 78 text=True, encoding="utf-8") 79 task_thread = threading.Thread(target=task2, args=()) 80 task_thread.start() 81 task_thread.join() 82 subprocess.run(r'hdc file recv /data/local/tmp/test_hisysevent2.htrace ..\outputfiles', 83 text=True, encoding="utf-8") 84 subprocess.run( 85 r'..\inputfiles\trace_streamer_db.exe ..\outputfiles\test_hisysevent2.htrace -e ..\outputfiles\test_hisysevent2.db') 86 87 # 连接数据库文件 88 conn = sqlite3.connect(r'..\outputfiles\test_hisysevent2.db') 89 90 # # 创建游标对象 91 cursor = conn.cursor() 92 93 # # 执行SQL查询 94 check = False 95 db_relative_path = "..\\outputfiles\\test_hisysevent2.htrace" 96 absolute_path = os.path.abspath(db_relative_path) 97 db_size = os.path.getsize(absolute_path) 98 # # 判断数据库大小 99 assert db_size > SMALL_FILE_EXPECTED_SIZE 100 table_list = [a for a in cursor.execute('SELECT name FROM sqlite_master WHERE type = "table"')] 101 cursor.execute('SELECT * FROM hisys_all_event') 102 result = cursor.fetchall() 103 if len(result): 104 check = True 105 assert check 106 107 @pytest.mark.L0 108 def test_hisysevent_plugin3(self): 109 110 # # 参数设置为false 111 subprocess.run(r'hdc file send ..\inputfiles\hisysevent_plugin\config_hisysevent3.txt /data/local/tmp', 112 text=True, encoding="utf-8") 113 task_thread = threading.Thread(target=task3, args=()) 114 task_thread.start() 115 task_thread.join() 116 subprocess.run(r'hdc file recv /data/local/tmp/test_hisysevent3.htrace ..\outputfiles', 117 text=True, encoding="utf-8") 118 119 subprocess.run( 120 r'..\inputfiles\trace_streamer_db.exe ..\outputfiles\test_hisysevent3.htrace -e ..\outputfiles\test_hisysevent3.db') 121 122 # 连接数据库文件 123 conn = sqlite3.connect(r'..\outputfiles\test_hisysevent3.db') 124 125 # # 创建游标对象 126 cursor = conn.cursor() 127 128 # # 执行SQL查询 129 check = False 130 db_relative_path = "..\\outputfiles\\test_hisysevent3.htrace" 131 absolute_path = os.path.abspath(db_relative_path) 132 db_size = os.path.getsize(absolute_path) 133 # # 判断数据库大小 134 assert db_size > SMALL_FILE_EXPECTED_SIZE 135 table_list = [a for a in cursor.execute('SELECT name FROM sqlite_master WHERE type = "table"')] 136 cursor.execute('SELECT * FROM hisys_all_event') 137 result = cursor.fetchall() 138 if len(result) == 0: 139 check = True 140 assert check