• Home
  • Line#
  • Scopes#
  • Navigate#
  • Raw
  • Download
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_TRACE_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 15 -s -k"')
37
38
39def task4():
40    subprocess.check_output(f'hdc shell "hiprofiler_cmd -c /data/local/tmp/config_hisysevent4.txt -o /data/local/tmp/test_hisysevent4.htrace -t 15 -s -k"')
41
42
43class TestHiprofilerHisyseventPlugin:
44
45    # 未设置subscribe_domain参数，检查trace大小及数据
46    @pytest.mark.L0
47    def test_hisysevent_plugin(self):
48        subprocess.run(r'hdc file send ..\inputfiles\hisysevent_plugin\config_hisysevent.txt /data/local/tmp',
49                                text=True, encoding="utf-8")
50
51        task_thread = threading.Thread(target=task, args=())
52        task_thread.start()
53        task_thread.join()
54        subprocess.run(r'hdc file recv /data/local/tmp/test_hisysevent.htrace ..\outputfiles',
55                         text=True, encoding="utf-8")
56        subprocess.run(
57            r'..\inputfiles\trace_streamer_db.exe ..\outputfiles\test_hisysevent.htrace -e ..\outputfiles\test_hisysevent.db')
58
59        # 连接数据库文件
60        conn = sqlite3.connect(r'..\outputfiles\test_hisysevent.db')
61        # # 创建游标对象
62        cursor = conn.cursor()
63
64        # # 执行SQL查询
65        check = False
66        db_relative_path = "..\\outputfiles\\test_hisysevent.htrace"
67        absolute_path = os.path.abspath(db_relative_path)
68        db_size = os.path.getsize(absolute_path)
69        # # 判断数据库大小
70        assert db_size > SMALL_TRACE_EXPECTED_SIZE
71        table_list = [a for a in cursor.execute('SELECT name FROM sqlite_master WHERE type = "table"')]
72        cursor.execute('SELECT * FROM hisys_all_event')
73        result = cursor.fetchall()
74        if len(result):
75            check = True
76        assert check
77
78        # # 判断数据库中 domain_id pid event_id 数据
79        for row in result:
80            # domain_id
81            assert (row[1] > 0)
82            # pid
83            assert (row[6] > 0)
84            # event_id
85            assert (len(row[11]) > 0)
86
87    # domain为KERNEL_VENDOR，检查trace大小及数据
88    @pytest.mark.L0
89    def test_hisysevent_plugin2(self):
90
91        # # 抓特定进程1
92        subprocess.run(r'hdc file send ..\inputfiles\hisysevent_plugin\config_hisysevent2.txt /data/local/tmp',
93                                text=True, encoding="utf-8")
94        task_thread = threading.Thread(target=task2, args=())
95        task_thread.start()
96        task_thread.join()
97        subprocess.run(r'hdc file recv /data/local/tmp/test_hisysevent2.htrace ..\outputfiles',
98                         text=True, encoding="utf-8")
99        subprocess.run(
100            r'..\inputfiles\trace_streamer_db.exe ..\outputfiles\test_hisysevent2.htrace -e ..\outputfiles\test_hisysevent2.db')
101
102        # 连接数据库文件
103        conn = sqlite3.connect(r'..\outputfiles\test_hisysevent2.db')
104
105        # # 创建游标对象
106        cursor = conn.cursor()
107
108        # # 执行SQL查询
109        check = False
110        db_relative_path = "..\\outputfiles\\test_hisysevent2.htrace"
111        absolute_path = os.path.abspath(db_relative_path)
112        db_size = os.path.getsize(absolute_path)
113        # # 判断数据库大小
114        assert db_size > SMALL_TRACE_EXPECTED_SIZE
115        table_list = [a for a in cursor.execute('SELECT name FROM sqlite_master WHERE type = "table"')]
116        cursor.execute('SELECT * FROM hisys_all_event')
117        result = cursor.fetchall()
118        if len(result):
119            check = True
120        assert check
121
122    # domain为PROFILER，检查trace大小及数据
123    @pytest.mark.L0
124    def test_hisysevent_plugin3(self):
125
126        # # 参数设置为false
127        subprocess.run(r'hdc file send ..\inputfiles\hisysevent_plugin\config_hisysevent3.txt /data/local/tmp',
128                                text=True, encoding="utf-8")
129        task_thread = threading.Thread(target=task3, args=())
130        task_thread.start()
131        task_thread.join()
132        subprocess.run(r'hdc file recv /data/local/tmp/test_hisysevent3.htrace ..\outputfiles',
133                         text=True, encoding="utf-8")
134
135        subprocess.run(
136            r'..\inputfiles\trace_streamer_db.exe ..\outputfiles\test_hisysevent3.htrace -e ..\outputfiles\test_hisysevent3.db')
137
138        # 连接数据库文件
139        conn = sqlite3.connect(r'..\outputfiles\test_hisysevent3.db')
140
141        # # 创建游标对象
142        cursor = conn.cursor()
143
144        # # 执行SQL查询
145        check = False
146        db_relative_path = "..\\outputfiles\\test_hisysevent3.htrace"
147        absolute_path = os.path.abspath(db_relative_path)
148        db_size = os.path.getsize(absolute_path)
149        # # 判断数据库大小
150        assert db_size > SMALL_TRACE_EXPECTED_SIZE
151        table_list = [a for a in cursor.execute('SELECT name FROM sqlite_master WHERE type = "table"')]
152        cursor.execute('SELECT * FROM hisys_all_event')
153        result = cursor.fetchall()
154        if len(result) == 0:
155            check = True
156        assert check
157
158    # domain为异常值，检查trace大小及数据
159    @pytest.mark.L0
160    def test_hisysevent_plugin4(self):
161
162        # # 参数设置为false
163        subprocess.run(r'hdc file send ..\inputfiles\hisysevent_plugin\config_hisysevent4.txt /data/local/tmp',
164                                text=True, encoding="utf-8")
165        task_thread = threading.Thread(target=task4, args=())
166        task_thread.start()
167        task_thread.join()
168        subprocess.run(r'hdc file recv /data/local/tmp/test_hisysevent4.htrace ..\outputfiles',
169                         text=True, encoding="utf-8")
170
171        subprocess.run(
172            r'..\inputfiles\trace_streamer_db.exe ..\outputfiles\test_hisysevent4.htrace -e ..\outputfiles\test_hisysevent4.db')
173
174        # 连接数据库文件
175        conn = sqlite3.connect(r'..\outputfiles\test_hisysevent4.db')
176
177        # # 创建游标对象
178        cursor = conn.cursor()
179
180        # # 执行SQL查询
181        check = False
182        db_relative_path = "..\\outputfiles\\test_hisysevent4.htrace"
183        absolute_path = os.path.abspath(db_relative_path)
184        db_size = os.path.getsize(absolute_path)
185        # # 判断数据库大小
186        assert db_size > SMALL_TRACE_EXPECTED_SIZE
187        table_list = [a for a in cursor.execute('SELECT name FROM sqlite_master WHERE type = "table"')]
188        cursor.execute('SELECT * FROM hisys_all_event')
189        result = cursor.fetchall()
190        if len(result) == 0:
191            check = True
192        assert check