(self, client_port)
| 27 | @pytest.mark.skipif(os.environ.get("TRAVIS") == "true" or not has_ipython, reason="Too flaky on Travis (and requires IPython).") |
| 28 | class Test(unittest.TestCase): |
| 29 | def start_client_thread(self, client_port): |
| 30 | class ClientThread(threading.Thread): |
| 31 | def __init__(self, client_port): |
| 32 | threading.Thread.__init__(self) |
| 33 | self.client_port = client_port |
| 34 | |
| 35 | def run(self): |
| 36 | class HandleRequestInput: |
| 37 | def RequestInput(self): |
| 38 | client_thread.requested_input = True |
| 39 | return "RequestInput: OK" |
| 40 | |
| 41 | def NotifyFinished(self, *args, **kwargs): |
| 42 | client_thread.notified_finished += 1 |
| 43 | return 1 |
| 44 | |
| 45 | handle_request_input = HandleRequestInput() |
| 46 | |
| 47 | from _pydev_bundle import pydev_localhost |
| 48 | |
| 49 | self.client_server = client_server = SimpleXMLRPCServer( |
| 50 | (pydev_localhost.get_localhost(), self.client_port), logRequests=False |
| 51 | ) |
| 52 | client_server.register_function(handle_request_input.RequestInput) |
| 53 | client_server.register_function(handle_request_input.NotifyFinished) |
| 54 | client_server.serve_forever() |
| 55 | |
| 56 | def shutdown(self): |
| 57 | return |
| 58 | self.client_server.shutdown() |
| 59 | |
| 60 | client_thread = ClientThread(client_port) |
| 61 | client_thread.requested_input = False |
| 62 | client_thread.notified_finished = 0 |
| 63 | client_thread.daemon = True |
| 64 | client_thread.start() |
| 65 | return client_thread |
| 66 | |
| 67 | def get_free_addresses(self): |
| 68 | from _pydev_bundle.pydev_localhost import get_socket_names |
no test coverage detected