• 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 MID_TRACE_EXPECTED_SIZE_2, SMALL_TRACE_EXPECTED_SIZE_2
24
25
26def task():
27    subprocess.check_output(f'hdc shell "hiprofiler_cmd -c /data/local/tmp/config_hiperf.txt -o /data/local/tmp/test_hiperf.htrace -t 10 -s -k"')
28
29
30def task2():
31    subprocess.check_output(f'hdc shell "hiprofiler_cmd -c /data/local/tmp/config_hiperf2.txt -o /data/local/tmp/test_hiperf2.htrace -t 10 -s -k"')
32
33
34def task3():
35    subprocess.check_output(f'hdc shell "hiprofiler_cmd -c /data/local/tmp/config_hiperf3.txt -o /data/local/tmp/test_hiperf3.htrace -t 10 -s -k"')
36
37
38def task4():
39    subprocess.check_output(f'hdc shell "hiprofiler_cmd -c /data/local/tmp/config_hiperf4.txt -o /data/local/tmp/test_hiperf4.htrace -t 10 -s -k"')
40
41
42def task5():
43    subprocess.check_output(f'hdc shell "hiprofiler_cmd -c /data/local/tmp/config_hiperf5.txt -o /data/local/tmp/test_hiperf5.htrace -t 10 -s -k"')
44
45
46class TestHiprofilerhiperfPlugin:
47    # 配置is_root为false，检查trace数据正确性
48    @pytest.mark.L0
49    def test_hiperf_plugin(self):
50        subprocess.run(r'hdc file send ..\inputfiles\hiperf_plugin\config_hiperf.txt /data/local/tmp',
51                                text=True, encoding="utf-8")
52        task_thread = threading.Thread(target=task, args=())
53        task_thread.start()
54        task_thread.join()
55        subprocess.run(r'hdc file recv /data/local/tmp/test_hiperf.htrace ..\outputfiles',
56                         text=True, encoding="utf-8")
57
58        subprocess.run(
59            r'..\inputfiles\trace_streamer_db.exe ..\outputfiles\test_hiperf.htrace -e ..\outputfiles\test_hiperf.db')
60
61        # 连接数据库文件
62        conn = sqlite3.connect(r'..\outputfiles\test_hiperf.db')
63
64        # # 创建游标对象
65        cursor = conn.cursor()
66
67        # # 执行SQL查询
68        check = False
69        db_relative_path = "..\\outputfiles\\test_hiperf.htrace"
70        absolute_path = os.path.abspath(db_relative_path)
71        db_size = os.path.getsize(absolute_path)
72        # # 判断数据库大小
73        assert db_size > MID_TRACE_EXPECTED_SIZE_2
74        table_list = [a for a in cursor.execute('SELECT name FROM sqlite_master WHERE type = "table"')]
75        cursor.execute('SELECT * FROM perf_callchain order by ip limit 0,10')
76        result = cursor.fetchall()
77        if len(result):
78            check = True
79
80        # # 判断数据库中 callchain_id name数据
81        for row in result:
82            assert (row[1] > 0)
83            assert (row[7] > 0)
84
85        cursor.execute('SELECT * FROM perf_files')
86        result2 = cursor.fetchall()
87        if len(result2):
88            check = True
89        assert check
90
91    # 输入不存在的outfile_name文件路径检查trace数据正确性
92    @pytest.mark.L0
93    def test_hiperf_plugin2(self):
94        subprocess.run(r'hdc file send ..\inputfiles\hiperf_plugin\config_hiperf2.txt /data/local/tmp',
95                                text=True, encoding="utf-8")
96        task_thread = threading.Thread(target=task2, args=())
97        task_thread.start()
98        task_thread.join()
99        subprocess.run(r'hdc file recv /data/local/tmp/test_hiperf2.htrace ..\outputfiles',
100                         text=True, encoding="utf-8")
101
102        subprocess.run(
103            r'..\inputfiles\trace_streamer_db.exe ..\outputfiles\test_hiperf2.htrace -e ..\outputfiles\test_hiperf2.db')
104
105        # 连接数据库文件
106        conn = sqlite3.connect(r'..\outputfiles\test_hiperf2.db')
107
108        # # 创建游标对象
109        cursor = conn.cursor()
110
111        # # 执行SQL查询
112        check = False
113        db_relative_path = "..\\outputfiles\\test_hiperf2.htrace"
114        absolute_path = os.path.abspath(db_relative_path)
115        db_size = os.path.getsize(absolute_path)
116        # # 判断数据库大小
117        assert db_size < SMALL_TRACE_EXPECTED_SIZE_2
118        table_list = [a for a in cursor.execute('SELECT name FROM sqlite_master WHERE type = "table"')]
119        cursor.execute('SELECT * FROM perf_callchain')
120        result = cursor.fetchall()
121        if len(result) == 0:
122            check = True
123        assert check
124
125        cursor.execute('SELECT * FROM perf_files')
126        result = cursor.fetchall()
127        if len(result) == 0:
128            check = True
129        assert check
130
131    # 配置is_root为true，检查trace数据正确性
132    @pytest.mark.L0
133    def test_hiperf_plugin3(self):
134
135        # # 参数设置为false
136        subprocess.run(r'hdc file send ..\inputfiles\hiperf_plugin\config_hiperf3.txt /data/local/tmp',
137                                text=True, encoding="utf-8")
138        task_thread = threading.Thread(target=task3, args=())
139        task_thread.start()
140        task_thread.join()
141        subprocess.run(r'hdc file recv /data/local/tmp/test_hiperf3.htrace ..\outputfiles',
142                         text=True, encoding="utf-8")
143
144        subprocess.run(
145            r'..\inputfiles\trace_streamer_db.exe ..\outputfiles\test_hiperf3.htrace -e ..\outputfiles\test_hiperf3.db')
146
147        # 连接数据库文件
148        conn = sqlite3.connect(r'..\outputfiles\test_hiperf3.db')
149
150        # # 创建游标对象
151        cursor = conn.cursor()
152
153        # # 执行SQL查询
154        check = False
155        db_relative_path = "..\\outputfiles\\test_hiperf3.htrace"
156        absolute_path = os.path.abspath(db_relative_path)
157        db_size = os.path.getsize(absolute_path)
158        # # 判断数据库大小
159        assert db_size < SMALL_TRACE_EXPECTED_SIZE_2
160        table_list = [a for a in cursor.execute('SELECT name FROM sqlite_master WHERE type = "table"')]
161        cursor.execute('SELECT * FROM perf_callchain')
162        result = cursor.fetchall()
163        if len(result) == 0:
164            check = True
165        assert check
166
167        cursor.execute('SELECT * FROM perf_files')
168        result = cursor.fetchall()
169        if len(result) == 0:
170            check = True
171        assert check
172
173    # 未输入record_args参数，检查trace数据正确性
174    @pytest.mark.L0
175    def test_hiperf_plugin4(self):
176
177        # # 参数设置为false
178        subprocess.run(r'hdc file send ..\inputfiles\hiperf_plugin\config_hiperf4.txt /data/local/tmp',
179                                text=True, encoding="utf-8")
180        task_thread = threading.Thread(target=task4, args=())
181        task_thread.start()
182        task_thread.join()
183        subprocess.run(r'hdc file recv /data/local/tmp/test_hiperf4.htrace ..\outputfiles',
184                         text=True, encoding="utf-8")
185
186        subprocess.run(
187            r'..\inputfiles\trace_streamer_db.exe ..\outputfiles\test_hiperf4.htrace -e ..\outputfiles\test_hiperf4.db')
188
189        # 连接数据库文件
190        conn = sqlite3.connect(r'..\outputfiles\test_hiperf4.db')
191
192        # # 创建游标对象
193        cursor = conn.cursor()
194
195        # # 执行SQL查询
196        check = False
197        db_relative_path = "..\\outputfiles\\test_hiperf4.htrace"
198        absolute_path = os.path.abspath(db_relative_path)
199        db_size = os.path.getsize(absolute_path)
200        # # 判断数据库大小
201        assert db_size < SMALL_TRACE_EXPECTED_SIZE_2
202        table_list = [a for a in cursor.execute('SELECT name FROM sqlite_master WHERE type = "table"')]
203        cursor.execute('SELECT * FROM perf_callchain')
204        result = cursor.fetchall()
205        if len(result) == 0:
206            check = True
207        assert check
208
209        cursor.execute('SELECT * FROM perf_files')
210        result = cursor.fetchall()
211        if len(result) == 0:
212            check = True
213        assert check
214
215    # 配置log_level为1，检查trace数据正确性
216    @pytest.mark.L0
217    def test_hiperf_plugin5(self):
218
219        # # 参数设置为false
220        subprocess.run(r'hdc file send ..\inputfiles\hiperf_plugin\config_hiperf5.txt /data/local/tmp',
221                                text=True, encoding="utf-8")
222        task_thread = threading.Thread(target=task5, args=())
223        task_thread.start()
224        task_thread.join()
225        subprocess.run(r'hdc file recv /data/local/tmp/test_hiperf5.htrace ..\outputfiles',
226                         text=True, encoding="utf-8")
227
228        subprocess.run(
229            r'..\inputfiles\trace_streamer_db.exe ..\outputfiles\test_hiperf5.htrace -e ..\outputfiles\test_hiperf5.db')
230
231        # 连接数据库文件
232        conn = sqlite3.connect(r'..\outputfiles\test_hiperf5.db')
233
234        # # 创建游标对象
235        cursor = conn.cursor()
236
237        # # 执行SQL查询
238        check = False
239        db_relative_path = "..\\outputfiles\\test_hiperf5.htrace"
240        absolute_path = os.path.abspath(db_relative_path)
241        db_size = os.path.getsize(absolute_path)
242        # # 判断数据库大小
243        assert db_size > MID_TRACE_EXPECTED_SIZE_2
244        table_list = [a for a in cursor.execute('SELECT name FROM sqlite_master WHERE type = "table"')]
245        cursor.execute('SELECT * FROM perf_callchain order by ip limit 0,10')
246        result = cursor.fetchall()
247        if len(result):
248            check = True
249        assert check
250
251        # # 判断数据库中 callchain_id name数据
252        for row in result:
253            assert (row[1] > 0)
254            assert (row[7] > 0)