• 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_3, SMALL_TRACE_EXPECTED_SIZE_2
24
25
26def task():
27    subprocess.check_output(f'hdc shell "hiprofiler_cmd -c /data/local/tmp/config_cpu.txt -o /data/local/tmp/test_cpu.htrace -t 15 -s -k"')
28
29
30def task_invalid_pid():
31    subprocess.check_output(f'hdc shell "hiprofiler_cmd -c /data/local/tmp/config_cpu_invalid_pid.txt -o /data/local/tmp/test_cpu2.htrace -t 15 -s -k"')
32
33
34def task_no_pid():
35    subprocess.check_output(f'hdc shell "hiprofiler_cmd -c /data/local/tmp/config_cpu_no_pid.txt -o /data/local/tmp/test_cpu3.htrace -t 15 -s -k"')
36
37
38def task_process_is_false():
39    subprocess.check_output(f'hdc shell "hiprofiler_cmd -c /data/local/tmp/config_cpu_process_is_false.txt -o /data/local/tmp/test_cpu4.htrace -t 15 -s -k"')
40
41
42class TestHiprofilerCpuPlugin:
43
44    @pytest.mark.L0
45    def test_cpu_plugin(self):
46        subprocess.run(r'hdc file send ..\inputfiles\cpu_plugin\config_cpu.txt /data/local/tmp',
47                                text=True, encoding="utf-8")
48        task_thread = threading.Thread(target=task, args=())
49        task_thread.start()
50        task_thread.join()
51        subprocess.run(r'hdc file recv /data/local/tmp/test_cpu.htrace ..\outputfiles',
52                         text=True, encoding="utf-8")
53
54        subprocess.run(
55            r'..\inputfiles\trace_streamer_db.exe ..\outputfiles\test_cpu.htrace -e ..\outputfiles\test_cpu.db')
56
57        # 连接数据库文件
58        conn = sqlite3.connect(r'..\outputfiles\test_cpu.db')
59
60        # # 创建游标对象
61        cursor = conn.cursor()
62        # # 执行SQL查询
63        check = False
64        db_relative_path = "..\\outputfiles\\test_cpu.htrace"
65        absolute_path = os.path.abspath(db_relative_path)
66        db_size = os.path.getsize(absolute_path)
67        assert db_size > SMALL_TRACE_EXPECTED_SIZE_3
68
69        table_list = [a for a in cursor.execute('SELECT name FROM sqlite_master WHERE type = "table"')]
70        cursor.execute('SELECT * FROM cpu_usage')
71        result = cursor.fetchall()
72        if len(result):
73            check = True
74        assert check
75        # # 判断数据库中user_load system_load 数据
76        for row in result:
77            assert (row[3] > 0)
78            assert (row[4] > 0)
79
80    @pytest.mark.L0
81    def test_cpu_plugin_invalid_pid(self):
82        subprocess.run(r'hdc file send ..\inputfiles\cpu_plugin\config_cpu_invalid_pid.txt /data/local/tmp',
83                                text=True, encoding="utf-8")
84        task_thread = threading.Thread(target=task_invalid_pid, args=())
85        task_thread.start()
86        task_thread.join()
87        subprocess.run(r'hdc file recv /data/local/tmp/test_cpu2.htrace ..\outputfiles',
88                         text=True, encoding="utf-8")
89
90        subprocess.run(
91            r'..\inputfiles\trace_streamer_db.exe ..\outputfiles\test_cpu2.htrace -e ..\outputfiles\test_cpu2.db')
92
93        # 连接数据库文件
94        conn = sqlite3.connect(r'..\outputfiles\test_cpu2.db')
95
96        # # 创建游标对象
97        cursor = conn.cursor()
98
99        # # 执行SQL查询
100        check = False
101        db_relative_path = "..\\outputfiles\\test_cpu2.htrace"
102        absolute_path = os.path.abspath(db_relative_path)
103        db_size = os.path.getsize(absolute_path)
104        assert db_size < SMALL_TRACE_EXPECTED_SIZE_2
105
106        table_list = [a for a in cursor.execute('SELECT name FROM sqlite_master WHERE type = "table"')]
107        cursor.execute('SELECT * FROM cpu_usage')
108        result = cursor.fetchall()
109        if len(result) == 0:
110            check = True
111        assert check
112
113    @pytest.mark.L0
114    def test_cpu_plugin_no_pid(self):
115        subprocess.run(r'hdc file send ..\inputfiles\cpu_plugin\config_cpu_no_pid.txt /data/local/tmp',
116                                text=True, encoding="utf-8")
117        task_thread = threading.Thread(target=task_no_pid, args=())
118        task_thread.start()
119        task_thread.join()
120        subprocess.run(r'hdc file recv /data/local/tmp/test_cpu3.htrace ..\outputfiles',
121                         text=True, encoding="utf-8")
122
123        subprocess.run(
124            r'..\inputfiles\trace_streamer_db.exe ..\outputfiles\test_cpu3.htrace -e ..\outputfiles\test_cpu3.db')
125
126        # 连接数据库文件
127        conn = sqlite3.connect(r'..\outputfiles\test_cpu3.db')
128
129        # # 创建游标对象
130        cursor = conn.cursor()
131
132        # # 执行SQL查询
133        check = False
134        db_relative_path = "..\\outputfiles\\test_cpu3.htrace"
135        absolute_path = os.path.abspath(db_relative_path)
136        db_size = os.path.getsize(absolute_path)
137        assert db_size < SMALL_TRACE_EXPECTED_SIZE_2
138
139        table_list = [a for a in cursor.execute('SELECT name FROM sqlite_master WHERE type = "table"')]
140        cursor.execute('SELECT * FROM cpu_usage')
141        result = cursor.fetchall()
142        if len(result) == 0:
143            check = True
144        assert check
145
146    @pytest.mark.L0
147    def test_cpu_plugin_process_info_is_false(self):
148        subprocess.run(r'hdc file send ..\inputfiles\cpu_plugin\config_cpu_process_is_false.txt /data/local/tmp',
149                                text=True, encoding="utf-8")
150        task_thread = threading.Thread(target=task_process_is_false, args=())
151        task_thread.start()
152        task_thread.join()
153        subprocess.run(r'hdc file recv /data/local/tmp/test_cpu4.htrace ..\outputfiles',
154                         text=True, encoding="utf-8")
155
156        subprocess.run(
157            r'..\inputfiles\trace_streamer_db.exe ..\outputfiles\test_cpu4.htrace -e ..\outputfiles\test_cpu4.db')
158
159        # 连接数据库文件
160        conn = sqlite3.connect(r'..\outputfiles\test_cpu4.db')
161
162        # # 创建游标对象
163        cursor = conn.cursor()
164        # # 执行SQL查询
165        check = False
166        db_relative_path = "..\\outputfiles\\test_cpu4.htrace"
167        absolute_path = os.path.abspath(db_relative_path)
168        db_size = os.path.getsize(absolute_path)
169        assert db_size > SMALL_TRACE_EXPECTED_SIZE_3
170
171        table_list = [a for a in cursor.execute('SELECT name FROM sqlite_master WHERE type = "table"')]
172        cursor.execute('SELECT * FROM cpu_usage')
173        result = cursor.fetchall()
174        if len(result):
175            check = True
176        assert check
177        # # 判断数据库中user_load system_load 数据
178        for row in result:
179            assert (row[3] == 0)
180            assert (row[4] == 0)