• 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 pytest
17import subprocess
18import re
19import time
20import threading
21import sqlite3
22import os
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_hidumper.txt -o /data/local/tmp/test_hidumper.htrace -t 30 -s -k"')
32
33
34def task_nosec():
35    subprocess.check_output(f'hdc shell "hiprofiler_cmd -c /data/local/tmp/config_hidumper_nosec.txt -o /data/local/tmp/test_hidumper_nosec.htrace -t 30 -s -k"')
36
37
38def task_section_60():
39    subprocess.check_output(f'hdc shell "hiprofiler_cmd -c /data/local/tmp/config_hidumper_60.txt -o /data/local/tmp/test_hidumper_60.htrace -t 30 -s -k"')
40
41
42def task_fps():
43    subprocess.check_output(f'hdc shell "hiprofiler_cmd -c /data/local/tmp/config_hidumper_fps.txt -o /data/local/tmp/test_hidumper_fps.htrace -t 30 -s -k"')
44
45
46def check_process():
47    count = 0
48    while (count < 5):
49        output_text = subprocess.run(f'hdc shell "ps -ef | grep SP_daemon"', stdout=subprocess.PIPE, text=True, check=True)
50        process_info = output_text.stdout
51        lines = process_info.strip().split('\n')
52        check_index = False
53        for line in lines:
54            if line.find("SP_daemon -profilerfps") != -1:
55                check_index = True
56        assert (check_index)
57        time.sleep(10)
58        count = count + 1
59
60
61class TestHiprofilerHidumpPlugin:
62    # 将一秒分成10段 section 为10
63    @pytest.mark.L0
64    def test_hidump_plugin(self):
65        subprocess.check_output(f"hdc file send ..\\inputfiles\\hidumper_plugin\\config_hidumper.txt /data/local/tmp/", shell=False,
66                                text=True, encoding="utf-8")
67        task_thread = threading.Thread(target=task, args=())
68        task_thread.start()
69        # 唤醒屏幕
70        subprocess.check_call("hdc shell power-shell wakeup", shell=False)
71        # 设置屏幕常亮
72        subprocess.check_call("hdc shell power-shell setmode 602", shell=False)
73        time.sleep(3)
74        # 解锁屏幕
75        subprocess.check_call("hdc shell uinput -T -g 100 100 500 500", shell=False)
76        time.sleep(3)
77        subprocess.check_output(f"hdc shell uitest uiInput keyEvent Home", shell=False,
78                                text=True, encoding="utf-8")
79        time.sleep(5)
80        subprocess.check_output(f"hdc shell uinput -T -c 650 2447", shell=False, text=True, encoding="utf-8")
81        time.sleep(1)
82        subprocess.check_output(f"hdc shell uinput -T -c 104 1532", shell=False, text=True, encoding="utf-8")
83
84        task_thread.join()
85        subprocess.run(f'hdc file recv /data/local/tmp/test_hidumper.htrace ../outputfiles/', shell=False,
86                       text=True, encoding="utf-8")
87        # 检查文件大小
88        file_size = get_file_size(f"../outputfiles/test_hidumper.htrace")
89        assert (file_size > 1024)
90        subprocess.check_output(
91            r"../inputfiles/trace_streamer_db.exe ../outputfiles/test_hidumper.htrace -e ../outputfiles/test_hidumper.db")
92        # 连接数据库文件
93        conn = sqlite3.connect(r'../outputfiles/test_hidumper.db')
94        # # 创建游标对象
95        cursor = conn.cursor()
96        cursor.execute('SELECT * FROM hidump order by ts limit 0,10')
97        result = cursor.fetchall()
98        row_count = len(result)
99        #检查获得FPS数据是否正确
100        assert(row_count == 10)
101        for row in result:
102            assert(row[2] >= 0)
103        # 检查分段有没有成功
104
105        last_row = result[0][1]
106        for row in result[1:]:
107            assert((row[1] - last_row) == 100 * 1000 * 1000 or (row[1] - last_row) == 1000 * 1000 * 1000)
108            last_row = row[1]
109        cursor.close()
110        conn.close()
111
112    #不分段场景
113    @pytest.mark.L0
114    def test_hidump_plugin_nosec(self):
115        subprocess.check_output(f"hdc file send ..\\inputfiles\\hidumper_plugin\\config_hidumper_nosec.txt /data/local/tmp/", shell=False,
116                                 text=True, encoding="utf-8")
117        task_thread = threading.Thread(target=task_nosec, args=())
118        task_thread.start()
119        task_thread.join()
120        subprocess.run(f'hdc file recv /data/local/tmp/test_hidumper_nosec.htrace ../outputfiles/', shell=False,
121                         text=True, encoding="utf-8")
122        # 检查文件大小 能正常抓到trace
123        file_size = get_file_size(f"../outputfiles/test_hidumper_nosec.htrace")
124        assert (file_size > 1024)
125        subprocess.check_output(
126            r"../inputfiles/trace_streamer_db.exe ../outputfiles/test_hidumper_nosec.htrace -e ../outputfiles/test_hidumper_nosec.db")
127        # 连接数据库文件
128        conn = sqlite3.connect(r'../outputfiles/test_hidumper_nosec.db')
129        # # 创建游标对象
130        cursor = conn.cursor()
131        cursor.execute('SELECT * FROM hidump order by ts limit 0,10')
132        result = cursor.fetchall()
133        row_count = len(result)
134        assert(row_count > 0)
135        for row in result:
136            assert(row[2] >= 0)
137        cursor.close()
138        conn.close()
139
140    #验证hidumper进程和Sp_damon 进程能否正常拉起和结束
141    @pytest.mark.L0
142    def test_hidumper_process(self):
143        subprocess.check_output(f"hdc file send ..\\inputfiles\\hidumper_plugin\\config_hidumper.txt /data/local/tmp/", shell=False,
144                                text=True, encoding="utf-8")
145        task_thread = threading.Thread(target=task, args=())
146        task_thread.start()
147        time.sleep(2)
148        task_check = threading.Thread(target=check_process, args=())
149        task_check.start()
150        task_thread.join()
151        task_check.join()
152        #检查结束后，子进程是否结束
153        output_text = subprocess.run(f'hdc shell "ps -ef | grep SP_daemon"', stdout=subprocess.PIPE, text=True, check=True)
154        process_info = output_text.stdout
155        lines = process_info.strip().split('\n')
156        check_index = False
157        for line in lines:
158            if line.find("SP_daemon -profilerfps") != -1:
159                check_index = True
160        assert(check_index == False)
161        #检查trace 文件大小
162        subprocess.run(f'hdc file recv /data/local/tmp/test_hidumper.htrace ../outputfiles/', shell=False,
163                        text=True, encoding="utf-8")
164        # 检查文件大小
165        file_size = get_file_size(f"../outputfiles/test_hidumper.htrace")
166        assert (file_size > 1024)
167
168    # 将一秒分成60段 section 为60
169    @pytest.mark.L0
170    def test_hidump_plugin_section_60(self):
171        subprocess.check_output(f"hdc file send ..\\inputfiles\\hidumper_plugin\\config_hidumper_60.txt /data/local/tmp/", shell=False,
172                                text=True, encoding="utf-8")
173        task_thread = threading.Thread(target=task_section_60, args=())
174        task_thread.start()
175        # 唤醒屏幕
176        subprocess.check_call("hdc shell power-shell wakeup", shell=False)
177        # 设置屏幕常亮
178        subprocess.check_call("hdc shell power-shell setmode 602", shell=False)
179        time.sleep(3)
180        # 解锁屏幕
181        subprocess.check_call("hdc shell uinput -T -g 100 100 500 500", shell=False)
182        time.sleep(3)
183        subprocess.check_output(f"hdc shell uitest uiInput keyEvent Home", shell=False,
184                                text=True, encoding="utf-8")
185        time.sleep(5)
186        subprocess.check_output(f"hdc shell uinput -T -c 650 2447", shell=False, text=True, encoding="utf-8")
187        time.sleep(1)
188        subprocess.check_output(f"hdc shell uinput -T -c 104 1532", shell=False, text=True, encoding="utf-8")
189
190        task_thread.join()
191        subprocess.run(f'hdc file recv /data/local/tmp/test_hidumper_60.htrace ../outputfiles/', shell=False,
192                       text=True, encoding="utf-8")
193        # 检查文件大小
194        file_size = get_file_size(f"../outputfiles/test_hidumper_60.htrace")
195        assert (file_size > 1024)
196        subprocess.check_output(
197            r"../inputfiles/trace_streamer_db.exe ../outputfiles/test_hidumper_60.htrace -e ../outputfiles/test_hidumper_60.db")
198        # 连接数据库文件
199        conn = sqlite3.connect(r'../outputfiles/test_hidumper_60.db')
200        # # 创建游标对象
201        cursor = conn.cursor()
202        cursor.execute('SELECT * FROM hidump order by ts limit 0,10')
203        result = cursor.fetchall()
204        row_count = len(result)
205        #检查获得FPS数据是否正确
206        assert(row_count == 0)
207
208    #report_fps 为false
209    @pytest.mark.L0
210    def test_hidump_plugin_fps(self):
211        subprocess.check_output(f"hdc file send ..\\inputfiles\\hidumper_plugin\\config_hidumper_fps.txt /data/local/tmp/", shell=False,
212                                 text=True, encoding="utf-8")
213        task_thread = threading.Thread(target=task_fps, args=())
214        task_thread.start()
215        task_thread.join()
216        subprocess.run(f'hdc file recv /data/local/tmp/test_hidumper_fps.htrace ../outputfiles/', shell=False,
217                         text=True, encoding="utf-8")
218        # 检查文件大小 能正常抓到trace
219        file_size = get_file_size(f"../outputfiles/test_hidumper_fps.htrace")
220        assert (file_size > 1024)
221        subprocess.check_output(
222            r"../inputfiles/trace_streamer_db.exe ../outputfiles/test_hidumper_fps.htrace -e ../outputfiles/test_hidumper_fps.db")
223        # 连接数据库文件
224        conn = sqlite3.connect(r'../outputfiles/test_hidumper_fps.db')
225        # # 创建游标对象
226        cursor = conn.cursor()
227        cursor.execute('SELECT * FROM hidump order by ts limit 0,10')
228        result = cursor.fetchall()
229        row_count = len(result)
230        assert(row_count > 0)
231        for row in result:
232            assert(row[2] == 0)
233        cursor.close()
234        conn.close()