add profiler tests

Signed-off-by: zyxzyx <zhangyixin19@huawei.com>
This commit is contained in:
zyxzyx 2024-11-13 16:49:26 +08:00
parent 416abff943
commit 34574158d6
36 changed files with 3068 additions and 0 deletions

View File

@ -123,6 +123,50 @@ ohos_executable("nativetest_c") {
part_name = "${OHOS_PROFILER_PART_NAME}"
}
ohos_executable("malloctest") {
output_name = "malloctest"
sources = [ "test/malloctest.cpp" ]
ldflags = [ "-rdynamic" ]
install_enable = false
subsystem_name = "${OHOS_PROFILER_SUBSYS_NAME}"
part_name = "${OHOS_PROFILER_PART_NAME}"
}
ohos_executable("hookDecoder") {
output_name = "hookDecoder"
sources = [ "test/hook_decoder.cpp" ]
if (current_toolchain != host_toolchain) {
defines = [ "HAVE_HILOG" ]
external_deps = [
"abseil-cpp:absl_sync",
"bounds_checking_function:libsec_shared",
"grpc:grpc",
"grpc:grpcxx",
"hilog:libhilog",
"openssl:libcrypto_shared",
"protobuf:protobuf",
"protobuf:protobuf_lite",
]
}
cflags = [
"-Wno-error=inline-asm",
"-O3",
]
deps = [
"${OHOS_PROFILER_DIR}/device/services/profiler_service:profiler_service",
"${OHOS_PROFILER_DIR}/protos/services:plugin_service_proto",
"${OHOS_PROFILER_DIR}/protos/services:profiler_service_all_type_source",
"${OHOS_PROFILER_DIR}/protos/types/plugins/native_hook:native_hook_cpp",
"${OHOS_PROFILER_DIR}/protos/types/plugins/native_hook:native_hook_cpp_standard",
]
ldflags = [ "-rdynamic" ]
install_enable = false
subsystem_name = "${OHOS_PROFILER_SUBSYS_NAME}"
part_name = "${OHOS_PROFILER_PART_NAME}"
}
ohos_executable("nativetest_cpp") {
output_name = "nativetest_cpp"
sources = [ "test/hook_test.cpp" ]

View File

@ -0,0 +1,97 @@
/*
* Copyright (c) Huawei Technologies Co., Ltd. 2024. All rights reserved.
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
#include <sys/file.h>
#include <unistd.h>
#include "common_types.pb.h"
#include "trace_file_reader.h"
#include "trace_file_header.h"
#include "native_hook_result_standard.pb.h"
#include "native_hook_config_standard.pb.h"
#include "google/protobuf/text_format.h"
namespace {
using UsageHandle = std::function<void(void)>;
static std::map<std::string, std::string> params;
int ParseArgs(int argc, char** argv, UsageHandle usage)
{
params.clear();
for (int i = 1; i < argc;) {
std::string key = argv[i];
i++;
if (i >= argc) {
if (usage) usage();
break;
}
std::string val = argv[i];
i++;
params.insert(std::make_pair(key, val));
}
return params.size();
}
int GetStringArg(const char* name, std::string& val, const char* defaultVal)
{
val = params[std::string(name)];
if (val.empty()) {
val = defaultVal;
}
return val.size();
}
void Usage()
{
printf("usage: hookdecoder [-f filepath] \n");
}
} // namespace
int main(int argc, char* argv[])
{
std::string filePath;
int ret = ParseArgs(argc, argv, Usage);
if (ret == 0) {
std::cout << "parse parameters error!" << std::endl;
return 0;
}
GetStringArg("-f", filePath, "/data/local/tmp/hiprofiler_data.htrace");
auto reader = std::make_shared<TraceFileReader>();
if (!reader->Open(filePath)) {
std::cout << "open file :" << filePath << "failed!" << std::endl;
return 0;
}
long bytes = 0;
do {
ProfilerPluginData data{};
bytes = reader->Read(data);
if (data.data().size() <= 0) {
continue;
}
std::cout << "name=" << data.name() << ",status=" << data.status() << ",tv_sec=" << data.tv_sec()
<< ",tv_nsec=" << data.tv_nsec() << ",version=" << data.version() << std::endl;
std::string str;
ForStandard::BatchNativeHookData StandardStackData;
if (!StandardStackData.ParseFromArray(data.data().data(), data.data().size())) {
std::cout << "parse profiler plugin data failed!" << std::endl;
continue;
}
google::protobuf::TextFormat::PrintToString(StandardStackData, &str);
std::cout << str << std::endl;
} while (bytes > 0);
return 0;
}

View File

@ -0,0 +1,103 @@
/*
* Copyright (c) Huawei Technologies Co., Ltd. 2024. All rights reserved.
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
#include <iostream>
#include <thread>
#include <mutex>
#include <chrono>
#include <unistd.h>
#include <cstring>
#pragma clang optimize off
std::recursive_mutex mtx;
constexpr uint64_t S_TO_NS = 1000 * 1000 * 1000;
constexpr uint64_t SLEEP_TIME = 5;
constexpr uint64_t COUNT_INDEX = 3;
constexpr uint64_t SIZE_INDEX = 2;
constexpr int MAX_SIZE = 1024 * 1024 * 1024;
void AllocateMemory(int depth, int size)
{
if (size > MAX_SIZE || (size == 0)) {
return;
}
if (depth == 0) {
char* mem = new char[size];
if (mem == nullptr) {
return;
}
mem[0] = 'a';
delete[] mem;
return;
}
AllocateMemory(depth - 1, size);
}
void ThreadFunc(int depth, int count, int size)
{
for (int i = 0; i < count; ++i) {
AllocateMemory(depth, size);
}
}
int main(int argc, char* argv[])
{
int threadCount = 10;
int depth = 10;
int count = 10000;
int mallocSize = 1;
depth = atoi(argv[1]);
mallocSize = atoi(argv[SIZE_INDEX]);
count = atoi(argv[COUNT_INDEX]);
if (depth <= 0) {
std::cout << "invalid depth" << std::endl;
return 0;
}
if (count <= 0) {
std::cout << "invalid count" << std::endl;
return 0;
}
if (mallocSize < 1 || mallocSize >= MAX_SIZE) {
std::cout << "invalid size" << std::endl;
return 0;
}
std::cout << "starting memory allocation..." << std::endl;
sleep(SLEEP_TIME);
std::cout << "starting hook..." << std::endl;
void* ptr = malloc(1);
free(ptr);
sleep(SLEEP_TIME);
std::thread threads[threadCount];
std::cout << "Running..." << std::endl;
struct timespec start = {};
clock_gettime(CLOCK_REALTIME, &start);
for (int i = 0; i < threadCount; ++i) {
threads[i] = std::thread(ThreadFunc, depth, count, mallocSize);
}
for (int i = 0; i < threadCount; i++) {
threads[i].join();
}
struct timespec end = {};
clock_gettime(CLOCK_REALTIME, &end);
std::cout << "Total cost time: " << (end.tv_sec - start.tv_sec) * S_TO_NS + (end.tv_nsec - start.tv_nsec)
<< " ns" << std::endl;
sleep(SLEEP_TIME);
return 0;
}

View File

@ -0,0 +1,21 @@
request_id: 1
session_config {
buffers {
pages: 16384
}
}
plugin_configs {
plugin_name: "ftrace-plugin"
sample_interval: 1000
config_data {
ftrace_events: "sched/sched_switch"
ftrace_events: "sched/sched_blocked_reason"
buffer_size_kb: 20480
flush_interval_ms: 1000
flush_threshold_kb: 4096
parse_ksyms: true
clock: "boot"
trace_period_ms: 200
debug_on: false
}
}

View File

@ -0,0 +1,261 @@
request_id: 1
session_config {
buffers {
pages: 16384
}
}
plugin_configs {
plugin_name: "ftrace-plugin"
sample_interval: 1000
config_data {
ftrace_events: "sched/sched_switch"
ftrace_events: "power/suspend_resume"
ftrace_events: "sched/sched_wakeup"
ftrace_events: "sched/sched_wakeup_new"
ftrace_events: "sched/sched_waking"
ftrace_events: "sched/sched_process_exit"
ftrace_events: "sched/sched_process_free"
ftrace_events: "task/task_newtask"
ftrace_events: "task/task_rename"
ftrace_events: "power/cpu_frequency"
ftrace_events: "power/cpu_idle"
hitrace_categories: "ability"
hitrace_categories: "ace"
hitrace_categories: "app"
hitrace_categories: "ark"
hitrace_categories: "binder"
hitrace_categories: "disk"
hitrace_categories: "freq"
hitrace_categories: "graphic"
hitrace_categories: "idle"
hitrace_categories: "irq"
hitrace_categories: "memreclaim"
hitrace_categories: "mmc"
hitrace_categories: "multimodalinput"
hitrace_categories: "ohos"
hitrace_categories: "pagecache"
hitrace_categories: "rpc"
hitrace_categories: "sched"
hitrace_categories: "sync"
hitrace_categories: "window"
hitrace_categories: "workq"
hitrace_categories: "zaudio"
hitrace_categories: "zcamera"
hitrace_categories: "zimage"
hitrace_categories: "zmedia"
buffer_size_kb: 2048
flush_interval_ms: 1000
flush_threshold_kb: 4096
parse_ksyms: true
clock: "boot"
trace_period_ms: 200
debug_on: false
hitrace_time: 30
}
}
plugin_configs {
plugin_name: "process-plugin"
sample_interval: 1000
config_data {
report_process_tree: true
report_cpu: true
report_diskio: true
report_pss: true
}
}
plugin_configs {
plugin_name: "cpu-plugin"
sample_interval: 1000
config_data {
report_process_info: true
}
}
plugin_configs {
plugin_name: "diskio-plugin"
sample_interval: 1000
config_data {
report_io_stats: IO_REPORT
}
}
plugin_configs {
plugin_name: "network-plugin"
sample_interval: 1000
config_data {
test_file: "/data/local/tmp/"
}
}
plugin_configs {
plugin_name: "memory-plugin"
sample_interval: 5000
config_data {
report_process_tree: true
report_sysmem_mem_info: true
sys_meminfo_counters: PMEM_ACTIVE
sys_meminfo_counters: PMEM_ACTIVE_ANON
sys_meminfo_counters: PMEM_ACTIVE_FILE
sys_meminfo_counters: PMEM_ANON_PAGES
sys_meminfo_counters: PMEM_BUFFERS
sys_meminfo_counters: PMEM_CACHED
sys_meminfo_counters: PMEM_CMA_FREE
sys_meminfo_counters: PMEM_CMA_TOTAL
sys_meminfo_counters: PMEM_COMMIT_LIMIT
sys_meminfo_counters: PMEM_COMMITED_AS
sys_meminfo_counters: PMEM_DIRTY
sys_meminfo_counters: PMEM_INACTIVE
sys_meminfo_counters: PMEM_INACTIVE_ANON
sys_meminfo_counters: PMEM_INACTIVE_FILE
sys_meminfo_counters: PMEM_KERNEL_STACK
sys_meminfo_counters: PMEM_MAPPED
sys_meminfo_counters: PMEM_MEM_AVAILABLE
sys_meminfo_counters: PMEM_MEM_FREE
sys_meminfo_counters: PMEM_MEM_TOTAL
sys_meminfo_counters: PMEM_MLOCKED
sys_meminfo_counters: PMEM_PAGE_TABLES
sys_meminfo_counters: PMEM_SHMEM
sys_meminfo_counters: PMEM_SLAB
sys_meminfo_counters: PMEM_SLAB_RECLAIMABLE
sys_meminfo_counters: PMEM_SLAB_UNRECLAIMABLE
sys_meminfo_counters: PMEM_SWAP_CACHED
sys_meminfo_counters: PMEM_SWAP_FREE
sys_meminfo_counters: PMEM_SWAP_TOTAL
sys_meminfo_counters: PMEM_UNEVICTABLE
sys_meminfo_counters: PMEM_VMALLOC_CHUNK
sys_meminfo_counters: PMEM_VMALLOC_TOTAL
sys_meminfo_counters: PMEM_VMALLOC_USED
sys_meminfo_counters: PMEM_WRITEBACK
sys_meminfo_counters: PMEM_KERNEL_RECLAIMABLE
report_sysmem_vmem_info: true
sys_vmeminfo_counters: VMEMINFO_UNSPECIFIED
sys_vmeminfo_counters: VMEMINFO_NR_FREE_PAGES
sys_vmeminfo_counters: VMEMINFO_NR_ALLOC_BATCH
sys_vmeminfo_counters: VMEMINFO_NR_INACTIVE_ANON
sys_vmeminfo_counters: VMEMINFO_NR_ACTIVE_ANON
sys_vmeminfo_counters: VMEMINFO_NR_INACTIVE_FILE
sys_vmeminfo_counters: VMEMINFO_NR_ACTIVE_FILE
sys_vmeminfo_counters: VMEMINFO_NR_UNEVICTABLE
sys_vmeminfo_counters: VMEMINFO_NR_MLOCK
sys_vmeminfo_counters: VMEMINFO_NR_ANON_PAGES
sys_vmeminfo_counters: VMEMINFO_NR_MAPPED
sys_vmeminfo_counters: VMEMINFO_NR_FILE_PAGES
sys_vmeminfo_counters: VMEMINFO_NR_DIRTY
sys_vmeminfo_counters: VMEMINFO_NR_WRITEBACK
sys_vmeminfo_counters: VMEMINFO_NR_SLAB_RECLAIMABLE
sys_vmeminfo_counters: VMEMINFO_NR_SLAB_UNRECLAIMABLE
sys_vmeminfo_counters: VMEMINFO_NR_PAGE_TABLE_PAGES
sys_vmeminfo_counters: VMEMINFO_NR_KERNEL_STACK
sys_vmeminfo_counters: VMEMINFO_NR_OVERHEAD
sys_vmeminfo_counters: VMEMINFO_NR_UNSTABLE
sys_vmeminfo_counters: VMEMINFO_NR_BOUNCE
sys_vmeminfo_counters: VMEMINFO_NR_VMSCAN_WRITE
sys_vmeminfo_counters: VMEMINFO_NR_VMSCAN_IMMEDIATE_RECLAIM
sys_vmeminfo_counters: VMEMINFO_NR_WRITEBACK_TEMP
sys_vmeminfo_counters: VMEMINFO_NR_ISOLATED_ANON
sys_vmeminfo_counters: VMEMINFO_NR_ISOLATED_FILE
sys_vmeminfo_counters: VMEMINFO_NR_SHMEM
sys_vmeminfo_counters: VMEMINFO_NR_DIRTIED
sys_vmeminfo_counters: VMEMINFO_NR_WRITTEN
sys_vmeminfo_counters: VMEMINFO_NR_PAGES_SCANNED
sys_vmeminfo_counters: VMEMINFO_WORKINGSET_REFAULT
sys_vmeminfo_counters: VMEMINFO_WORKINGSET_ACTIVATE
sys_vmeminfo_counters: VMEMINFO_WORKINGSET_NODERECLAIM
sys_vmeminfo_counters: VMEMINFO_NR_ANON_TRANSPARENT_HUGEPAGES
sys_vmeminfo_counters: VMEMINFO_NR_FREE_CMA
sys_vmeminfo_counters: VMEMINFO_NR_SWAPCACHE
sys_vmeminfo_counters: VMEMINFO_NR_DIRTY_THRESHOLD
sys_vmeminfo_counters: VMEMINFO_NR_DIRTY_BACKGROUND_THRESHOLD
sys_vmeminfo_counters: VMEMINFO_PGPGIN
sys_vmeminfo_counters: VMEMINFO_PGPGOUT
sys_vmeminfo_counters: VMEMINFO_PGPGOUTCLEAN
sys_vmeminfo_counters: VMEMINFO_PSWPIN
sys_vmeminfo_counters: VMEMINFO_PSWPOUT
sys_vmeminfo_counters: VMEMINFO_PGALLOC_DMA
sys_vmeminfo_counters: VMEMINFO_PGALLOC_NORMAL
sys_vmeminfo_counters: VMEMINFO_PGALLOC_MOVABLE
sys_vmeminfo_counters: VMEMINFO_PGFREE
sys_vmeminfo_counters: VMEMINFO_PGACTIVATE
sys_vmeminfo_counters: VMEMINFO_PGDEACTIVATE
sys_vmeminfo_counters: VMEMINFO_PGFAULT
sys_vmeminfo_counters: VMEMINFO_PGMAJFAULT
sys_vmeminfo_counters: VMEMINFO_PGREFILL_DMA
sys_vmeminfo_counters: VMEMINFO_PGREFILL_NORMAL
sys_vmeminfo_counters: VMEMINFO_PGREFILL_MOVABLE
sys_vmeminfo_counters: VMEMINFO_PGSTEAL_KSWAPD_DMA
sys_vmeminfo_counters: VMEMINFO_PGSTEAL_KSWAPD_NORMAL
sys_vmeminfo_counters: VMEMINFO_PGSTEAL_KSWAPD_MOVABLE
sys_vmeminfo_counters: VMEMINFO_PGSTEAL_DIRECT_DMA
sys_vmeminfo_counters: VMEMINFO_PGSTEAL_DIRECT_NORMAL
sys_vmeminfo_counters: VMEMINFO_PGSTEAL_DIRECT_MOVABLE
sys_vmeminfo_counters: VMEMINFO_PGSCAN_KSWAPD_DMA
sys_vmeminfo_counters: VMEMINFO_PGSCAN_KSWAPD_NORMAL
sys_vmeminfo_counters: VMEMINFO_PGSCAN_KSWAPD_MOVABLE
sys_vmeminfo_counters: VMEMINFO_PGSCAN_DIRECT_DMA
sys_vmeminfo_counters: VMEMINFO_PGSCAN_DIRECT_NORMAL
sys_vmeminfo_counters: VMEMINFO_PGSCAN_DIRECT_MOVABLE
sys_vmeminfo_counters: VMEMINFO_PGSCAN_DIRECT_THROTTLE
sys_vmeminfo_counters: VMEMINFO_PGINODESTEAL
sys_vmeminfo_counters: VMEMINFO_SLABS_SCANNED
sys_vmeminfo_counters: VMEMINFO_KSWAPD_INODESTEAL
sys_vmeminfo_counters: VMEMINFO_KSWAPD_LOW_WMARK_HIT_QUICKLY
sys_vmeminfo_counters: VMEMINFO_KSWAPD_HIGH_WMARK_HIT_QUICKLY
sys_vmeminfo_counters: VMEMINFO_PAGEOUTRUN
sys_vmeminfo_counters: VMEMINFO_ALLOCSTALL
sys_vmeminfo_counters: VMEMINFO_PGROTATED
sys_vmeminfo_counters: VMEMINFO_DROP_PAGECACHE
sys_vmeminfo_counters: VMEMINFO_DROP_SLAB
sys_vmeminfo_counters: VMEMINFO_PGMIGRATE_SUCCESS
sys_vmeminfo_counters: VMEMINFO_PGMIGRATE_FAIL
sys_vmeminfo_counters: VMEMINFO_COMPACT_MIGRATE_SCANNED
sys_vmeminfo_counters: VMEMINFO_COMPACT_FREE_SCANNED
sys_vmeminfo_counters: VMEMINFO_COMPACT_ISOLATED
sys_vmeminfo_counters: VMEMINFO_COMPACT_STALL
sys_vmeminfo_counters: VMEMINFO_COMPACT_FAIL
sys_vmeminfo_counters: VMEMINFO_COMPACT_SUCCESS
sys_vmeminfo_counters: VMEMINFO_COMPACT_DAEMON_WAKE
sys_vmeminfo_counters: VMEMINFO_UNEVICTABLE_PGS_CULLED
sys_vmeminfo_counters: VMEMINFO_UNEVICTABLE_PGS_SCANNED
sys_vmeminfo_counters: VMEMINFO_UNEVICTABLE_PGS_RESCUED
sys_vmeminfo_counters: VMEMINFO_UNEVICTABLE_PGS_MLOCKED
sys_vmeminfo_counters: VMEMINFO_UNEVICTABLE_PGS_MUNLOCKED
sys_vmeminfo_counters: VMEMINFO_UNEVICTABLE_PGS_CLEARED
sys_vmeminfo_counters: VMEMINFO_UNEVICTABLE_PGS_STRANDED
sys_vmeminfo_counters: VMEMINFO_NR_ZSPAGES
sys_vmeminfo_counters: VMEMINFO_NR_ION_HEAP
sys_vmeminfo_counters: VMEMINFO_NR_GPU_HEAP
sys_vmeminfo_counters: VMEMINFO_ALLOCSTALL_DMA
sys_vmeminfo_counters: VMEMINFO_ALLOCSTALL_MOVABLE
sys_vmeminfo_counters: VMEMINFO_ALLOCSTALL_NORMAL
sys_vmeminfo_counters: VMEMINFO_COMPACT_DAEMON_FREE_SCANNED
sys_vmeminfo_counters: VMEMINFO_COMPACT_DAEMON_MIGRATE_SCANNED
sys_vmeminfo_counters: VMEMINFO_NR_FASTRPC
sys_vmeminfo_counters: VMEMINFO_NR_INDIRECTLY_RECLAIMABLE
sys_vmeminfo_counters: VMEMINFO_NR_ION_HEAP_POOL
sys_vmeminfo_counters: VMEMINFO_NR_KERNEL_MISC_RECLAIMABLE
sys_vmeminfo_counters: VMEMINFO_NR_SHADOW_CALL_STACK_BYTES
sys_vmeminfo_counters: VMEMINFO_NR_SHMEM_HUGEPAGES
sys_vmeminfo_counters: VMEMINFO_NR_SHMEM_PMDMAPPED
sys_vmeminfo_counters: VMEMINFO_NR_UNRECLAIMABLE_PAGES
sys_vmeminfo_counters: VMEMINFO_NR_ZONE_ACTIVE_ANON
sys_vmeminfo_counters: VMEMINFO_NR_ZONE_ACTIVE_FILE
sys_vmeminfo_counters: VMEMINFO_NR_ZONE_INACTIVE_ANON
sys_vmeminfo_counters: VMEMINFO_NR_ZONE_INACTIVE_FILE
sys_vmeminfo_counters: VMEMINFO_NR_ZONE_UNEVICTABLE
sys_vmeminfo_counters: VMEMINFO_NR_ZONE_WRITE_PENDING
sys_vmeminfo_counters: VMEMINFO_OOM_KILL
sys_vmeminfo_counters: VMEMINFO_PGLAZYFREE
sys_vmeminfo_counters: VMEMINFO_PGLAZYFREED
sys_vmeminfo_counters: VMEMINFO_PGREFILL
sys_vmeminfo_counters: VMEMINFO_PGSCAN_DIRECT
sys_vmeminfo_counters: VMEMINFO_PGSCAN_KSWAPD
sys_vmeminfo_counters: VMEMINFO_PGSKIP_DMA
sys_vmeminfo_counters: VMEMINFO_PGSKIP_MOVABLE
sys_vmeminfo_counters: VMEMINFO_PGSKIP_NORMAL
sys_vmeminfo_counters: VMEMINFO_PGSTEAL_DIRECT
sys_vmeminfo_counters: VMEMINFO_PGSTEAL_KSWAPD
sys_vmeminfo_counters: VMEMINFO_SWAP_RA
sys_vmeminfo_counters: VMEMINFO_SWAP_RA_HIT
sys_vmeminfo_counters: VMEMINFO_WORKINGSET_RESTORE
report_process_mem_info: true
report_app_mem_info: false
report_app_mem_by_memory_service: false
}
}

View File

@ -0,0 +1,26 @@
request_id: 1
session_config {
buffers {
pages: 16384
}
}
plugin_configs {
plugin_name: "nativehook"
sample_interval: 5000
config_data {
save_file: false
filter_size: 4096
smb_pages: 16384
max_stack_depth: 20
process_name: "com.example.insight_test_stage"
string_compressed: true
fp_unwind: true
blocked: true
callframe_compress: true
record_accurately: true
offline_symbolization: false
startup_mode: true
js_stack_report: 1
max_js_stack_depth: 20
}
}

View File

@ -0,0 +1,29 @@
request_id: 1
session_config {
buffers {
pages: 16384
}
}
plugin_configs {
plugin_name: "nativehook"
sample_interval: 5000
config_data {
save_file: false
filter_size: 4096
smb_pages: 16384
max_stack_depth: 20
process_name: "com.huawei.hmos.settings"
string_compressed: true
fp_unwind: true
blocked: true
callframe_compress: true
record_accurately: true
offline_symbolization: false
js_stack_report: 1
max_js_stack_depth: 5
filter_size: 500
startup_mode: true
statistics_interval: 10
sample_interval: 256
}
}

View File

@ -0,0 +1,28 @@
request_id: 1
session_config {
buffers {
pages: 16384
}
}
plugin_configs {
plugin_name: "nativehook"
sample_interval: 5000
config_data {
save_file: false
filter_size: 4096
smb_pages: 16384
max_stack_depth: 20
process_name: "com.example.insight_test_stage"
string_compressed: true
fp_unwind: true
blocked: true
callframe_compress: true
record_accurately: true
offline_symbolization: false
statistics_interval: 10
startup_mode: true
sample_interval: 256
js_stack_report: 1
max_js_stack_depth: 20
}
}

View File

@ -0,0 +1,27 @@
request_id: 1
session_config {
buffers {
pages: 16384
}
}
plugin_configs {
plugin_name: "nativehook"
sample_interval: 5000
config_data {
save_file: false
filter_size: 4096
smb_pages: 16384
max_stack_depth: 20
process_name: "com.example.insight_test_stage"
string_compressed: true
fp_unwind: false
blocked: true
callframe_compress: true
record_accurately: true
offline_symbolization: true
sample_interval: 256
statistics_interval: 10
js_stack_report: 1
max_js_stack_depth: 20
}
}

View File

@ -0,0 +1,26 @@
request_id: 1
session_config {
buffers {
pages: 16384
}
}
plugin_configs {
plugin_name: "nativehook"
sample_interval: 5000
config_data {
save_file: false
filter_size: 4096
smb_pages: 16384
max_stack_depth: 20
process_name: "com.example.insight_test_stage"
string_compressed: true
fp_unwind: true
blocked: true
callframe_compress: true
record_accurately: true
offline_symbolization: false
malloc_free_matching_interval: 10
js_stack_report: 1
max_js_stack_depth: 20
}
}

View File

@ -0,0 +1,27 @@
request_id: 1
session_config {
buffers {
pages: 16384
}
}
plugin_configs {
plugin_name: "nativehook"
sample_interval: 5000
config_data {
save_file: false
filter_size: 4096
smb_pages: 16384
max_stack_depth: 20
process_name: "com.example.insight_test_stage"
string_compressed: true
fp_unwind: true
blocked: true
callframe_compress: true
record_accurately: true
offline_symbolization: false
statistics_interval: 10
sample_interval: 50000
js_stack_report: 1
max_js_stack_depth: 20
}
}

View File

@ -0,0 +1,26 @@
request_id: 1
session_config {
buffers {
pages: 16384
}
}
plugin_configs {
plugin_name: "nativehook"
sample_interval: 5000
config_data {
save_file: false
filter_size: 4096
smb_pages: 16384
max_stack_depth: 20
process_name: "malloctest"
string_compressed: true
fp_unwind: true
blocked: true
callframe_compress: true
record_accurately: true
offline_symbolization: true
statistics_interval: 10
js_stack_report: 1
max_js_stack_depth: 20
}
}

View File

@ -0,0 +1,25 @@
request_id: 1
session_config {
buffers {
pages: 16384
}
}
plugin_configs {
plugin_name: "nativehook"
sample_interval: 5000
config_data {
save_file: false
filter_size: 4096
smb_pages: 16384
max_stack_depth: 20
process_name: "com.ohos.sceneboard"
string_compressed: true
fp_unwind: true
blocked: true
callframe_compress: true
record_accurately: true
offline_symbolization: false
js_stack_report: 1
max_js_stack_depth: 5
}
}

View File

@ -0,0 +1,26 @@
request_id: 1
session_config {
buffers {
pages: 16384
}
}
plugin_configs {
plugin_name: "nativehook"
sample_interval: 5000
config_data {
save_file: false
filter_size: 4096
smb_pages: 16384
max_stack_depth: 20
process_name: "com.ohos.sceneboard"
string_compressed: true
fp_unwind: true
blocked: true
callframe_compress: true
record_accurately: true
offline_symbolization: false
js_stack_report: 1
max_js_stack_depth: 5
filter_size: 50000
}
}

View File

@ -0,0 +1,29 @@
request_id: 1
session_config {
buffers {
pages: 16384
}
}
plugin_configs {
plugin_name: "nativehook"
sample_interval: 5000
config_data {
save_file: false
smb_pages: 16384
max_stack_depth: 20
process_name: "com.example.insight_test_stage"
string_compressed: true
fp_unwind: true
blocked: true
callframe_compress: true
record_accurately: true
offline_symbolization: true
js_stack_report: 1
max_js_stack_depth: 20
filter_size: 500
startup_mode: true
statistics_interval: 10
sample_interval: 256
expand_pids: 0
}
}

View File

@ -0,0 +1,29 @@
request_id: 1
session_config {
buffers {
pages: 16384
}
}
plugin_configs {
plugin_name: "nativehook"
sample_interval: 5000
config_data {
save_file: false
smb_pages: 16384
max_stack_depth: 20
process_name: "com.example.insight_test_stage"
string_compressed: true
fp_unwind: true
blocked: true
callframe_compress: true
record_accurately: true
offline_symbolization: true
js_stack_report: 1
max_js_stack_depth: 20
filter_size: 500
startup_mode: true
statistics_interval: 10
sample_interval: 256
response_library_mode: false
}
}

View File

@ -0,0 +1,16 @@
request_id: 1
session_config {
buffers {
pages: 16384
}
}
plugin_configs {
plugin_name: "network-profiler"
sample_interval: 1000
config_data {
startup_process_name : "com.example.myapplication523"
flush_interval : 1
smb_pages: 16384
clock_id : 1
}
}

View File

@ -0,0 +1,16 @@
request_id: 1
session_config {
buffers {
pages: 16384
}
}
plugin_configs {
plugin_name: "network-profiler"
sample_interval: 1000
config_data {
startup_process_name : "com.tencent.mtthm"
flush_interval : 1
smb_pages: 16384
clock_id : 1
}
}

52
test/scripts/main.py Normal file
View File

@ -0,0 +1,52 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# Copyright (C) 2024 Huawei Device Co., Ltd.
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
import subprocess
import pkg_resources
import pytest
import os
import time
GET_VERSION = "V1.0.0"
def check_library_installation(library_name):
try:
pkg_resources.get_distribution(library_name)
return 0
except pkg_resources.DistributionNotFound:
print(f"\n\n{library_name} is not installed.\n\n")
print(f"try to use command below:")
print(f"pip install {library_name}")
return 1
if __name__ == '__main__':
if check_library_installation("pytest"):
subprocess.check_call(["pip", "install", "-r", "requirements.txt"])
if check_library_installation("pytest"):
exit(1)
start_time = time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(time.time()))
pytest.main()
end_time = time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(time.time()))
report_time = time.strftime('%Y-%m-%d_%H_%M_%S', time.localtime(time.time()))
report_dir = os.path.join(os.getcwd(), "reports")
report_file = os.path.join(report_dir, f"{report_time}report.html")
print(f"Test over, the script version is {GET_VERSION},"
f" start at {start_time}, end at {end_time} \n"
f"=======>{report_file} is saved. \n"
)
input("=======>press [Enter] key to Show logs.")

View File

26
test/scripts/prepare.py Normal file
View File

@ -0,0 +1,26 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# Copyright (C) 2024 Huawei Device Co., Ltd.
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
import subprocess
def install_dependencies(requirements_file):
try:
subprocess.check_call(["pip", "install", "-r", requirements_file])
print(f"install requirements.txt success")
except subprocess.CalledProcessError as e:
print(f"install dependence fail: {str(e)}")
install_dependencies("requirements.txt")

20
test/scripts/pytest.ini Normal file
View File

@ -0,0 +1,20 @@
[pytest]
# 命令行规则,空格分隔
# -m "L0 or L1"
# --alluredir ./temp
addopts = -vs --html=./reports/report.html --self-contained-html
# 测试用例路径
testpaths = ../py
# 模块名的规则
python_files = test*.py
# 类名的规则
python_classes = Test*
# 方法名的规格
python_functions = test*
# pytest执行顺序默认从上到下需要改变顺序进行order参数控制
#用例分类
markers =
L0:冒烟用例
L1:基础用例
L2:扩展用例

36
test/scripts/readme.md Normal file
View File

@ -0,0 +1,36 @@
# pytest命令行参数
-s: 显示输出调试信息包括print打印的信息
-v: 显示更详细的信息
-n支持多线程运行脚本需要保持用例彼此独立
--reruns NUM失败用例重跑次数
-x表示只要有一个用例报错那么测试停止
-k模糊匹配字符串进行用例跑测
## 测试用例执行
前置条件:
1. 将onebox中的trace_streamer_nativehook.exe, process_resource_limit.json放入inputfiles文件夹
2. 打开并运行onebox中的profiler-test和network-profiler-test这两个IDE工程
3. 将所有的 "text=True"替换为"shell=True, text=True"
windows环境下执行测试用例
进入scripts目录
### 方式一:
```
python main.py
```
执行参数在pytest.main中配置
### 方式二:
执行所有用例
```
pytest ./
```
执行指定测试文件
```
pytest ./testRoot/test_nativehook.py
```
```
## 测试报告
执行python main.py后会在reports目录下生成测试报告

View File

@ -0,0 +1,319 @@
body {
font-family: Helvetica, Arial, sans-serif;
font-size: 12px;
/* do not increase min-width as some may use split screens */
min-width: 800px;
color: #999;
}
h1 {
font-size: 24px;
color: black;
}
h2 {
font-size: 16px;
color: black;
}
p {
color: black;
}
a {
color: #999;
}
table {
border-collapse: collapse;
}
/******************************
* SUMMARY INFORMATION
******************************/
#environment td {
padding: 5px;
border: 1px solid #e6e6e6;
vertical-align: top;
}
#environment tr:nth-child(odd) {
background-color: #f6f6f6;
}
#environment ul {
margin: 0;
padding: 0 20px;
}
/******************************
* TEST RESULT COLORS
******************************/
span.passed,
.passed .col-result {
color: green;
}
span.skipped,
span.xfailed,
span.rerun,
.skipped .col-result,
.xfailed .col-result,
.rerun .col-result {
color: orange;
}
span.error,
span.failed,
span.xpassed,
.error .col-result,
.failed .col-result,
.xpassed .col-result {
color: red;
}
.col-links__extra {
margin-right: 3px;
}
/******************************
* RESULTS TABLE
*
* 1. Table Layout
* 2. Extra
* 3. Sorting items
*
******************************/
/*------------------
* 1. Table Layout
*------------------*/
#results-table {
border: 1px solid #e6e6e6;
color: #999;
font-size: 12px;
width: 100%;
}
#results-table th,
#results-table td {
padding: 5px;
border: 1px solid #e6e6e6;
text-align: left;
}
#results-table th {
font-weight: bold;
}
/*------------------
* 2. Extra
*------------------*/
.logwrapper {
max-height: 230px;
overflow-y: scroll;
background-color: #e6e6e6;
}
.logwrapper.expanded {
max-height: none;
}
.logwrapper.expanded .logexpander:after {
content: "collapse [-]";
}
.logwrapper .logexpander {
z-index: 1;
position: sticky;
top: 10px;
width: max-content;
border: 1px solid;
border-radius: 3px;
padding: 5px 7px;
margin: 10px 0 10px calc(100% - 80px);
cursor: pointer;
background-color: #e6e6e6;
}
.logwrapper .logexpander:after {
content: "expand [+]";
}
.logwrapper .logexpander:hover {
color: #000;
border-color: #000;
}
.logwrapper .log {
min-height: 40px;
position: relative;
top: -50px;
height: calc(100% + 50px);
border: 1px solid #e6e6e6;
color: black;
display: block;
font-family: "Courier New", Courier, monospace;
padding: 5px;
padding-right: 80px;
white-space: pre-wrap;
}
div.media {
border: 1px solid #e6e6e6;
float: right;
height: 240px;
margin: 0 5px;
overflow: hidden;
width: 320px;
}
.media-container {
display: grid;
grid-template-columns: 25px auto 25px;
align-items: center;
flex: 1 1;
overflow: hidden;
height: 200px;
}
.media-container--fullscreen {
grid-template-columns: 0px auto 0px;
}
.media-container__nav--right,
.media-container__nav--left {
text-align: center;
cursor: pointer;
}
.media-container__viewport {
cursor: pointer;
text-align: center;
height: inherit;
}
.media-container__viewport img,
.media-container__viewport video {
object-fit: cover;
width: 100%;
max-height: 100%;
}
.media__name,
.media__counter {
display: flex;
flex-direction: row;
justify-content: space-around;
flex: 0 0 25px;
align-items: center;
}
.collapsible td:not(.col-links) {
cursor: pointer;
}
.collapsible td:not(.col-links):hover::after {
color: #bbb;
font-style: italic;
cursor: pointer;
}
.col-result {
width: 130px;
}
.col-result:hover::after {
content: " (hide details)";
}
.col-result.collapsed:hover::after {
content: " (show details)";
}
#environment-header h2:hover::after {
content: " (hide details)";
color: #bbb;
font-style: italic;
cursor: pointer;
font-size: 12px;
}
#environment-header.collapsed h2:hover::after {
content: " (show details)";
color: #bbb;
font-style: italic;
cursor: pointer;
font-size: 12px;
}
/*------------------
* 3. Sorting items
*------------------*/
.sortable {
cursor: pointer;
}
.sortable.desc:after {
content: " ";
position: relative;
left: 5px;
bottom: -12.5px;
border: 10px solid #4caf50;
border-bottom: 0;
border-left-color: transparent;
border-right-color: transparent;
}
.sortable.asc:after {
content: " ";
position: relative;
left: 5px;
bottom: 12.5px;
border: 10px solid #4caf50;
border-top: 0;
border-left-color: transparent;
border-right-color: transparent;
}
.hidden, .summary__reload__button.hidden {
display: none;
}
.summary__data {
flex: 0 0 550px;
}
.summary__reload {
flex: 1 1;
display: flex;
justify-content: center;
}
.summary__reload__button {
flex: 0 0 300px;
display: flex;
color: white;
font-weight: bold;
background-color: #4caf50;
text-align: center;
justify-content: center;
align-items: center;
border-radius: 3px;
cursor: pointer;
}
.summary__reload__button:hover {
background-color: #46a049;
}
.summary__spacer {
flex: 0 0 550px;
}
.controls {
display: flex;
justify-content: space-between;
}
.filters,
.collapse {
display: flex;
align-items: center;
}
.filters button,
.collapse button {
color: #999;
border: none;
background: none;
cursor: pointer;
text-decoration: underline;
}
.filters button:hover,
.collapse button:hover {
color: #ccc;
}
.filter__label {
margin-right: 10px;
}

View File

@ -0,0 +1,7 @@
pytest
allure-pytest
pytest-xdist
pytest-ordering
pytest-rerunfailures
pytest-html
pytest-repeat

View File

@ -0,0 +1,167 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# Copyright (C) 2024 Huawei Device Co., Ltd.
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
import pytest
import subprocess
import re
import time
import sys
sys.path.append("..")
from tools.utils import *
import threading
OUTPUT_PATH = "testRoot/output"
LIB_PATH = "/system/lib"
THRESH = 25000000000
CLICK_TIMES = 150
SWIPE_TIMES = 20
SLEEP_TWENTY = 20
SLEEP_FIVE = 5
SLEEP_FOUR = 4
SLEEP_TWO = 2
SETTING_INDEX = 13
GC_INTERVAL = 10
SLEEP_LONG = 195
WAIT_TIMES = 7
HOOK_SETTINGS_TIMES = 5
def task_cmd(index):
indexstr = str(index)
subprocess.check_output(f"hdc shell hiprofiler_cmd -c /data/local/tmp/config" + indexstr + ".txt -o /data/local/tmp/test" + indexstr + ".htrace -t 20 -s -k")
class TestHiprofilerReliability:
@pytest.mark.L0
def test_appfreeze_sceneboard_sa(self):
subprocess.check_output("hdc shell rm /data/local/tmp/test.htrace", text=True, encoding="utf-8")
subprocess.check_output("hdc shell rm /data/log/reliability/resource_leak/memory_leak/*", text=True, encoding="utf-8")
subprocess.check_output(r"del .\..\outputfiles\nativehook.db ", text=True, encoding="utf-8")
subprocess.check_output(r"del .\..\outputfiles\test.htrace", text=True, encoding="utf-8")
subprocess.check_output("hdc target mount", text=True, encoding="utf-8")
subprocess.check_output(f"hdc file send .\..\inputfiles\process_resource_limit_reliability.json /data/local/tmp/", text=True, encoding="utf-8")
subprocess.check_output(f"hdc shell mv /data/local/tmp/process_resource_limit_reliability.json /data/local/tmp/process_resource_limit.json", text=True, encoding="utf-8")
subprocess.check_output(f"hdc shell cp /data/local/tmp/process_resource_limit.json /system/variant/phone/base/etc/efficiency_manager/", text=True, encoding="utf-8")
subprocess.check_output("hdc shell reboot", text=True, encoding="utf-8")
time.sleep(SLEEP_TWENTY)
j = 0
while j < WAIT_TIMES:
output = subprocess.check_output("hdc list targets", text=True, encoding="utf-8")
if output == '[Empty]\n\n':
time.sleep(SLEEP_FIVE)
j += 1
else:
break
#解除锁屏
subprocess.check_output("hdc shell uitest uinput drag 100 500 100 100 1000")
time.sleep(SLEEP_FIVE)
subprocess.check_output("hdc shell uitest uinput drag 100 500 100 100 1000")
time.sleep(SLEEP_FIVE)
subprocess.check_output("hdc shell uitest uinput drag 100 500 100 100 1000")
subprocess.check_output("hdc shell power-shell setmode 602")
subprocess.check_output("hdc shell killall com.example.insight_test_stage")
subprocess.check_output("hdc shell param set hiview.memleak.test disable")
subprocess.check_output("hdc shell killall hiview")
sceneboard = get_pid("com.ohos.sceneboard")
i = 0
while i < CLICK_TIMES:
subprocess.check_output("hdc shell uinput -T -m 200 1500 2000 1500")
subprocess.check_output("hdc shell uinput -T -m 2000 1500 200 1500")
time.sleep(SLEEP_FIVE)
if ((i % GC_INTERVAL) == 0):
subprocess.check_output("hdc shell hidumper --mem-jsheap " + str(sceneboard))
i += 1
subprocess.check_output("hdc shell ls -lh /data/log/faultlog/faultlogger/ > /data/local/tmp/faultlog.txt")
subprocess.check_output(f"hdc file recv /data/local/tmp/faultlog.txt .\..\outputfiles\ ", text=True, encoding="utf-8")
check = True
with open(r'.\..\outputfiles\faultlog.txt', 'r') as file:
lines = file.readlines()
for line in lines:
if "com.ohos.sceneboard" in line and ("syswarning" not in line):
check = False
if "render_service" in line:
check = False
assert check
def test_appfreeze_profiler_test(self):
subprocess.check_output("hdc shell rm /data/log/reliability/resource_leak/memory_leak/*", text=True, encoding="utf-8")
subprocess.check_output("hdc shell killall com.example.insight_test_stage")
subprocess.check_output("hdc shell param set hiview.memleak.test disable")
subprocess.check_output("hdc shell killall hiview")
subprocess.check_output("hdc shell aa start -a EntryAbility -b com.example.insight_test_stage")
time.sleep(SLEEP_FOUR)
touch_button("模板测试")
time.sleep(1)
subprocess.check_output("hdc shell uitest uinput drag 100 800 100 100 1000")
time.sleep(1)
touch_button("Allocations_Js_Depth")
i = 0
while i < CLICK_TIMES:
touch_button("malloc-release(depth 6)")
touch_button("small-malloc(depth 7)")
i += 1
time.sleep(SLEEP_FIVE)
subprocess.check_output("hdc shell ls -lh /data/log/faultlog/faultlogger/ > /data/local/tmp/faultlog.txt")
subprocess.check_output(f"hdc file recv /data/local/tmp/faultlog.txt .\..\outputfiles\ ", text=True, encoding="utf-8")
check = True
with open(r'.\..\outputfiles\faultlog.txt', 'r') as file:
lines = file.readlines()
for line in lines:
if "com.ohos.sceneboard" in line and ("syswarning" not in line):
check = False
if "render_service" in line:
check = False
if "com.example.insight_test_stage" in line:
check = False
assert check
def test_appfreeze_cmd_settings(self):
subprocess.check_output(f"hdc file send .\..\inputfiles\nativehook\config13.txt /data/local/tmp/", text=True, encoding="utf-8")
j = 0
i = 0
time.sleep(SLEEP_LONG)
while j < HOOK_SETTINGS_TIMES:
j += 1
task_thread = threading.Thread(target=task_cmd, args=(SETTING_INDEX, ))
task_thread.start()
i = 0
time.sleep(SLEEP_TWO)
subprocess.check_output("hdc shell killall com.huawei.hmos.settings")
subprocess.check_output("hdc shell aa start -a EntryAbility -b com.huawei.hmos.settings")
time.sleep(SLEEP_FIVE)
while (i < SWIPE_TIMES):
subprocess.check_output("hdc shell uinput -T -m 200 1500 200 200")
i += 1
task_thread.join()
j += 1
subprocess.check_output("hdc shell ls -lh /data/log/faultlog/faultlogger/ > /data/local/tmp/faultlog.txt")
subprocess.check_output(f"hdc file recv /data/local/tmp/faultlog.txt .\..\outputfiles\ ", text=True, encoding="utf-8")
check = True
with open(r'.\..\outputfiles\faultlog.txt', 'r') as file:
lines = file.readlines()
for line in lines:
if "com.huawei.hmos.settings" in line:
check = False
if "render_service" in line:
check = False
assert check == True

View File

@ -0,0 +1,100 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# Copyright (C) 2024 Huawei Device Co., Ltd.
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
import pytest
import subprocess
import re
import time
import sys
sys.path.append("..")
from tools.utils import *
import threading
WAIT_TIMES = 7
SLEEP_TWENTY = 20
SLEEP_FIVE = 5
GET_PID_TIME = 30
TOUCH_TIMES = 67
def get_daemon_pid():
subprocess.check_output(f"hdc shell ps -ef | grep daemon > /data/local/tmp/daemon.txt")
subprocess.check_output(f"hdc file recv /data/local/tmp/daemon.txt .\..\outputfiles\ ", text=True, encoding="utf-8")
check = True
with open(r'.\..\outputfiles\daemon.txt', 'r') as file:
lines = file.readlines()
for line in lines:
if "native_daemon sa" in line:
return line.split()[1]
return -1
class TestHiprofilerReliability:
@pytest.mark.L0
def test_badfd(self):
subprocess.check_output("hdc shell rm /data/local/tmp/test.htrace")
subprocess.check_output(r"del .\..\outputfiles\test.htrace", text=True, encoding="utf-8")
subprocess.check_output(r"del .\..\outputfiles\nativehook.db", text=True, encoding="utf-8")
subprocess.check_output("hdc target mount", text=True, encoding="utf-8")
subprocess.check_output(f"hdc file send .\..\inputfiles\process_resource_limit.json /system/variant/phone/base/etc/efficiency_manager", text=True, encoding="utf-8")
subprocess.check_output("hdc shell reboot", text=True, encoding="utf-8")
time.sleep(SLEEP_TWENTY)
j = 0
while j < WAIT_TIMES:
output = subprocess.check_output("hdc list targets", text=True, encoding="utf-8")
if output == '[Empty]\n\n':
time.sleep(SLEEP_FIVE)
j += 1
else:
break
#解除锁屏
subprocess.check_output("hdc shell uitest uiInput drag 100 500 100 100 1000")
time.sleep(SLEEP_FIVE)
subprocess.check_output("hdc shell uitest uiInput drag 100 500 100 100 1000")
time.sleep(SLEEP_FIVE)
subprocess.check_output("hdc shell uitest uiInput drag 100 500 100 100 1000")
subprocess.check_output("hdc shell power-shell setmode 602")
subprocess.check_output("hdc shell killall com.example.insight_test_stage")
subprocess.check_output("hdc shell param set hiview.memleak.test enable")
subprocess.check_output("hdc shell killall hiview")
subprocess.check_output("hdc shell aa start -a EntryAbility -b com.example.insight_test_stage")
process = subprocess.Popen(['hdc', 'shell', 'dmesg -w | grep avc > /data/local/tmp/avc.txt'])
time.sleep(1)
touch_button("模板测试")
time.sleep(1)
subprocess.check_output("hdc shell uitest uiInput drag 100 800 100 100 1000")
time.sleep(1)
touch_button("Allocations_Js_Depth")
i = 0
daemon_pid = 0
while i < TOUCH_TIMES:
touch_button("malloc-release(depth 6)")
if (i == GET_PID_TIME):
daemon_pid = get_daemon_pid()
touch_button("small-malloc(depth 7)")
i += 1
process.terminate()
subprocess.check_output(f"hdc file recv /data/local/tmp/avc.txt .\..\outputfiles\ ", text=True, encoding="utf-8")
check = True
with open(r'.\..\outputfiles\avc.txt', 'r') as file:
lines = file.readlines()
for line in lines:
if "BADFD" in line:
if "pid=" + str(daemon_pid) + " tid=" in line:
check = False
assert check == True

View File

@ -0,0 +1,42 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# Copyright (C) 2024 Huawei Device Co., Ltd.
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
import pytest
import subprocess
import re
import sys
sys.path.append("..")
from tools.utils import *
import threading
class TestHiprofilerReliability:
@pytest.mark.L0
def test_reliability_nocrash(self):
check = True
with open(r'.\..\outputfiles\faultlog.txt', 'r') as file:
lines = file.readlines()
for line in lines:
if "hiprofilerd" in line:
check = False
if "hiprofiler_plugins" in line:
check = False
if "native_daemon" in line:
check = False
if "com.example.insight_test_stage" in line:
check = False
if "com.ohos.sceneboard" in line:
check = False
assert check == True

View File

@ -0,0 +1,60 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# Copyright (C) 2024 Huawei Device Co., Ltd.
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
import pytest
import subprocess
import re
import sys
import time
sys.path.append("..")
from tools.utils import *
import threading
LIB_PATH = "/system/lib"
THRESH = 25000000000
SLEEP_TIME = 2
CONFIG_INDEX = 8
def task(index):
indexstr = str(index)
subprocess.check_output("hdc shell hiprofiler_cmd -c /data/local/tmp/config" + indexstr + ".txt -o /data/local/tmp/test" + indexstr + ".htrace -t 20 -s -k")
class TestHiprofilerFtrace:
@pytest.mark.L0
def test_allplugin(self):
subprocess.check_output(r"hdc file send .\..\inputfiles\ftrace\config8.txt /data/local/tmp/", text=True, encoding="utf-8")
task_thread = threading.Thread(target=task, args=(CONFIG_INDEX, ))
time.sleep(SLEEP_TIME)
task_thread.start()
subprocess.check_output(f"hdc shell uitest uiInput drag 100 100 800 100 1000")
subprocess.check_output(f"hdc shell uitest uiInput drag 800 100 100 100 1000")
task_thread.join()
subprocess.check_output("hdc shell chmod 777 /data/local/tmp/hookDecoder")
subprocess.check_output("hdc shell ./data/local/tmp/hookDecoder -f /data/local/tmp/test8.htrace > /data/local/tmp/test8_result.txt")
subprocess.check_output(f"hdc file recv /data/local/tmp/test8.htrace .\..\outputfiles\ ", text=True, encoding="utf-8")
subprocess.check_output(f"hdc file recv /data/local/tmp/test8_result.txt .\..\outputfiles\ ", text=True, encoding="utf-8")
check_sceneboard = False
check_cpu = False
with open(r'.\..\outputfiles\test8_result.txt', 'r') as file:
lines = file.readlines()
for line in lines:
if "com.ohos.sceneboard" in line:
check_sceneboard = True
if "cpu5" in line:
check_cpu = True
assert (check_sceneboard and check_cpu)

View File

@ -0,0 +1,57 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# Copyright (C) 2024 Huawei Device Co., Ltd.
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
import pytest
import subprocess
import re
import sys
sys.path.append("..")
from tools.utils import *
import threading
OUTPUT_PATH = "testModule/output"
LIB_PATH = "/system/lib"
THRESH = 25000000000
SLEEP_TIME = 2
CONFIG_INDEX = 11
def task(index):
indexstr = str(index)
subprocess.check_output(f"hdc shell hiprofiler_cmd -c /data/local/tmp/config" + indexstr + ".txt -o /data/local/tmp/test" + indexstr + ".htrace -t 20 -s -k")
class TestHiprofilerMalloctime:
@pytest.mark.L0
def test_sched_blocked_reason(self):
subprocess.check_output(r"hdc file send .\..\inputfiles\ftrace\config11.txt /data/local/tmp/", text=True, encoding="utf-8")
task_thread = threading.Thread(target=task, args=(CONFIG_INDEX, ))
time.sleep(SLEEP_TIME)
task_thread.start()
subprocess.check_output(f"hdc shell uitest uiInput drag 100 100 800 100 1000")
subprocess.check_output(f"hdc shell uitest uiInput drag 800 100 100 100 1000")
task_thread.join()
subprocess.check_output(f"hdc shell chmod 777 /data/local/tmp/hookDecoder")
subprocess.check_output(f"hdc shell ./data/local/tmp/hookDecoder -f /data/local/tmp/test11.htrace > /data/local/tmp/test11_result.txt")
subprocess.check_output(f"hdc file recv /data/local/tmp/test11.htrace .\..\outputfiles\ ", text=True, encoding="utf-8")
subprocess.check_output(f"hdc file recv /data/local/tmp/test11_result.txt .\..\outputfiles\ ", text=True, encoding="utf-8")
sched_blocked_reason = False
with open(r'.\..\outputfiles\test11_result.txt', 'r') as file:
lines = file.readlines()
for line in lines:
if "+0x" in line and "/0x" in line:
sched_blocked_reason = True
assert sched_blocked_reason

View File

@ -0,0 +1,102 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# Copyright (C) 2024 Huawei Device Co., Ltd.
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
import pytest
import subprocess
import re
import sys
sys.path.append("..")
from tools.utils import *
OUTPUT_PATH = "testModule/output"
LIB_PATH = "/system/lib64"
MB_SIZE = 1024
ROM_THRESH = 9000
SIZE_INDEX = 4
def check_rom(output):
result = output.split()[SIZE_INDEX]
multi = False
if (result[-1] == 'M'):
multi = True
result = float(result[:-1])
if multi:
result *= MB_SIZE
return result
class TestHiprofilerRom:
@pytest.mark.L0
def test_rom(self):
# 校验命令行输出
rom_cpu = subprocess.check_output(f"hdc shell ls -lh /system/lib64/libcpudataplugin*", text=True, encoding="utf-8")
rom_cpu = check_rom(rom_cpu)
rom_gpu = subprocess.check_output(f"hdc shell ls -lh /system/lib64/libgpudataplugin*", text=True, encoding="utf-8")
rom_gpu = check_rom(rom_gpu)
rom_disk = subprocess.check_output(f"hdc shell ls -lh /system/lib64/libdiskiodataplugin*", text=True, encoding="utf-8")
rom_disk = check_rom(rom_disk)
rom_ftrace = subprocess.check_output(f"hdc shell ls -lh /system/lib64/libftrace_plugin*", text=True, encoding="utf-8")
rom_ftrace = check_rom(rom_ftrace)
rom_hidump = subprocess.check_output(f"hdc shell ls -lh /system/lib64/libhidumpplugin*", text=True, encoding="utf-8")
rom_hidump = check_rom(rom_hidump)
rom_hiebpf = subprocess.check_output(f"hdc shell ls -lh /system/lib64/libhiebpfplugin*", text=True, encoding="utf-8")
rom_hiebpf = check_rom(rom_hiebpf)
rom_hilog = subprocess.check_output(f"hdc shell ls -lh /system/lib64/libhilogplugin*", text=True, encoding="utf-8")
rom_hilog = check_rom(rom_hilog)
rom_hiperf = subprocess.check_output(f"hdc shell ls -lh /system/lib64/libhiperfplugin*", text=True, encoding="utf-8")
rom_hiperf = check_rom(rom_hiperf)
rom_hisys = subprocess.check_output(f"hdc shell ls -lh /system/lib64/libhisyseventplugin*", text=True, encoding="utf-8")
rom_hisys = check_rom(rom_hisys)
rom_memory = subprocess.check_output(f"hdc shell ls -lh /system/lib64/libmemdataplugin*", text=True, encoding="utf-8")
rom_memory = check_rom(rom_memory)
rom_network = subprocess.check_output(f"hdc shell ls -lh /system/lib64/libnetworkplugin*", text=True, encoding="utf-8")
rom_network = check_rom(rom_network)
rom_process = subprocess.check_output(f"hdc shell ls -lh /system/lib64/libprocessplugin*", text=True, encoding="utf-8")
rom_process = check_rom(rom_process)
rom_xpower = subprocess.check_output(f"hdc shell ls -lh /system/lib64/libxpowerplugin*", text=True, encoding="utf-8")
rom_xpower = check_rom(rom_xpower)
rom_hook = subprocess.check_output(f"hdc shell ls -lh /system/lib64/libnative_hook*", text=True, encoding="utf-8")
rom_hook = check_rom(rom_hook)
rom_netprofiler = subprocess.check_output(f"hdc shell ls -lh /system/lib64/libnetwork_profiler*", text=True, encoding="utf-8")
rom_netprofiler = check_rom(rom_netprofiler)
rom_daemon = subprocess.check_output(f"hdc shell ls -lh /system/bin/native_daemon*", text=True, encoding="utf-8")
rom_daemon = check_rom(rom_daemon)
rom_hiprofilerd = subprocess.check_output(f"hdc shell ls -lh /system/bin/hiprofilerd*", text=True, encoding="utf-8")
rom_hiprofilerd = check_rom(rom_hiprofilerd)
rom_cmd = subprocess.check_output(f"hdc shell ls -lh /system/bin/hiprofiler_cmd*", text=True, encoding="utf-8")
rom_cmd = check_rom(rom_cmd)
rom_plugins = subprocess.check_output(f"hdc shell ls -lh /system/bin/hiprofiler_plugins*", text=True, encoding="utf-8")
rom_plugins = check_rom(rom_plugins)
assert(rom_cpu + rom_gpu + rom_disk + rom_ftrace + rom_hidump + rom_hiebpf + rom_hilog + rom_hiperf + rom_hisys + rom_memory + rom_network + rom_process + romXpower + rom_hook + rom_daemon + rom_hiprofilerd + rom_cmd + rom_plugins + rom_netprofiler < ROM_THRESH)

View File

@ -0,0 +1,785 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# Copyright (C) 2024 Huawei Device Co., Ltd.
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
import pytest
import subprocess
import re
import time
import sys
sys.path.append("..")
from tools.utils import *
import threading
import sqlite3
import datetime
import os
import stat
DESTROY_SIZE = 41943040
EXIST_SIZE = 40960
SLEEP_TWO = 2
SLEEP_FOUR = 4
SLEEP_FIVE = 5
SLEEP_TWENTY = 20
SYMBOL_INDEX = 4
APPLY_INDEX = 8
RELEASE_INDEX = 9
ALLOC_INDEX = 10
TYPE_INDEX = 4
MALLOC_TIMES = 10
ADDR_INDEX = 9
FILTER_THRESH = 5000
DEPTH_FIVE = 5
DEPTH_TEN = 10
DEPTH_FIFTEEN = 15
DEPTH_TWENTY = 20
DEPTH_THIRTY = 30
DEPTH_FIFTY = 50
CALLSTACKID_INDEX = 4
IPID_INDEX = 2
PID_INDEX = 2
MALLOC_THRESH = 1000
SA_CLICK_TIMES = 67
SA_WAIT_TIMES = 7
SA_STATISTICS = 300
SA_SAMPLE = 512
SAMPLE_SMALL = 512
SAMPLE_LARGE = 51200
FILTER_SMALL = 256
FILTER_LARGE = 10000
CLICK_TWICE = 2
CLICK_THREETIMES = 3
STATISTICS_INTERVAL = 10
MATCH_INTERVAL = 10
def task_template():
subprocess.check_output("hdc shell hiprofiler_cmd -c /data/local/tmp/config.txt -o /data/local/tmp/test.htrace -t 20 -s -k")
def task_multiple_template():
subprocess.check_output("hdc shell hiprofiler_cmd -c /data/local/tmp/config_multipleprocess.txt -o /data/local/tmp/test.htrace -t 25 -s -k")
def get_target_stack(result):
malloc_release_stack = [0, 0]
small_malloc_stack = [0, 0]
for row in result:
if 'Add(napi_env__*, napi_callback_info__*)' in row[1]:
small_malloc_stack[0] = row[0]
malloc_release_stack[0] = row[0]
if 'js_depthmr6' in row[1]:
malloc_release_stack[1] = row[0]
if 'js_depthsm7' in row[1]:
small_malloc_stack[1] = row[0]
return malloc_release_stack, small_malloc_stack
def check_library_result(statistics, startup, offline, sample_interval, dwarf, filtersize, depth, touchtimes, malloc_match_interval):
conn = sqlite3.connect(r'./../outputfiles/nativehook.db')
cursor = conn.cursor()
cursor.execute('SELECT * FROM data_dict')
result = cursor.fetchall()
callstack_destroyed = []
callstack_exists = []
symbol_destroy = 0
symbol_exist = 0
for row in result:
if "createAndReleaseHeap" in row[1]:
symbol_destroy = row[0]
if "createMemory" in row[1]:
symbol_exist = row[0]
cursor.execute('SELECT * FROM native_hook_frame')
result = cursor.fetchall()
for row in result:
if row[SYMBOL_INDEX] == symbol_destroy:
callstack_destroyed.append(row[1])
if row[SYMBOL_INDEX] == symbol_exist:
callstack_exists.append(row[1])
check_destroyed = False
check_exists = False
if statistics > 0:
cursor.execute('SELECT * FROM native_hook_statistic')
result = cursor.fetchall()
if touchtimes != 0:
for row in result:
for callstackid in callstack_destroyed:
if row[1] == callstackid:
if row[APPLY_INDEX] == DESTROY_SIZE * touchtimes and row[RELEASE_INDEX] == DESTROY_SIZE * touchtimes:
check_destroyed = True
for callstackid in callstack_exists:
if row[1] == callstackid:
if row[APPLY_INDEX] == EXIST_SIZE * touchtimes and row[RELEASE_INDEX] == 0:
check_exists = True
else:
for row in result:
for callstackid in callstack_destroyed:
if row[1] == callstackid:
if (row[APPLY_INDEX] % DESTROY_SIZE == 0) and row[RELEASE_INDEX] == row[APPLY_INDEX]:
check_destroyed = True
check_exists = True
else:
cursor.execute('SELECT * FROM native_hook')
result = cursor.fetchall()
times_destroyed = 0
times_exists = 0
malloc_addrs = []
for row in result:
for callstackid in callstack_destroyed:
if row[1] == callstackid and row[ALLOC_INDEX] == (DESTROY_SIZE / MALLOC_TIMES) and row[TYPE_INDEX] == "AllocEvent":
times_destroyed += 1
malloc_addrs.append(row[ADDR_INDEX])
for callstackid in callstack_exists:
if row[1] == callstackid and row[ALLOC_INDEX] == (EXIST_SIZE / MALLOC_TIMES) and row[TYPE_INDEX] == "AllocEvent":
times_exists += 1
if malloc_match_interval != 0:
if times_destroyed != 0:
return False
elif times_destroyed != (touchtimes * MALLOC_TIMES):
return False
for row in result:
if row[ADDR_INDEX] in malloc_addrs and row[ALLOC_INDEX] == (DESTROY_SIZE / MALLOC_TIMES) and row[TYPE_INDEX] == "FreeEvent":
times_destroyed -= 1
malloc_addrs.remove(row[ADDR_INDEX])
check_destroyed = (times_destroyed == 0)
check_exists = (times_exists == (touchtimes * MALLOC_TIMES))
if (sample_interval > FILTER_THRESH) or (filtersize > FILTER_THRESH):
check_exists = True
if malloc_match_interval > 0:
check_destroyed = True
cursor.close()
conn.close()
return check_destroyed and check_exists
def check_result(statistics, startup, offline, sample_interval, dwarf, filtersize, depth, touchtimes, malloc_match_interval):
conn = sqlite3.connect(r'./../outputfiles/nativehook.db')
cursor = conn.cursor()
cursor.execute('SELECT * FROM data_dict')
result = cursor.fetchall()
malloc_release_stack, small_malloc_stack = get_target_stack(result)
cursor.execute('SELECT * FROM native_hook_frame')
result = cursor.fetchall()
callstack_ids_destroyed = []
callstack_ids_exists = []
callstack_ids_native = []
for row in result:
if row[CALLSTACKID_INDEX] == malloc_release_stack[1]:
callstack_ids_destroyed.append(row[1])
if row[CALLSTACKID_INDEX] == small_malloc_stack[1]:
callstack_ids_exists.append(row[1])
if row[CALLSTACKID_INDEX] == malloc_release_stack[0]:
callstack_ids_native.append(row[1])
callstack_ids_destroyed = list(set(callstack_ids_destroyed) & set(callstack_ids_native))
callstack_ids_exists = list(set(callstack_ids_exists) & set(callstack_ids_native))
if depth == ADDR_INDEX and (not dwarf):
if len(callstack_ids_destroyed) != 0:
return False
if len(callstack_ids_exists) != 0:
return False
return True
if len(callstack_ids_destroyed) == 0 and (malloc_match_interval != 0):
return False
if (sample_interval < FILTER_THRESH and filtersize < FILTER_THRESH) and len(callstack_ids_exists) == 0:
return False
if (sample_interval >= FILTER_THRESH and filtersize >= FILTER_THRESH) and len(callstack_ids_exists) != 0:
return False
check_destroyed = False
check_exists = False
if statistics > 0:
cursor.execute('SELECT * FROM native_hook_statistic')
result = cursor.fetchall()
if touchtimes != 0:
for row in result:
for callstackid in callstack_ids_destroyed:
if row[1] == callstackid:
if row[APPLY_INDEX] == DESTROY_SIZE * touchtimes and row[RELEASE_INDEX] == DESTROY_SIZE * touchtimes:
check_destroyed = True
for callstackid in callstack_ids_exists:
if row[1] == callstackid:
if row[APPLY_INDEX] == EXIST_SIZE * touchtimes and row[RELEASE_INDEX] == 0:
check_exists = True
else:
for row in result:
for callstackid in callstack_ids_destroyed:
if row[1] == callstackid:
if (row[APPLY_INDEX] % DESTROY_SIZE == 0) and row[RELEASE_INDEX] == row[APPLY_INDEX]:
check_destroyed = True
check_exists = True
else:
cursor.execute('SELECT * FROM native_hook')
result = cursor.fetchall()
times_destroyed = 0
times_exists = 0
malloc_addrs = []
for row in result:
for callstackid in callstack_ids_destroyed:
if row[1] == callstackid and row[ALLOC_INDEX] == (DESTROY_SIZE / MALLOC_TIMES) and row[TYPE_INDEX] == "AllocEvent":
times_destroyed += 1
malloc_addrs.append(row[ADDR_INDEX])
for callstackid in callstack_ids_exists:
if row[1] == callstackid and row[ALLOC_INDEX] == (EXIST_SIZE / MALLOC_TIMES) and row[TYPE_INDEX] == "AllocEvent":
times_exists += 1
if malloc_match_interval != 0:
if times_destroyed != 0:
return False
elif times_destroyed != (touchtimes * MALLOC_TIMES):
return False
for row in result:
if row[ADDR_INDEX] in malloc_addrs and row[ALLOC_INDEX] == (DESTROY_SIZE / MALLOC_TIMES) and row[TYPE_INDEX] == "FreeEvent":
times_destroyed -= 1
malloc_addrs.remove(row[ADDR_INDEX])
check_destroyed = (times_destroyed == 0)
check_exists = (times_exists == (touchtimes * MALLOC_TIMES))
if (sample_interval > FILTER_THRESH) or (filtersize > FILTER_THRESH):
check_exists = True
if malloc_match_interval > 0:
check_destroyed = True
cursor.close()
conn.close()
return check_destroyed and check_exists
def check_nativehook_result(statistics, startup, offline, sample_interval, dwarf, filtersize, depth, touchtimes, malloc_match_interval=0, response_library=False):
try:
subprocess.check_output(r"del .\..\inputfiles\nativehook\config.txt", text=True, encoding="utf-8")
subprocess.check_output(r"del .\..\outputfiles\test.htrace", text=True, encoding="utf-8")
subprocess.check_output(r"del .\..\inputfiles\layout.json", text=True, encoding="utf-8")
subprocess.check_output(r"del .\..\outputfiles\nativehook.db", text=True, encoding="utf-8")
subprocess.check_output("hdc shell rm /data/local/tmp/test.htrace")
except Exception as e:
print(f"An error occurred: {e}")
pass
with open(r".\..\inputfiles\nativehook\config_template.txt", 'r') as file:
content = file.read()
subprocess.check_output("hdc shell power-shell setmode 602")
modified_content = content.replace('sample_interval: 256', 'sample_interval: ' + str(sample_interval))
if malloc_match_interval == 0:
modified_content = modified_content.replace('statistics_interval: 10', 'statistics_interval: ' + str(statistics))
else:
modified_content = modified_content.replace('statistics_interval: 10', 'statistics_interval: ' + str(statistics) + '\n' +
" malloc_free_matching_interval: " + str(malloc_match_interval))
modified_content = modified_content.replace('filter_size: 500', 'filter_size: ' + str(filtersize))
modified_content = modified_content.replace('max_js_stack_depth: 20', 'max_js_stack_depth: ' + str(depth))
if not offline:
modified_content = modified_content.replace('offline_symbolization: true', 'offline_symbolization: false')
if not startup:
modified_content = modified_content.replace('startup_mode: true', 'startup_mode: false')
if dwarf:
modified_content = modified_content.replace('fp_unwind: true', 'fp_unwind: false')
if response_library:
modified_content = modified_content.replace('response_library_mode: false', 'response_library_mode: true')
flags = os.O_WRONLY | os.O_CREAT | os.O_EXCL
mode = stat.S_IWUSR | stat.S_IRUSR
with os.fdopen(os.open(r".\..\inputfiles\nativehook\config.txt", flags, mode), 'w') as file:
file.write(modified_content)
subprocess.check_output(r"hdc file send .\..\inputfiles\nativehook\config.txt /data/local/tmp/", text=True, encoding="utf-8")
task_thread = threading.Thread(target=task_template, args=())
task_thread.start()
time.sleep(SLEEP_TWO)
if (startup):
subprocess.check_output("hdc shell killall com.example.insight_test_stage")
subprocess.check_output("hdc shell aa start -a EntryAbility -b com.example.insight_test_stage")
time.sleep(SLEEP_FOUR)
touch_button("模板测试")
time.sleep(1)
subprocess.check_output("hdc shell uitest uiInput drag 100 800 100 100 1000")
time.sleep(1)
touch_button("Allocations_Js_Depth")
i = 0
while i < touchtimes:
touch_button("malloc-release(depth 6)")
touch_button("small-malloc(depth 7)")
i += 1
task_thread.join()
subprocess.check_output(r"hdc file recv /data/local/tmp/test.htrace .\..\outputfiles\ ", text=True, encoding="utf-8")
subprocess.check_output(r".\..\inputfiles\trace_streamer_nativehook.exe .\..\outputfiles\test.htrace -e .\..\outputfiles\nativehook.db", text=True, encoding="utf-8")
if response_library:
return check_library_result(statistics, startup, offline, sample_interval, dwarf, filtersize, depth, touchtimes, malloc_match_interval)
return check_result(statistics, startup, offline, sample_interval, dwarf, filtersize, depth, touchtimes, malloc_match_interval)
def check_nativehook_multipleprocess(statistics, startup, offline, sample_interval, dwarf, filtersize, depth, touchtimes, malloc_match_interval=0, response_library=False):
subprocess.check_output(r"del .\..\inputfiles\nativehook\config_multipleprocess.txt", text=True, encoding="utf-8")
subprocess.check_output(r"del .\..\outputfiles\test.htrace", text=True, encoding="utf-8")
subprocess.check_output(r"del .\..\inputfiles\layout.json", text=True, encoding="utf-8")
subprocess.check_output(r"del .\..\outputfiles\nativehook.db", text=True, encoding="utf-8")
subprocess.check_output("hdc shell rm /data/local/tmp/test.htrace")
with open(r".\..\inputfiles\nativehook\config_multipleprocess_template.txt", 'r') as file:
content = file.read()
subprocess.check_output("hdc shell power-shell setmode 602")
sceneboard = get_pid("com.ohos.sceneboard")
modified_content = content.replace('sample_interval: 256', 'sample_interval: ' + str(sample_interval))
if malloc_match_interval == 0:
modified_content = modified_content.replace('statistics_interval: 10', 'statistics_interval: ' + str(statistics))
else:
modified_content = modified_content.replace('statistics_interval: 10', 'statistics_interval: ' + str(statistics) + '\n' +
" malloc_free_matching_interval: " + str(malloc_match_interval))
modified_content = modified_content.replace('filter_size: 500', 'filter_size: ' + str(filtersize))
modified_content = modified_content.replace('max_js_stack_depth: 20', 'max_js_stack_depth: ' + str(depth))
modified_content = modified_content.replace('expand_pids: 0', 'expand_pids: ' + str(sceneboard))
if not offline:
modified_content = modified_content.replace('offline_symbolization: true', 'offline_symbolization: false')
if not startup:
modified_content = modified_content.replace('startup_mode: true', 'startup_mode: false')
if dwarf:
modified_content = modified_content.replace('fp_unwind: true', 'fp_unwind: false')
if response_library:
modified_content = modified_content.replace('response_library_mode: false', 'response_library_mode: true')
flags = os.O_WRONLY | os.O_CREAT | os.O_EXCL
mode = stat.S_IWUSR | stat.S_IRUSR
with os.fdopen(os.open(r".\..\inputfiles\nativehook\config_multipleprocess.txt", flags, mode), 'w') as file:
file.write(modified_content)
subprocess.check_output(r"hdc file send .\..\inputfiles\nativehook\config_multipleprocess.txt /data/local/tmp/", text=True, encoding="utf-8")
task_thread = threading.Thread(target=task_multiple_template, args=())
task_thread.start()
time.sleep(SLEEP_TWO)
if (startup):
subprocess.check_output("hdc shell killall com.example.insight_test_stage")
subprocess.check_output("hdc shell aa start -a EntryAbility -b com.example.insight_test_stage")
time.sleep(SLEEP_FOUR)
touch_button("模板测试")
time.sleep(1)
subprocess.check_output("hdc shell uitest uiInput drag 100 800 100 100 1000")
time.sleep(1)
touch_button("Allocations_Js_Depth")
i = 0
while i < touchtimes:
touch_button("malloc-release(depth 6)")
touch_button("small-malloc(depth 7)")
i += 1
task_thread.join()
subprocess.check_output(r"hdc file recv /data/local/tmp/test.htrace .\..\outputfiles\ ", text=True, encoding="utf-8")
subprocess.check_output(r".\..\inputfiles\trace_streamer_nativehook.exe .\..\outputfiles\test.htrace -e .\..\outputfiles\nativehook.db", text=True, encoding="utf-8")
first_process = False
if response_library:
first_process = check_library_result(statistics, startup, offline, sample_interval, dwarf, filtersize, depth, touchtimes, malloc_match_interval)
else:
first_process = check_result(statistics, startup, offline, sample_interval, dwarf, filtersize, depth, touchtimes, malloc_match_interval)
conn = sqlite3.connect(r'./../outputfiles/nativehook.db')
cursor = conn.cursor()
cursor.execute('SELECT * FROM process')
result = cursor.fetchall()
ipid = 0
sceneboard = get_pid("com.ohos.sceneboard")
for row in result:
if row[PID_INDEX] == int(sceneboard):
ipid = row[1]
if ipid == 0:
return False
second_process = False
if statistics > 0:
cursor.execute('SELECT * FROM native_hook_statistic')
result = cursor.fetchall()
for row in result:
if row[IPID_INDEX] == ipid and row[APPLY_INDEX] >= MALLOC_THRESH:
second_process = True
else:
cursor.execute('SELECT * FROM native_hook')
result = cursor.fetchall()
for row in result:
if row[IPID_INDEX] == ipid and row[ALLOC_INDEX] >= MALLOC_THRESH:
second_process = True
cursor.close()
conn.close()
return first_process and second_process
def get_profiler_test_trace(process):
subprocess.check_output("hdc shell ls -lh /data/log/reliability/resource_leak/memory_leak/ > /data/local/tmp/leak.txt")
subprocess.check_output(r"hdc file recv /data/local/tmp/leak.txt .\..\outputfiles\ ", text=True, encoding="utf-8")
with open(r'.\..\outputfiles\leak.txt', 'r') as file:
lines = file.readlines()
for line in lines:
if process in line and ("smaps" not in line) and ("sample" not in line):
return line.split()[len(line.split()) - 1]
return ""
def check_sa_result():
subprocess.check_output(r"hdc shell rm /data/local/tmp/test.htrace")
subprocess.check_output("hdc shell rm /data/log/reliability/resource_leak/memory_leak/*")
subprocess.check_output(r"del .\..\outputfiles\nativehook.db ", text=True, encoding="utf-8")
subprocess.check_output(r"del .\..\outputfiles\test.htrace", text=True, encoding="utf-8")
subprocess.check_output("hdc target mount")
subprocess.check_output(f"hdc file send .\..\inputfiles\process_resource_limit.json /system/variant/phone/base/etc/efficiency_manager", text=True, encoding="utf-8")
subprocess.check_output("hdc shell reboot", text=True, encoding="utf-8")
time.sleep(SLEEP_TWENTY)
j = 0
while j < SA_WAIT_TIMES:
output = subprocess.check_output(r"hdc list targets", text=True, encoding="utf-8")
if output == '[Empty]\n\n':
time.sleep(SLEEP_FIVE)
j += 1
else:
break
#解除锁屏
subprocess.check_output("hdc shell uitest uiInput drag 100 500 100 100 1000")
time.sleep(SLEEP_FIVE)
subprocess.check_output("hdc shell uitest uiInput drag 100 500 100 100 1000")
time.sleep(SLEEP_FIVE)
subprocess.check_output("hdc shell uitest uiInput drag 100 500 100 100 1000")
subprocess.check_output("hdc shell power-shell setmode 602")
subprocess.check_output("hdc shell killall com.example.insight_test_stage")
subprocess.check_output("hdc shell param set hiview.memleak.test enable")
subprocess.check_output("hdc shell killall hiview")
subprocess.check_output("hdc shell aa start -a EntryAbility -b com.example.insight_test_stage")
time.sleep(SLEEP_FOUR)
touch_button("模板测试")
time.sleep(1)
subprocess.check_output("hdc shell uitest uiInput drag 100 800 100 100 1000")
time.sleep(1)
touch_button("Allocations_Js_Depth")
i = 0
while i < SA_CLICK_TIMES:
touch_button("malloc-release(depth 6)")
touch_button("small-malloc(depth 7)")
i += 1
filename = get_profiler_test_trace("com.example.insight_test_stage")
subprocess.check_output("hdc shell cp /data/log/reliability/resource_leak/memory_leak/" + filename + " /data/local/tmp/test.htrace")
subprocess.check_output(r"hdc file recv /data/local/tmp/test.htrace .\..\outputfiles\ ", text=True, encoding="utf-8")
subprocess.check_output(r".\..\inputfiles\trace_streamer_nativehook.exe .\..\outputfiles\test.htrace -e .\..\outputfiles\nativehook.db", text=True, encoding="utf-8")
return check_result(SA_STATISTICS, False, True, SA_SAMPLE, False, 0, DEPTH_TWENTY, 0)
class TestNativehook:
@pytest.mark.L0
def test_nativehook_startup(self):
assert check_nativehook_result(STATISTICS_INTERVAL, True, True, SAMPLE_SMALL, False, 0, DEPTH_TEN, 1)
@pytest.mark.L0
def test_nativehook_nonstatistics_sample(self):
assert check_nativehook_result(0, False, True, SAMPLE_LARGE, False, 0, DEPTH_TEN, CLICK_TWICE)
@pytest.mark.L0
def test_nativehook_statistics_dwarf(self):
assert check_nativehook_result(STATISTICS_INTERVAL, False, True, SAMPLE_SMALL, True, 0, DEPTH_TEN, CLICK_THREETIMES)
@pytest.mark.L0
def test_nativehook_nonstatistics_match(self):
assert check_nativehook_result(0, False, False, SAMPLE_SMALL, False, 0, DEPTH_TEN, 1, MATCH_INTERVAL)
@pytest.mark.L0
def test_nativehook_startup_nonstatistics(self):
assert check_nativehook_result(0, True, True, SAMPLE_SMALL, False, 0, DEPTH_TEN, CLICK_TWICE)
@pytest.mark.L0
def test_nativehook_statistics_response_library(self):
assert check_nativehook_result(STATISTICS_INTERVAL, False, False, SAMPLE_SMALL, False, 0, DEPTH_TEN, CLICK_THREETIMES, 0, True)
@pytest.mark.L0
def test_dwarf_online(self):
assert check_nativehook_result(STATISTICS_INTERVAL, False, False, SAMPLE_SMALL, True, FILTER_SMALL, DEPTH_TEN, 1)
@pytest.mark.L0
def test_dwarf_startup(self):
assert check_nativehook_result(STATISTICS_INTERVAL, True, False, SAMPLE_SMALL, True, 0, DEPTH_TEN, CLICK_TWICE)
@pytest.mark.L0
def test_dwarf_sample_interval(self):
assert check_nativehook_result(STATISTICS_INTERVAL, False, False, SAMPLE_LARGE, True, 0, DEPTH_TEN, CLICK_THREETIMES)
@pytest.mark.L0
def test_dwarf_filtersize(self):
assert check_nativehook_result(STATISTICS_INTERVAL, False, False, SAMPLE_SMALL, True, FILTER_LARGE, DEPTH_TEN, CLICK_TWICE)
@pytest.mark.L0
def test_dwarf_nonstatistics(self):
assert check_nativehook_result(0, False, False, SAMPLE_SMALL, True, 0, DEPTH_TEN, 1)
@pytest.mark.L0
def test_dwarf_nonstatistics_match(self):
assert check_nativehook_result(0, False, False, SAMPLE_SMALL, True, 0, FILTER_SMALL, 1, DEPTH_TEN, False)
@pytest.mark.L0
def test_dwarf_response_library(self):
assert check_nativehook_result(STATISTICS_INTERVAL, False, False, SAMPLE_SMALL, True, 0, DEPTH_TEN, 1, 0, True)
@pytest.mark.L0
def test_nonstatistics_response_library(self):
assert check_nativehook_result(0, False, False, SAMPLE_SMALL, True, 0, DEPTH_TEN, 1, 0, True)
@pytest.mark.L0
def test_nonstatistics_match_response_library(self):
assert check_nativehook_result(0, False, False, SAMPLE_SMALL, True, 0, DEPTH_TEN, 1, MATCH_INTERVAL, True)
@pytest.mark.L0
def test_startup_filter(self):
assert check_nativehook_result(STATISTICS_INTERVAL, True, True, SAMPLE_SMALL, False, FILTER_LARGE, DEPTH_TEN, CLICK_TWICE)
@pytest.mark.L0
def test_startup_non_statistics(self):
assert check_nativehook_result(0, True, True, SAMPLE_SMALL, False, 0, DEPTH_TEN, CLICK_TWICE)
@pytest.mark.L0
def test_startup_online(self):
assert check_nativehook_result(0, True, False, SAMPLE_SMALL, False, 0, DEPTH_TEN, CLICK_TWICE)
@pytest.mark.L0
def test_startup_nonstatistics_match(self):
assert check_nativehook_result(0, True, False, SAMPLE_SMALL, False, 0, DEPTH_TEN, CLICK_TWICE, MATCH_INTERVAL, False)
@pytest.mark.L0
def test_startup_sample_interval(self):
assert check_nativehook_result(STATISTICS_INTERVAL, True, False, SAMPLE_LARGE, False, 0, DEPTH_TEN, CLICK_THREETIMES)
@pytest.mark.L0
def test_startup_response_library(self):
assert check_nativehook_result(STATISTICS_INTERVAL, True, False, SAMPLE_SMALL, False, 0, DEPTH_TEN, CLICK_TWICE, 0, True)
@pytest.mark.L0
def test_startup_response_library_sample_interval(self):
assert check_nativehook_result(STATISTICS_INTERVAL, True, False, SAMPLE_LARGE, False, 0, DEPTH_TEN, CLICK_THREETIMES, 0, True)
@pytest.mark.L0
def test_startup_response_library_filter(self):
assert check_nativehook_result(STATISTICS_INTERVAL, True, False, SAMPLE_SMALL, False, FILTER_LARGE, DEPTH_TEN, CLICK_TWICE, 0, True)
@pytest.mark.L0
def test_startup_sample_interval_filter(self):
assert check_nativehook_result(STATISTICS_INTERVAL, True, False, SAMPLE_LARGE, False, FILTER_LARGE, DEPTH_TEN, CLICK_TWICE)
@pytest.mark.L0
def test_startup_sample_interval_nonstatistics(self):
assert check_nativehook_result(0, True, False, SAMPLE_LARGE, False, 0, DEPTH_TEN, CLICK_TWICE)
@pytest.mark.L0
def test_startup_sample_interval_nonstatistics_match(self):
assert check_nativehook_result(0, True, False, SAMPLE_LARGE, False, 0, DEPTH_TEN, CLICK_TWICE, MATCH_INTERVAL, False)
@pytest.mark.L0
def test_startup_filter_nonstatistics_match(self):
assert check_nativehook_result(0, True, False, SAMPLE_SMALL, False, FILTER_LARGE, DEPTH_TEN, CLICK_THREETIMES, MATCH_INTERVAL, False)
@pytest.mark.L0
def test_startup_filter_nonstatistics_match(self):
assert check_nativehook_result(0, True, False, SAMPLE_SMALL, False, FILTER_LARGE, DEPTH_TEN, CLICK_TWICE, MATCH_INTERVAL, False)
@pytest.mark.L0
def test_online_filtersize(self):
assert check_nativehook_result(STATISTICS_INTERVAL, False, False, SAMPLE_SMALL, True, FILTER_LARGE, DEPTH_TEN, 1)
@pytest.mark.L0
def test_online_sample_interval(self):
assert check_nativehook_result(STATISTICS_INTERVAL, False, False, SAMPLE_LARGE, False, FILTER_SMALL, DEPTH_TEN, 1)
@pytest.mark.L0
def test_online_sample_interval_filter(self):
assert check_nativehook_result(STATISTICS_INTERVAL, False, False, SAMPLE_LARGE, False, FILTER_LARGE, DEPTH_TEN, 1)
@pytest.mark.L0
def test_online_response_library_filter(self):
assert check_nativehook_result(STATISTICS_INTERVAL, False, False, SAMPLE_SMALL, False, FILTER_LARGE, DEPTH_TEN, 1, 0, True)
@pytest.mark.L0
def test_online_nonstatistics(self):
assert check_nativehook_result(STATISTICS_INTERVAL, False, False, SAMPLE_SMALL, False, FILTER_SMALL, DEPTH_TEN, CLICK_THREETIMES)
@pytest.mark.L0
def test_online_nonstatistics_match(self):
assert check_nativehook_result(0, False, False, SAMPLE_SMALL, False, FILTER_SMALL, MATCH_INTERVAL, 1, DEPTH_TEN, False)
@pytest.mark.L0
def test_online_nonstatistics_response_library(self):
assert check_nativehook_result(STATISTICS_INTERVAL, False, False, SAMPLE_SMALL, False, FILTER_SMALL, DEPTH_TEN, 1, 0, True)
@pytest.mark.L0
def test_online_nonstatistics_match_filter(self):
assert check_nativehook_result(STATISTICS_INTERVAL, False, False, SAMPLE_SMALL, False, FILTER_LARGE, DEPTH_TEN, 1, MATCH_INTERVAL, False)
@pytest.mark.L0
def test_online_nonstatistics_filter(self):
assert check_nativehook_result(0, False, False, SAMPLE_SMALL, False, FILTER_LARGE, DEPTH_TEN, 1)
@pytest.mark.L0
def test_no_dataqueue(self):
assert check_nativehook_result(STATISTICS_INTERVAL, False, True, SAMPLE_SMALL, False, 0, DEPTH_TEN, CLICK_THREETIMES)
@pytest.mark.L0
def test_no_dataqueue_startup(self):
assert check_nativehook_result(STATISTICS_INTERVAL, True, True, SAMPLE_SMALL, False, 0, DEPTH_TEN, CLICK_TWICE)
@pytest.mark.L0
def test_no_dataqueue_online(self):
assert check_nativehook_result(STATISTICS_INTERVAL, False, False, SAMPLE_SMALL, False, 0, DEPTH_TEN, CLICK_TWICE)
@pytest.mark.L0
def test_no_dataqueue_dwarf(self):
assert check_nativehook_result(STATISTICS_INTERVAL, False, False, SAMPLE_SMALL, True, 0, DEPTH_TEN, CLICK_TWICE)
@pytest.mark.L0
def test_depth_five_dwarf(self):
assert check_nativehook_result(STATISTICS_INTERVAL, False, False, SAMPLE_SMALL, True, 0, DEPTH_FIVE, 1)
@pytest.mark.L0
def test_depth_five(self):
assert check_nativehook_result(STATISTICS_INTERVAL, False, True, SAMPLE_SMALL, False, 0, DEPTH_FIVE, CLICK_THREETIMES)
@pytest.mark.L0
def test_depth_five_startup(self):
assert check_nativehook_result(STATISTICS_INTERVAL, True, True, SAMPLE_SMALL, False, 0, DEPTH_FIVE, CLICK_THREETIMES)
@pytest.mark.L0
def test_depth_five_online(self):
assert check_nativehook_result(STATISTICS_INTERVAL, False, False, SAMPLE_SMALL, False, 0, DEPTH_FIVE, CLICK_TWICE)
@pytest.mark.L0
def test_depth_five_filtersize(self):
assert check_nativehook_result(STATISTICS_INTERVAL, False, False, SAMPLE_SMALL, False, FILTER_LARGE, DEPTH_FIVE, CLICK_THREETIMES)
@pytest.mark.L0
def test_depth_five_nonstatistics(self):
assert check_nativehook_result(0, False, False, SAMPLE_SMALL, False, 0, DEPTH_FIVE, CLICK_THREETIMES)
@pytest.mark.L0
def test_depth_fifteen(self):
assert check_nativehook_result(STATISTICS_INTERVAL, False, False, SAMPLE_SMALL, True, 0, DEPTH_FIFTEEN, CLICK_THREETIMES)
@pytest.mark.L0
def test_depth_twenty(self):
assert check_nativehook_result(STATISTICS_INTERVAL, False, False, SAMPLE_SMALL, True, 0, DEPTH_TWENTY, CLICK_THREETIMES)
@pytest.mark.L0
def test_depth_thirty(self):
assert check_nativehook_result(STATISTICS_INTERVAL, False, False, SAMPLE_SMALL, True, 0, DEPTH_THIRTY, CLICK_TWICE)
@pytest.mark.L0
def test_depth_fifty(self):
assert check_nativehook_result(STATISTICS_INTERVAL, False, False, SAMPLE_SMALL, True, 0, DEPTH_FIFTY, CLICK_THREETIMES)
@pytest.mark.L0
def test_nativehook_sa(self):
assert check_sa_result()
@pytest.mark.L0
def test_usermode_nondebug_app_startup(self):
assert check_nativehook_multipleprocess(STATISTICS_INTERVAL, False, True, 0, False, 0, DEPTH_TEN, 1)
@pytest.mark.L0
def test_multipleprocess_online(self):
assert check_nativehook_multipleprocess(STATISTICS_INTERVAL, False, False, 0, False, 0, DEPTH_TEN, 1)
@pytest.mark.L0
def test_multipleprocess_statistics(self):
assert check_nativehook_multipleprocess(STATISTICS_INTERVAL, False, False, 0, False, 0, DEPTH_TEN, 1)
@pytest.mark.L0
def test_multipleprocess_offline(self):
assert check_nativehook_multipleprocess(STATISTICS_INTERVAL, False, True, SAMPLE_SMALL, False, 0, DEPTH_TWENTY, CLICK_TWICE)
@pytest.mark.L0
def test_multipleprocess_dwarf(self):
assert check_nativehook_multipleprocess(STATISTICS_INTERVAL, False, False, SAMPLE_SMALL, True, 0, DEPTH_THIRTY, CLICK_THREETIMES)
@pytest.mark.L0
def test_multipleprocess_offline_sample(self):
assert check_nativehook_multipleprocess(STATISTICS_INTERVAL, False, True, SAMPLE_LARGE, False, 0, DEPTH_THIRTY, CLICK_TWICE)
@pytest.mark.L0
def test_multipleprocess_dwarf_nonstatistics(self):
assert check_nativehook_multipleprocess(0, False, True, SAMPLE_SMALL, False, 0, DEPTH_THIRTY, CLICK_THREETIMES)
@pytest.mark.L0
def test_multipleprocess_dwarf_response_library(self):
assert check_nativehook_multipleprocess(STATISTICS_INTERVAL, False, True, SAMPLE_SMALL, True, 0, DEPTH_TEN, CLICK_TWICE, 0, True)
@pytest.mark.L0
def test_multipleprocess_nonstatics_response_library(self):
assert check_nativehook_multipleprocess(0, False, True, SAMPLE_SMALL, False, 0, DEPTH_TEN, CLICK_THREETIMES, 0, True)
@pytest.mark.L0
def test_multipleprocess_nodataqueue_dwarf(self):
assert check_nativehook_multipleprocess(STATISTICS_INTERVAL, False, True, SAMPLE_SMALL, True, 0, DEPTH_TEN, CLICK_TWICE)
@pytest.mark.L0
def test_multipleprocess_depth_five_dwarf(self):
assert check_nativehook_multipleprocess(STATISTICS_INTERVAL, False, True, SAMPLE_SMALL, True, 0, DEPTH_FIVE, CLICK_TWICE)
@pytest.mark.L0
def test_multipleprocess_depth_five(self):
assert check_nativehook_multipleprocess(STATISTICS_INTERVAL, False, True, SAMPLE_SMALL, False, 0, DEPTH_FIVE, CLICK_THREETIMES)
@pytest.mark.L0
def test_multipleprocess_depth_five_filter(self):
assert check_nativehook_multipleprocess(STATISTICS_INTERVAL, False, True, SAMPLE_SMALL, False, FILTER_LARGE, DEPTH_FIVE, 1)
@pytest.mark.L0
def test_multipleprocess_depth_five_online(self):
assert check_nativehook_multipleprocess(STATISTICS_INTERVAL, False, False, SAMPLE_SMALL, False, 0, DEPTH_FIVE, 1)
@pytest.mark.L0
def test_multipleprocess_depth_five_filtersize(self):
assert check_nativehook_multipleprocess(STATISTICS_INTERVAL, False, True, SAMPLE_SMALL, False, FILTER_LARGE, DEPTH_FIVE, 1)
@pytest.mark.L0
def test_multipleprocess_depth_five_nonstatistics(self):
assert check_nativehook_multipleprocess(0, False, True, SAMPLE_SMALL, False, 0, DEPTH_FIVE, 1)
@pytest.mark.L0
def test_appfreeze(self):
subprocess.check_output(f"hdc shell ls -lh /data/log/faultlog/faultlogger/ > /data/local/tmp/faultlog.txt")
subprocess.check_output(r"hdc file recv /data/local/tmp/faultlog.txt .\..\outputfiles\ ", text=True, encoding="utf-8")
check = True
with open(r'.\..\outputfiles\faultlog.txt', 'r') as file:
lines = file.readlines()
for line in lines:
if "com.ohos.sceneboard" in line and ("syswarning" not in line):
check = False
if "com.example.insight_test_stage" in line:
check = False
assert check == True
@pytest.mark.L0
def test_nocrash(self):
check = True
with open(r'.\..\outputfiles\faultlog.txt', 'r') as file:
lines = file.readlines()
for line in lines:
if "hiprofilerd" in line:
check = False
if "hiprofiler_plugins" in line:
check = False
if "native_daemon" in line:
check = False
assert check == True

View File

@ -0,0 +1,97 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# Copyright (C) 2024 Huawei Device Co., Ltd.
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
import pytest
import subprocess
import re
import sys
sys.path.append("..")
from tools.utils import *
import time
import threading
SLEEP_TWO = 2
SLEEP_THREE = 3
SLEEP_FIVE = 5
MULTIPLE_RESULT = 3
ONCE_RESULT = 2
CONFIG_INDEX = 10
THRESH = 25000000000
def task(index):
indexstr = str(index)
subprocess.check_output("hdc shell hiprofiler_cmd -c /data/local/tmp/config10.txt -o /data/local/tmp/test" + indexstr + ".htrace -t 30 -s -k")
class TestHiprofilerNetworkProfiler:
@pytest.mark.L0
def test_network_profiler_multiple_times(self):
subprocess.check_output(f"hdc file send .\..\inputfiles\network_profiler\config10.txt /data/local/tmp/", text=True, encoding="utf-8")
subprocess.check_output("hdc shell killall com.example.myapplication523")
task_thread = threading.Thread(target=task, args=(CONFIG_INDEX, ))
time.sleep(SLEEP_TWO)
task_thread.start()
time.sleep(SLEEP_THREE)
subprocess.check_output("hdc shell aa start -a EntryAbility -b com.example.myapplication523")
time.sleep(SLEEP_FIVE)
touch_button("http_request")
time.sleep(SLEEP_TWO)
touch_button("http_request")
time.sleep(SLEEP_TWO)
touch_button("http_request")
task_thread.join()
subprocess.check_output("hdc shell chmod 777 /data/local/tmp/hookDecoder")
subprocess.check_output("hdc shell ./data/local/tmp/hookDecoder -f /data/local/tmp/test10.htrace > /data/local/tmp/test10_result.txt")
subprocess.check_output(f"hdc file recv /data/local/tmp/test10.htrace .\..\outputfiles\ ", text=True, encoding="utf-8")
subprocess.check_output(f"hdc file recv /data/local/tmp/test10_result.txt .\..\outputfiles\ ", text=True, encoding="utf-8")
count = 0
with open(r'.\..\outputfiles\test10_result.txt', 'r') as file:
lines = file.readlines()
for line in lines:
if "tv_nsec" in line:
count += 1
assert count == MULTIPLE_RESULT
# 第二次请求和第三次请求一起被写入trace文件
@pytest.mark.L0
def test_network_profiler_one_time(self):
subprocess.check_output("hdc shell rm /data/local/tmp/test10.htrace")
subprocess.check_output("hdc shell rm /data/local/tmp/test10_result.txt")
subprocess.check_output("hdc shell killall com.example.myapplication523")
subprocess.check_output(f"hdc file send .\..\inputfiles\network_profiler\config10.txt /data/local/tmp/", text=True, encoding="utf-8")
subprocess.check_output("hdc shell killall com.example.myapplication523")
task_thread = threading.Thread(target=task, args=(CONFIG_INDEX, ))
time.sleep(SLEEP_TWO)
task_thread.start()
time.sleep(SLEEP_THREE)
subprocess.check_output("hdc shell aa start -a EntryAbility -b com.example.myapplication523")
time.sleep(SLEEP_FIVE)
touch_button("http_request")
time.sleep(SLEEP_TWO)
task_thread.join()
subprocess.check_output("hdc shell chmod 777 /data/local/tmp/hookDecoder")
subprocess.check_output("hdc shell ./data/local/tmp/hookDecoder -f /data/local/tmp/test10.htrace > /data/local/tmp/test10_result.txt")
subprocess.check_output(f"hdc file recv /data/local/tmp/test10.htrace .\..\outputfiles\ ", text=True, encoding="utf-8")
subprocess.check_output(f"hdc file recv /data/local/tmp/test10_result.txt .\..\outputfiles\ ", text=True, encoding="utf-8")
count = 0
with open(r'.\..\outputfiles\test10_result.txt', 'r') as file:
lines = file.readlines()
for line in lines:
if "tv_nsec" in line:
count += 1
assert count == ONCE_RESULT
# 包括文件头

View File

@ -0,0 +1,89 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# Copyright (C) 2024 Huawei Device Co., Ltd.
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
import pytest
import subprocess
import re
import sys
sys.path.append("..")
from tools.utils import *
import threading
LIB_PATH = "/system/lib"
THRESH = 25000000000
SLEEP_TWO = 2
SLEEP_THIRTY = 30
CONFIG_INDEX = 6
PORT_INDEX = 5
MALLOCTIME_INDEX = 3
LISTENURI_INDEX = 3
def task(index):
indexstr = str(index)
subprocess.check_output("hdc shell hiprofiler_cmd -c /data/local/tmp/config" + indexstr + ".txt -o /data/local/tmp/test" + indexstr + ".htrace -t 20 -s -k")
def malloctest():
subprocess.check_output("hdc shell chmod 777 /data/local/tmp/malloctest")
subprocess.check_output("hdc shell ./data/local/tmp/malloctest 10 1024 1000000 > /data/local/tmp/malloctest.txt")
class TestHiprofilerMalloctime:
@pytest.mark.L0
def test_malloctime(self):
subprocess.check_output(f"hdc file send .\inputfiles\nativehook\config6.txt /data/local/tmp/", text=True, encoding="utf-8")
malloc_thread = threading.Thread(target=malloctest)
task_thread = threading.Thread(target=task, args=(CONFIG_INDEX, ))
malloc_thread.start()
time.sleep(SLEEP_TWO)
task_thread.start()
time.sleep(SLEEP_THIRTY)
malloc_thread.join()
task_thread.join()
subprocess.check_output(f"hdc file recv /data/local/tmp/malloctest.txt .\..\outputfiles\ ", text=True, encoding="utf-8")
malloctime = 0
with open(r'.\..\outputfiles\malloctest.txt', 'r') as file:
lines = file.readlines()
malloctime = int(lines[MALLOCTIME_INDEX].split()[MALLOCTIME_INDEX])
assert(malloctime < THRESH)
@pytest.mark.L0
def test_listenuri(self):
port = 0
subprocess.check_output("hdc shell hiprofiler_cmd -q > /data/local/tmp/cmdtmp.txt")
subprocess.check_output(f"hdc file recv /data/local/tmp/cmdtmp.txt .\..\outputfiles\ ", text=True, encoding="utf-8")
check = False
with open(r'.\..\outputfiles\cmdtmp.txt', 'r') as file:
lines = file.readlines()
for line in lines:
if "port" in line:
port = int(line[PORT_INDEX:])
subprocess.check_output("hdc shell ls -lh netstat -anp | grep " + str(port) + " > /data/local/tmp/uri.txt")
subprocess.check_output(f"hdc file recv /data/local/tmp/uri.txt .\..\outputfiles\ ", text=True, encoding="utf-8")
check = False
with open(r'.\..\outputfiles\uri.txt', 'r') as file:
lines = file.readlines()
for line in lines:
if "LISTEN" in line:
target = line.split()[LISTENURI_INDEX]
if "127.0.0.1" in target:
check = True
assert check

View File

@ -0,0 +1,167 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# Copyright (C) 2024 Huawei Device Co., Ltd.
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
import pytest
import subprocess
import re
import time
import sys
sys.path.append("..")
from tools.utils import *
import threading
import sqlite3
LIB_PATH = "/system/lib"
NETWORK_PROFILER_RESULT = 2
SLEEP_TWO = 2
SLEEP_THREE = 3
SLEEP_FOUR = 4
CONFIG_NETWORK_PROFILER = 10
CONFIG_NETWORK_PROFILER_NONDEBUG = 12
CONFIG_SCENEBOARD = 7
SIZE_INDEX = 4
def task(index):
indexstr = str(index)
subprocess.check_output("hdc shell hiprofiler_cmd -c /data/local/tmp/config" + indexstr + ".txt -o /data/local/tmp/test" + indexstr + ".htrace -t 20 -s -k")
def malloctest():
subprocess.check_output("hdc shell chmod 777 /data/local/tmp/malloctest")
subprocess.check_output("hdc shell ./data/local/tmp/malloctest 10 1024 1000000 > /data/local/tmp/malloctest.txt")
class TestHiprofilerUserMode:
@pytest.mark.L0
def test_usermode_nativehook_debug_app(self):
subprocess.check_output(r"hdc file send .\..\inputfiles\nativehook\config1.txt /data/local/tmp/", text=True, encoding="utf-8")
subprocess.check_output("hdc shell power-shell setmode 602")
subprocess.check_output("hdc shell killall com.example.insight_test_stage")
subprocess.check_output("hdc shell rm /data/local/tmp/test1.htrace")
task_thread = threading.Thread(target=task, args=(1, ))
task_thread.start()
time.sleep(SLEEP_TWO)
subprocess.check_output("hdc shell aa start -a EntryAbility -b com.example.insight_test_stage")
time.sleep(SLEEP_FOUR)
touch_button("模板测试")
time.sleep(SLEEP_TWO)
subprocess.check_output("hdc shell uitest uiInput drag 100 800 100 100 1000")
time.sleep(1)
touch_button("Allocations_Js_Depth")
touch_button("malloc-release(depth 6)")
touch_button("small-malloc(depth 7)")
task_thread.join()
subprocess.check_output(f"hdc file recv /data/local/tmp/test1.htrace .\..\outputfiles\ ", text=True, encoding="utf-8")
subprocess.check_output("hdc shell ls -lh /data/local/tmp/ > /data/local/tmp/tmp.txt")
subprocess.check_output(f"hdc file recv /data/local/tmp/tmp.txt .\..\outputfiles\ ", text=True, encoding="utf-8")
result = False
with open(r'.\..\outputfiles\tmp.txt', 'r') as file:
lines = file.readlines()
for line in lines:
if "test1.htrace" in line:
result = (line.split()[SIZE_INDEX][-1] == 'M')
assert result
@pytest.mark.L0
def test_usermode_nondebug_app(self):
# 校验命令行输出
subprocess.check_output("hdc shell rm /data/local/tmp/test7.htrace")
subprocess.check_output("hdc shell rm /data/local/tmp/tmp.txt")
subprocess.check_output(r"hdc file send .\inputfiles\network_profiler\config7.txt /data/local/tmp/", text=True, encoding="utf-8")
task_thread = threading.Thread(target=task, args=(CONFIG_SCENEBOARD, ))
task_thread.start()
time.sleep(SLEEP_TWO)
subprocess.check_output("hdc shell uitest uinput drag 100 800 100 100 1000")
time.sleep(1)
subprocess.check_output("hdc shell uitest uinput drag 100 100 800 100 1000")
task_thread.join()
subprocess.check_output(f"hdc file recv /data/local/tmp/test7.htrace .\..\outputfiles\ ", text=True, encoding="utf-8")
subprocess.check_output("hdc shell ls -lh /data/local/tmp/ > /data/local/tmp/tmp.txt")
subprocess.check_output(f"hdc file recv /data/local/tmp/tmp.txt .\..\outputfiles\ ", text=True, encoding="utf-8")
result = False
with open(r'.\..\outputfiles\tmp.txt', 'r') as file:
lines = file.readlines()
for line in lines:
if "test7.htrace" in line:
result = (line.split()[SIZE_INDEX][:-1] == "1.0")
assert result
@pytest.mark.L0
def test_usermode_network_profiler_debugapp(self):
subprocess.check_output(r"hdc file send .\..\inputfiles\network_profiler\config10.txt /data/local/tmp/", text=True, encoding="utf-8")
subprocess.check_output("hdc shell killall com.example.myapplication523")
task_thread = threading.Thread(target=task, args=(CONFIG_NETWORK_PROFILER, ))
time.sleep(SLEEP_TWO)
task_thread.start()
time.sleep(SLEEP_TWO)
subprocess.check_output("hdc shell aa start -a EntryAbility -b com.example.myapplication523")
time.sleep(SLEEP_THREE)
touch_button("http_request")
task_thread.join()
subprocess.check_output("hdc shell chmod 777 /data/local/tmp/hookDecoder")
subprocess.check_output("hdc shell ./data/local/tmp/hookDecoder -f /data/local/tmp/test10.htrace > /data/local/tmp/test10_result.txt")
subprocess.check_output(f"hdc file recv /data/local/tmp/test10.htrace .\..\outputfiles\ ", text=True, encoding="utf-8")
subprocess.check_output(f"hdc file recv /data/local/tmp/test10_result.txt .\..\outputfiles\ ", text=True, encoding="utf-8")
count = 0
with open(r'.\..\outputfiles\test10_result.txt', 'r') as file:
lines = file.readlines()
for line in lines:
if "tv_nsec" in line:
count += 1
assert count == NETWORK_PROFILER_RESULT
#包括文件头
@pytest.mark.L0
def test_usermode_network_profiler_nondebugapp(self):
subprocess.check_output("hdc shell rm /data/local/tmp/test12.htrace")
subprocess.check_output(r"hdc file send .\..\inputfiles\network_profiler\config12.txt /data/local/tmp/", text=True, encoding="utf-8")
task_thread = threading.Thread(target=task, args=(CONFIG_NETWORK_PROFILER_NONDEBUG, ))
time.sleep(SLEEP_TWO)
task_thread.start()
time.sleep(SLEEP_THREE)
subprocess.check_output("hdc shell aa start -a EntryAbility -b com.tencent.mtthm")
time.sleep(SLEEP_FIVE)
subprocess.check_output("hdc shell uinput -T -c 850 1550")
time.sleep(SLEEP_TWO)
touch_button("微信")
task_thread.join()
subprocess.check_output("hdc shell chmod 777 /data/local/tmp/hookDecoder")
subprocess.check_output("hdc shell ./data/local/tmp/hookDecoder -f /data/local/tmp/test12.htrace > /data/local/tmp/test12_result.txt")
subprocess.check_output(f"hdc file recv /data/local/tmp/test12.htrace .\..\outputfiles\ ", text=True, encoding="utf-8")
subprocess.check_output(f"hdc file recv /data/local/tmp/test12_result.txt .\..\outputfiles\ ", text=True, encoding="utf-8")
count = 0
with open(r'.\..\outputfiles\test12_result.txt', 'r') as file:
lines = file.readlines()
for line in lines:
if "tv_nsec" in line:
count += 1
assert count == 0
@pytest.mark.L0
def test_usermode_kernel_symbols(self):
subprocess.check_output(r"hdc file recv /data/local/tmp/test1.htrace .\..\outputfiles\ ", text=True, encoding="utf-8")
subprocess.check_output(r".\..\inputfiles\trace_streamer_nativehook.exe .\..\outputfiles\test1.htrace -e .\..\outputfiles\nativehook.db", text=True, encoding="utf-8")
conn = sqlite3.connect(r'./../outputfiles/nativehook.db')
cursor = conn.cursor()
cursor.execute('SELECT * FROM data_dict')
result = cursor.fetchall()
check = True
for row in result:
if 'kallsyms' in row[1]:
check = False
assert check

View File

@ -0,0 +1,86 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# Copyright (C) 2024 Huawei Device Co., Ltd.
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
import os
import zipfile
import subprocess
import re
import time
import json
OUTPUT_PATH = "testModule/output"
PID_INDEX = 7
def get_path_by_attribute(tree, key, value):
attributes = tree['attributes']
if attributes is None:
print("tree contains no attributes")
return None
path = []
if attributes.get(key) == value:
return path
for index, child in enumerate(tree['children']):
child_path = path + [index]
result = get_path_by_attribute(child, key, value)
if result is not None:
return child_path + result
return None
def get_element_by_path(tree, path):
if len(path) == 1:
return tree['children'][path[0]]
return get_element_by_path(tree['children'][path[0]], path[1:])
def get_location_by_text(tree, text):
path = get_path_by_attribute(tree, "text", text)
if path is None or len(path) == 0:
print("text not found in layout file")
element = get_element_by_path(tree, path)
locations = element['attributes']['bounds'].replace('[', '').replace(']', ' ').replace(',', ' ').strip().split()
return int((int(locations[0]) + int(locations[2])) / 2), int((int(locations[1]) + int(locations[3])) / 2)
def touch(dx, dy):
output = subprocess.check_output(f"hdc shell uitest uiInput click {dx} {dy}")
def get_layout_tree():
output = subprocess.check_output("hdc shell uitest dumpLayout", text=True)
path = output.strip().split(":")[-1]
subprocess.check_output(f"hdc file recv {path} .\..\inputfiles\layout.json")
subprocess.check_output("hdc shell rm " + path)
with open(".\..\inputfiles\layout.json", encoding="utf-8") as f:
tree = json.load(f)
return tree
def touch_button(text):
layout_tree = get_layout_tree()
location = get_location_by_text(layout_tree, text)
touch(location[0], location[1])
def get_pid(name):
subprocess.check_output("hdc shell ps -ef | grep " + name + " > /data/local/tmp/pids.txt")
subprocess.check_output(f"hdc file recv /data/local/tmp/pids.txt .\..\outputfiles\ ", text=True, encoding="utf-8")
with open(r'.\..\outputfiles\pids.txt', 'r') as file:
lines = file.readlines()
for line in lines:
if line.split()[PID_INDEX] == name:
return line.split()[1]
return 0