• 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/inputfiles/hidumper_plugin/config_hidumper.txt -o /data/local/tmp/test_hidumper.htrace -t 60 -s -k"')
32
33
34def task_nosec():
35    subprocess.check_output(f'hdc shell "hiprofiler_cmd -c /data/local/tmp/inputfiles/hidumper_plugin/config_hidumper_nosec.txt -o /data/local/tmp/test_hidumper_nosec.htrace -t 60 -s -k"')
36
37
38def check_process():
39    count = 0
40    while (count < 5):
41        output_text = subprocess.run(f'hdc shell "ps -ef | grep SP_daemon"', stdout=subprocess.PIPE, text=True, check=True)
42        process_info = output_text.stdout
43        lines = process_info.strip().split('\n')
44        check_index = False
45        for line in lines:
46            if line.find("SP_daemon -profilerfps") != -1:
47                check_index = True
48        assert (check_index)
49        time.sleep(10)
50        count = count + 1
51
52
53class TestHiprofilerHidumpPlugin:
54    # 将一秒分成10段 section 为10
55    @pytest.mark.L0
56    def test_hidump_plugin(self):
57        subprocess.check_output(f"hdc file send ./inputfiles/hidumper_plugin/config_hidumper.txt /data/local/tmp/", shell=False,
58                                text=True, encoding="utf-8")
59        task_thread = threading.Thread(target=task, args=())
60        task_thread.start()
61        # 唤醒屏幕
62        subprocess.check_call("hdc shell power-shell wakeup", shell=False)
63        # 设置屏幕常亮
64        subprocess.check_call("hdc shell power-shell setmode 602", shell=False)
65        time.sleep(3)
66        # 解锁屏幕
67        subprocess.check_call("hdc shell uinput -T -g 100 100 500 500", shell=False)
68        time.sleep(3)
69        subprocess.check_output(f"hdc shell uitest uiInput keyEvent Home", shell=False,
70                                text=True, encoding="utf-8")
71        subprocess.check_output(f"hdc shell aa start -a com.huawei.hmos.settings.MainAbility -b com.huawei.hmos.settings", shell=False,
72                                text=True, encoding="utf-8")
73        time.sleep(5)
74        subprocess.check_output(f"hdc shell aa start -a MainAbility -b com.huawei.hmos.calendar", shell=False,
75                                text=True, encoding="utf-8")
76        subprocess.check_output(f"hdc shell uinput -T -c 650 2447", shell=False, text=True, encoding="utf-8")
77        time.sleep(1)
78        subprocess.check_output(f"hdc shell uinput -T -c 104 1532", shell=False, text=True, encoding="utf-8")
79
80        task_thread.join()
81        subprocess.run(f'hdc file recv /data/local/tmp/test_hidumper.htrace ./outputfiles/', shell=False,
82                       text=True, encoding="utf-8")
83        # 检查文件大小
84        file_size = get_file_size(f"./outputfiles/test_hidumper.htrace")
85        assert (file_size > 1024)
86        subprocess.check_output(
87            r"./inputfiles/trace_streamer_db.exe ./outputfiles/test_hidumper.htrace -e ./outputfiles/test_hidumper.db")
88        # 连接数据库文件
89        conn = sqlite3.connect(r'./outputfiles/test_hidumper.db')
90        # # 创建游标对象
91        cursor = conn.cursor()
92        cursor.execute('SELECT * FROM hidump order by ts limit 0,10')
93        result = cursor.fetchall()
94        row_count = len(result)
95        #检查获得FPS数据是否正确
96        assert (row_count == 10)
97        for row in result:
98            assert (row[2] >= 0)
99        # 检查分段有没有成功
100        last_row = result[0][1]
101        for row in result[1:]:
102            print(row[1] - last_row)
103            assert ((row[1] - last_row) == 100 * 1000 * 1000 or (row[1] - last_row) == 1000 * 1000 * 1000)
104            last_row = row[1]
105        cursor.close()
106        conn.close()
107
108    #不分段场景
109    @pytest.mark.L0
110    def test_hidump_plugin_nosec(self):
111        subprocess.check_output(f"hdc file send ./inputfiles/hidumper_plugin/config_hidumper_nosec.txt /data/local/tmp/", shell=False,
112                                text=True, encoding="utf-8")
113        task_thread = threading.Thread(target=task_nosec, args=())
114        task_thread.start()
115        task_thread.join()
116        subprocess.run(f'hdc file recv /data/local/tmp/test_hidumper_nosec.htrace ./outputfiles/', shell=False,
117                         text=True, encoding="utf-8")
118        # 检查文件大小 能正常抓到trace
119        file_size = get_file_size(f"./outputfiles/test_hidumper_nosec.htrace")
120        assert (file_size > 1024)
121        subprocess.check_output(
122            r"./inputfiles/trace_streamer_db.exe ./outputfiles/test_hidumper_nosec.htrace -e ./outputfiles/test_hidumper_nosec.db")
123        # 连接数据库文件
124        conn = sqlite3.connect(r'./outputfiles/test_hidumper_nosec.db')
125        # # 创建游标对象
126        cursor = conn.cursor()
127        cursor.execute('SELECT * FROM hidump order by ts limit 0,10')
128        result = cursor.fetchall()
129        row_count = len(result)
130        assert (row_count > 0)
131        for row in result:
132            assert (row[2] >= 0)
133        cursor.close()
134        conn.close()
135
136    #验证hidumper进程和Sp_damon 进程能否正常拉起和结束
137    @pytest.mark.L0
138    def test_hidumper_process(self):
139        subprocess.check_output(f"hdc file send ./inputfiles/hidumper_plugin/config_hidumper.txt /data/local/tmp/", shell=False,
140                                text=True, encoding="utf-8")
141        task_thread = threading.Thread(target=task, args=())
142        task_thread.start()
143        time.sleep(2)
144        task_check = threading.Thread(target=check_process, args=())
145        task_check.start()
146        task_thread.join()
147        task_check.join()
148        #检查结束后,子进程是否结束
149        output_text = subprocess.run(f'hdc shell "ps -ef | grep SP_daemon"', stdout=subprocess.PIPE, text=True, check=True)
150        process_info = output_text.stdout
151        lines = process_info.strip().split('\n')
152        check_index = False
153        for line in lines:
154            if line.find("SP_daemon -profilerfps") != -1:
155                check_index = True
156        assert (check_index == False)
157        #检查trace 文件大小
158        subprocess.run(f'hdc file recv /data/local/tmp/test_hidumper.htrace ./outputfiles/', shell=False,
159                         text=True, encoding="utf-8")
160        # 检查文件大小
161        file_size = get_file_size(f"./outputfiles/test_hidumper.htrace")
162        assert (file_size > 1024)