(case_setup_remote_attach_to_dap)
| 4332 | |
| 4333 | @pytest.mark.skipif(not TEST_GEVENT, reason="Gevent not installed.") |
| 4334 | def test_wait_for_attach_gevent(case_setup_remote_attach_to_dap): |
| 4335 | host_port = get_socket_name(close=True) |
| 4336 | |
| 4337 | def get_environ(writer): |
| 4338 | env = os.environ.copy() |
| 4339 | env["GEVENT_SUPPORT"] = "True" |
| 4340 | return env |
| 4341 | |
| 4342 | def check_thread_events(json_facade): |
| 4343 | json_facade.write_list_threads() |
| 4344 | # Check that we have the started thread event (whenever we reconnect). |
| 4345 | started_events = json_facade.mark_messages(ThreadEvent, lambda x: x.body.reason == "started") |
| 4346 | assert len(started_events) == 1 |
| 4347 | |
| 4348 | with case_setup_remote_attach_to_dap.test_file( |
| 4349 | "_debugger_case_gevent.py", host_port[1], additional_args=["remote", "as-server"], get_environ=get_environ |
| 4350 | ) as writer: |
| 4351 | writer.TEST_FILE = debugger_unittest._get_debugger_test_file("_debugger_case_gevent.py") |
| 4352 | time.sleep(0.5) # Give some time for it to pass the first breakpoint and wait. |
| 4353 | writer.start_socket_client(*host_port) |
| 4354 | |
| 4355 | json_facade = JsonFacade(writer) |
| 4356 | check_thread_events(json_facade) |
| 4357 | |
| 4358 | break1_line = writer.get_line_index_with_content("break here") |
| 4359 | json_facade.write_set_breakpoints(break1_line) |
| 4360 | json_facade.write_make_initial_run() |
| 4361 | json_facade.wait_for_thread_stopped(line=break1_line) |
| 4362 | |
| 4363 | json_facade.write_disconnect() |
| 4364 | writer.finished_ok = True |
| 4365 | |
| 4366 | |
| 4367 | @pytest.mark.skipif(not TEST_GEVENT, reason="Gevent not installed.") |
nothing calls this directly
no test coverage detected