code.grnet.gr Git - kamaki/blob - kamaki/clients/livetest/image.py

   1 # Copyright 2012-2013 GRNET S.A. All rights reserved.
   2 #
   3 # Redistribution and use in source and binary forms, with or
   4 # without modification, are permitted provided that the following
   5 # conditions are met:
   6 #
   7 #   1. Redistributions of source code must retain the above
   8 #      copyright notice, this list of conditions and the following
   9 #      disclaimer.
  10 #
  11 #   2. Redistributions in binary form must reproduce the above
  12 #      copyright notice, this list of conditions and the following
  13 #      disclaimer in the documentation and/or other materials
  14 #      provided with the distribution.
  15 #
  16 # THIS SOFTWARE IS PROVIDED BY GRNET S.A. ``AS IS'' AND ANY EXPRESS
  17 # OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
  18 # WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR
  19 # PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL GRNET S.A OR
  20 # CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
  21 # SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
  22 # LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF
  23 # USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED
  24 # AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT
  25 # LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN
  26 # ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
  27 # POSSIBILITY OF SUCH DAMAGE.
  28 #
  29 # The views and conclusions contained in the software and
  30 # documentation are those of the authors and should not be
  31 # interpreted as representing official policies, either expressed
  32 # or implied, of GRNET S.A.
  33
  34 import time
  35
  36 from kamaki.clients import livetest
  37 from kamaki.clients.cyclades import CycladesClient
  38 from kamaki.clients.image import ImageClient
  39 from kamaki.clients import ClientError
  40
  41
  42 class Image(livetest.Generic):
  43     def setUp(self):
  44         self.now = time.mktime(time.gmtime())
  45
  46         self.imgname = 'img_%s' % self.now
  47         url = self['image', 'url']
  48         self.client = ImageClient(url, self['token'])
  49         cyclades_url = self['compute', 'url']
  50         self.cyclades = CycladesClient(cyclades_url, self['token'])
  51         self._imglist = {}
  52
  53     def test_000(self):
  54         self._prepare_img()
  55         super(self.__class__, self).test_000()
  56
  57     def _prepare_img(self):
  58         f = open(self['image', 'local_path'], 'rb')
  59         (token, uuid) = (self['token'], self['store', 'account'])
  60         if not uuid:
  61             from kamaki.clients.astakos import AstakosClient
  62             uuid = AstakosClient(self['astakos', 'url'], token).term('uuid')
  63         from kamaki.clients.pithos import PithosClient
  64         self.pithcli = PithosClient(self['store', 'url'], token, uuid)
  65         cont = 'cont_%s' % self.now
  66         self.pithcli.container = cont
  67         self.obj = 'obj_%s' % self.now
  68         print('\t- Create container %s on Pithos server' % cont)
  69         self.pithcli.container_put()
  70         self.location = 'pithos://%s/%s/%s' % (uuid, cont, self.obj)
  71         print('\t- Upload an image at %s...' % self.location)
  72         self.pithcli.upload_object(self.obj, f)
  73         print('\t- ok')
  74         f.close()
  75
  76         self.client.register(
  77             self.imgname,
  78             self.location,
  79             params=dict(is_public=True))
  80         img = self._get_img_by_name(self.imgname)
  81         self._imglist[self.imgname] = img
  82
  83     def tearDown(self):
  84         for img in self._imglist.values():
  85             print('\tDeleting image %s' % img['id'])
  86             self.cyclades.delete_image(img['id'])
  87         if hasattr(self, 'pithcli'):
  88             print('\tDeleting container %s' % self.pithcli.container)
  89             try:
  90                 self.pithcli.del_container(delimiter='/')
  91                 self.pithcli.purge_container()
  92             except ClientError:
  93                 pass
  94
  95     def _get_img_by_name(self, name):
  96         r = self.cyclades.list_images()
  97         for img in r:
  98             if img['name'] == name:
  99                 return img
 100         return None
 101
 102     def assert_dicts_are_deeply_equal(self, d1, d2):
 103         for k, v in d1.items():
 104             self.assertTrue(k in d2)
 105             if isinstance(v, dict):
 106                 self.assert_dicts_are_deeply_equal(v, d2[k])
 107             else:
 108                 self.assertEqual(unicode(v), unicode(d2[k]))
 109
 110     def test_list_public(self):
 111         """Test list_public"""
 112         self._test_list_public()
 113
 114     def _test_list_public(self):
 115         r = self.client.list_public()
 116         r0 = self.client.list_public(order='-')
 117         self.assertTrue(len(r) > 0)
 118         for img in r:
 119             for term in (
 120                     'status',
 121                     'name',
 122                     'container_format',
 123                     'disk_format',
 124                     'id',
 125                     'size'):
 126                 self.assertTrue(term in img)
 127         self.assertTrue(r, r0)
 128         r0.reverse()
 129         for i, img in enumerate(r):
 130             self.assert_dicts_are_deeply_equal(img, r0[i])
 131         r1 = self.client.list_public(detail=True)
 132         for img in r1:
 133             for term in (
 134                     'status',
 135                     'name',
 136                     'checksum',
 137                     'created_at',
 138                     'disk_format',
 139                     'updated_at',
 140                     'id',
 141                     'location',
 142                     'container_format',
 143                     'owner',
 144                     'is_public',
 145                     'deleted_at',
 146                     'properties',
 147                     'size'):
 148                 self.assertTrue(term in img)
 149                 if img['properties']:
 150                     for interm in (
 151                             'osfamily',
 152                             'users',
 153                             'os',
 154                             'root_partition',
 155                             'description'):
 156                         self.assertTrue(interm in img['properties'])
 157         size_max = 1000000000
 158         r2 = self.client.list_public(filters=dict(size_max=size_max))
 159         self.assertTrue(len(r2) <= len(r))
 160         for img in r2:
 161             self.assertTrue(int(img['size']) <= size_max)
 162
 163     def test_get_meta(self):
 164         """Test get_meta"""
 165         self._test_get_meta()
 166
 167     def _test_get_meta(self):
 168         r = self.client.get_meta(self['image', 'id'])
 169         self.assertEqual(r['id'], self['image', 'id'])
 170         for term in (
 171                 'status',
 172                 'name',
 173                 'checksum',
 174                 'updated-at',
 175                 'created-at',
 176                 'deleted-at',
 177                 'location',
 178                 'is-public',
 179                 'owner',
 180                 'disk-format',
 181                 'size',
 182                 'container-format'):
 183             self.assertTrue(term in r)
 184             for interm in (
 185                     'kernel',
 186                     'osfamily',
 187                     'users',
 188                     'gui', 'sortorder',
 189                     'root-partition',
 190                     'os',
 191                     'description'):
 192                 self.assertTrue(interm in r['properties'])
 193
 194     def test_register(self):
 195         """Test register"""
 196         self._prepare_img()
 197         self._test_register()
 198
 199     def _test_register(self):
 200         self.assertTrue(self._imglist)
 201         for img in self._imglist.values():
 202             self.assertTrue(img is not None)
 203
 204     def test_reregister(self):
 205         """Test reregister"""
 206         self._prepare_img()
 207         self._test_reregister()
 208
 209     def _test_reregister(self):
 210         self.client.reregister(
 211             self.location,
 212             properties=dict(my_property='some_value'))
 213
 214     def test_set_members(self):
 215         """Test set_members"""
 216         self._prepare_img()
 217         self._test_set_members()
 218
 219     def _test_set_members(self):
 220         members = ['%s@fake.net' % self.now]
 221         for img in self._imglist.values():
 222             self.client.set_members(img['id'], members)
 223             r = self.client.list_members(img['id'])
 224             self.assertEqual(r[0]['member_id'], members[0])
 225
 226     def test_list_members(self):
 227         """Test list_members"""
 228         self._test_list_members()
 229
 230     def _test_list_members(self):
 231         self._test_set_members()
 232
 233     def test_remove_members(self):
 234         """Test remove_members - NO CHECK"""
 235         self._prepare_img()
 236         self._test_remove_members()
 237
 238     def _test_remove_members(self):
 239         return
 240         members = ['%s@fake.net' % self.now, '%s_v2@fake.net' % self.now]
 241         for img in self._imglist.values():
 242             self.client.set_members(img['id'], members)
 243             r = self.client.list_members(img['id'])
 244             self.assertTrue(len(r) > 1)
 245             self.client.remove_member(img['id'], members[0])
 246             r0 = self.client.list_members(img['id'])
 247             self.assertEqual(len(r), 1 + len(r0))
 248             self.assertEqual(r0[0]['member_id'], members[1])
 249
 250     def test_list_shared(self):
 251         """Test list_shared - NOT CHECKED"""
 252         self._test_list_shared()
 253
 254     def _test_list_shared(self):
 255         #No way to test this, if I dont have member images
 256         pass