Synnefo » ./kamaki

class Client(object):

    MAX_THREADS = 7

    LOG_TOKEN = False

    CONNECTION_RETRY_LIMIT = 0

    def __init__(self, base_url, token):

    def _has_status(self, servid, status):

        r = self.client.get_server_details(servid)

        return r['status'] == status

    def _wait_for_status(self, servid, status):

        (wait_bar, wait_cb) = self._safe_progress_bar(

            'Server %s in %s' % (servid, status))

        self._safe_progress_bar_finish(wait_bar)

    def test_parallel_creation(self):

        Do not use this in regular livetest

        """

        from kamaki.clients import SilentEvent

    def test_create_server(self):

        """Test create_server"""

        self.server1 = self._create_server(

        self._wait_for_status(self.server1['id'], 'BUILD')

        self._test_0010_create_server()

    def test_list_servers(self):

        """Test list servers"""

        self.server1 = self._create_server(

        self.server2 = self._create_server(

        self._test_0020_list_servers()

    def _test_0020_list_servers(self):

        """detailed and simple are same size"""

        self.assertEqual(len(dservers), len(servers))

        """detailed and simple contain same names"""

        names = sorted(map(lambda x: x["name"], servers))

    def test_get_server_details(self):

        """Test get_server_details"""

        self.server1 = self._create_server(

        self._wait_for_status(self.server1['id'], 'BUILD')

        self._test_0040_get_server_details()

    def test_update_server_name(self):

        """Test update_server_name"""

        self.server1 = self._create_server(

        self._test_0050_update_server_name()

    def _test_0050_update_server_name(self):

        new_name = self.servname1 + '_new_name'

        self.client.update_server_name(self.server1['id'], new_name)

        r = self.client.get_server_details(

        self.assertEqual(r['name'], new_name)

        changed = self.servers.pop(self.servname1)

        changed['name'] = new_name

    def test_reboot_server(self):

        """Test reboot server"""

        self.server1 = self._create_server(

        self._wait_for_status(self.server1['id'], 'BUILD')

        self.server2 = self._create_server(

        self._wait_for_status(self.server2['id'], 'BUILD')

        self._test_0060_reboot_server()

        self._wait_for_status(self.server1['id'], 'REBOOT')

        self.server1 = self._create_server(

            self.servname1, self.flavorid, self.img)

        srv = self.client.get_flavor_details(self.server1['id'])

    def _test_0070_wait_test_servers_to_reboot(self):

        """Pseudo-test to wait for VMs to load"""

    def test_create_server_metadata(self):

        """Test create_server_metadata"""

        self.server1 = self._create_server(

        self._test_0080_create_server_metadata()

    def _test_0080_create_server_metadata(self):

        r1 = self.client.create_server_metadata(

        self.assertTrue('mymeta' in r1)

        r2 = self.client.get_server_metadata(self.server1['id'], 'mymeta')

        self.assert_dicts_are_equal(r1, r2)

    def test_get_server_metadata(self):

        """Test get server_metadata"""

        self.server1 = self._create_server(

        self._test_0090_get_server_metadata()

    def _test_0090_get_server_metadata(self):

    def test_update_server_metadata(self):

        """Test update_server_metadata"""

        self.server1 = self._create_server(

        self._test_0100_update_server_metadata()

    def _test_0100_update_server_metadata(self):

    def test_shutdown_server(self):

        """Test shutdown_server"""

        self.server1 = self._create_server(

        self._wait_for_status(self.server1['id'], 'BUILD')

        self._test_0170_shutdown_server()

    def test_start_server(self):

        """Test start_server"""

        self.server1 = self._create_server(

        self._wait_for_status(self.server1['id'], 'BUILD')

        self.client.shutdown_server(self.server1['id'])

        self._wait_for_status(self.server1['id'], 'ACTIVE')

    def test_get_server_console(self):

        """Test get_server_console"""

        self.server2 = self._create_server(

        self._wait_for_status(self.server2['id'], 'BUILD')

        self._test_0190_get_server_console()

    def test_get_firewall_profile(self):

        """Test get_firewall_profile"""

        self.server1 = self._create_server(

        self._test_0200_get_firewall_profile()

    def _test_0200_get_firewall_profile(self):

    def test_set_firewall_profile(self):

        """Test set_firewall_profile"""

        self.server1 = self._create_server(

        self._test_0210_set_firewall_profile()

    def _test_0210_set_firewall_profile(self):

    def test_get_server_stats(self):

        self.server1 = self._create_server(

        self._test_0220_get_server_stats()

    def _test_0220_get_server_stats(self):

    def test_connect_server(self):

        """Test connect_server"""

        self.server1 = self._create_server(

        self.network1 = self._create_network(self.netname1)

        self._wait_for_status(self.server1['id'], 'BUILD')

        self._wait_for_network(self.network1['id'], 'ACTIVE')

    def _test_0250_connect_server(self):

        self.client.connect_server(self.server1['id'], self.network1['id'])

    def test_disconnect_server(self):

        """Test disconnect_server"""

    def _test_0250_disconnect_server(self):

        self.client.disconnect_server(self.server1['id'], self.network1['id'])

        self.assertTrue(self._wait_for_nic(

    def _test_0260_wait_for_second_network(self):

        self.network2 = self._create_network(self.netname2)

    def test_list_server_nics(self):

        """Test list_server_nics"""

        self.server1 = self._create_server(

        self.network2 = self._create_network(self.netname2)

        self._wait_for_status(self.server1['id'], 'BUILD')

        self._wait_for_network(self.network2['id'], 'ACTIVE')

        r = self.client.list_server_nics(self.server1['id'])

        len0 = len(r)

        self.client.connect_server(self.server1['id'], self.network2['id'])

        r = self.client.list_server_nics(self.server1['id'])

        self.assertTrue(len(r) > len0)

    def test_list_network_nics(self):

        """Test list_server_nics"""

        self.server1 = self._create_server(

        self.network1 = self._create_network(self.netname1)

        self.network2 = self._create_network(self.netname2)

        self._wait_for_status(self.server1['id'], 'BUILD')

        self._test_0293_list_network_nics()

    def _test_0293_list_network_nics(self):

        r = self.client.list_network_nics(netid1)

        expected = ['nic-%s-1' % self.server1['id']]

        self.assertEqual(r, expected)

    def _test_0130_object_post(self):

        self.client.container = self.c2

        obj = 'test2'

        """create a filesystem file"""

        self.files.append(NamedTemporaryFile())

        newf = self.files[-1]

            'ello!\n',

            'This is a test line\n',

            'inside a test file\n'])

        """create a file on container"""

        r = self.client.object_put(

            obj,

        """Check if_etag_(not)match"""

        etag = r['etag']

        r = self.client.object_post(

            obj,

            update=True,

            public=True,

            if_etag_not_match=etag,

            success=(412, 202, 204))

        r = self.client.object_post(

            obj,

        self._assert_account()

        if groups:

            for group, usernames in groups.items():

        """

        self._assert_container()

        self.set_param('format', format, iff=format)

        self.set_header('X-Container-Policy-Quota', quota)

        self.set_header('Content-Type', content_type)

        self.set_header('Content-Encoding', content_encoding)

        self.set_header('Content-Disposition', content_disposition)

        self.set_header('X-Object-Public', public, public is not None)

        if metadata:

            for key, val in metadata.items():

        self._assert_container()

        self.set_param('format', format, iff=format)

        self.set_header('If-Match', if_etag_match)

        self.set_header('If-None-Match', if_etag_not_match)

        self.set_header(

        self.set_header('Content-Type', content_type)

        self.set_header('Content-Range', content_range)

        self.set_header('Transfer-Encoding', transfer_encoding)

        self.set_header('X-Source-Version', source_version)

        self.set_header('X-Object-Bytes', object_bytes)

        self.set_header('X-Object-Manifest', manifest)

        self.set_header('X-Object-Public', public, public is not None)

        for key, val in metadata.items():

            self.set_header('X-Object-Meta-' + key, val)