• Home
  • Line#
  • Scopes#
  • Navigate#
  • Raw
  • Download
1#!/usr/bin/env python3
2# -*- coding: utf-8 -*-
3"""
4Copyright (c) 2024 Huawei Device Co., Ltd.
5Licensed under the Apache License, Version 2.0 (the "License");
6you may not use this file except in compliance with the License.
7You may obtain a copy of the License at
8
9    http://www.apache.org/licenses/LICENSE-2.0
10
11Unless required by applicable law or agreed to in writing, software
12distributed under the License is distributed on an "AS IS" BASIS,
13WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14See the License for the specific language governing permissions and
15limitations under the License.
16
17Description: Python Profiler Domain Interfaces
18"""
19
20import re
21import sys
22import json
23from pathlib import Path
24
25sys.path.append(str(Path(__file__).parent.parent)) # add aw path to sys.path
26
27from all_utils import CommonUtils
28from implement_api.protocol_api import ProtocolImpl
29from cdp import profiler
30from customized_types import ProtocolType
31
32comm_with_debugger_server = CommonUtils.communicate_with_debugger_server
33
34
35class ProfilerImpl(ProtocolImpl):
36
37    def __init__(self, id_generator, websocket):
38        super().__init__(id_generator, websocket)
39        self.dispatch_table = {"start": (self.start, ProtocolType.send),
40                               "stop": (self.stop, ProtocolType.send),
41                               "setSamplingInterval": (self.set_sampling_interval, ProtocolType.send)}
42
43    async def start(self, message_id, connection, params):
44        response = await comm_with_debugger_server(self.websocket, connection,
45                                                   profiler.start(), message_id)
46        CommonUtils.assert_equal(json.loads(response), {"id": message_id, "result": {}})
47
48    async def stop(self, message_id, connection, params):
49        response = await comm_with_debugger_server(self.websocket, connection,
50                                                   profiler.stop(), message_id)
51        response = json.loads(response)
52        CommonUtils.assert_equal(response['id'], message_id)
53        time_deltas = response['result']['profile']['timeDeltas']
54        assert all(i >= 0 for i in time_deltas), \
55            f"TimeDeltas are not correct: {time_deltas}"
56        nodes = response['result']['profile']['nodes']
57        # NAPI方法名需要遵循格式: 方法名(地址)(NAPI)
58        pattern = r'^[\w.]+\([A-Za-z0-9]+\)\(NAPI\)$'
59        for node in nodes:
60            func_name = node['callFrame']['functionName']
61            if func_name.endswith('(NAPI)'):
62                assert re.match(pattern, func_name), \
63                    f"The function name '{func_name}' is not match the pattern '{pattern}'"
64        return response
65
66    async def set_sampling_interval(self, message_id, connection, params):
67        response = await comm_with_debugger_server(self.websocket, connection,
68                                                   profiler.set_sampling_interval(params), message_id)
69        CommonUtils.assert_equal(json.loads(response), {"id": message_id, "result": {}})
70