Synnefo » ./kamaki

# Copyright 2011 GRNET S.A. All rights reserved.

#

# Redistribution and use in source and binary forms, with or

# without modification, are permitted provided that the following

# conditions are met:

#

#   1. Redistributions of source code must retain the above

#      copyright notice, this list of conditions and the following

#      disclaimer.

#

#   2. Redistributions in binary form must reproduce the above

#      copyright notice, this list of conditions and the following

#      disclaimer in the documentation and/or other materials

#      provided with the distribution.

#

# THIS SOFTWARE IS PROVIDED BY GRNET S.A. ``AS IS'' AND ANY EXPRESS

# OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED

# WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR

# PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL GRNET S.A OR

# CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,

# SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT

# LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF

# USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED

# AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT

# LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN

# ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE

# POSSIBILITY OF SUCH DAMAGE.

#

# The views and conclusions contained in the software and

# documentation are those of the authors and should not be

# interpreted as representing official policies, either expressed

# or implied, of GRNET S.A.

from argparse import ArgumentParser

import unittest

import time

import datetime

import os

import sys

from kamaki.clients import ClientError

from kamaki.clients.pithos import PithosClient as pithos

from kamaki.clients.cyclades import CycladesClient as cyclades

from kamaki.clients.image import ImageClient as image

from kamaki.clients.astakos import AstakosClient as astakos

from kamaki.cli.config import Config

TEST_ALL = False

cnf = Config('/home/saxtouri/.kamakirc')

global_username = cnf.get('global', 'account')

token = cnf.get('global', 'token')

class testAstakos(unittest.TestCase):

    def setUp(self):

        global cnf

        url = cnf.get('astakos', 'url')

        global token

        self.client = astakos(url, token)

    def tearDown(self):

        pass

    def test_authenticate(self):

        r = self.client.authenticate()

        for term in ('username',

            'auth_token_expires',

            'auth_token',

            'auth_token_created',

            'groups',

            'uniq',

            'has_credits',

            'has_signed_terms'):

            self.assertTrue(term in r)

class testImage(unittest.TestCase):

    def setUp(self):

        global cnf

        cyclades_url = cnf.get('compute', 'url')

        url = cnf.get('image', 'url')

        global token

        self.token = token

        self.imgid = 'b2dffe52-64a4-48c3-8a4c-8214cc3165cf'

        self.now = time.mktime(time.gmtime())

        self.imgname = 'img_%s' % self.now

        global global_username

        self.imglocation = 'pithos://%s/pithos/my.img'\

        % global_username

        self.client = image(url, self.token)

        self.cyclades = cyclades(cyclades_url, self.token)

        self._imglist = {}

    def _prepare_img(self):

        global cnf

        global global_username

        username = global_username.split('@')[0]

        imglocalpath =\

        '/home/%s/src/kamaki-settings/files/centos.diskdump' % username

        f = open(imglocalpath, 'rb')

        pithcli = pithos(cnf.get('store', 'url'),

            self.token,

            global_username,

            'pithos')

        print('\t- Upload an image at %s...' % imglocalpath)

        pithcli.upload_object('my.img', f)

        print('\t- ok')

        f.close()

        self.client.register(self.imgname,

            self.imglocation,

            params=dict(is_public=True))

        img = self._get_img_by_name(self.imgname)

        self._imglist[self.imgname] = img

    def tearDown(self):

        for img in self._imglist.values():

            self.cyclades.delete_image(img['id'])

    def _get_img_by_name(self, name):

        r = self.cyclades.list_images()

        for img in r:

            if img['name'] == name:

                return img

        return None

    def assert_dicts_are_deeply_equal(self, d1, d2):

        for k, v in d1.items():

            self.assertTrue(k in d2)

            if isinstance(v, dict):

                self.assert_dicts_are_deeply_equal(v, d2[k])

            else:

                self.assertEqual(unicode(v), unicode(d2[k]))

    def test_list_public(self):

        """Test list_public"""

        r = self.client.list_public()

        r0 = self.client.list_public(order='-')

        self.assertTrue(len(r) > 0)

        for img in r:

            for term in ('status',

                'name',

                'container_format',

                'disk_format',

                'id',

                'size'):

                self.assertTrue(term in img)

        self.assertTrue(len(r), len(r0))

        r0.reverse()

        for i, img in enumerate(r):

            self.assert_dicts_are_deeply_equal(img, r0[i])

        r1 = self.client.list_public(detail=True)

        for img in r1:

            for term in ('status',

                'name',

                'checksum',

                'created_at',

                'disk_format',

                'updated_at',

                'id',

                'location',

                'container_format',

                'owner',

                'is_public',

                'deleted_at',

                'properties',

                'size'):

                self.assertTrue(term in img)

                for interm in (

                    'osfamily',

                    'users',

                    'os',

                    'root_partition',

                    'description'):

                    self.assertTrue(interm in img['properties'])

        size_max = 1000000000

        r2 = self.client.list_public(filters=dict(size_max=size_max))

        self.assertTrue(len(r2) <= len(r))

        for img in r2:

            self.assertTrue(int(img['size']) <= size_max)

    def test_get_meta(self):

        """Test get_meta"""

        r = self.client.get_meta(self.imgid)

        self.assertEqual(r['id'], self.imgid)

        for term in ('status',

            'name',

            'checksum',

            'updated-at',

            'created-at',

            'deleted-at',

            'location',

            'is-public',

            'owner',

            'disk-format',

            'size',

            'container-format'):

            self.assertTrue(term in r)

            for interm in ('kernel',

                'osfamily',

                'users',

                'partition-table',

                'gui', 'sortorder',

                'root-partition',

                'os',

                'description'):

                self.assertTrue(interm in r['properties'])

    def test_register(self):

        """Test register"""

        self._prepare_img()

        self.assertTrue(len(self._imglist) > 0)

        for img in self._imglist.values():

            self.assertTrue(img != None)

    def test_set_members(self):

        """Test set_members"""

        self._prepare_img()

        members = ['%s@fake.net' % self.now]

        for img in self._imglist.values():

            self.client.set_members(img['id'], members)

            r = self.client.list_members(img['id'])

            self.assertEqual(r[0]['member_id'], members[0])

    def test_list_members(self):

        """Test list_members"""

        self.test_set_members()

    def test_remove_members(self):

        """Test remove_members - NO CHECK"""

        return

        self._prepare_img()

        members = ['%s@fake.net' % self.now, '%s_v2@fake.net' % self.now]

        for img in self._imglist.values():

            self.client.set_members(img['id'], members)

            r = self.client.list_members(img['id'])

            self.assertTrue(len(r) > 1)

            self.client.remove_member(img['id'], members[0])

            r0 = self.client.list_members(img['id'])

            self.assertEqual(len(r), 1 + len(r0))

            self.assertEqual(r0[0]['member_id'], members[1])

    def test_list_shared(self):

        """Test list_shared - NOT CHECKED"""

        #No way to test this, if I dont have member images

        pass

class testCyclades(unittest.TestCase):

    """Set up a Cyclades thorough test"""

    def setUp(self):

        """okeanos"""

        url = 'https://cyclades.okeanos.grnet.gr/api/v1.1'

        global token

        global global_username

        self.img = 'b2dffe52-64a4-48c3-8a4c-8214cc3165cf'

        self.img_details = {

            u'status': u'ACTIVE',

            u'updated': u'2012-10-16T09:04:17+00:00',

            u'name': u'Debian Base',

            u'created': u'2012-10-16T09:03:12+00:00',

            u'progress': 100,

            u'id': 'b2dffe52-64a4-48c3-8a4c-8214cc3165cf',

            u'metadata': {

                u'values': {

                    u'kernel': u'2.6.32',

                    u'osfamily': u'linux',

                    u'users': u'root',

                    u'gui': u'No GUI',

                    u'sortorder': u'1',

                    u'os': u'debian',

                    u'root_partition': u'1',

                    u'description': u'Debian 6.0.6 (Squeeze) Base System',

                    u'partition_table': u'msdos'}

                }

            }

        self.flavor_details = {u'name': u'C1R1024D20',

            u'ram': 1024,

            u'id': 1,

            u'SNF:disk_template': u'drbd',

            u'disk': 20,

            u'cpu': 1}

        self.PROFILES = ('ENABLED', 'DISABLED', 'PROTECTED')

        """okeanos.io"""

        #url = 'https://cyclades.okeanos.io/api/v1.1'

        #global token

        #self.img = '43cc8497-61c3-4c46-ae8d-3e33861f8527'

        #self.img_details= {

        #    u'status': u'ACTIVE',

        #    u'updated': u'2012-08-21T12:57:39+00:00',

        #    u'name': u'Debian Base',

        #    u'created': u'2012-08-21T12:56:53+00:00',

        #    u'progress': 100,

        #    u'id': u'43cc8497-61c3-4c46-ae8d-3e33861f8527',

        #    u'metadata': {

        #        u'values': {

        #            u'kernel': u'2.6.32',

        #            u'osfamily': u'linux',

        #            u'users': u'root',

        #            u'gui': u'No GUI',

        #            u'sortorder': u'1',

        #            u'os': u'debian',

        #            u'root_partition':

        #            u'1', u'description':

        #            u'Debian Squeeze Base System'}

        #        }

        #    }

        #flavorid = 1

        self.servers = {}

        self.now = time.mktime(time.gmtime())

        self.servname1 = 'serv' + unicode(self.now)

        self.servname2 = self.servname1 + '_v2'

        self.flavorid = 1

        #servers have to be created at the begining...

        self.networks = {}

        self.netname1 = 'net' + unicode(self.now)

        self.netname2 = 'net' + unicode(self.now) + '_v2'

        self.client = cyclades(url, token)

        pass

    @if_not_all

    def test_parallel_creation(self):

        """test create with multiple threads"""

        from kamaki.clients import SilentEvent

        c1 = SilentEvent(self._create_server,

            self.servname1,

            self.flavorid,

            self.img)

        c2 = SilentEvent(self._create_server,

            self.servname2,

            self.flavorid + 2,

            self.img)

        c3 = SilentEvent(self._create_server,

            self.servname1,

            self.flavorid,

            self.img)

        c4 = SilentEvent(self._create_server,

            self.servname2,

            self.flavorid + 2,

            self.img)

        c5 = SilentEvent(self._create_server,

            self.servname1,

            self.flavorid,

            self.img)

        c6 = SilentEvent(self._create_server,

            self.servname2,

            self.flavorid + 2,

            self.img)

        c7 = SilentEvent(self._create_server,

            self.servname1,

            self.flavorid,

            self.img)

        c8 = SilentEvent(self._create_server,

            self.servname2,

            self.flavorid + 2,

            self.img)

        print('START THREADS')

        c1.start()

        c2.start()

        c3.start()

        c4.start()

        c5.start()

        c6.start()

        c7.start()

        c8.start()

        print('KEWL')

    def tearDown(self):

        """Destoy servers used in testing"""

        print

        for netid in self.networks.keys():

            self._delete_network(netid)

        if 0 >= len(self.servers):

            return

        there_are_servers_running = True

        deleted_servers = {}

        waitime = 0

        print('-> Found %s servers to delete' % len(self.servers))

        while there_are_servers_running:

            there_are_servers_running = False

            if waitime > 0:

                c = ['|', '/', '-', '\\']

                suffix = ''

                sys.stdout.write('\t. . . wait %s seconds: ' % waitime)

                for i in range(4 * waitime):

                    oldlen = len(suffix)

                    suffix = '%ss %s' % (i / 4, c[i % 4])

                    sys.stdout.write(oldlen * '\b' + suffix)

                    sys.stdout.flush()

                    time.sleep(0.25)

                oldlen = len(': ' + suffix)

                print(oldlen * '\b' + oldlen * ' ')

                sys.stdout.flush()

            waitime += 3

            dservers = self.client.list_servers(detail=True)

            for server in dservers:

                if server['name'] in self.servers.keys():

                    there_are_servers_running = True

                    sys.stdout.write('\t%s status:%s '\

                        % (server['name'], server['status']))

                    if server['status'] == 'BUILD':

                        print('\twait...')

                    else:

                        print('\tDELETE %s' % server['name'])

                        self._delete_server(server['id'])

                        self.servers.pop(server['name'])

                        deleted_servers[server['name']] = 0

                        waitime = 0

                elif server['name'] in deleted_servers.keys():

                    there_are_servers_running = True

                    sys.stdout.write('\t%s status:%s '\

                        % (server['name'], server['status']))

                    retries = deleted_servers[server['name']]

                    if retries > 10:

                        print('\tretry DELETE %s'\

                            % server['name'])

                        self._delete_server(server['id'])

                        retries = 0

                        waitime = 0

                    else:

                        print('\tnot deleted yet ...')

                    deleted_servers[server['name']] = retries + 1

    def _create_server(self, servername, flavorid, imageid, personality=None):

        server = self.client.create_server(servername,

            flavorid,

            imageid,

            personality)

        self.servers[servername] = server

        return server

    def _delete_server(self, servid):

        self.client.delete_server(servid)

    def _create_network(self, netname, **kwargs):

        net = self.client.create_network(netname, **kwargs)

        self.networks[net['id']] = net

        return net

    def _delete_network(self, netid):

        sys.stdout.write('\tDelete network %s ' % netid)

        self.client.disconnect_network_nics(netid)

        wait = 3

        while True:

            try:

                self.client.delete_network(netid)

                print('\n\tSUCCESFULL COMMIT delete network %s' % netid)

                break

            except ClientError as err:

                self.assertEqual(err.status, 421)

                time.sleep(wait)

                wait += 3

                sys.stdout.write('.')

    def if_not_all(foo):

        global TEST_ALL

        if TEST_ALL:

            return None

        return foo

    def assert_dicts_are_deeply_equal(self, d1, d2):

        for k, v in d1.items():

            self.assertTrue(k in d2)

            if isinstance(v, dict):

                self.assert_dicts_are_deeply_equal(v, d2[k])

            else:

                self.assertEqual(unicode(v), unicode(d2[k]))

    def test_000(self):

        "Prepare a full Cyclades test scenario"

        global TEST_ALL

        TEST_ALL = True

        self.server1 = self._create_server(self.servname1,

            self.flavorid,

            self.img)

        self.server2 = self._create_server(self.servname2,

            self.flavorid + 2,

            self.img)

        print('testing')

        sys.stdout.write(' test create server')

        self._test_create_server()

        print('...ok')

        sys.stdout.write(' test list servers')

        self._test_list_servers()

        print('...ok')

        print('- wait for test servers to build')

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

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

        print('- ok')

        sys.stdout.write(' test get server details')

        self._test_get_server_details()

        print('...ok')

        sys.stdout.write(' test get image details')

        self._test_get_image_details()

        print('...ok')

        sys.stdout.write(' test update_server_name')

        self._test_update_server_name()

        print('...ok')

        sys.stdout.write(' test reboot_server')

        self._test_reboot_server()

        print('...ok')

        print('- wait for test servers to boot')

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

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

        print('- ok')

        sys.stdout.write(' test create_server_metadata')

        self._test_create_server_metadata()

        print('...ok')

        sys.stdout.write(' test get_server_metadata')

        self._test_get_server_metadata()

        print('...ok')

        sys.stdout.write(' test update_server_metadata')

        self._test_update_server_metadata()

        print('...ok')

        sys.stdout.write(' test delete_server_metadata')

        self._test_delete_server_metadata()

        print('...ok')

        sys.stdout.write(' test list_flavors')

        self._test_list_flavors()

        print('...ok')

        sys.stdout.write(' test get_flavor_details')

        self._test_get_flavor_details()

        print('...ok')

        sys.stdout.write(' test list_images')

        self._test_list_images()

        print('...ok')

        sys.stdout.write(' test get_image_details')

        self._test_get_image_details()

        print('...ok')

        sys.stdout.write(' test get_image_metadata')

        self._test_get_image_metadata()

        print('...ok')

        sys.stdout.write(' test shutdown_server')

        self._test_shutdown_server()

        print('...ok')

        sys.stdout.write(' test start_server')

        self._test_start_server()

        print('...ok')

        sys.stdout.write(' test get_server_console')

        self._test_get_server_console()

        print('...ok')

        sys.stdout.write(' test get_firewall_profile')

        self._test_get_firewall_profile()

        print('...ok')

        sys.stdout.write(' test set_firewall_profile')

        self._test_set_firewall_profile()

        print('...ok')

        sys.stdout.write(' test get_server_stats')

        self._test_get_server_stats()

        print('...ok')

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

        sys.stdout.write(' test create_network')

        self._test_create_network()

        print('...ok')

        print('- wait for network to be activated')

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

        print('- ok')

        sys.stdout.write(' test connect_server')

        self._test_connect_server()

        print('...ok')

        sys.stdout.write(' test disconnect_server')

        self._test_disconnect_server()

        print('...ok')

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

        print('- wait for network to be activated')

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

        print('- ok')

        sys.stdout.write(' test list_server_nics')

        self._test_list_server_nics()

        print('...ok')

        sys.stdout.write(' test list_networks')

        self._test_list_networks()

        print('...ok')

        sys.stdout.write(' test get_network_details')

        self._test_get_network_details()

        print('...ok')

        sys.stdout.write(' test update_network_name')

        self._test_update_network_name()

        print('...ok')

        """Don't have auth for these:

        sys.stdout.write(' test delete_image')

        self._test_delete_image()

        print('...ok')

        sys.stdout.write(' test create_image_metadata')

        self._test_create_image_metadata()

        print('...ok')

        sys.stdout.write(' test update_image_metadata')

        self._test_update_image_metadata()

        print('...ok')

        sys.stdout.write(' test delete_image_metadata')

        self._test_delete_image_metadata()

        print('...ok')

        """

    def _wait_for_network(self, netid, status):

        wait = 3

        limit = 50

        c = ['|', '/', '-', '\\']

        sys.stdout.write('\t- make net %s %s  ' % (netid, status))

        while wait < limit:

            r = self.client.get_network_details(netid)

            if r['status'] == status:

                print('\tOK')

                return True

            sys.stdout.write('\tit is now %s, wait %ss  '\

                % (r['status'], wait))

            for i in range(wait * 4):

                sys.stdout.write('\b%s' % c[i % 4])

                sys.stdout.flush()

                time.sleep(0.25)

            print('\b ')

            wait += 3

        return False

    def _wait_for_nic(self, netid, servid, in_creation=True):

        self._wait_for_network(netid, 'ACTIVE')

        c = ['|', '/', '-', '\\']

        limit = 50

        wait = 3

        largetry = 0

        while wait < limit:

            nics = self.client.list_server_nics(servid)

            for net in nics:

                found_nic = net['network_id'] == netid

                if (in_creation and found_nic)\

                or not (in_creation or found_nic):

                    return True

            dis = '' if in_creation else 'dis'

            sys.stdout.write('\twait nic %s to %sconnect to %s: %ss  '\

                % (netid, dis, servid, wait))

            for i in range(wait * 4):

                sys.stdout.write('\b%s' % c[i % 4])

                sys.stdout.flush()

                time.sleep(0.25)

            print('\b ')

            wait += 3

            if wait >= limit and largetry < 3:

                wait = 3

                largetry += 1

        return False

    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 = 0

        c = ['|', '/', '-', '\\']

        while self._has_status(servid, status):

            if wait:

                sys.stdout.write('\tServer %s in %s. Wait %ss  '\

                    % (servid, status, wait))

                for i in range(4 * wait):

                    sys.stdout.write('\b%s' % c[i % 4])

                    sys.stdout.flush()

                    time.sleep(0.25)

                print('\b ')

            wait = (wait + 3) if wait < 60 else 0

    @if_not_all

    def test_list_servers(self):

        """Test list servers"""

        self.server1 = self._create_server(self.servname1,

            self.flavorid,

            self.img)

        self.server2 = self._create_server(self.servname2,

            self.flavorid + 1,

            self.img)

        self._test_list_servers()

    def _test_list_servers(self):

        servers = self.client.list_servers()

        dservers = self.client.list_servers(detail=True)

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

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

        for i in range(len(servers)):

            for field in ('created',

            'flavorRef',

            'hostId',

            'imageRef',