Revision 45c0bcf8 astakosclient/astakosclient/tests.py
b/astakosclient/astakosclient/tests.py | ||
---|---|---|
42 | 42 |
|
43 | 43 |
import re |
44 | 44 |
import sys |
45 |
import socket |
|
46 | 45 |
import simplejson |
47 |
from mock import patch |
|
48 |
from contextlib import contextmanager |
|
49 | 46 |
|
50 | 47 |
import astakosclient |
51 | 48 |
from astakosclient import AstakosClient |
... | ... | |
68 | 65 |
auth_url = "https://example.org/identity/v2.0" |
69 | 66 |
account_prefix = "/account_prefix" |
70 | 67 |
ui_prefix = "/ui_prefix" |
71 |
api_authenticate = join_urls(account_prefix, "authenticate")
|
|
68 |
api_tokens = "/identity/v2.0/tokens"
|
|
72 | 69 |
api_usercatalogs = join_urls(account_prefix, "user_catalogs") |
73 | 70 |
api_resources = join_urls(account_prefix, "resources") |
74 | 71 |
api_quotas = join_urls(account_prefix, "quotas") |
... | ... | |
76 | 73 |
|
77 | 74 |
# -------------------------------------- |
78 | 75 |
# Local users |
79 |
token_1 = "skzleaFlBl+fasFdaf24sx" |
|
80 |
user_1 = \ |
|
81 |
{"username": "user1@example.com", |
|
82 |
"name": "Example User One", |
|
83 |
"email": ["user1@example.com"], |
|
84 |
"uuid": "73917abc-abcd-477e-a1f1-1763abcdefab"} |
|
76 |
token = { |
|
77 |
'id': "skzleaFlBl+fasFdaf24sx", |
|
78 |
'tenant': { |
|
79 |
'id': "73917abc-abcd-477e-a1f1-1763abcdefab", |
|
80 |
'name': "Example User One", |
|
81 |
}, |
|
82 |
} |
|
83 |
|
|
84 |
user = { |
|
85 |
'id': "73917abc-abcd-477e-a1f1-1763abcdefab", |
|
86 |
'name': "Example User One", |
|
87 |
'roles': [{u'id': 1, u'name': u'default'}, |
|
88 |
{u'id': 5, u'name': u'academic-login-users'}], |
|
89 |
'roles_links': [] |
|
90 |
} |
|
85 | 91 |
|
86 | 92 |
resources = { |
87 | 93 |
"cyclades.ram": { |
... | ... | |
103 | 109 |
} |
104 | 110 |
} |
105 | 111 |
|
112 |
endpoints_with_info = dict(endpoints) |
|
113 |
endpoints_with_info['access']['token'] = dict(token) |
|
114 |
endpoints_with_info['access']['user'] = dict(user) |
|
115 |
|
|
106 | 116 |
quotas = { |
107 | 117 |
"system": { |
108 | 118 |
"cyclades.ram": { |
... | ... | |
197 | 207 |
|
198 | 208 |
# ---------------------------- |
199 | 209 |
# These functions will be used as mocked requests |
200 |
def _request_offline(conn, method, url, **kwargs): |
|
201 |
"""This request behaves as we were offline""" |
|
202 |
raise socket.gaierror |
|
203 |
|
|
204 |
|
|
205 | 210 |
def _request_status_302(conn, method, url, **kwargs): |
206 | 211 |
"""This request returns 302""" |
207 | 212 |
message = "FOUND" |
... | ... | |
242 | 247 |
return (message, data, status) |
243 | 248 |
|
244 | 249 |
|
245 |
def _request_ok(conn, method, url, **kwargs):
|
|
250 |
def _mock_request(conn, method, url, **kwargs):
|
|
246 | 251 |
"""This request behaves like original Astakos does""" |
247 |
if api_authenticate == url:
|
|
248 |
return _req_authenticate(conn, method, url, **kwargs)
|
|
252 |
if api_tokens == url:
|
|
253 |
return _req_tokens(conn, method, url, **kwargs)
|
|
249 | 254 |
elif api_usercatalogs == url: |
250 | 255 |
return _req_catalogs(conn, method, url, **kwargs) |
251 | 256 |
elif api_resources == url: |
... | ... | |
258 | 263 |
return _request_status_404(conn, method, url, **kwargs) |
259 | 264 |
|
260 | 265 |
|
261 |
def _req_authenticate(conn, method, url, **kwargs):
|
|
262 |
"""Check if user exists and return his profile"""
|
|
263 |
global user_1, token_1
|
|
266 |
def _req_tokens(conn, method, url, **kwargs):
|
|
267 |
"""Return endpoints"""
|
|
268 |
global token, user, endpoints
|
|
264 | 269 |
|
265 | 270 |
# Check input |
266 | 271 |
if conn.__class__.__name__ != "HTTPSConnection": |
267 | 272 |
return _request_status_302(conn, method, url, **kwargs) |
268 |
if method != "GET":
|
|
273 |
if method != "POST":
|
|
269 | 274 |
return _request_status_400(conn, method, url, **kwargs) |
270 |
token = kwargs['headers'].get('X-Auth-Token') |
|
271 |
if token == token_1: |
|
272 |
user = dict(user_1) |
|
273 |
return ("", simplejson.dumps(user), 200) |
|
274 |
else: |
|
275 |
# No user found |
|
275 |
req_token = kwargs['headers'].get('X-Auth-Token') |
|
276 |
if req_token != token['id']: |
|
276 | 277 |
return _request_status_401(conn, method, url, **kwargs) |
277 | 278 |
|
279 |
if 'body' in kwargs: |
|
280 |
# Return endpoints with authenticate info |
|
281 |
return ("", simplejson.dumps(endpoints_with_info), 200) |
|
282 |
else: |
|
283 |
# Return endpoints without authenticate info |
|
284 |
return ("", simplejson.dumps(endpoints), 200) |
|
285 |
|
|
278 | 286 |
|
279 | 287 |
def _req_catalogs(conn, method, url, **kwargs): |
280 | 288 |
"""Return user catalogs""" |
281 |
global token_1, token_2, user_1, user_2
|
|
289 |
global token, user
|
|
282 | 290 |
|
283 | 291 |
# Check input |
284 | 292 |
if conn.__class__.__name__ != "HTTPSConnection": |
285 | 293 |
return _request_status_302(conn, method, url, **kwargs) |
286 | 294 |
if method != "POST": |
287 | 295 |
return _request_status_400(conn, method, url, **kwargs) |
288 |
token = kwargs['headers'].get('X-Auth-Token') |
|
289 |
if token != token_1:
|
|
296 |
req_token = kwargs['headers'].get('X-Auth-Token')
|
|
297 |
if req_token != token['id']:
|
|
290 | 298 |
return _request_status_401(conn, method, url, **kwargs) |
291 | 299 |
|
292 | 300 |
# Return |
... | ... | |
295 | 303 |
# Return uuid_catalog |
296 | 304 |
uuids = body['uuids'] |
297 | 305 |
catalogs = {} |
298 |
if user_1['uuid'] in uuids:
|
|
299 |
catalogs[user_1['uuid']] = user_1['username']
|
|
306 |
if user['id'] in uuids:
|
|
307 |
catalogs[user['id']] = user['name']
|
|
300 | 308 |
return_catalog = {"displayname_catalog": {}, "uuid_catalog": catalogs} |
301 | 309 |
elif 'displaynames' in body: |
302 | 310 |
# Return displayname_catalog |
303 | 311 |
names = body['displaynames'] |
304 | 312 |
catalogs = {} |
305 |
if user_1['username'] in names:
|
|
306 |
catalogs[user_1['username']] = user_1['uuid']
|
|
313 |
if user['name'] in names:
|
|
314 |
catalogs[user['name']] = user['id']
|
|
307 | 315 |
return_catalog = {"displayname_catalog": catalogs, "uuid_catalog": {}} |
308 | 316 |
else: |
309 | 317 |
return_catalog = {"displayname_catalog": {}, "uuid_catalog": {}} |
... | ... | |
326 | 334 |
|
327 | 335 |
def _req_quotas(conn, method, url, **kwargs): |
328 | 336 |
"""Return quotas for user_1""" |
329 |
global token_1, quotas
|
|
337 |
global token, quotas |
|
330 | 338 |
|
331 | 339 |
# Check input |
332 | 340 |
if conn.__class__.__name__ != "HTTPSConnection": |
333 | 341 |
return _request_status_302(conn, method, url, **kwargs) |
334 | 342 |
if method != "GET": |
335 | 343 |
return _request_status_400(conn, method, url, **kwargs) |
336 |
token = kwargs['headers'].get('X-Auth-Token') |
|
337 |
if token != token_1:
|
|
344 |
req_token = kwargs['headers'].get('X-Auth-Token')
|
|
345 |
if req_token != token['id']:
|
|
338 | 346 |
return _request_status_401(conn, method, url, **kwargs) |
339 | 347 |
|
340 | 348 |
# Return |
... | ... | |
343 | 351 |
|
344 | 352 |
def _req_commission(conn, method, url, **kwargs): |
345 | 353 |
"""Perform a commission for user_1""" |
346 |
global token_1, pending_commissions, \
|
|
354 |
global token, pending_commissions, \ |
|
347 | 355 |
commission_successful_response, commission_failure_response |
348 | 356 |
|
349 | 357 |
# Check input |
350 | 358 |
if conn.__class__.__name__ != "HTTPSConnection": |
351 | 359 |
return _request_status_302(conn, method, url, **kwargs) |
352 |
token = kwargs['headers'].get('X-Auth-Token') |
|
353 |
if token != token_1:
|
|
360 |
req_token = kwargs['headers'].get('X-Auth-Token')
|
|
361 |
if req_token != token['id']:
|
|
354 | 362 |
return _request_status_401(conn, method, url, **kwargs) |
355 | 363 |
|
356 | 364 |
if method == "POST": |
... | ... | |
399 | 407 |
return _request_status_400(conn, method, url, **kwargs) |
400 | 408 |
|
401 | 409 |
|
402 |
# ---------------------------- |
|
403 |
# Mock the actual _doRequest |
|
404 |
def _mock_request(new_requests): |
|
405 |
"""Mock the actual request |
|
406 |
|
|
407 |
Given a list of requests to use (in rotation), |
|
408 |
replace the original _doRequest function with |
|
409 |
a new one |
|
410 |
|
|
411 |
""" |
|
412 |
def _mock(conn, method, url, **kwargs): |
|
413 |
# Get first request |
|
414 |
request = _mock.requests[0] |
|
415 |
# Rotate requests |
|
416 |
_mock.requests = _mock.requests[1:] + _mock.requests[:1] |
|
417 |
# Use first request |
|
418 |
return request(conn, method, url, **kwargs) |
|
419 |
|
|
420 |
_mock.requests = new_requests |
|
421 |
# Replace `_doRequest' with our `_mock' |
|
422 |
astakosclient._do_request = _mock |
|
423 |
|
|
424 |
|
|
425 |
# -------------------------------------- |
|
426 |
# Mock the get_endpoints method |
|
427 |
@contextmanager |
|
428 |
def patch_astakosclient(new_requests): |
|
429 |
_mock_request(new_requests) |
|
430 |
with patch('astakosclient.AstakosClient.get_endpoints') as patcher: |
|
431 |
patcher.return_value = endpoints |
|
432 |
yield |
|
433 |
|
|
434 |
|
|
435 | 410 |
# -------------------------------------------------------------------- |
436 | 411 |
# The actual tests |
437 | 412 |
|
438 | 413 |
class TestCallAstakos(unittest.TestCase): |
439 | 414 |
"""Test cases for function _callAstakos""" |
440 | 415 |
|
441 |
# ---------------------------------- |
|
442 |
# Test the response we get if we don't have internet access |
|
443 |
def _offline(self, pool): |
|
444 |
global token_1, auth_url |
|
445 |
_mock_request([_request_offline]) |
|
446 |
try: |
|
447 |
client = AstakosClient(token_1, auth_url, use_pool=pool) |
|
448 |
client._call_astakos("offline") |
|
449 |
except AstakosClientException: |
|
450 |
pass |
|
451 |
else: |
|
452 |
self.fail("Should have raised AstakosClientException") |
|
453 |
|
|
454 |
def test_offline(self): |
|
455 |
"""Test _offline without pool""" |
|
456 |
self._offline(False) |
|
457 |
|
|
458 |
def test_offline_pool(self): |
|
459 |
"""Test _offline using pool""" |
|
460 |
self._offline(True) |
|
416 |
# Patch astakosclient's _do_request function |
|
417 |
def setUp(self): |
|
418 |
astakosclient._do_request = _mock_request |
|
461 | 419 |
|
462 | 420 |
# ---------------------------------- |
463 | 421 |
# Test the response we get if we send invalid token |
... | ... | |
465 | 423 |
global auth_url |
466 | 424 |
token = "skaksaFlBl+fasFdaf24sx" |
467 | 425 |
try: |
468 |
with patch_astakosclient([_request_ok]): |
|
469 |
client = AstakosClient(token, auth_url, use_pool=pool) |
|
470 |
client.get_user_info() |
|
426 |
client = AstakosClient(token, auth_url, use_pool=pool) |
|
427 |
client.authenticate() |
|
471 | 428 |
except Unauthorized: |
472 | 429 |
pass |
473 | 430 |
except Exception: |
... | ... | |
486 | 443 |
# ---------------------------------- |
487 | 444 |
# Test the response we get if we send invalid url |
488 | 445 |
def _invalid_url(self, pool): |
489 |
global token_1, auth_url
|
|
446 |
global token, auth_url |
|
490 | 447 |
try: |
491 |
with patch_astakosclient([_request_ok]): |
|
492 |
client = AstakosClient(token_1, auth_url, use_pool=pool) |
|
493 |
client._call_astakos("/astakos/api/misspelled") |
|
448 |
client = AstakosClient(token['id'], auth_url, use_pool=pool) |
|
449 |
client._call_astakos("/astakos/api/misspelled") |
|
494 | 450 |
except NotFound: |
495 | 451 |
pass |
496 | 452 |
except Exception, e: |
... | ... | |
509 | 465 |
# ---------------------------------- |
510 | 466 |
# Test the response we get if we use an unsupported scheme |
511 | 467 |
def _unsupported_scheme(self, pool): |
512 |
global token_1, auth_url
|
|
468 |
global token, auth_url |
|
513 | 469 |
try: |
514 |
with patch_astakosclient([_request_ok]): |
|
515 |
client = AstakosClient(token_1, "ftp://example.com", |
|
516 |
use_pool=pool) |
|
517 |
client.get_user_info() |
|
470 |
client = AstakosClient( |
|
471 |
token['id'], "ftp://example.com", use_pool=pool) |
|
472 |
client.authenticate() |
|
518 | 473 |
except BadValue: |
519 | 474 |
pass |
520 | 475 |
except Exception: |
... | ... | |
533 | 488 |
# ---------------------------------- |
534 | 489 |
# Test the response we get if we use http instead of https |
535 | 490 |
def _http_scheme(self, pool): |
536 |
global token_1
|
|
491 |
global token |
|
537 | 492 |
http_auth_url = "http://example.org/identity/v2.0" |
538 | 493 |
try: |
539 |
with patch_astakosclient([_request_ok]): |
|
540 |
client = AstakosClient(token_1, http_auth_url, use_pool=pool) |
|
541 |
client.get_user_info() |
|
494 |
client = AstakosClient(token['id'], http_auth_url, use_pool=pool) |
|
495 |
client.authenticate() |
|
542 | 496 |
except AstakosClientException as err: |
543 | 497 |
if err.status != 302: |
544 | 498 |
self.fail("Should have returned 302 (Found)") |
... | ... | |
554 | 508 |
self._http_scheme(True) |
555 | 509 |
|
556 | 510 |
# ---------------------------------- |
557 |
# Test the response we get if we use authenticate with POST
|
|
558 |
def _post_authenticate(self, pool):
|
|
559 |
global token_1, auth_url
|
|
511 |
# Test the response we get if we use authenticate with GET
|
|
512 |
def _get_authenticate(self, pool):
|
|
513 |
global token, auth_url |
|
560 | 514 |
try: |
561 |
with patch_astakosclient([_request_ok]): |
|
562 |
client = AstakosClient(token_1, auth_url, use_pool=pool) |
|
563 |
client._call_astakos(api_authenticate, method="POST") |
|
515 |
client = AstakosClient(token['id'], auth_url, use_pool=pool) |
|
516 |
client._call_astakos(api_tokens, method="GET") |
|
564 | 517 |
except BadRequest: |
565 | 518 |
pass |
566 | 519 |
except Exception: |
... | ... | |
568 | 521 |
else: |
569 | 522 |
self.fail("Should have returned 400 (Method not allowed)") |
570 | 523 |
|
571 |
def test_post_authenticate(self):
|
|
572 |
"""Test _post_authenticate without pool"""
|
|
573 |
self._post_authenticate(False)
|
|
524 |
def test_get_authenticate(self):
|
|
525 |
"""Test _get_authenticate without pool"""
|
|
526 |
self._get_authenticate(False)
|
|
574 | 527 |
|
575 |
def test_post_authenticate_pool(self):
|
|
576 |
"""Test _post_authenticate using pool"""
|
|
577 |
self._post_authenticate(True)
|
|
528 |
def test_get_authenticate_pool(self):
|
|
529 |
"""Test _get_authenticate using pool"""
|
|
530 |
self._get_authenticate(True)
|
|
578 | 531 |
|
579 | 532 |
# ---------------------------------- |
580 | 533 |
# Test the response if we request user_catalogs with GET |
581 | 534 |
def _get_user_catalogs(self, pool): |
582 |
global token_1, auth_url, api_usercatalogs
|
|
535 |
global token, auth_url, api_usercatalogs |
|
583 | 536 |
try: |
584 |
with patch_astakosclient([_request_ok]): |
|
585 |
client = AstakosClient(token_1, auth_url, use_pool=pool) |
|
586 |
client._call_astakos(api_usercatalogs) |
|
537 |
client = AstakosClient(token['id'], auth_url, use_pool=pool) |
|
538 |
client._call_astakos(api_usercatalogs) |
|
587 | 539 |
except BadRequest: |
588 | 540 |
pass |
589 | 541 |
except Exception: |
... | ... | |
603 | 555 |
class TestAuthenticate(unittest.TestCase): |
604 | 556 |
"""Test cases for function getUserInfo""" |
605 | 557 |
|
606 |
# ---------------------------------- |
|
607 |
# Test the response we get if we don't have internet access |
|
608 |
def test_offline(self): |
|
609 |
"""Test offline after 3 retries""" |
|
610 |
global token_1, auth_url |
|
611 |
try: |
|
612 |
with patch_astakosclient([_request_offline]): |
|
613 |
client = AstakosClient(token_1, auth_url, retry=3) |
|
614 |
client.get_user_info() |
|
615 |
except AstakosClientException: |
|
616 |
pass |
|
617 |
else: |
|
618 |
self.fail("Should have raised AstakosClientException exception") |
|
558 |
# Patch astakosclient's _do_request function |
|
559 |
def setUp(self): |
|
560 |
astakosclient._do_request = _mock_request |
|
619 | 561 |
|
620 | 562 |
# ---------------------------------- |
621 | 563 |
# Test the response we get for invalid token |
... | ... | |
623 | 565 |
global auth_url |
624 | 566 |
token = "skaksaFlBl+fasFdaf24sx" |
625 | 567 |
try: |
626 |
with patch_astakosclient([_request_ok]): |
|
627 |
client = AstakosClient(token, auth_url, use_pool=pool) |
|
628 |
client.get_user_info() |
|
568 |
client = AstakosClient(token, auth_url, use_pool=pool) |
|
569 |
client.authenticate() |
|
629 | 570 |
except Unauthorized: |
630 | 571 |
pass |
631 | 572 |
except Exception: |
... | ... | |
641 | 582 |
"""Test _invalid_token using pool""" |
642 | 583 |
self._invalid_token(True) |
643 | 584 |
|
644 |
#- ---------------------------------
|
|
585 |
# ----------------------------------
|
|
645 | 586 |
# Test response for user |
646 |
def _auth_user(self, token, user_info, pool):
|
|
647 |
global auth_url |
|
587 |
def _auth_user(self, pool): |
|
588 |
global token, endpoints_with_info, auth_url
|
|
648 | 589 |
try: |
649 |
with patch_astakosclient([_request_ok]): |
|
650 |
client = AstakosClient(token, auth_url, use_pool=pool) |
|
651 |
auth_info = client.get_user_info() |
|
590 |
client = AstakosClient(token['id'], auth_url, use_pool=pool) |
|
591 |
auth_info = client.authenticate() |
|
652 | 592 |
except: |
653 | 593 |
self.fail("Shouldn't raise an Exception") |
654 |
self.assertEqual(user_info, auth_info)
|
|
594 |
self.assertEqual(endpoints_with_info, auth_info)
|
|
655 | 595 |
|
656 | 596 |
def test_auth_user(self): |
657 | 597 |
"""Test _auth_user without pool""" |
658 |
global token_1, user_1 |
|
659 |
user_info = dict(user_1) |
|
660 |
self._auth_user(token_1, user_info, False) |
|
598 |
self._auth_user(False) |
|
661 | 599 |
|
662 | 600 |
def test_auth_user_pool(self): |
663 | 601 |
"""Test _auth_user for User 1 using pool, with usage""" |
664 |
global token_1, user_1 |
|
665 |
self._auth_user(token_1, user_1, True) |
|
666 |
|
|
667 |
# ---------------------------------- |
|
668 |
# Test retry functionality |
|
669 |
def test_offline_retry(self): |
|
670 |
"""Test retry functionality for getUserInfo""" |
|
671 |
global token_1, user_1, auth_url |
|
672 |
_mock_request([_request_offline, _request_offline, _request_ok]) |
|
673 |
try: |
|
674 |
with patch_astakosclient([_request_offline, _request_offline, |
|
675 |
_request_ok]): |
|
676 |
client = AstakosClient(token_1, auth_url, retry=2) |
|
677 |
auth_info = client.get_user_info() |
|
678 |
except Exception, e: |
|
679 |
self.fail("Shouldn't raise an Exception \"%s\"" % e) |
|
680 |
self.assertEqual(user_1, auth_info) |
|
602 |
self._auth_user(True) |
|
681 | 603 |
|
682 | 604 |
|
683 | 605 |
class TestDisplayNames(unittest.TestCase): |
684 | 606 |
"""Test cases for functions getDisplayNames/getDisplayName""" |
685 | 607 |
|
608 |
# Patch astakosclient's _do_request function |
|
609 |
def setUp(self): |
|
610 |
astakosclient._do_request = _mock_request |
|
611 |
|
|
686 | 612 |
# ---------------------------------- |
687 | 613 |
# Test the response we get for invalid token |
688 | 614 |
def test_invalid_token(self): |
689 | 615 |
"""Test the response we get for invalid token (without pool)""" |
690 |
global user_1, auth_url
|
|
616 |
global auth_url |
|
691 | 617 |
token = "skaksaFlBl+fasFdaf24sx" |
692 | 618 |
try: |
693 |
with patch_astakosclient([_request_ok]): |
|
694 |
client = AstakosClient(token, auth_url) |
|
695 |
client.get_usernames([user_1['uuid']]) |
|
619 |
client = AstakosClient(token, auth_url) |
|
620 |
client.get_usernames(["12412351"]) |
|
696 | 621 |
except Unauthorized: |
697 | 622 |
pass |
698 | 623 |
except Exception: |
... | ... | |
701 | 626 |
self.fail("Should have returned 401 (Invalid X-Auth-Token)") |
702 | 627 |
|
703 | 628 |
# ---------------------------------- |
704 |
# Get info for user 1 |
|
705 |
def test_username_user_one(self): |
|
706 |
"""Test get_username for User One""" |
|
707 |
global token_1, user_1, auth_url |
|
708 |
try: |
|
709 |
with patch_astakosclient([_request_offline, _request_ok]): |
|
710 |
client = AstakosClient(token_1, auth_url, |
|
711 |
use_pool=True, retry=2) |
|
712 |
info = client.get_username(user_1['uuid']) |
|
713 |
except: |
|
714 |
self.fail("Shouldn't raise an Exception") |
|
715 |
self.assertEqual(info, user_1['username']) |
|
629 |
# Get username |
|
630 |
def test_username(self): |
|
631 |
"""Test get_username""" |
|
632 |
global token, user, auth_url |
|
633 |
try: |
|
634 |
client = AstakosClient(token['id'], auth_url, |
|
635 |
use_pool=False, retry=2) |
|
636 |
info = client.get_username(user['id']) |
|
637 |
except Exception, e: |
|
638 |
self.fail("Shouldn't raise an Exception: %s" % e) |
|
639 |
self.assertEqual(info, user['name']) |
|
716 | 640 |
|
717 | 641 |
# ---------------------------------- |
718 | 642 |
# Get info with wrong uuid |
719 | 643 |
def test_no_username(self): |
720 |
global token_1, auth_url
|
|
644 |
global token, auth_url |
|
721 | 645 |
try: |
722 |
with patch_astakosclient([_request_ok]): |
|
723 |
client = AstakosClient(token_1, auth_url) |
|
724 |
client.get_username("1234") |
|
646 |
client = AstakosClient(token['id'], auth_url) |
|
647 |
client.get_username("1234") |
|
725 | 648 |
except NoUserName: |
726 | 649 |
pass |
727 | 650 |
except: |
... | ... | |
733 | 656 |
class TestGetUUIDs(unittest.TestCase): |
734 | 657 |
"""Test cases for functions getUUIDs/getUUID""" |
735 | 658 |
|
659 |
# Patch astakosclient's _do_request function |
|
660 |
def setUp(self): |
|
661 |
astakosclient._do_request = _mock_request |
|
662 |
|
|
736 | 663 |
# ---------------------------------- |
737 | 664 |
# Test the response we get for invalid token |
738 | 665 |
def test_invalid_token(self): |
739 | 666 |
"""Test the response we get for invalid token (using pool)""" |
740 |
global user_1, auth_url
|
|
667 |
global user, auth_url |
|
741 | 668 |
token = "skaksaFlBl+fasFdaf24sx" |
742 | 669 |
try: |
743 |
with patch_astakosclient([_request_ok]): |
|
744 |
client = AstakosClient(token, auth_url) |
|
745 |
client.get_uuids([user_1['username']]) |
|
670 |
client = AstakosClient(token, auth_url) |
|
671 |
client.get_uuids([user['name']]) |
|
746 | 672 |
except Unauthorized: |
747 | 673 |
pass |
748 | 674 |
except Exception: |
... | ... | |
751 | 677 |
self.fail("Should have returned 401 (Invalid X-Auth-Token)") |
752 | 678 |
|
753 | 679 |
# ---------------------------------- |
754 |
# Get uuid for user 1
|
|
755 |
def test_get_uuid_user_two(self):
|
|
756 |
"""Test get_uuid for User Two"""
|
|
757 |
global token_1, user_1, auth_url
|
|
680 |
# Get uuid |
|
681 |
def test_get_uuid(self): |
|
682 |
"""Test get_uuid""" |
|
683 |
global token, user, auth_url
|
|
758 | 684 |
try: |
759 |
with patch_astakosclient([_request_ok]): |
|
760 |
client = AstakosClient(token_1, auth_url, retry=1) |
|
761 |
catalog = client.get_uuids([user_1['username']]) |
|
685 |
client = AstakosClient(token['id'], auth_url, retry=1) |
|
686 |
catalog = client.get_uuids([user['name']]) |
|
762 | 687 |
except: |
763 | 688 |
self.fail("Shouldn't raise an Exception") |
764 |
self.assertEqual(catalog[user_1['username']], user_1['uuid'])
|
|
689 |
self.assertEqual(catalog[user['name']], user['id'])
|
|
765 | 690 |
|
766 | 691 |
# ---------------------------------- |
767 | 692 |
# Get uuid with wrong username |
768 | 693 |
def test_no_uuid(self): |
769 |
global token_1, auth_url
|
|
694 |
global token, auth_url |
|
770 | 695 |
try: |
771 |
with patch_astakosclient([_request_ok]): |
|
772 |
client = AstakosClient(token_1, auth_url) |
|
773 |
client.get_uuid("1234") |
|
696 |
client = AstakosClient(token['id'], auth_url) |
|
697 |
client.get_uuid("1234") |
|
774 | 698 |
except NoUUID: |
775 | 699 |
pass |
776 | 700 |
except: |
... | ... | |
782 | 706 |
class TestResources(unittest.TestCase): |
783 | 707 |
"""Test cases for function get_resources""" |
784 | 708 |
|
709 |
# Patch astakosclient's _do_request function |
|
710 |
def setUp(self): |
|
711 |
astakosclient._do_request = _mock_request |
|
712 |
|
|
785 | 713 |
# ---------------------------------- |
786 | 714 |
def test_get_resources(self): |
787 | 715 |
"""Test function call of get_resources""" |
788 |
global resources, auth_url, token_1
|
|
716 |
global resources, auth_url, token |
|
789 | 717 |
try: |
790 |
with patch_astakosclient([_request_offline, _request_ok]): |
|
791 |
client = AstakosClient(token_1, auth_url, retry=1) |
|
792 |
result = client.get_resources() |
|
718 |
client = AstakosClient(token['id'], auth_url, retry=1) |
|
719 |
result = client.get_resources() |
|
793 | 720 |
except Exception as err: |
794 | 721 |
self.fail("Shouldn't raise Exception %s" % err) |
795 | 722 |
self.assertEqual(resources, result) |
... | ... | |
798 | 725 |
class TestQuotas(unittest.TestCase): |
799 | 726 |
"""Test cases for function get_quotas""" |
800 | 727 |
|
728 |
# Patch astakosclient's _do_request function |
|
729 |
def setUp(self): |
|
730 |
astakosclient._do_request = _mock_request |
|
731 |
|
|
801 | 732 |
# ---------------------------------- |
802 | 733 |
def test_get_quotas(self): |
803 | 734 |
"""Test function call of get_quotas""" |
804 |
global quotas, token_1, auth_url
|
|
735 |
global quotas, token, auth_url |
|
805 | 736 |
try: |
806 |
with patch_astakosclient([_request_ok]): |
|
807 |
client = AstakosClient(token_1, auth_url) |
|
808 |
result = client.get_quotas() |
|
737 |
client = AstakosClient(token['id'], auth_url) |
|
738 |
result = client.get_quotas() |
|
809 | 739 |
except Exception as err: |
810 | 740 |
self.fail("Shouldn't raise Exception %s" % err) |
811 | 741 |
self.assertEqual(quotas, result) |
... | ... | |
816 | 746 |
global auth_url |
817 | 747 |
token = "buahfhsda" |
818 | 748 |
try: |
819 |
with patch_astakosclient([_request_ok]): |
|
820 |
client = AstakosClient(token, auth_url) |
|
821 |
client.get_quotas() |
|
749 |
client = AstakosClient(token, auth_url) |
|
750 |
client.get_quotas() |
|
822 | 751 |
except Unauthorized: |
823 | 752 |
pass |
824 | 753 |
except Exception as err: |
... | ... | |
830 | 759 |
class TestCommissions(unittest.TestCase): |
831 | 760 |
"""Test cases for quota commissions""" |
832 | 761 |
|
762 |
# Patch astakosclient's _do_request function |
|
763 |
def setUp(self): |
|
764 |
astakosclient._do_request = _mock_request |
|
765 |
|
|
833 | 766 |
# ---------------------------------- |
834 | 767 |
def test_issue_commission(self): |
835 | 768 |
"""Test function call of issue_commission""" |
836 |
global token_1, commission_request, commission_successful_reqsponse
|
|
769 |
global token, commission_request, commission_successful_reqsponse |
|
837 | 770 |
global auth_url |
838 | 771 |
try: |
839 |
with patch_astakosclient([_request_ok]): |
|
840 |
client = AstakosClient(token_1, auth_url) |
|
841 |
response = client.issue_commission(commission_request) |
|
772 |
client = AstakosClient(token['id'], auth_url) |
|
773 |
response = client.issue_commission(commission_request) |
|
842 | 774 |
except Exception as err: |
843 | 775 |
self.fail("Shouldn't raise Exception %s" % err) |
844 | 776 |
self.assertEqual(response, commission_successful_response['serial']) |
... | ... | |
846 | 778 |
# ---------------------------------- |
847 | 779 |
def test_issue_commission_quota_limit(self): |
848 | 780 |
"""Test function call of issue_commission with limit exceeded""" |
849 |
global token_1, commission_request, commission_failure_response
|
|
781 |
global token, commission_request, commission_failure_response |
|
850 | 782 |
global auth_url |
851 | 783 |
new_request = dict(commission_request) |
852 | 784 |
new_request['provisions'][1]['quantity'] = 520000000 |
853 | 785 |
try: |
854 |
with patch_astakosclient([_request_ok]): |
|
855 |
client = AstakosClient(token_1, auth_url) |
|
856 |
client.issue_commission(new_request) |
|
786 |
client = AstakosClient(token['id'], auth_url) |
|
787 |
client.issue_commission(new_request) |
|
857 | 788 |
except QuotaLimit: |
858 | 789 |
pass |
859 | 790 |
except Exception as err: |
... | ... | |
864 | 795 |
# ---------------------------------- |
865 | 796 |
def test_issue_one_commission(self): |
866 | 797 |
"""Test function call of issue_one_commission""" |
867 |
global token_1, commission_successful_response, auth_url
|
|
798 |
global token, commission_successful_response, auth_url |
|
868 | 799 |
try: |
869 |
with patch_astakosclient([_request_ok]): |
|
870 |
client = AstakosClient(token_1, auth_url) |
|
871 |
response = client.issue_one_commission( |
|
872 |
"c02f315b-7d84-45bc-a383-552a3f97d2ad", |
|
873 |
"system", {"cyclades.vm": 1, "cyclades.ram": 30000}) |
|
800 |
client = AstakosClient(token['id'], auth_url) |
|
801 |
response = client.issue_one_commission( |
|
802 |
"c02f315b-7d84-45bc-a383-552a3f97d2ad", |
|
803 |
"system", {"cyclades.vm": 1, "cyclades.ram": 30000}) |
|
874 | 804 |
except Exception as err: |
875 | 805 |
self.fail("Shouldn't have raised Exception %s" % err) |
876 | 806 |
self.assertEqual(response, commission_successful_response['serial']) |
... | ... | |
878 | 808 |
# ---------------------------------- |
879 | 809 |
def test_get_pending_commissions(self): |
880 | 810 |
"""Test function call of get_pending_commissions""" |
881 |
global token_1, pending_commissions, auth_url
|
|
811 |
global token, pending_commissions, auth_url |
|
882 | 812 |
try: |
883 |
with patch_astakosclient([_request_ok]): |
|
884 |
client = AstakosClient(token_1, auth_url) |
|
885 |
response = client.get_pending_commissions() |
|
813 |
client = AstakosClient(token['id'], auth_url) |
|
814 |
response = client.get_pending_commissions() |
|
886 | 815 |
except Exception as err: |
887 | 816 |
self.fail("Shouldn't raise Exception %s" % err) |
888 | 817 |
self.assertEqual(response, pending_commissions) |
... | ... | |
890 | 819 |
# ---------------------------------- |
891 | 820 |
def test_get_commission_info(self): |
892 | 821 |
"""Test function call of get_commission_info""" |
893 |
global token_1, commission_description, auth_url
|
|
822 |
global token, commission_description, auth_url |
|
894 | 823 |
try: |
895 |
with patch_astakosclient([_request_ok]): |
|
896 |
client = AstakosClient(token_1, auth_url, |
|
897 |
use_pool=True, pool_size=2) |
|
898 |
response = client.get_commission_info(57) |
|
824 |
client = AstakosClient(token['id'], auth_url, |
|
825 |
use_pool=True, pool_size=2) |
|
826 |
response = client.get_commission_info(57) |
|
899 | 827 |
except Exception as err: |
900 | 828 |
self.fail("Shouldn't raise Exception %s" % err) |
901 | 829 |
self.assertEqual(response, commission_description) |
... | ... | |
903 | 831 |
# ---------------------------------- |
904 | 832 |
def test_get_commission_info_not_found(self): |
905 | 833 |
"""Test function call of get_commission_info with invalid serial""" |
906 |
global token_1, auth_url |
|
907 |
_mock_request([_request_ok]) |
|
834 |
global token, auth_url |
|
908 | 835 |
try: |
909 |
with patch_astakosclient([_request_ok]): |
|
910 |
client = AstakosClient(token_1, auth_url) |
|
911 |
client.get_commission_info("57lala") |
|
836 |
client = AstakosClient(token['id'], auth_url) |
|
837 |
client.get_commission_info("57lala") |
|
912 | 838 |
except NotFound: |
913 | 839 |
pass |
914 | 840 |
except Exception as err: |
... | ... | |
919 | 845 |
# ---------------------------------- |
920 | 846 |
def test_get_commission_info_without_serial(self): |
921 | 847 |
"""Test function call of get_commission_info without serial""" |
922 |
global token_1, auth_url
|
|
848 |
global token, auth_url |
|
923 | 849 |
try: |
924 |
with patch_astakosclient([_request_ok]): |
|
925 |
client = AstakosClient(token_1, auth_url) |
|
926 |
client.get_commission_info(None) |
|
850 |
client = AstakosClient(token['id'], auth_url) |
|
851 |
client.get_commission_info(None) |
|
927 | 852 |
except BadValue: |
928 | 853 |
pass |
929 | 854 |
except Exception as err: |
... | ... | |
934 | 859 |
# ---------------------------------- |
935 | 860 |
def test_commision_action(self): |
936 | 861 |
"""Test function call of commision_action with wrong action""" |
937 |
global token_1, auth_url
|
|
862 |
global token, auth_url |
|
938 | 863 |
try: |
939 |
with patch_astakosclient([_request_ok]): |
|
940 |
client = AstakosClient(token_1, auth_url) |
|
941 |
client.commission_action(57, "lala") |
|
864 |
client = AstakosClient(token['id'], auth_url) |
|
865 |
client.commission_action(57, "lala") |
|
942 | 866 |
except BadRequest: |
943 | 867 |
pass |
944 | 868 |
except Exception as err: |
... | ... | |
949 | 873 |
# ---------------------------------- |
950 | 874 |
def test_accept_commission(self): |
951 | 875 |
"""Test function call of accept_commission""" |
952 |
global token_1, auth_url
|
|
876 |
global token, auth_url |
|
953 | 877 |
try: |
954 |
with patch_astakosclient([_request_ok]): |
|
955 |
client = AstakosClient(token_1, auth_url) |
|
956 |
client.accept_commission(57) |
|
878 |
client = AstakosClient(token['id'], auth_url) |
|
879 |
client.accept_commission(57) |
|
957 | 880 |
except Exception as err: |
958 | 881 |
self.fail("Shouldn't raise Exception %s" % err) |
959 | 882 |
|
960 | 883 |
# ---------------------------------- |
961 | 884 |
def test_reject_commission(self): |
962 | 885 |
"""Test function call of reject_commission""" |
963 |
global token_1, auth_url
|
|
886 |
global token, auth_url |
|
964 | 887 |
try: |
965 |
with patch_astakosclient([_request_ok]): |
|
966 |
client = AstakosClient(token_1, auth_url) |
|
967 |
client.reject_commission(57) |
|
888 |
client = AstakosClient(token['id'], auth_url) |
|
889 |
client.reject_commission(57) |
|
968 | 890 |
except Exception as err: |
969 | 891 |
self.fail("Shouldn't raise Exception %s" % err) |
970 | 892 |
|
971 | 893 |
# ---------------------------------- |
972 | 894 |
def test_accept_commission_not_found(self): |
973 | 895 |
"""Test function call of accept_commission with wrong serial""" |
974 |
global token_1, auth_url
|
|
896 |
global token, auth_url |
|
975 | 897 |
try: |
976 |
with patch_astakosclient([_request_ok]): |
|
977 |
client = AstakosClient(token_1, auth_url) |
|
978 |
client.reject_commission(20) |
|
898 |
client = AstakosClient(token['id'], auth_url) |
|
899 |
client.reject_commission(20) |
|
979 | 900 |
except NotFound: |
980 | 901 |
pass |
981 | 902 |
except Exception as err: |
... | ... | |
986 | 907 |
# ---------------------------------- |
987 | 908 |
def test_resolve_commissions(self): |
988 | 909 |
"""Test function call of resolve_commissions""" |
989 |
global token_1, auth_url
|
|
910 |
global token, auth_url |
|
990 | 911 |
try: |
991 |
with patch_astakosclient([_request_ok]): |
|
992 |
client = AstakosClient(token_1, auth_url) |
|
993 |
result = client.resolve_commissions([56, 57], [56, 58, 59]) |
|
912 |
client = AstakosClient(token['id'], auth_url) |
|
913 |
result = client.resolve_commissions([56, 57], [56, 58, 59]) |
|
994 | 914 |
except Exception as err: |
995 | 915 |
self.fail("Shouldn't raise Exception %s" % err) |
996 | 916 |
self.assertEqual(result, resolve_commissions_rep) |
... | ... | |
999 | 919 |
# ---------------------------- |
1000 | 920 |
# Run tests |
1001 | 921 |
if __name__ == "__main__": |
1002 |
unittest.main() |
|
922 |
unittest.main() |
Also available in: Unified diff