Synnefo

    # def test_003a_get_server_oob_console(self):

    #     """Test getting OOB server console over VNC

    #     Implementation of RFB protocol follows

    #     http://www.realvnc.com/docs/rfbproto.pdf.

    #     """

    #     console = self.cyclades.get_server_console(self.serverid)

    #     self.assertEquals(console['type'], "vnc")

    #     sock = self._insist_on_tcp_connection(socket.AF_INET,

    #                                     console["host"], console["port"])

    #     # Step 1. ProtocolVersion message (par. 6.1.1)

    #     version = sock.recv(1024)

    #     self.assertEquals(version, 'RFB 003.008\n')

    #     sock.send(version)

    #     # Step 2. Security (par 6.1.2): Only VNC Authentication supported

    #     sec = sock.recv(1024)

    #     self.assertEquals(list(sec), ['\x01', '\x02'])

    #     # Step 3. Request VNC Authentication (par 6.1.2)

    #     sock.send('\x02')

    #     # Step 4. Receive Challenge (par 6.2.2)

    #     challenge = sock.recv(1024)

    #     self.assertEquals(len(challenge), 16)

    #     # Step 5. DES-Encrypt challenge, use password as key (par 6.2.2)

    #     response = d3des_generate_response(

    #         (console["password"] + '\0' * 8)[:8], challenge)

    #     sock.send(response)

    #     # Step 6. SecurityResult (par 6.1.3)

    #     result = sock.recv(4)

    #     self.assertEquals(list(result), ['\x00', '\x00', '\x00', '\x00'])

    #     sock.close()

                runner = unittest.TextTestRunner(f, verbosity=2, failfast = True)

            for s in stale:

                c.delete_server(s["id"])

def cleanup_networks(timeout, query_interval, delete_stale=False):

        fail_tmout = time.time() + timeout

        cleanup_servers(delete_stale=opts.delete_stale)

        log.info('Parallel spawn:')