Synnefo

    def _verify_server_status(self, current_status, new_status):

        """Verify a server has switched to a specified status"""

        server = self.client.get_server_details(self.serverid)

        self.assertIn(server["status"], (current_status, new_status))

        self.assertEquals(server["status"], new_status)

    def _ping_once(self, ipv6, ip):

        """Test server responds to a single IPv4 or IPv6 ping"""

        cmd = "ping%s -c 2 -w 3 %s" % ("6" if ipv6 else "", ip)

        ping = subprocess.Popen(cmd, shell=True,

                                stdout=subprocess.PIPE, stderr=subprocess.PIPE)

        stdin, stdout, stderr = ssh.exec_command("hostname")

        lines = stdout.readlines()

        self.assertEqual(len(lines), 1)

    def _try_until_timeout_expires(self, warn_timeout, fail_timeout,

                                   opmsg, callable, *args, **kwargs):

        warn_tmout = time.time() + warn_timeout

        fail_tmout = time.time() + fail_timeout

        while True:

            if time.time() > warn_tmout:

            try:

                return callable(*args, **kwargs)

            except AssertionError:

                pass

            time.sleep(self.query_interval)

    def _insist_on_tcp_connection(self, family, host, port):

    def _insist_on_status_transition(self, current_status, new_status,

                                    fail_timeout, warn_timeout=None):

        if warn_timeout is None:

            warn_timeout = fail_timeout

        self._try_until_timeout_expires(warn_timeout, fail_timeout,

                                        current_status, new_status)

    def _insist_on_ssh_hostname(self, hostip, username, password):

        hostname = self._try_until_timeout_expires(

                self.action_timeout, self.action_timeout,

                msg, self._get_hostname_over_ssh,

        ip = self._get_ipv4(server)

        self._try_until_timeout_expires(self.action_timeout,

                                        self.action_timeout,

                                        False, ip)

    def test_007_server_responds_to_ping_IPv6(self):

        ip = self._get_ipv6(server)

        self._try_until_timeout_expires(self.action_timeout,

                                        self.action_timeout,

                                        True, ip)

    def test_008_submit_shutdown_request(self):

        self._skipIf(self.is_windows, "only implemented for Linux servers")

        self.assertTrue(False, "test not implemented, will fail")

def _spawn_server_test_case(**kwargs):

    """Construct a new unit test case class from SpawnServerTestCase"""

    servers = c.list_servers()

    stale = [s for s in servers if s["name"].startswith(SNF_TEST_PREFIX)]

    print >> sys.stderr, "Found these stale servers from previous runs:"

    print "    " + \

          "\n    ".join(["%d: %s" % (s['id'], s['name']) for s in stale])