Class ClientThread

tests/test_check_pydevconsole.py:30–58

class Test(unittest.TestCase):
    def start_client_thread(self, client_port):
        class ClientThread(threading.Thread):
            def __init__(self, client_port):
                threading.Thread.__init__(self)
                self.client_port = client_port

            def run(self):
                class HandleRequestInput:
                    def RequestInput(self):
                        client_thread.requested_input = True
                        return "RequestInput: OK"

                    def NotifyFinished(self, *args, **kwargs):
                        client_thread.notified_finished += 1
                        return 1

                handle_request_input = HandleRequestInput()

                from _pydev_bundle import pydev_localhost

                self.client_server = client_server = SimpleXMLRPCServer(
                    (pydev_localhost.get_localhost(), self.client_port), logRequests=False
                )
                client_server.register_function(handle_request_input.RequestInput)
                client_server.register_function(handle_request_input.NotifyFinished)
                client_server.serve_forever()

            def shutdown(self):
                return
                self.client_server.shutdown()

        client_thread = ClientThread(client_port)
        client_thread.requested_input = False