95 lines · python
1"""2Test lldb-dap cancel request3"""4 5import time6 7from lldbsuite.test.decorators import *8from lldbsuite.test.lldbtest import *9import lldbdap_testcase10 11 12class TestDAP_cancel(lldbdap_testcase.DAPTestCaseBase):13 def send_async_req(self, command: str, arguments: dict = {}) -> int:14 return self.dap_server.send_packet(15 {16 "type": "request",17 "command": command,18 "arguments": arguments,19 }20 )21 22 def async_blocking_request(self, duration: float) -> int:23 """24 Sends an evaluate request that will sleep for the specified duration to25 block the request handling thread.26 """27 return self.send_async_req(28 command="evaluate",29 arguments={30 "expression": '`script import time; print("starting sleep", file=lldb.debugger.GetOutputFileHandle()); time.sleep({})'.format(31 duration32 ),33 "context": "repl",34 },35 )36 37 def async_cancel(self, requestId: int) -> int:38 return self.send_async_req(command="cancel", arguments={"requestId": requestId})39 40 def test_pending_request(self):41 """42 Tests cancelling a pending request.43 """44 program = self.getBuildArtifact("a.out")45 self.build_and_launch(program)46 47 # Use a relatively short timeout since this is only to ensure the48 # following request is queued.49 blocking_seq = self.async_blocking_request(duration=self.DEFAULT_TIMEOUT / 10)50 # Use a longer timeout to ensure we catch if the request was interrupted51 # properly.52 pending_seq = self.async_blocking_request(duration=self.DEFAULT_TIMEOUT / 2)53 cancel_seq = self.async_cancel(requestId=pending_seq)54 55 blocking_resp = self.dap_server.receive_response(blocking_seq)56 self.assertEqual(blocking_resp["request_seq"], blocking_seq)57 self.assertEqual(blocking_resp["command"], "evaluate")58 self.assertEqual(blocking_resp["success"], True)59 60 pending_resp = self.dap_server.receive_response(pending_seq)61 self.assertEqual(pending_resp["request_seq"], pending_seq)62 self.assertEqual(pending_resp["command"], "evaluate")63 self.assertEqual(pending_resp["success"], False)64 self.assertEqual(pending_resp["message"], "cancelled")65 66 cancel_resp = self.dap_server.receive_response(cancel_seq)67 self.assertEqual(cancel_resp["request_seq"], cancel_seq)68 self.assertEqual(cancel_resp["command"], "cancel")69 self.assertEqual(cancel_resp["success"], True)70 self.continue_to_exit()71 72 def test_inflight_request(self):73 """74 Tests cancelling an inflight request.75 """76 program = self.getBuildArtifact("a.out")77 self.build_and_launch(program)78 79 blocking_seq = self.async_blocking_request(duration=self.DEFAULT_TIMEOUT / 2)80 # Wait for the sleep to start to cancel the inflight request.81 self.collect_console(pattern="starting sleep")82 cancel_seq = self.async_cancel(requestId=blocking_seq)83 84 blocking_resp = self.dap_server.receive_response(blocking_seq)85 self.assertEqual(blocking_resp["request_seq"], blocking_seq)86 self.assertEqual(blocking_resp["command"], "evaluate")87 self.assertEqual(blocking_resp["success"], False)88 self.assertEqual(blocking_resp["message"], "cancelled")89 90 cancel_resp = self.dap_server.receive_response(cancel_seq)91 self.assertEqual(cancel_resp["request_seq"], cancel_seq)92 self.assertEqual(cancel_resp["command"], "cancel")93 self.assertEqual(cancel_resp["success"], True)94 self.continue_to_exit()95