tests.py 16.8 KB
Newer Older
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
#!/usr/bin/env python
#
# Copyright (C) 2012, 2013 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.

"""Unit Tests for the astakos-client module

Provides unit tests for the code implementing
the astakos client library

"""

43
import sys
44
import socket
45
import simplejson
46
47

import astakosclient
48
from astakosclient import AstakosClient, AstakosClientException
49
50

# Use backported unittest functionality if Python < 2.7
51
52
53
54
55
56
try:
    import unittest2 as unittest
except ImportError:
    if sys.version_info < (2, 7):
        raise Exception("The unittest2 package is required for Python < 2.7")
    import unittest
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74


# --------------------------------------------------------------------
# Helper functions

# ----------------------------
# This functions will be used as mocked requests
def _requestOffline(conn, method, url, **kwargs):
    """This request behaves as we were offline"""
    raise socket.gaierror


def _requestStatus302(conn, method, url, **kwargs):
    """This request returns 302"""
    status = 302
    data = '<html>\r\n<head><title>302 Found</title></head>\r\n' \
        '<body bgcolor="white">\r\n<center><h1>302 Found</h1></center>\r\n' \
        '<hr><center>nginx/0.7.67</center>\r\n</body>\r\n</html>\r\n'
75
    return (data, status)
76
77
78
79
80
81
82
83
84


def _requestStatus404(conn, method, url, **kwargs):
    """This request returns 404"""
    status = 404
    data = '<html><head><title>404 Not Found</title></head>' \
        '<body><h1>Not Found</h1><p>The requested URL /foo was ' \
        'not found on this server.</p><hr><address>Apache Server ' \
        'at example.com Port 80</address></body></html>'
85
    return (data, status)
86
87
88
89
90
91


def _requestStatus401(conn, method, url, **kwargs):
    """This request returns 401"""
    status = 401
    data = "Invalid X-Auth-Token\n"
92
    return (data, status)
93
94
95
96
97
98


def _requestStatus400(conn, method, url, **kwargs):
    """This request returns 400"""
    status = 400
    data = "Method not allowed.\n"
99
    return (data, status)
100
101
102
103


def _requestOk(conn, method, url, **kwargs):
    """This request behaves like original Astakos does"""
104
    if url[0:16] == "/im/authenticate":
105
        return _reqAuthenticate(conn, method, url, **kwargs)
106
    elif url[0:14] == "/user_catalogs":
107
108
109
110
111
112
113
        return _reqCatalogs(conn, method, url, **kwargs)
    else:
        return _requestStatus404(conn, method, url, **kwargs)


def _reqAuthenticate(conn, method, url, **kwargs):
    """Check if user exists and return his profile"""
114
    global user_1, user_2
115
116
117
118
119
120
121
122
123

    if conn.__class__.__name__ != "HTTPSConnection":
        return _requestStatus302(conn, method, url, **kwargs)

    if method != "GET":
        return _requestStatus400(conn, method, url, **kwargs)

    token = kwargs['headers']['X-Auth-Token']
    if token == token_1:
124
        user = dict(user_1)
125
    elif token == token_2:
126
        user = dict(user_2)
127
128
129
130
    else:
        # No user found
        return _requestStatus401(conn, method, url, **kwargs)

131
    if "usage=1" not in url:
132
        # Strip `usage' key from `user'
133
        del user['usage']
134
    return (simplejson.dumps(user), 200)
135
136
137
138


def _reqCatalogs(conn, method, url, **kwargs):
    """Return user catalogs"""
139
140
141
142
143
144
145
146
147
    global token_1, token_2, user_1, user_2

    if conn.__class__.__name__ != "HTTPSConnection":
        return _requestStatus302(conn, method, url, **kwargs)

    if method != "POST":
        return _requestStatus400(conn, method, url, **kwargs)

    # XXX:
148
    return ({}, 200)
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166


# ----------------------------
# Mock the actual _doRequest
def _mockRequest(new_requests):
    """Mock the actual request

    Given a list of requests to use (in rotation),
    replace the original _doRequest function with
    a new one

    """
    def _mock(conn, method, url, **kwargs):
        # Get first request
        request = _mock.requests[0]
        # Rotate requests
        _mock.requests = _mock.requests[1:] + _mock.requests[:1]
        # Use first request
167
        return request(conn, method, url, **kwargs)
168
169
170
171
172
173
174
175
176

    _mock.requests = new_requests
    # Replace `_doRequest' with our `_mock'
    astakosclient._doRequest = _mock


# ----------------------------
# Local users
token_1 = "skzleaFlBl+fasFdaf24sx=="
177
user_1 = \
178
179
180
181
182
183
    {"username": "user1@example.com",
     "auth_token_created": 1359386939000,
     "name": "Example User One",
     "email": ["user1@example.com"],
     "auth_token_expires": 1361978939000,
     "id": 108,
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
     "uuid": "73917abc-abcd-477e-a1f1-1763abcdefab",
     "usage": [
         {"currValue": 42949672960,
          "display_name": "System Disk",
          "name": "cyclades.disk"},
         {"currValue": 4,
          "display_name": "CPU",
          "name": "cyclades.cpu"},
         {"currValue": 4294967296,
          "display_name": "RAM",
          "name": "cyclades.ram"},
         {"currValue": 3,
          "display_name": "VM",
          "name": "cyclades.vm"},
         {"currValue": 0,
          "display_name": "private network",
          "name": "cyclades.network.private"},
         {"currValue": 152,
          "display_name": "Storage Space",
          "name": "pithos+.diskspace"}]}
204
205

token_2 = "fasdfDSFdf98923DF+sdfk=="
206
user_2 = \
207
208
209
210
211
212
    {"username": "user2@example.com",
     "auth_token_created": 1358386938997,
     "name": "Example User Two",
     "email": ["user1@example.com"],
     "auth_token_expires": 1461998939000,
     "id": 109,
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
     "uuid": "73917bca-1234-5678-a1f1-1763abcdefab",
     "usage": [
         {"currValue": 68719476736,
          "display_name": "System Disk",
          "name": "cyclades.disk"},
         {"currValue": 1,
          "display_name": "CPU",
          "name": "cyclades.cpu"},
         {"currValue": 1073741824,
          "display_name": "RAM",
          "name": "cyclades.ram"},
         {"currValue": 2,
          "display_name": "VM",
          "name": "cyclades.vm"},
         {"currValue": 1,
          "display_name": "private network",
          "name": "cyclades.network.private"},
         {"currValue": 2341634510,
          "display_name": "Storage Space",
          "name": "pithos+.diskspace"}]}
233
234
235
236
237
238
239
240


# --------------------------------------------------------------------
# The actual tests

class TestCallAstakos(unittest.TestCase):
    """Test cases for function _callAstakos"""

241
242
243
    # ----------------------------------
    # Test the response we get if we don't have internet access
    def _offline(self, pool):
244
245
246
        global token_1
        _mockRequest([_requestOffline])
        try:
247
248
249
250
            client = AstakosClient(
                token_1, "https://example.com", use_pool=pool)
            client._callAstakos("/im/authenticate")
        except AstakosClientException:
251
252
            pass
        else:
253
            self.fail("Should have raised AstakosClientException")
254

255
256
257
258
    def test_Offline(self):
        """Test _offline without pool"""
        self._offline(False)

259
    def test_OfflinePool(self):
260
261
262
263
264
265
266
267
268
        """Test _offline using pool"""
        self._offline(True)

    # ----------------------------------
    # Test the response we get if we send invalid token
    def _invalidToken(self, pool):
        token = "skaksaFlBl+fasFdaf24sx=="
        _mockRequest([_requestOk])
        try:
269
270
271
272
273
            client = AstakosClient(
                token, "https://example.com", use_pool=pool)
            client._callAstakos("/im/authenticate")
        except AstakosClientException as err:
            if err.status != 401:
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
                self.fail("Should have returned 401 (Invalid X-Auth-Token)")
        else:
            self.fail("Should have returned 401 (Invalid X-Auth-Token)")

    def test_InvalidToken(self):
        """Test _invalidToken without pool"""
        self._invalidToken(False)

    def test_InvalidTokenPool(self):
        """Test _invalidToken using pool"""
        self._invalidToken(True)

    # ----------------------------------
    # Test the response we get if we send invalid url
    def _invalidUrl(self, pool):
289
        global token_1
290
        _mockRequest([_requestOk])
291
        try:
292
293
294
295
296
            client = AstakosClient(
                token_1, "https://example.com", use_pool=pool)
            client._callAstakos("/im/misspelled")
        except AstakosClientException as err:
            if err.status != 404:
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
                self.fail("Should have returned 404 (Not Found)")
        else:
            self.fail("Should have returned 404 (Not Found)")

    def test_InvalidUrl(self):
        """Test _invalidUrl without pool"""
        self._invalidUrl(False)

    def test_invalidUrlPool(self):
        """Test _invalidUrl using pool"""
        self._invalidUrl(True)

    # ----------------------------------
    # Test the response we get if we use an unsupported scheme
    def _unsupportedScheme(self, pool):
        global token_1
        _mockRequest([_requestOk])
        try:
315
316
317
            client = AstakosClient(
                token_1, "ftp://example.com", use_pool=pool)
            client._callAstakos("/im/authenticate")
318
        except ValueError:
319
320
            pass
        except Exception:
321
            self.fail("Should have raise ValueError Exception")
322
        else:
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
            self.fail("Should have raise ValueError Exception")

    def test_UnsupportedScheme(self):
        """Test _unsupportedScheme without pool"""
        self._unsupportedScheme(False)

    def test_UnsupportedSchemePool(self):
        """Test _unsupportedScheme using pool"""
        self._unsupportedScheme(True)

    # ----------------------------------
    # Test the response we get if we use http instead of https
    def _httpScheme(self, pool):
        global token_1
        _mockRequest([_requestOk])
        try:
339
340
341
342
343
            client = AstakosClient(
                token_1, "http://example.com", use_pool=pool)
            client._callAstakos("/im/authenticate")
        except AstakosClientException as err:
            if err.status != 302:
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
                self.fail("Should have returned 302 (Found)")
        else:
            self.fail("Should have returned 302 (Found)")

    def test_HttpScheme(self):
        """Test _httpScheme without pool"""
        self._httpScheme(False)

    def test_HttpSchemePool(self):
        """Test _httpScheme using pool"""
        self._httpScheme(True)

    # ----------------------------------
    # Test the response we get if we use authenticate with POST
    def _postAuthenticate(self, pool):
        global token_1
        _mockRequest([_requestOk])
        try:
362
363
364
365
366
            client = AstakosClient(
                token_1, "https://example.com", use_pool=pool)
            client._callAstakos("/im/authenticate", method="POST")
        except AstakosClientException as err:
            if err.status != 400:
367
368
369
370
371
372
373
374
375
376
377
                self.fail("Should have returned 400 (Method not allowed)")
        else:
            self.fail("Should have returned 400 (Method not allowed)")

    def test_PostAuthenticate(self):
        """Test _postAuthenticate without pool"""
        self._postAuthenticate(False)

    def test_PostAuthenticatePool(self):
        """Test _postAuthenticate using pool"""
        self._postAuthenticate(True)
378

379
380
381
382
383
384
    # ----------------------------------
    # Test the response if we request user_catalogs with GET
    def _getUserCatalogs(self, pool):
        global token_1
        _mockRequest([_requestOk])
        try:
385
386
387
388
389
            client = AstakosClient(
                token_1, "https://example.com", use_pool=pool)
            client._callAstakos("/user_catalogs")
        except AstakosClientException as err:
            if err.status != 400:
390
391
392
393
394
395
396
397
398
399
400
401
                self.fail("Should have returned 400 (Method not allowed)")
        else:
            self.fail("Should have returned 400 (Method not allowed)")

    def test_GetUserCatalogs(self):
        """Test _getUserCatalogs without pool"""
        self._getUserCatalogs(False)

    def test_GetUserCatalogsPool(self):
        """Test _getUserCatalogs using pool"""
        self._getUserCatalogs(True)

402

403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
class TestAuthenticate(unittest.TestCase):
    """Test cases for function authenticate"""

    # ----------------------------------
    # Test the response we get if we don't have internet access
    def test_Offline(self):
        """Test offline after 3 retries"""
        global token_1
        _mockRequest([_requestOffline])
        try:
            client = AstakosClient(
                token_1, "https://example.com", retry=3)
            client.authenticate()
        except AstakosClientException:
            pass
        else:
            self.fail("Should have raised AstakosClientException exception")

    # ----------------------------------
    # Test the response we get for invalid token
    def _invalidToken(self, pool):
        token = "skaksaFlBl+fasFdaf24sx=="
        _mockRequest([_requestOk])
        try:
            client = AstakosClient(
                token, "https://example.com", use_pool=pool)
            client.authenticate()
        except AstakosClientException as err:
            if err.status != 401:
                self.fail("Should have returned 401 (Invalide X-Auth-Token)")
        else:
            self.fail("Should have returned 401 (Invalide X-Auth-Token)")

    def test_InvalidToken(self):
        """Test _invalidToken without pool"""
        self._invalidToken(False)

    def test_InvalidTokenPool(self):
        """Test _invalidToken using pool"""
        self._invalidToken(True)

    #- ---------------------------------
    # Test response for user 1
    def _authUser(self, token, user_info, usage, pool):
        _mockRequest([_requestOk])
        try:
            client = AstakosClient(
                token, "https://example.com", use_pool=pool)
            auth_info = client.authenticate(usage=usage)
        except:
            self.fail("Shouldn't raise an Exception")
        self.assertEqual(user_info, auth_info)

    def test_AuthUserOne(self):
        """Test _authUser for User 1 without pool, without usage"""
        global token_1, user_1
        user_info = dict(user_1)
        del user_info['usage']
        self._authUser(token_1, user_info, False, False)

    def test_AuthUserOneUsage(self):
        """Test _authUser for User 1 without pool, with usage"""
        global token_1, user_1
        self._authUser(token_1, user_1, True, False)

    def test_AuthUserOneUsagePool(self):
        """Test _authUser for User 1 using pool, with usage"""
        global token_1, user_1
        self._authUser(token_1, user_1, True, True)

    def test_AuthUserTwo(self):
        """Test _authUser for User 2 without pool, without usage"""
        global token_2, user_2
        user_info = dict(user_2)
        del user_info['usage']
        self._authUser(token_2, user_info, False, False)

    def test_AuthUserTwoUsage(self):
        """Test _authUser for User 2 without pool, with usage"""
        global token_2, user_2
        self._authUser(token_2, user_2, True, False)

    def test_AuthUserTwoUsagePool(self):
        """Test _authUser for User 2 using pool, with usage"""
        global token_2, user_2
        self._authUser(token_2, user_2, True, True)

    # ----------------------------------
    # Test retry functionality
    def test_OfflineRetry(self):
        """Test retry functionality for authentication"""
        global token_1, user_1
        _mockRequest([_requestOffline, _requestOffline, _requestOk])
        try:
            client = AstakosClient(
                token_1, "https://example.com", retry=2)
            auth_info = client.authenticate(usage=True)
        except:
            self.fail("Shouldn't raise an Exception")
        self.assertEqual(user_1, auth_info)
503
504


505
506
507
508
# ----------------------------
# Run tests
if __name__ == "__main__":
    unittest.main()