Synnefo

    def test_006_server_responds_to_ping_IPv4(self):

        """Test server responds to ping on IPv4 address"""

        log.info("Testing if server responds to pings in IPv4")

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

        ip = self._get_ipv4(server)

        self._try_until_timeout_expires(self.action_timeout,

                                        self.action_timeout,

                                        "PING IPv4 to %s" % ip,

                                        self._ping_once,

                                        False, ip)

    def test_007_server_responds_to_ping_IPv6(self):

        """Test server responds to ping on IPv6 address"""

        self._skipIf(NO_IPV6, "--no-ipv6 flag enabled")

        log.info("Testing if server responds to pings in IPv6")

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

        ip = self._get_ipv6(server)

        self._try_until_timeout_expires(self.action_timeout,

                                        self.action_timeout,

                                        "PING IPv6 to %s" % ip,

                                        self._ping_once,

                                        True, ip)

    def test_011a_server_responds_to_ping_IPv4(self):

        """Test server OS is actually up and running again"""

        log.info("Testing if server is actually up and running")

        self.test_006_server_responds_to_ping_IPv4()

    def test_012_ssh_to_server_IPv4(self):

        """Test SSH to server public IPv4 works, verify hostname"""

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

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

        self._insist_on_ssh_hostname(self._get_ipv4(server),

                                     self.username, self.passwd)

    def test_013_ssh_to_server_IPv6(self):

        """Test SSH to server public IPv6 works, verify hostname"""

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

        self._skipIf(NO_IPV6, "--no-ipv6 flag enabled")

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

        self._insist_on_ssh_hostname(self._get_ipv6(server),

                                     self.username, self.passwd)

    def test_014_rdp_to_server_IPv4(self):

        "Test RDP connection to server public IPv4 works"""

        self._skipIf(not self.is_windows, "only valid for Windows servers")

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

        ipv4 = self._get_ipv4(server)

        sock = _insist_on_tcp_connection(socket.AF_INET, ipv4, 3389)

        # No actual RDP processing done. We assume the RDP server is there

        # if the connection to the RDP port is successful.

        # FIXME: Use rdesktop, analyze exit code? see manpage [costasd]

        sock.close()

    def test_015_rdp_to_server_IPv6(self):

        "Test RDP connection to server public IPv6 works"""

        self._skipIf(not self.is_windows, "only valid for Windows servers")

        self._skipIf(NO_IPV6, "--no-ipv6 flag enabled")

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

        ipv6 = self._get_ipv6(server)

        sock = _get_tcp_connection(socket.AF_INET6, ipv6, 3389)

        # No actual RDP processing done. We assume the RDP server is there

        # if the connection to the RDP port is successful.

        sock.close()

    def test_016_personality_is_enforced(self):

        """Test file injection for personality enforcement"""

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

        self._skipIf(self.personality == None, "No personality file selected")

        log.info("Trying to inject file for personality enforcement")

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

        for inj_file in self.personality:

            equal_files = self._check_file_through_ssh(self._get_ipv4(server),

                                                       inj_file['owner'],

                                                       self.passwd,

                                                       inj_file['path'],

                                                       inj_file['contents'])

            self.assertTrue(equal_files)

            netsA = [x['network_id'] for x in self.client.get_server_details(self.serverid['A'])['attachments']['values']]

            netsB = [x['network_id'] for x in self.client.get_server_details(self.serverid['B'])['attachments']['values']]

        #Adding private IPs to class attributes

        cls = type(self)

        cls.priv_ip = dict()

        nicsA = self.client.get_server_details(self.serverid['A'])['attachments']['values']

        nicsB = self.client.get_server_details(self.serverid['B'])['attachments']['values']

        if conn_exists:

            for nic in nicsA:

                if nic["network_id"] == self.networkid:

                    cls.priv_ip["A"] = nic["ipv4"]

            for nic in nicsB:

                if nic["network_id"] == self.networkid:

                    cls.priv_ip["B"] = nic["ipv4"]

    def test_002b_ping_server_A(self):

        "Test if server A responds to IPv4 pings"

        log.info("Testing if server A responds to IPv4 pings ")

        server = self.client.get_server_details(self.serverid['A'])

        ip = self._get_ipv4(server)

        fail_tmout = time.time() + self.action_timeout

        s = False

        while True:

            if self._ping_once(ip):

                s = True

                break

            elif time.time() > fail_tmout:

                self.assertLess(time.time(), fail_tmout)

            else:

                time.sleep(self.query_interval)

        self.assertTrue(s)

    def test_002d_ping_server_B(self):

        """Test if server B responds to IPv4 pings"""

        log.info("Testing if server B responds to IPv4 pings")

        server = self.client.get_server_details(self.serverid['B'])

        ip = self._get_ipv4(server)

        fail_tmout = time.time() + self.action_timeout

        s = False

        while True:

            if self._ping_once(ip):

                s = True

                break

            elif time.time() > fail_tmout:

                self.assertLess(time.time(), fail_tmout)

            else:

                time.sleep(self.query_interval)

        self.assertTrue(s)

    def test_003a_setup_interface_A(self):

        """Set up eth1 for server A"""

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

        log.info("Setting up interface eth1 for server A")

        server = self.client.get_server_details(self.serverid['A'])

        image = self.client.get_image_details(self.imageid)

        os = image['metadata']['values']['os']

        users = image["metadata"]["values"].get("users", None)

        userlist = users.split()

        if "root" in userlist:

            loginname = "root"

        elif users == None:

            loginname = self._connect_loginname(os)

        else:

            loginname = choice(userlist)

        hostip = self._get_ipv4(server)

        myPass = self.password['A']

        log.info("SSH in server A as %s/%s" % (loginname, myPass))

        res = False

        if loginname != "root":

            with settings(

                hide('warnings', 'running'),

                warn_only=True,

                host_string=hostip,

                user=loginname, password=myPass

                ):

                if len(sudo('ifconfig eth1 %s' % self.priv_ip["A"])) == 0:

                    res = True

        else:

            with settings(

                hide('warnings', 'running'),

                warn_only=True,

                host_string=hostip,

                user=loginname, password=myPass

                ):

                if len(run('ifconfig eth1 %s' % self.priv_ip["A"])) == 0:

                    res = True

        self.assertTrue(res)

    def test_003b_setup_interface_B(self):

        """Setup eth1 for server B"""

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

        log.info("Setting up interface eth1 for server B")

        server = self.client.get_server_details(self.serverid['B'])

        image = self.client.get_image_details(self.imageid)

        os = image['metadata']['values']['os']

        users = image["metadata"]["values"].get("users", None)

        userlist = users.split()

        if "root" in userlist:

            loginname = "root"

        elif users == None:

            loginname = self._connect_loginname(os)

        else:

            loginname = choice(userlist)

        hostip = self._get_ipv4(server)

        myPass = self.password['B']

        log.info("SSH in server B as %s/%s" % (loginname, myPass))

        res = False

        if loginname != "root":

            with settings(

                hide('warnings', 'running'),

                warn_only=True,

                host_string=hostip,

                user=loginname, password=myPass

                ):

                if len(sudo('ifconfig eth1 %s' % self.priv_ip["B"])) == 0:

                    res = True

        else:

            with settings(

                hide('warnings', 'running'),

                warn_only=True,

                host_string=hostip,

                user=loginname, password=myPass

                ):

                if len(run('ifconfig eth1 %s' % self.priv_ip["B"])) == 0:

                    res = True

        self.assertTrue(res)

    def test_003c_test_connection_exists(self):

        """Ping server B from server A to test if connection exists"""

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

        log.info("Testing if server A is actually connected to server B")

        server = self.client.get_server_details(self.serverid['A'])

        image = self.client.get_image_details(self.imageid)

        os = image['metadata']['values']['os']

        hostip = self._get_ipv4(server)

        users = image["metadata"]["values"].get("users", None)

        userlist = users.split()

        if "root" in userlist:

            loginname = "root"

        elif users == None:

            loginname = self._connect_loginname(os)

        else:

            loginname = choice(userlist)

        myPass = self.password['A']

        try:

            ssh = paramiko.SSHClient()

            ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())

            ssh.connect(hostip, username=loginname, password=myPass)

        except socket.error:

            raise AssertionError

        cmd = "if ping -c 2 -w 3 %s >/dev/null; \

               then echo \'True\'; fi;" % self.priv_ip["B"]

        stdin, stdout, stderr = ssh.exec_command(cmd)

        lines = stdout.readlines()

        exists = False

        if 'True\n' in lines:

            exists = True

        self.assertTrue(exists)