-
Ilias Tsitsimpis authored
All Exceptions that AstakosClient is raising should be under AstakosClientException class
6dee206f
tests.py 26.56 KiB
#!/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
"""
import sys
import socket
import simplejson
import astakosclient
from astakosclient import AstakosClient
from astakosclient.errors import \
AstakosClientException, Unauthorized, BadRequest, NotFound, \
NoUserName, NoUUID, BadValue
# Use backported unittest functionality if Python < 2.7
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
# --------------------------------------------------------------------
# Helper functions
# ----------------------------
# This functions will be used as mocked requests
def _request_offline(conn, method, url, **kwargs):
"""This request behaves as we were offline"""
raise socket.gaierror
def _request_status_302(conn, method, url, **kwargs):
"""This request returns 302"""
message = "FOUND"
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'
return (message, data, status)
def _request_status_404(conn, method, url, **kwargs):
"""This request returns 404"""
message = "Not Found"
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>'
return (message, data, status)
def _request_status_401(conn, method, url, **kwargs):
"""This request returns 401"""
message = "UNAUTHORIZED"
status = 401
data = "Invalid X-Auth-Token\n"
return (message, data, status)
def _request_status_400(conn, method, url, **kwargs):
"""This request returns 400"""
message = "BAD REQUEST"
status = 400
data = "Method not allowed.\n"
return (message, data, status)
def _request_ok(conn, method, url, **kwargs):
"""This request behaves like original Astakos does"""
if url.startswith("/im/authenticate"):
return _req_authenticate(conn, method, url, **kwargs)
elif url.startswith("/user_catalogs"):
return _req_catalogs(conn, method, url, **kwargs)
elif url.startswith("/astakos/api/resources"):
return _req_resources(conn, method, url, **kwargs)
elif url.startswith("/astakos/api/quotas"):
return _req_quotas(conn, method, url, **kwargs)
else:
return _request_status_404(conn, method, url, **kwargs)
def _req_authenticate(conn, method, url, **kwargs):
"""Check if user exists and return his profile"""
global user_1, user_2, token_1, token_2
# Check input
if conn.__class__.__name__ != "HTTPSConnection":
return _request_status_302(conn, method, url, **kwargs)
if method != "GET":
return _request_status_400(conn, method, url, **kwargs)
token = kwargs['headers'].get('X-Auth-Token')
if token == token_1:
user = dict(user_1)
elif token == token_2:
user = dict(user_2)
else:
# No user found
return _request_status_401(conn, method, url, **kwargs)
# Return
if "usage=1" not in url:
# Strip `usage' key from `user'
del user['usage']
return ("", simplejson.dumps(user), 200)
def _req_catalogs(conn, method, url, **kwargs):
"""Return user catalogs"""
global token_1, token_2, user_1, user_2
# Check input
if conn.__class__.__name__ != "HTTPSConnection":
return _request_status_302(conn, method, url, **kwargs)
if method != "POST":
return _request_status_400(conn, method, url, **kwargs)
token = kwargs['headers'].get('X-Auth-Token')
if token != token_1 and token != token_2:
return _request_status_401(conn, method, url, **kwargs)
# Return
body = simplejson.loads(kwargs['body'])
if 'uuids' in body:
# Return uuid_catalog
uuids = body['uuids']
catalogs = {}
if user_1['uuid'] in uuids:
catalogs[user_1['uuid']] = user_1['username']
if user_2['uuid'] in uuids:
catalogs[user_2['uuid']] = user_2['username']
return_catalog = {"displayname_catalog": {}, "uuid_catalog": catalogs}
elif 'displaynames' in body:
# Return displayname_catalog
names = body['displaynames']
catalogs = {}
if user_1['username'] in names:
catalogs[user_1['username']] = user_1['uuid']
if user_2['username'] in names:
catalogs[user_2['username']] = user_2['uuid']
return_catalog = {"displayname_catalog": catalogs, "uuid_catalog": {}}
else:
return_catalog = {"displayname_catalog": {}, "uuid_catalog": {}}
return ("", simplejson.dumps(return_catalog), 200)
def _req_resources(conn, method, url, **kwargs):
"""Return quota resources"""
global resources
# Check input
if conn.__class__.__name__ != "HTTPSConnection":
return _request_status_302(conn, method, url, **kwargs)
if method != "GET":
return _request_status_400(conn, method, url, **kwargs)
# Return
return ("", simplejson.dumps(resources), 200)
def _req_quotas(conn, method, url, **kwargs):
"""Return quotas for user_1"""
global token_1, quotas
# Check input
if conn.__class__.__name__ != "HTTPSConnection":
return _request_status_302(conn, method, url, **kwargs)
if method != "GET":
return _request_status_400(conn, method, url, **kwargs)
token = kwargs['headers'].get('X-Auth-Token')
if token != token_1:
return _request_status_401(conn, method, url, **kwargs)
# Return
return ("", simplejson.dumps(quotas), 200)
# ----------------------------
# Mock the actual _doRequest
def _mock_request(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
return request(conn, method, url, **kwargs)
_mock.requests = new_requests
# Replace `_doRequest' with our `_mock'
astakosclient._do_request = _mock
# ----------------------------
# Local users
token_1 = "skzleaFlBl+fasFdaf24sx=="
user_1 = \
{"username": "user1@example.com",
"auth_token_created": 1359386939000,
"name": "Example User One",
"email": ["user1@example.com"],
"auth_token_expires": 1361978939000,
"id": 108,
"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"}]}
token_2 = "fasdfDSFdf98923DF+sdfk=="
user_2 = \
{"username": "user2@example.com",
"auth_token_created": 1358386938997,
"name": "Example User Two",
"email": ["user1@example.com"],
"auth_token_expires": 1461998939000,
"id": 109,
"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"}]}
resources = {
"cyclades.vm": {
"unit": None,
"description": "Number of virtual machines",
"service": "cyclades"},
"cyclades.ram": {
"unit": "bytes",
"description": "Virtual machine memory",
"service": "cyclades"}}
quotas = {
"system": {
"cyclades.ram": {
"available": 536870912,
"limit": 1073741824,
"used": 536870912},
"cyclades.vm": {
"available": 0,
"limit": 2,
"used": 2}},
"project:1": {
"cyclades.ram": {
"available": 0,
"limit": 2147483648,
"used": 2147483648},
"cyclades.vm": {
"available": 3,
"limit": 5,
"used": 2}}}
# --------------------------------------------------------------------
# The actual tests
class TestCallAstakos(unittest.TestCase):
"""Test cases for function _callAstakos"""
# ----------------------------------
# Test the response we get if we don't have internet access
def _offline(self, pool):
global token_1
_mock_request([_request_offline])
try:
client = AstakosClient("https://example.com", use_pool=pool)
client._call_astakos(token_1, "/im/authenticate")
except AstakosClientException:
pass
else:
self.fail("Should have raised AstakosClientException")
def test_offline(self):
"""Test _offline without pool"""
self._offline(False)
def test_offline_pool(self):
"""Test _offline using pool"""
self._offline(True)
# ----------------------------------
# Test the response we get if we send invalid token
def _invalid_token(self, pool):
token = "skaksaFlBl+fasFdaf24sx=="
_mock_request([_request_ok])
try:
client = AstakosClient("https://example.com", use_pool=pool)
client._call_astakos(token, "/im/authenticate")
except Unauthorized:
pass
except Exception:
self.fail("Should have returned 401 (Invalid X-Auth-Token)")
else:
self.fail("Should have returned 401 (Invalid X-Auth-Token)")
def test_invalid_token(self):
"""Test _invalid_token without pool"""
self._invalid_token(False)
def test_invalid_token_pool(self):
"""Test _invalid_token using pool"""
self._invalid_token(True)
# ----------------------------------
# Test the response we get if we send invalid url
def _invalid_url(self, pool):
global token_1
_mock_request([_request_ok])
try:
client = AstakosClient("https://example.com", use_pool=pool)
client._call_astakos(token_1, "/im/misspelled")
except NotFound:
pass
except Exception:
self.fail("Should have returned 404 (Not Found)")
else:
self.fail("Should have returned 404 (Not Found)")
def test_invalid_url(self):
"""Test _invalid_url without pool"""
self._invalid_url(False)
def test_invalid_url_pool(self):
"""Test _invalid_url using pool"""
self._invalid_url(True)
# ----------------------------------
# Test the response we get if we use an unsupported scheme
def _unsupported_scheme(self, pool):
global token_1
_mock_request([_request_ok])
try:
client = AstakosClient("ftp://example.com", use_pool=pool)
client._call_astakos(token_1, "/im/authenticate")
except BadValue:
pass
except Exception:
self.fail("Should have raise BadValue Exception")
else:
self.fail("Should have raise BadValue Exception")
def test_unsupported_scheme(self):
"""Test _unsupported_scheme without pool"""
self._unsupported_scheme(False)
def test_unsupported_scheme_pool(self):
"""Test _unsupported_scheme using pool"""
self._unsupported_scheme(True)
# ----------------------------------
# Test the response we get if we use http instead of https
def _http_scheme(self, pool):
global token_1
_mock_request([_request_ok])
try:
client = AstakosClient("http://example.com", use_pool=pool)
client._call_astakos(token_1, "/im/authenticate")
except AstakosClientException as err:
if err.status != 302:
self.fail("Should have returned 302 (Found)")
else:
self.fail("Should have returned 302 (Found)")
def test_http_scheme(self):
"""Test _http_scheme without pool"""
self._http_scheme(False)
def test_http_scheme_pool(self):
"""Test _http_scheme using pool"""
self._http_scheme(True)
# ----------------------------------
# Test the response we get if we use authenticate with POST
def _post_authenticate(self, pool):
global token_1
_mock_request([_request_ok])
try:
client = AstakosClient("https://example.com", use_pool=pool)
client._call_astakos(token_1, "/im/authenticate", method="POST")
except BadRequest:
pass
except Exception:
self.fail("Should have returned 400 (Method not allowed)")
else:
self.fail("Should have returned 400 (Method not allowed)")
def test_post_authenticate(self):
"""Test _post_authenticate without pool"""
self._post_authenticate(False)
def test_post_authenticate_pool(self):
"""Test _post_authenticate using pool"""
self._post_authenticate(True)
# ----------------------------------
# Test the response if we request user_catalogs with GET
def _get_user_catalogs(self, pool):
global token_1
_mock_request([_request_ok])
try:
client = AstakosClient("https://example.com", use_pool=pool)
client._call_astakos(token_1, "/user_catalogs")
except BadRequest:
pass
except Exception:
self.fail("Should have returned 400 (Method not allowed)")
else:
self.fail("Should have returned 400 (Method not allowed)")
def test_get_user_catalogs(self):
"""Test _get_user_catalogs without pool"""
self._get_user_catalogs(False)
def test_get_user_catalogs_pool(self):
"""Test _get_user_catalogs using pool"""
self._get_user_catalogs(True)
class TestAuthenticate(unittest.TestCase):
"""Test cases for function getUserInfo"""
# ----------------------------------
# Test the response we get if we don't have internet access
def test_offline(self):
"""Test offline after 3 retries"""
global token_1
_mock_request([_request_offline])
try:
client = AstakosClient("https://example.com", retry=3)
client.get_user_info(token_1)
except AstakosClientException:
pass
else:
self.fail("Should have raised AstakosClientException exception")
# ----------------------------------
# Test the response we get for invalid token
def _invalid_token(self, pool):
token = "skaksaFlBl+fasFdaf24sx=="
_mock_request([_request_ok])
try:
client = AstakosClient("https://example.com", use_pool=pool)
client.get_user_info(token)
except Unauthorized:
pass
except Exception:
self.fail("Should have returned 401 (Invalid X-Auth-Token)")
else:
self.fail("Should have returned 401 (Invalid X-Auth-Token)")
def test_invalid_token(self):
"""Test _invalid_token without pool"""
self._invalid_token(False)
def test_invalid_token_pool(self):
"""Test _invalid_token using pool"""
self._invalid_token(True)
#- ---------------------------------
# Test response for user 1
def _auth_user(self, token, user_info, usage, pool):
_mock_request([_request_ok])
try:
client = AstakosClient("https://example.com", use_pool=pool)
auth_info = client.get_user_info(token, usage=usage)
except:
self.fail("Shouldn't raise an Exception")
self.assertEqual(user_info, auth_info)
def test_auth_user_one(self):
"""Test _auth_user for User 1 without pool, without usage"""
global token_1, user_1
user_info = dict(user_1)
del user_info['usage']
self._auth_user(token_1, user_info, False, False)
def test_auth_user_one_usage(self):
"""Test _auth_user for User 1 without pool, with usage"""
global token_1, user_1
self._auth_user(token_1, user_1, True, False)
def test_auth_user_one_usage_pool(self):
"""Test _auth_user for User 1 using pool, with usage"""
global token_1, user_1
self._auth_user(token_1, user_1, True, True)
def test_auth_user_two(self):
"""Test _auth_user for User 2 without pool, without usage"""
global token_2, user_2
user_info = dict(user_2)
del user_info['usage']
self._auth_user(token_2, user_info, False, False)
def test_auth_user_two_usage(self):
"""Test _auth_user for User 2 without pool, with usage"""
global token_2, user_2
self._auth_user(token_2, user_2, True, False)
def test_auth_user_two_usage_pool(self):
"""Test _auth_user for User 2 using pool, with usage"""
global token_2, user_2
self._auth_user(token_2, user_2, True, True)
# ----------------------------------
# Test retry functionality
def test_offline_retry(self):
"""Test retry functionality for getUserInfo"""
global token_1, user_1
_mock_request([_request_offline, _request_offline, _request_ok])
try:
client = AstakosClient("https://example.com", retry=2)
auth_info = client.get_user_info(token_1, usage=True)
except:
self.fail("Shouldn't raise an Exception")
self.assertEqual(user_1, auth_info)
class TestDisplayNames(unittest.TestCase):
"""Test cases for functions getDisplayNames/getDisplayName"""
# ----------------------------------
# Test the response we get for invalid token
def test_invalid_token(self):
"""Test the response we get for invalid token (without pool)"""
global user_1
token = "skaksaFlBl+fasFdaf24sx=="
_mock_request([_request_ok])
try:
client = AstakosClient("https://example.com")
client.get_usernames(token, [user_1['uuid']])
except Unauthorized:
pass
except Exception:
self.fail("Should have returned 401 (Invalid X-Auth-Token)")
else:
self.fail("Should have returned 401 (Invalid X-Auth-Token)")
# ----------------------------------
# Get Info for both users
def test_usernames(self):
"""Test get_usernames with both users"""
global token_1, user_1, user_2
_mock_request([_request_ok])
try:
client = AstakosClient("https://example.com")
catalog = client.get_usernames(
token_1, [user_1['uuid'], user_2['uuid']])
except:
self.fail("Shouldn't raise an Exception")
self.assertEqual(catalog[user_1['uuid']], user_1['username'])
self.assertEqual(catalog[user_2['uuid']], user_2['username'])
# ----------------------------------
# Get info for user 1
def test_username_user_one(self):
"""Test get_username for User One"""
global token_2, user_1
_mock_request([_request_offline, _request_ok])
try:
client = AstakosClient(
"https://example.com", use_pool=True, retry=2)
info = client.get_username(token_2, user_1['uuid'])
except:
self.fail("Shouldn't raise an Exception")
self.assertEqual(info, user_1['username'])
# ----------------------------------
# Get info with wrong uuid
def test_no_username(self):
global token_1
_mock_request([_request_ok])
try:
client = AstakosClient("https://example.com")
client.get_username(token_1, "1234")
except NoUserName:
pass
except:
self.fail("Should have raised NoDisplayName exception")
else:
self.fail("Should have raised NoDisplayName exception")
class TestGetUUIDs(unittest.TestCase):
"""Test cases for functions getUUIDs/getUUID"""
# ----------------------------------
# Test the response we get for invalid token
def test_invalid_token(self):
"""Test the response we get for invalid token (using pool)"""
global user_1
token = "skaksaFlBl+fasFdaf24sx=="
_mock_request([_request_ok])
try:
client = AstakosClient("https://example.com")
client.get_uuids(token, [user_1['username']])
except Unauthorized:
pass
except Exception:
self.fail("Should have returned 401 (Invalid X-Auth-Token)")
else:
self.fail("Should have returned 401 (Invalid X-Auth-Token)")
# ----------------------------------
# Get info for both users
def test_uuids(self):
"""Test get_uuids with both users"""
global token_1, user_1, user_2
_mock_request([_request_ok])
try:
client = AstakosClient("https://example.com")
catalog = client.get_uuids(
token_1, [user_1['username'], user_2['username']])
except:
self.fail("Shouldn't raise an Exception")
self.assertEqual(catalog[user_1['username']], user_1['uuid'])
self.assertEqual(catalog[user_2['username']], user_2['uuid'])
# ----------------------------------
# Get uuid for user 2
def test_get_uuid_user_two(self):
"""Test get_uuid for User Two"""
global token_1, user_2
_mock_request([_request_offline, _request_ok])
try:
client = AstakosClient("https://example.com", retry=1)
info = client.get_uuid(token_2, user_1['username'])
except:
self.fail("Shouldn't raise an Exception")
self.assertEqual(info, user_1['uuid'])
# ----------------------------------
# Get uuid with wrong username
def test_no_uuid(self):
global token_1
_mock_request([_request_ok])
try:
client = AstakosClient("https://example.com")
client.get_uuid(token_1, "1234")
except NoUUID:
pass
except:
self.fail("Should have raised NoUUID exception")
else:
self.fail("Should have raised NoUUID exception")
class TestResources(unittest.TestCase):
"""Test cases for function get_resources"""
# ----------------------------------
def test_get_resources(self):
"""Test function call of get_resources"""
global resources
_mock_request([_request_offline, _request_ok])
try:
client = AstakosClient("https://example.com", retry=1)
result = client.get_resources()
except Exception as err:
self.fail("Shouldn't raise Exception %s" % err)
self.assertEqual(resources, result)
class TestQuotas(unittest.TestCase):
"""Test cases for function get_quotas"""
# ----------------------------------
def test_get_quotas(self):
"""Test function call of get_quotas"""
global quotas, token_1
_mock_request([_request_ok])
try:
client = AstakosClient("https://example.com")
result = client.get_quotas(token_1)
except Exception as err:
self.fail("Shouldn't raise Exception %s" % err)
self.assertEqual(quotas, result)
# -----------------------------------
def test_get_quotas_unauthorized(self):
"""Test function call of get_quotas with wrong token"""
global token_2
_mock_request([_request_ok])
try:
client = AstakosClient("https://example.com")
client.get_quotas(token_2)
except Unauthorized:
pass
except Exception as err:
self.fail("Shouldn't raise Exception %s" % err)
else:
self.fail("Should have raised Unauthorized Exception")
# ----------------------------------
def test_get_quotas_without_token(self):
_mock_request([_request_ok])
try:
client = AstakosClient("https://example.com")
client.get_quotas(None)
except Unauthorized:
pass
except Exception as err:
self.fail("Shouldn't raise Exception %s" % err)
else:
self.fail("Should have raised Unauthorized Exception")
# ----------------------------
# Run tests
if __name__ == "__main__":
unittest.main()