Method test_server

tests/test_check_pydevconsole.py:73–121

(self)

        return [socket_name[1] for socket_name in socket_names]

    def test_server(self):
        # Just making sure that the singleton is created in this thread.
        from _pydev_bundle.pydev_ipython_console_011 import get_pydev_frontend

        get_pydev_frontend(get_localhost(), 0)

        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

                print("Starting server with:", pydev_localhost.get_localhost(), self.server_port, self.client_port)
                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

        try:
            import time

            time.sleep(0.3)  # let's give it some time to start the threads

            from _pydev_bundle import pydev_localhost

            server = xmlrpclib.Server("http://%s:%s" % (pydev_localhost.get_localhost(), server_port))
            server.execLine("import sys; print('Running with: %s %s' % (sys.executable or sys.platform, sys.version))")
            server.execLine("class Foo:")
            server.execLine("    pass")
            server.execLine("")
            server.execLine("foo = Foo()")
            server.execLine("a = %s()" % raw_input_name)
            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)
            frame_xml = server.getFrame()
            self.assertTrue("RequestInput" in frame_xml, "Did not fid RequestInput in:\n%s" % (frame_xml,))
        finally:
            client_thread.shutdown()