#!/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 TestDebug24: """ 测试用例：单实例异常断点的设置/取消 测试步骤： 1. 连接 connect server 和主线程 debugger server 2. 主线程使能 Runtime 和 Debugger 3. 主线程运行等待的调试器 4. 主线程解析 entry_ability 和 index 文件 5. 设置异常捕获状态 ALL 6. 主线程 resume(Debugger.resume)命中异常断点,验证异常断点信息 CAUGHT 类型异常 7. 主线程 resume(Debugger.resume)命中异常断点,验证异常断点信息 UNCAUGHT 类型异常 8. 关闭主线程 debugger server 和 connect server 连接 测试代码： function main() { try { throw new Error("This is a caught error"); } catch (error) { console.log("After error"); } throw new Error("This is a uncaught error"); } main(); """ def setup_method(self): logging.info('Start running TestDebug24: setup') self.log_path = rf'{os.path.dirname(__file__)}\..\log' self.hilog_file_name = 'test_debug_24.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('TestDebug24 done') def test(self, test_suite_main_instance_04_debug): logging.info('Start running TestDebug24: test') self.config = test_suite_main_instance_04_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.setPauseOnExceptions ################################################################################################################ params = debugger.PauseOnExceptionsState.ALL await self.debugger_impl.send("Debugger.setPauseOnExceptions", 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']['reason'] == 'exception' assert 'This is a caught error' in response['params']['data']['unserializableValue'] ################################################################################################################ # 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']['reason'] == 'exception' assert 'This is a uncaught error' in response['params']['data']['unserializableValue'] ################################################################################################################ # main thread: Debugger.resume ################################################################################################################ await self.debugger_impl.send("Debugger.resume", 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') ################################################################################################################