Commit e75c9698 authored by Ilias Tsitsimpis's avatar Ilias Tsitsimpis Committed by Christos Stavrakakis
Browse files

Correct unit tests to run with AstakosClient class

parent fcc0cf60
......@@ -45,6 +45,7 @@ import socket
import simplejson
import astakosclient
from astakosclient import AstakosClient, AstakosClientException
# Use backported unittest functionality if Python < 2.7
try:
......@@ -71,7 +72,7 @@ def _requestStatus302(conn, method, url, **kwargs):
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 (status, data)
return (data, status)
def _requestStatus404(conn, method, url, **kwargs):
......@@ -81,28 +82,28 @@ def _requestStatus404(conn, method, url, **kwargs):
'<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 (status, data)
return (data, status)
def _requestStatus401(conn, method, url, **kwargs):
"""This request returns 401"""
status = 401
data = "Invalid X-Auth-Token\n"
return (status, data)
return (data, status)
def _requestStatus400(conn, method, url, **kwargs):
"""This request returns 400"""
status = 400
data = "Method not allowed.\n"
return (status, data)
return (data, status)
def _requestOk(conn, method, url, **kwargs):
"""This request behaves like original Astakos does"""
if url[0:17] == "/im/authenticate?":
if url[0:16] == "/im/authenticate":
return _reqAuthenticate(conn, method, url, **kwargs)
elif url[0:15] == "/user_catalogs?":
elif url[0:14] == "/user_catalogs":
return _reqCatalogs(conn, method, url, **kwargs)
else:
return _requestStatus404(conn, method, url, **kwargs)
......@@ -130,7 +131,7 @@ def _reqAuthenticate(conn, method, url, **kwargs):
if "usage=1" not in url:
# Strip `usage' key from `user'
del user['usage']
return (200, simplejson.dumps(user))
return (simplejson.dumps(user), 200)
def _reqCatalogs(conn, method, url, **kwargs):
......@@ -144,7 +145,7 @@ def _reqCatalogs(conn, method, url, **kwargs):
return _requestStatus400(conn, method, url, **kwargs)
# XXX:
return (200, {})
return ({}, 200)
# ----------------------------
......@@ -243,14 +244,13 @@ class TestCallAstakos(unittest.TestCase):
global token_1
_mockRequest([_requestOffline])
try:
astakosclient._callAstakos(
token_1, "https://example.com/im/authenticate", use_pool=pool)
except socket.error:
client = AstakosClient(
token_1, "https://example.com", use_pool=pool)
client._callAstakos("/im/authenticate")
except AstakosClientException:
pass
except Exception:
self.fail("Should have raised socket exception")
else:
self.fail("Shouldn't succeed")
self.fail("Should have raised AstakosClientException")
def test_Offline(self):
"""Test _offline without pool"""
......@@ -266,10 +266,11 @@ class TestCallAstakos(unittest.TestCase):
token = "skaksaFlBl+fasFdaf24sx=="
_mockRequest([_requestOk])
try:
astakosclient._callAstakos(
token, "https://example.com/im/authenticate", use_pool=pool)
except Exception as (status, data):
if status != 401:
client = AstakosClient(
token, "https://example.com", use_pool=pool)
client._callAstakos("/im/authenticate")
except AstakosClientException as err:
if err.status != 401:
self.fail("Should have returned 401 (Invalid X-Auth-Token)")
else:
self.fail("Should have returned 401 (Invalid X-Auth-Token)")
......@@ -288,10 +289,11 @@ class TestCallAstakos(unittest.TestCase):
global token_1
_mockRequest([_requestOk])
try:
astakosclient._callAstakos(
token_1, "https://example.com/im/misspelled", use_pool=pool)
except Exception as (status, data):
if status != 404:
client = AstakosClient(
token_1, "https://example.com", use_pool=pool)
client._callAstakos("/im/misspelled")
except AstakosClientException as err:
if err.status != 404:
self.fail("Should have returned 404 (Not Found)")
else:
self.fail("Should have returned 404 (Not Found)")
......@@ -310,8 +312,9 @@ class TestCallAstakos(unittest.TestCase):
global token_1
_mockRequest([_requestOk])
try:
astakosclient._callAstakos(
token_1, "ftp://example.com/im/authenticate", use_pool=pool)
client = AstakosClient(
token_1, "ftp://example.com", use_pool=pool)
client._callAstakos("/im/authenticate")
except ValueError:
pass
except Exception:
......@@ -333,10 +336,11 @@ class TestCallAstakos(unittest.TestCase):
global token_1
_mockRequest([_requestOk])
try:
astakosclient._callAstakos(
token_1, "http://example.com/im/authenticate", use_pool=pool)
except Exception as (status, data):
if status != 302:
client = AstakosClient(
token_1, "http://example.com", use_pool=pool)
client._callAstakos("/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)")
......@@ -355,11 +359,11 @@ class TestCallAstakos(unittest.TestCase):
global token_1
_mockRequest([_requestOk])
try:
astakosclient._callAstakos(
token_1, "https://example.com/im/authenticate",
use_pool=pool, method="POST")
except Exception as (status, data):
if status != 400:
client = AstakosClient(
token_1, "https://example.com", use_pool=pool)
client._callAstakos("/im/authenticate", method="POST")
except AstakosClientException as err:
if err.status != 400:
self.fail("Should have returned 400 (Method not allowed)")
else:
self.fail("Should have returned 400 (Method not allowed)")
......@@ -378,10 +382,11 @@ class TestCallAstakos(unittest.TestCase):
global token_1
_mockRequest([_requestOk])
try:
astakosclient._callAstakos(
token_1, "https://example.com/user_catalogs", use_pool=pool)
except Exception as (status, data):
if status != 400:
client = AstakosClient(
token_1, "https://example.com", use_pool=pool)
client._callAstakos("/user_catalogs")
except AstakosClientException as err:
if err.status != 400:
self.fail("Should have returned 400 (Method not allowed)")
else:
self.fail("Should have returned 400 (Method not allowed)")
......@@ -395,103 +400,103 @@ class TestCallAstakos(unittest.TestCase):
self._getUserCatalogs(True)
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 replies"""
global token_1
_mockRequest([_requestOffline])
try:
astakosclient.authenticate(token_1, "https://example.com")
except socket.error:
pass
except Exception:
self.fail("Should have raised socket exception")
else:
self.fail("Shouldn't succeed")
# ----------------------------------
# Test the response we get for invalid token
def _invalidToken(self, pool):
token = "skaksaFlBl+fasFdaf24sx=="
_mockRequest([_requestOk])
try:
astakosclient.authenticate(
token, "https://example.com", use_pool=pool)
except Exception as (status, data):
if 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:
auth_info = astakosclient.authenticate(
token, "https://example.com/", usage=usage, use_pool=pool)
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:
auth_info = astakosclient.authenticate(
token_1, "https://example.com", usage=True)
except:
self.fail("Shouldn't raise an Exception")
self.assertEqual(user_1, auth_info)
#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 replies"""
# global token_1
# _mockRequest([_requestOffline])
# try:
# astakosclient.authenticate(token_1, "https://example.com")
# except socket.error:
# pass
# except Exception:
# self.fail("Should have raised socket exception")
# else:
# self.fail("Shouldn't succeed")
#
# # ----------------------------------
# # Test the response we get for invalid token
# def _invalidToken(self, pool):
# token = "skaksaFlBl+fasFdaf24sx=="
# _mockRequest([_requestOk])
# try:
# astakosclient.authenticate(
# token, "https://example.com", use_pool=pool)
# except Exception as (status, data):
# if 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:
# auth_info = astakosclient.authenticate(
# token, "https://example.com/", usage=usage, use_pool=pool)
# 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:
# auth_info = astakosclient.authenticate(
# token_1, "https://example.com", usage=True)
# except:
# self.fail("Shouldn't raise an Exception")
# self.assertEqual(user_1, auth_info)
# ----------------------------
......
Markdown is supported
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment