Method test_server

tests/test_pydevconsole.py:232–283

(self)

        return port0, port1

    def test_server(self):
        self.original_stdout = sys.stdout
        sys.stdout = pydevd_io.IOBuf()
        try:
            client_port, server_port = self.get_free_addresses()

            class ServerThread(threading.Thread):
                def __init__(self, client_port, server_port):
                    threading.Thread.__init__(self)
                    self.client_port = client_port
                    self.server_port = server_port

                def run(self):
                    from _pydev_bundle import pydev_localhost

                    pydevconsole.start_server(pydev_localhost.get_localhost(), self.server_port, self.client_port)

            server_thread = ServerThread(client_port, server_port)
            server_thread.daemon = True
            server_thread.start()

            client_thread = self.start_client_thread(client_port)  # @UnusedVariable

            import time

            time.sleep(0.3)  # let's give it some time to start the threads
            sys.stdout = pydevd_io.IOBuf()

            from _pydev_bundle import pydev_localhost

            server = xmlrpclib.Server("http://%s:%s" % (pydev_localhost.get_localhost(), server_port))
            server.execLine("class Foo:")
            server.execLine("    pass")
            server.execLine("")
            server.execLine("foo = Foo()")
            server.execLine("a = input()")
            server.execLine("print (a)")
            initial = time.time()
            while not client_thread.requested_input:
                if time.time() - initial > 2:
                    raise AssertionError("Did not get the return asked before the timeout.")
                time.sleep(0.1)

            found = sys.stdout.getvalue()
            while ["input_request"] != found.split():
                found += sys.stdout.getvalue()
                if time.time() - initial > 2:
                    break
                time.sleep(0.1)
            self.assertEqual(["input_request"], found.split())
        finally:
            sys.stdout = self.original_stdout