#!/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. Description: Scenario test case. """ import logging import os import time import pytest from aw import Application from aw import Utils from aw import debugger, runtime from aw.api import debugger_api, runtime_api @pytest.mark.debug @pytest.mark.timeout(60) class TestDebug23: """ 测试用例：单实例smartStepinto调试 测试步骤： 1. 连接 connect server 和主线程 debugger server 2. 主线程使能 Runtime 和 Debugger 3. 主线程运行等待的调试器 4. 主线程解析 entry_ability 和 index 文件 5. 主线程 Index.ts 文件设置断点并运行 6. 主线程 resume 验证是否正确命中断点 7. 主线程 smartStepInto 验证传回的函数信息 8. 关闭主线程 debugger server 和 connect server 连接 测试代码： function test1() { return taskNumber; } function test2() { return taskNumber + 5; } function main() { test1(); taskNumber = test1() + test2();// 实例debug23断点处 } main(); """ def setup_method(self): logging.info('Start running TestDebug23: setup') self.log_path = rf'{os.path.dirname(__file__)}\..\log' self.hilog_file_name = 'test_debug_23.hilog.txt' self.id_generator = Utils.message_id_generator() # receive the hilog before the test start Utils.clear_fault_log() self.hilog_process, self.write_thread = Utils.save_hilog(log_path=self.log_path, file_name=self.hilog_file_name, debug_on=True) def teardown_method(self): Application.uninstall(self.config['bundle_name']) # terminate the hilog receive process after the test done time.sleep(3) self.hilog_process.stdout.close() self.hilog_process.terminate() self.hilog_process.wait() self.write_thread.join() Utils.save_fault_log(log_path=self.log_path) logging.info('TestDebug23 done') def test(self, test_suite_main_instance_03_debug): logging.info('Start running TestDebug23: test') self.config = test_suite_main_instance_03_debug websocket = self.config['websocket'] taskpool = self.config['taskpool'] pid = self.config['pid'] self.debugger_impl = debugger_api.DebuggerImpl(self.id_generator, websocket) self.runtime_impl = runtime_api.RuntimeImpl(self.id_generator, websocket) taskpool.submit(websocket.main_task(taskpool, self.procedure, pid)) taskpool.await_taskpool() taskpool.task_join() if taskpool.task_exception: raise taskpool.task_exception async def procedure(self, websocket): ################################################################################################################ # main thread: connect the debugger server ################################################################################################################ main_thread = await self.debugger_impl.connect_to_debugger_server(self.config['pid'], True) logging.info(f'Connect to the debugger server of instance: {main_thread.instance_id}') ################################################################################################################ # main thread: Runtime.enable ################################################################################################################ await self.runtime_impl.send("Runtime.enable", main_thread) ################################################################################################################ # main thread: Debugger.enable ################################################################################################################ await self.debugger_impl.send("Debugger.enable", main_thread) ################################################################################################################ # main thread: Runtime.runIfWaitingForDebugger ################################################################################################################ await self.runtime_impl.send("Runtime.runIfWaitingForDebugger", main_thread) ################################################################################################################ # main thread: Debugger.scriptParsed ################################################################################################################ response = await self.debugger_impl.recv("Debugger.scriptParsed", main_thread) assert response['params']['url'] == self.config['file_path']['entry_ability'] assert response['params']['endLine'] == 0 ################################################################################################################ # main thread: Debugger.paused ################################################################################################################ response = await self.debugger_impl.recv("Debugger.paused", main_thread) assert response['params']['callFrames'][0]['url'] == self.config['file_path']['entry_ability'] assert response['params']['reason'] == 'Break on start' ################################################################################################################ # main thread: Debugger.resume ################################################################################################################ await self.debugger_impl.send("Debugger.resume", main_thread) ################################################################################################################ # main thread: Debugger.scriptParsed ################################################################################################################ response = await self.debugger_impl.recv("Debugger.scriptParsed", main_thread) assert response['params']['url'] == self.config['file_path']['index'] assert response['params']['endLine'] == 0 ################################################################################################################ # main thread: Debugger.paused ################################################################################################################ response = await self.debugger_impl.recv("Debugger.paused", main_thread) assert response['params']['callFrames'][0]['url'] == self.config['file_path']['index'] assert response['params']['reason'] == 'Break on start' ################################################################################################################ # main thread: Debugger.removeBreakpointsByUrl ################################################################################################################ params = debugger.RemoveBreakpointsUrl(self.config['file_path']['index']) await self.debugger_impl.send("Debugger.removeBreakpointsByUrl", main_thread, params) ################################################################################################################ # main thread: Debugger.getPossibleAndSetBreakpointByUrl ################################################################################################################ locations = [debugger.BreakLocationUrl(url=self.config['file_path']['index'], line_number=73)] params = debugger.SetBreakpointsLocations(locations) response = await self.debugger_impl.send("Debugger.getPossibleAndSetBreakpointsByUrl", main_thread, params) assert response['result']['locations'][0]['id'] == 'id:73:0:' + self.config['file_path']['index'] ################################################################################################################ # main thread: Debugger.resume ################################################################################################################ await self.debugger_impl.send("Debugger.resume", main_thread) ################################################################################################################ # main thread: Debugger.paused, hit breakpoint ################################################################################################################ response = await self.debugger_impl.recv("Debugger.paused", main_thread) assert response['params']['callFrames'][0]['url'] == self.config['file_path']['index'] assert response['params']['hitBreakpoints'] == ['id:73:17:' + self.config['file_path']['index']] ################################################################################################################ # main thread: Debugger.smartStepInto ################################################################################################################ params = debugger.SmartStepIntoParams(url=self.config['file_path']['index'], line_number=69) await self.debugger_impl.send("Debugger.smartStepInto", main_thread, params) ################################################################################################################ # main thread: Debugger.resume ################################################################################################################ await self.debugger_impl.send("Debugger.resume", main_thread) # main thread: Debugger.paused response = await self.debugger_impl.recv("Debugger.paused", main_thread) assert response['params']['callFrames'][0]['url'] == self.config['file_path']['index'] assert response['params']['callFrames'][0]['functionName'] == 'test2' assert response['params']['reason'] == 'other' assert response['params']['hitBreakpoints'] == ['id:69:11:' + self.config['file_path']['index']] ################################################################################################################ # main thread: Debugger.resume ################################################################################################################ await self.debugger_impl.send("Debugger.resume", main_thread) ################################################################################################################ # main thread: click on the screen ################################################################################################################ Application.click_on_middle() ################################################################################################################ # main thread: Debugger.disable ################################################################################################################ await self.debugger_impl.send("Debugger.disable", main_thread) ################################################################################################################ # close the websocket connections ################################################################################################################ await websocket.send_msg_to_debugger_server(main_thread.instance_id, main_thread.send_msg_queue, 'close') await websocket.send_msg_to_connect_server('close') ################################################################################################################