Statistics
| Branch: | Tag: | Revision:

root / kamaki / clients / livetest / image.py @ 2feb9477

History | View | Annotate | Download (9.3 kB)

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.astakos import AstakosClient as AstakosCachedClient
38
from kamaki.clients.cyclades import CycladesClient
39
from kamaki.clients.image import ImageClient
40
from kamaki.clients import ClientError
41

    
42

    
43
IMGMETA = set([
44
    'id', 'name', 'checksum', 'container-format', 'location', 'disk-format',
45
    'is-public', 'status', 'deleted-at', 'updated-at', 'created-at', 'owner',
46
    'size'])
47

    
48

    
49
class Image(livetest.Generic):
50
    def setUp(self):
51
        self.now = time.mktime(time.gmtime())
52
        self.cloud = 'cloud.%s' % self['testcloud']
53
        aurl, self.token = self[self.cloud, 'url'], self[self.cloud, 'token']
54
        self.auth_base = AstakosCachedClient(aurl, self.token)
55
        self.imgname = 'img_%s' % self.now
56
        url = self.auth_base.get_service_endpoints('image')['publicURL']
57
        self.token = self.auth_base.token
58
        self.client = ImageClient(url, self.token)
59
        cyclades_url = self.auth_base.get_service_endpoints(
60
            'compute')['publicURL']
61
        self.cyclades = CycladesClient(cyclades_url, self.token)
62
        self._imglist = {}
63
        self._imgdetails = {}
64

    
65
    def test_000(self):
66
        self._prepare_img()
67
        super(self.__class__, self).test_000()
68

    
69
    def _prepare_img(self):
70
        f = open(self['image', 'local_path'], 'rb')
71
        (token, uuid) = (self.token, self.auth_base.user_term('id'))
72
        purl = self.auth_base.get_service_endpoints(
73
            'object-store')['publicURL']
74
        from kamaki.clients.pithos import PithosClient
75
        self.pithcli = PithosClient(purl, token, uuid)
76
        cont = 'cont_%s' % self.now
77
        self.pithcli.container = cont
78
        self.obj = 'obj_%s' % self.now
79
        print('\t- Create container %s on Pithos server' % cont)
80
        self.pithcli.container_put()
81
        self.location = 'pithos://%s/%s/%s' % (uuid, cont, self.obj)
82
        print('\t- Upload an image at %s...\n' % self.location)
83
        self.pithcli.upload_object(self.obj, f)
84
        print('\t- ok')
85
        f.close()
86

    
87
        r = self.client.register(
88
            self.imgname,
89
            self.location,
90
            params=dict(is_public=True))
91
        self._imglist[self.imgname] = dict(
92
            name=r['name'], id=r['id'])
93
        self._imgdetails[self.imgname] = r
94

    
95
    def tearDown(self):
96
        for img in self._imglist.values():
97
            print('\tDeleting image %s' % img['id'])
98
            self.cyclades.delete_image(img['id'])
99
        if hasattr(self, 'pithcli'):
100
            print('\tDeleting container %s' % self.pithcli.container)
101
            try:
102
                self.pithcli.del_container(delimiter='/')
103
                self.pithcli.purge_container()
104
            except ClientError:
105
                pass
106

    
107
    def _get_img_by_name(self, name):
108
        r = self.cyclades.list_images()
109
        for img in r:
110
            if img['name'] == name:
111
                return img
112
        return None
113

    
114
    def test_list_public(self):
115
        """Test list_public"""
116
        self._test_list_public()
117

    
118
    def _test_list_public(self):
119
        r = self.client.list_public()
120
        r0 = self.client.list_public(order='-')
121
        self.assertTrue(len(r) > 0)
122
        for img in r:
123
            for term in (
124
                    'status',
125
                    'name',
126
                    'container_format',
127
                    'disk_format',
128
                    'id',
129
                    'size'):
130
                self.assertTrue(term in img)
131
        self.assertTrue(r, r0)
132
        r0.reverse()
133
        for i, img in enumerate(r):
134
            self.assert_dicts_are_equal(img, r0[i])
135
        r1 = self.client.list_public(detail=True)
136
        for img in r1:
137
            for term in (
138
                    'status',
139
                    'name',
140
                    'checksum',
141
                    'created_at',
142
                    'disk_format',
143
                    'updated_at',
144
                    'id',
145
                    'location',
146
                    'container_format',
147
                    'owner',
148
                    'is_public',
149
                    'deleted_at',
150
                    'properties',
151
                    'size'):
152
                self.assertTrue(term in img)
153
                if len(img['properties']):
154
                    for interm in ('osfamily', 'users', 'root_partition'):
155
                        self.assertTrue(interm in img['properties'])
156
        size_max = 1000000000
157
        r2 = self.client.list_public(filters=dict(size_max=size_max))
158
        self.assertTrue(len(r2) <= len(r))
159
        for img in r2:
160
            self.assertTrue(int(img['size']) <= size_max)
161

    
162
    def test_get_meta(self):
163
        """Test get_meta"""
164
        self._test_get_meta()
165

    
166
    def _test_get_meta(self):
167
        r = self.client.get_meta(self['image', 'id'])
168
        self.assertEqual(r['id'], self['image', 'id'])
169
        for term in (
170
                'status',
171
                'name',
172
                'checksum',
173
                'updated-at',
174
                'created-at',
175
                'deleted-at',
176
                'location',
177
                'is-public',
178
                'owner',
179
                'disk-format',
180
                'size',
181
                'container-format'):
182
            self.assertTrue(term in r)
183
            for interm in (
184
                    'KERNEL',
185
                    'OSFAMILY',
186
                    'USERS',
187
                    'GUI',
188
                    '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
            r = set(self._imgdetails[img['name']].keys())
204
            self.assertTrue(r.issubset(IMGMETA.union(['properties'])))
205

    
206
    def test_unregister(self):
207
        """Test unregister"""
208
        self._prepare_img()
209
        self._test_unregister()
210

    
211
    def _test_unregister(self):
212
        try:
213
            for img in self._imglist.values():
214
                self.client.unregister(img['id'])
215
                self._prepare_img()
216
                break
217
        except ClientError as ce:
218
            if ce.status in (405,):
219
                print 'IMAGE UNREGISTER is not supported by server: %s' % ce
220
            else:
221
                raise
222

    
223
    def test_set_members(self):
224
        """Test set_members"""
225
        self._prepare_img()
226
        self._test_set_members()
227

    
228
    def _test_set_members(self):
229
        members = ['%s@fake.net' % self.now]
230
        for img in self._imglist.values():
231
            self.client.set_members(img['id'], members)
232
            r = self.client.list_members(img['id'])
233
            self.assertEqual(r[0]['member_id'], members[0])
234

    
235
    def test_list_members(self):
236
        """Test list_members"""
237
        self._test_list_members()
238

    
239
    def _test_list_members(self):
240
        self._test_set_members()
241

    
242
    def test_remove_members(self):
243
        """Test remove_members - NO CHECK"""
244
        self._prepare_img()
245
        self._test_remove_members()
246

    
247
    def _test_remove_members(self):
248
        return
249
        members = ['%s@fake.net' % self.now, '%s_v2@fake.net' % self.now]
250
        for img in self._imglist.values():
251
            self.client.set_members(img['id'], members)
252
            r = self.client.list_members(img['id'])
253
            self.assertTrue(len(r) > 1)
254
            self.client.remove_member(img['id'], members[0])
255
            r0 = self.client.list_members(img['id'])
256
            self.assertEqual(len(r), 1 + len(r0))
257
            self.assertEqual(r0[0]['member_id'], members[1])
258

    
259
    def test_list_shared(self):
260
        """Test list_shared - NOT CHECKED"""
261
        self._test_list_shared()
262

    
263
    def _test_list_shared(self):
264
        #No way to test this, if I dont have member images
265
        pass