(case_setup_remote_attach_to_dap)
| 4249 | |
| 4250 | |
| 4251 | def test_wait_for_attach(case_setup_remote_attach_to_dap): |
| 4252 | host_port = get_socket_name(close=True) |
| 4253 | |
| 4254 | def check_thread_events(json_facade): |
| 4255 | json_facade.write_list_threads() |
| 4256 | # Check that we have the started thread event (whenever we reconnect). |
| 4257 | started_events = json_facade.mark_messages(ThreadEvent, lambda x: x.body.reason == "started") |
| 4258 | assert len(started_events) >= 1 |
| 4259 | |
| 4260 | def check_process_event(json_facade, start_method): |
| 4261 | if start_method == "attach": |
| 4262 | json_facade.write_attach() |
| 4263 | |
| 4264 | elif start_method == "launch": |
| 4265 | json_facade.write_launch() |
| 4266 | |
| 4267 | else: |
| 4268 | raise AssertionError("Unexpected: %s" % (start_method,)) |
| 4269 | |
| 4270 | process_events = json_facade.mark_messages(ProcessEvent) |
| 4271 | assert len(process_events) == 1 |
| 4272 | assert next(iter(process_events)).body.startMethod == start_method |
| 4273 | |
| 4274 | with case_setup_remote_attach_to_dap.test_file("_debugger_case_wait_for_attach.py", host_port[1]) as writer: |
| 4275 | writer.TEST_FILE = debugger_unittest._get_debugger_test_file("_debugger_case_wait_for_attach_impl.py") |
| 4276 | time.sleep(1) # Give some time for it to pass the first breakpoint and wait in 'wait_for_attach'. |
| 4277 | writer.start_socket_client(*host_port) |
| 4278 | |
| 4279 | json_facade = JsonFacade(writer) |
| 4280 | check_thread_events(json_facade) |
| 4281 | |
| 4282 | break1_line = writer.get_line_index_with_content("Break 1") |
| 4283 | break2_line = writer.get_line_index_with_content("Break 2") |
| 4284 | break3_line = writer.get_line_index_with_content("Break 3") |
| 4285 | |
| 4286 | pause1_line = writer.get_line_index_with_content("Pause 1") |
| 4287 | pause2_line = writer.get_line_index_with_content("Pause 2") |
| 4288 | |
| 4289 | check_process_event(json_facade, start_method="launch") |
| 4290 | json_facade.write_set_breakpoints([break1_line, break2_line, break3_line]) |
| 4291 | json_facade.write_make_initial_run() |
| 4292 | json_facade.wait_for_thread_stopped(line=break2_line) |
| 4293 | |
| 4294 | # Upon disconnect, all threads should be running again. |
| 4295 | json_facade.write_disconnect() |
| 4296 | |
| 4297 | # Connect back (socket should remain open). |
| 4298 | writer.start_socket_client(*host_port) |
| 4299 | json_facade = JsonFacade(writer) |
| 4300 | check_thread_events(json_facade) |
| 4301 | check_process_event(json_facade, start_method="attach") |
| 4302 | json_facade.write_set_breakpoints([break1_line, break2_line, break3_line]) |
| 4303 | json_facade.write_make_initial_run() |
| 4304 | json_facade.wait_for_thread_stopped(line=break3_line) |
| 4305 | |
| 4306 | # Upon disconnect, all threads should be running again. |
| 4307 | json_facade.write_disconnect() |
| 4308 |
nothing calls this directly
no test coverage detected