• 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 sys
20sys.path.append("..")
21from tools.utils import *
22import time
23import threading
24SLEEP_TWO = 2
25SLEEP_THREE = 3
26SLEEP_FIVE = 5
27MULTIPLE_RESULT = 3
28ONCE_RESULT = 2
29CONFIG_INDEX = 10
30THRESH = 25000000000
31
32
33def task(index):
34    indexstr = str(index)
35    subprocess.check_output("hdc shell hiprofiler_cmd -c /data/local/tmp/config10.txt -o /data/local/tmp/test" + indexstr + ".htrace -t 30 -s -k")
36
37
38class TestHiprofilerNetworkProfiler:
39    @pytest.mark.L0
40    def test_network_profiler_multiple_times(self):
41        subprocess.check_output(r"hdc file send .\..\inputfiles\network_profiler\config10.txt /data/local/tmp/", text=True, encoding="utf-8")
42        subprocess.check_output("hdc shell killall com.example.myapplication523")
43        task_thread = threading.Thread(target=task, args=(CONFIG_INDEX, ))
44        time.sleep(SLEEP_TWO)
45        task_thread.start()
46        time.sleep(SLEEP_THREE)
47        subprocess.check_output("hdc shell aa start -a EntryAbility -b com.example.myapplication523")
48        time.sleep(SLEEP_FIVE)
49        touch_button("http_request")
50        time.sleep(SLEEP_TWO)
51        touch_button("http_request")
52        time.sleep(SLEEP_TWO)
53        touch_button("http_request")
54        task_thread.join()
55
56        subprocess.check_output("hdc shell chmod 777 /data/local/tmp/hookDecoder")
57        subprocess.check_output("hdc shell ./data/local/tmp/hookDecoder -f /data/local/tmp/test10.htrace > /data/local/tmp/test10_result.txt")
58        subprocess.check_output(f"hdc file recv /data/local/tmp/test10.htrace .\..\outputfiles\ ", text=True, encoding="utf-8")
59        subprocess.check_output(f"hdc file recv /data/local/tmp/test10_result.txt .\..\outputfiles\ ", text=True, encoding="utf-8")
60        count = 0
61        with open(r'.\..\outputfiles\test10_result.txt', 'r') as file:
62            lines = file.readlines()
63            for line in lines:
64                if "tv_nsec" in line:
65                    count += 1
66        assert count == MULTIPLE_RESULT
67        # 第二次请求和第三次请求一起被写入trace文件
68
69    @pytest.mark.L0
70    def test_network_profiler_one_time(self):
71        subprocess.check_output("hdc shell rm /data/local/tmp/test10.htrace")
72        subprocess.check_output("hdc shell rm /data/local/tmp/test10_result.txt")
73        subprocess.check_output("hdc shell killall com.example.myapplication523")
74        subprocess.check_output(r"hdc file send .\..\inputfiles\network_profiler\config10.txt /data/local/tmp/", text=True, encoding="utf-8")
75        subprocess.check_output("hdc shell killall com.example.myapplication523")
76        task_thread = threading.Thread(target=task, args=(CONFIG_INDEX, ))
77        time.sleep(SLEEP_TWO)
78        task_thread.start()
79        time.sleep(SLEEP_THREE)
80        subprocess.check_output("hdc shell aa start -a EntryAbility -b com.example.myapplication523")
81        time.sleep(SLEEP_FIVE)
82        touch_button("http_request")
83        time.sleep(SLEEP_TWO)
84        task_thread.join()
85
86        subprocess.check_output("hdc shell chmod 777 /data/local/tmp/hookDecoder")
87        subprocess.check_output("hdc shell ./data/local/tmp/hookDecoder -f /data/local/tmp/test10.htrace > /data/local/tmp/test10_result.txt")
88        subprocess.check_output(f"hdc file recv /data/local/tmp/test10.htrace .\..\outputfiles\ ", text=True, encoding="utf-8")
89        subprocess.check_output(f"hdc file recv /data/local/tmp/test10_result.txt .\..\outputfiles\ ", text=True, encoding="utf-8")
90        count = 0
91        with open(r'.\..\outputfiles\test10_result.txt', 'r') as file:
92            lines = file.readlines()
93            for line in lines:
94                if "tv_nsec" in line:
95                    count += 1
96        assert count == ONCE_RESULT
97        # 包括文件头