• 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, SMALL_TRACE_EXPECTED_SIZE_2, SMALL_TRACE_EXPECTED_SIZE_3
24
25
26def task():
27    subprocess.check_output(f'hdc shell "hiprofiler_cmd -c /data/local/tmp/config_hilog.txt -o /data/local/tmp/test_hilog.htrace -t 10 -s -k"')
28
29
30def task2():
31    subprocess.check_output(f'hdc shell "hiprofiler_cmd -c /data/local/tmp/config_hilog2.txt -o /data/local/tmp/test_hilog2.htrace -t 10 -s -k"')
32
33
34def task3():
35    subprocess.check_output(f'hdc shell "hiprofiler_cmd -c /data/local/tmp/config_hilog3.txt -o /data/local/tmp/test_hilog3.htrace -t 10 -s -k"')
36
37
38def task4():
39    subprocess.check_output(f'hdc shell "hiprofiler_cmd -c /data/local/tmp/config_hilog4.txt -o /data/local/tmp/test_hilog4.htrace -t 10 -s -k"')
40
41
42class TestHiprofilerHilogPlugin:
43
44    @pytest.mark.L0
45    def test_hilog_plugin(self):
46        subprocess.run(r'hdc file send ..\inputfiles\hilog_plugin\config_hilog.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_hilog.htrace ..\outputfiles',
52                         text=True, encoding="utf-8")
53
54        subprocess.run(
55            r'..\inputfiles\trace_streamer_db.exe ..\outputfiles\test_hilog.htrace -e ..\outputfiles\test_hilog.db')
56
57        # 连接数据库文件
58        conn = sqlite3.connect(r'..\outputfiles\test_hilog.db')
59
60        # # 创建游标对象
61        cursor = conn.cursor()
62
63        # # 执行SQL查询
64        check = False
65        db_relative_path = "..\\outputfiles\\test_hilog.htrace"
66        absolute_path = os.path.abspath(db_relative_path)
67        db_size = os.path.getsize(absolute_path)
68        # # 判断数据库大小
69        assert db_size > SMALL_TRACE_EXPECTED_SIZE_3
70        table_list = [a for a in cursor.execute('SELECT name FROM sqlite_master WHERE type = "table"')]
71        cursor.execute('SELECT * FROM log order by ts limit 0,10')
72        result = cursor.fetchall()
73        if len(result):
74            check = True
75        assert check
76        # # 判断数据库中level tag 数据
77        for row in result:
78            assert (row[4] == 'I')
79            assert (len(row[5]) > 0)
80
81    @pytest.mark.L0
82    def test_hilog_plugin2(self):
83
84        # # 抓特定进程1 的log数据
85        subprocess.run(r'hdc file send ..\inputfiles\hilog_plugin\config_hilog2.txt /data/local/tmp',
86                                text=True, encoding="utf-8")
87        task_thread = threading.Thread(target=task2, args=())
88        task_thread.start()
89        task_thread.join()
90        subprocess.run(r'hdc file recv /data/local/tmp/test_hilog2.htrace ..\outputfiles',
91                         text=True, encoding="utf-8")
92
93        subprocess.run(
94            r'..\inputfiles\trace_streamer_db.exe ..\outputfiles\test_hilog2.htrace -e ..\outputfiles\test_hilog2.db')
95
96        # 连接数据库文件
97        conn = sqlite3.connect(r'..\outputfiles\test_hilog2.db')
98
99        # # 创建游标对象
100        cursor = conn.cursor()
101
102        # # 执行SQL查询
103        check = False
104        db_relative_path = "..\\outputfiles\\test_hilog2.htrace"
105        absolute_path = os.path.abspath(db_relative_path)
106        db_size = os.path.getsize(absolute_path)
107        # # 判断数据库大小
108        assert db_size < SMALL_TRACE_EXPECTED_SIZE_2
109        table_list = [a for a in cursor.execute('SELECT name FROM sqlite_master WHERE type = "table"')]
110        cursor.execute('SELECT * FROM log order by ts limit 0,10')
111        result = cursor.fetchall()
112        if len(result) == 0:
113            check = True
114        assert check
115
116    @pytest.mark.L0
117    def test_hilog_plugin3(self):
118
119        # # 日志等级设置为ERROR、参数设置为false
120        subprocess.run(r'hdc file send ..\inputfiles\hilog_plugin\config_hilog3.txt /data/local/tmp',
121                                text=True, encoding="utf-8")
122        task_thread = threading.Thread(target=task3, args=())
123        task_thread.start()
124        task_thread.join()
125        subprocess.run(r'hdc file recv /data/local/tmp/test_hilog3.htrace ..\outputfiles',
126                         text=True, encoding="utf-8")
127
128        subprocess.run(
129            r'..\inputfiles\trace_streamer_db.exe ..\outputfiles\test_hilog3.htrace -e ..\outputfiles\test_hilog3.db')
130
131        # 连接数据库文件
132        conn = sqlite3.connect(r'..\outputfiles\test_hilog3.db')
133
134        # # 创建游标对象
135        cursor = conn.cursor()
136
137        # # 执行SQL查询
138        check = False
139        db_relative_path = "..\\outputfiles\\test_hilog3.htrace"
140        absolute_path = os.path.abspath(db_relative_path)
141        db_size = os.path.getsize(absolute_path)
142        # # 判断数据库大小
143        assert db_size > SMALL_TRACE_EXPECTED_SIZE
144        table_list = [a for a in cursor.execute('SELECT name FROM sqlite_master WHERE type = "table"')]
145        cursor.execute('SELECT * FROM log order by ts limit 0,10')
146        result = cursor.fetchall()
147        if len(result):
148            check = True
149        assert check
150        # # 判断数据库中level tag 数据
151        for row in result:
152            assert (row[4] == 'E')
153            assert (len(row[5]) > 0)
154
155    @pytest.mark.L0
156    def test_hilog_plugin4(self):
157
158        # # pid设置为无效id、参数设置为false
159        subprocess.run(r'hdc file send ..\inputfiles\hilog_plugin\config_hilog4.txt /data/local/tmp',
160                                text=True, encoding="utf-8")
161        task_thread = threading.Thread(target=task4, args=())
162        task_thread.start()
163        task_thread.join()
164        subprocess.run(r'hdc file recv /data/local/tmp/test_hilog4.htrace ..\outputfiles',
165                         text=True, encoding="utf-8")
166
167        subprocess.run(
168            r'..\inputfiles\trace_streamer_db.exe ..\outputfiles\test_hilog4.htrace -e ..\outputfiles\test_hilog4.db')
169
170        # 连接数据库文件
171        conn = sqlite3.connect(r'..\outputfiles\test_hilog4.db')
172
173        # # 创建游标对象
174        cursor = conn.cursor()
175
176        # # 执行SQL查询
177        check = False
178        db_relative_path = "..\\outputfiles\\test_hilog4.htrace"
179        absolute_path = os.path.abspath(db_relative_path)
180        db_size = os.path.getsize(absolute_path)
181        # # 判断数据库大小
182        assert db_size > SMALL_TRACE_EXPECTED_SIZE
183        table_list = [a for a in cursor.execute('SELECT name FROM sqlite_master WHERE type = "table"')]
184        cursor.execute('SELECT * FROM log order by ts limit 0,10')
185        result = cursor.fetchall()
186        if len(result):
187            check = True
188        assert check