openstack/deb-python-requests-kerberos
Code Issues Proposed changes

Merge pull request #55 from AnthonyTruchet/master

Browse Source 
Uses kerberos_sspi on windows
This commit is contained in:
Ian Cordasco 2015-08-26 07:46:35 -05:00
commit eecb004bd1
2 changed files with 29 additions and 21 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
setup.py
View File

@ -35,7 +35,7 @@ def get_version():
""" """
reg = re.compile(r'__version__ = [\'"]([^\'"]*)[\'"]') reg = re.compile(r'__version__ = [\'"]([^\'"]*)[\'"]')
with open('requests_kerberos/__init__.py') as fd: with open('requests_kerberos/__init__.py') as fd:
matches = filter(lambda x: x, map(reg.match, fd)) matches = list(filter(lambda x: x, map(reg.match, fd)))
if not matches: if not matches:
raise RuntimeError( raise RuntimeError(

46
test_requests_kerberos.py
View File

@ -5,7 +5,15 @@
from mock import Mock, patch from mock import Mock, patch
import requests import requests
import kerberos
try:
import kerberos
kerberos_module_name='kerberos'
except ImportError:
import kerberos_sspi as kerberos # On Windows
kerberos_module_name='kerberos_sspi'
import requests_kerberos import requests_kerberos
import unittest import unittest
@ -67,7 +75,7 @@ class KerberosTestCase(unittest.TestCase):
) )
def test_generate_request_header(self): def test_generate_request_header(self):
with patch.multiple('kerberos', with patch.multiple(kerberos_module_name,
authGSSClientInit=clientInit_complete, authGSSClientInit=clientInit_complete,
authGSSClientResponse=clientResponse, authGSSClientResponse=clientResponse,
authGSSClientStep=clientStep_continue): authGSSClientStep=clientStep_continue):
@ -84,7 +92,7 @@ class KerberosTestCase(unittest.TestCase):
clientResponse.assert_called_with("CTX") clientResponse.assert_called_with("CTX")
def test_generate_request_header_init_error(self): def test_generate_request_header_init_error(self):
with patch.multiple('kerberos', with patch.multiple(kerberos_module_name,
authGSSClientInit=clientInit_error, authGSSClientInit=clientInit_error,
authGSSClientResponse=clientResponse, authGSSClientResponse=clientResponse,
authGSSClientStep=clientStep_continue): authGSSClientStep=clientStep_continue):
@ -101,7 +109,7 @@ class KerberosTestCase(unittest.TestCase):
self.assertFalse(clientResponse.called) self.assertFalse(clientResponse.called)
def test_generate_request_header_step_error(self): def test_generate_request_header_step_error(self):
with patch.multiple('kerberos', with patch.multiple(kerberos_module_name,
authGSSClientInit=clientInit_complete, authGSSClientInit=clientInit_complete,
authGSSClientResponse=clientResponse, authGSSClientResponse=clientResponse,
authGSSClientStep=clientStep_error): authGSSClientStep=clientStep_error):
@ -118,7 +126,7 @@ class KerberosTestCase(unittest.TestCase):
self.assertFalse(clientResponse.called) self.assertFalse(clientResponse.called)
def test_authenticate_user(self): def test_authenticate_user(self):
with patch.multiple('kerberos', with patch.multiple(kerberos_module_name,
authGSSClientInit=clientInit_complete, authGSSClientInit=clientInit_complete,
authGSSClientResponse=clientResponse, authGSSClientResponse=clientResponse,
authGSSClientStep=clientStep_continue): authGSSClientStep=clientStep_continue):
@ -157,7 +165,7 @@ class KerberosTestCase(unittest.TestCase):
def test_handle_401(self): def test_handle_401(self):
with patch.multiple('kerberos', with patch.multiple(kerberos_module_name,
authGSSClientInit=clientInit_complete, authGSSClientInit=clientInit_complete,
authGSSClientResponse=clientResponse, authGSSClientResponse=clientResponse,
authGSSClientStep=clientStep_continue): authGSSClientStep=clientStep_continue):
@ -195,7 +203,7 @@ class KerberosTestCase(unittest.TestCase):
clientResponse.assert_called_with("CTX") clientResponse.assert_called_with("CTX")
def test_authenticate_server(self): def test_authenticate_server(self):
with patch.multiple('kerberos', authGSSClientStep=clientStep_complete): with patch.multiple(kerberos_module_name, authGSSClientStep=clientStep_complete):
response_ok = requests.Response() response_ok = requests.Response()
response_ok.url = "http://www.example.org/" response_ok.url = "http://www.example.org/"
@ -212,7 +220,7 @@ class KerberosTestCase(unittest.TestCase):
clientStep_complete.assert_called_with("CTX", "servertoken") clientStep_complete.assert_called_with("CTX", "servertoken")
def test_handle_other(self): def test_handle_other(self):
with patch('kerberos.authGSSClientStep', clientStep_complete): with patch(kerberos_module_name+'.authGSSClientStep', clientStep_complete):
response_ok = requests.Response() response_ok = requests.Response()
response_ok.url = "http://www.example.org/" response_ok.url = "http://www.example.org/"
@ -230,7 +238,7 @@ class KerberosTestCase(unittest.TestCase):
clientStep_complete.assert_called_with("CTX", "servertoken") clientStep_complete.assert_called_with("CTX", "servertoken")
def test_handle_response_200(self): def test_handle_response_200(self):
with patch('kerberos.authGSSClientStep', clientStep_complete): with patch(kerberos_module_name+'.authGSSClientStep', clientStep_complete):
response_ok = requests.Response() response_ok = requests.Response()
response_ok.url = "http://www.example.org/" response_ok.url = "http://www.example.org/"
@ -248,7 +256,7 @@ class KerberosTestCase(unittest.TestCase):
clientStep_complete.assert_called_with("CTX", "servertoken") clientStep_complete.assert_called_with("CTX", "servertoken")
def test_handle_response_200_mutual_auth_required_failure(self): def test_handle_response_200_mutual_auth_required_failure(self):
with patch('kerberos.authGSSClientStep', clientStep_error): with patch(kerberos_module_name+'.authGSSClientStep', clientStep_error):
response_ok = requests.Response() response_ok = requests.Response()
response_ok.url = "http://www.example.org/" response_ok.url = "http://www.example.org/"
@ -265,7 +273,7 @@ class KerberosTestCase(unittest.TestCase):
self.assertFalse(clientStep_error.called) self.assertFalse(clientStep_error.called)
def test_handle_response_200_mutual_auth_required_failure_2(self): def test_handle_response_200_mutual_auth_required_failure_2(self):
with patch('kerberos.authGSSClientStep', clientStep_exception): with patch(kerberos_module_name+'.authGSSClientStep', clientStep_exception):
response_ok = requests.Response() response_ok = requests.Response()
response_ok.url = "http://www.example.org/" response_ok.url = "http://www.example.org/"
@ -284,7 +292,7 @@ class KerberosTestCase(unittest.TestCase):
clientStep_exception.assert_called_with("CTX", "servertoken") clientStep_exception.assert_called_with("CTX", "servertoken")
def test_handle_response_200_mutual_auth_optional_hard_failure(self): def test_handle_response_200_mutual_auth_optional_hard_failure(self):
with patch('kerberos.authGSSClientStep', clientStep_error): with patch(kerberos_module_name+'.authGSSClientStep', clientStep_error):
response_ok = requests.Response() response_ok = requests.Response()
response_ok.url = "http://www.example.org/" response_ok.url = "http://www.example.org/"
@ -305,7 +313,7 @@ class KerberosTestCase(unittest.TestCase):
def test_handle_response_200_mutual_auth_optional_soft_failure(self): def test_handle_response_200_mutual_auth_optional_soft_failure(self):
with patch('kerberos.authGSSClientStep', clientStep_error): with patch(kerberos_module_name+'.authGSSClientStep', clientStep_error):
response_ok = requests.Response() response_ok = requests.Response()
response_ok.url = "http://www.example.org/" response_ok.url = "http://www.example.org/"
@ -321,7 +329,7 @@ class KerberosTestCase(unittest.TestCase):
self.assertFalse(clientStep_error.called) self.assertFalse(clientStep_error.called)
def test_handle_response_500_mutual_auth_required_failure(self): def test_handle_response_500_mutual_auth_required_failure(self):
with patch('kerberos.authGSSClientStep', clientStep_error): with patch(kerberos_module_name+'.authGSSClientStep', clientStep_error):
response_500 = requests.Response() response_500 = requests.Response()
response_500.url = "http://www.example.org/" response_500.url = "http://www.example.org/"
@ -347,13 +355,13 @@ class KerberosTestCase(unittest.TestCase):
self.assertEqual(r.url, response_500.url) self.assertEqual(r.url, response_500.url)
self.assertEqual(r.reason, response_500.reason) self.assertEqual(r.reason, response_500.reason)
self.assertEqual(r.connection, response_500.connection) self.assertEqual(r.connection, response_500.connection)
self.assertEqual(r.content, b'') self.assertEqual(r.content, '')
self.assertNotEqual(r.cookies, response_500.cookies) self.assertNotEqual(r.cookies, response_500.cookies)
self.assertFalse(clientStep_error.called) self.assertFalse(clientStep_error.called)
def test_handle_response_500_mutual_auth_optional_failure(self): def test_handle_response_500_mutual_auth_optional_failure(self):
with patch('kerberos.authGSSClientStep', clientStep_error): with patch(kerberos_module_name+'.authGSSClientStep', clientStep_error):
response_500 = requests.Response() response_500 = requests.Response()
response_500.url = "http://www.example.org/" response_500.url = "http://www.example.org/"
@ -378,7 +386,7 @@ class KerberosTestCase(unittest.TestCase):
def test_handle_response_401(self): def test_handle_response_401(self):
# Get a 401 from server, authenticate, and get a 200 back. # Get a 401 from server, authenticate, and get a 200 back.
with patch.multiple('kerberos', with patch.multiple(kerberos_module_name,
authGSSClientInit=clientInit_complete, authGSSClientInit=clientInit_complete,
authGSSClientResponse=clientResponse, authGSSClientResponse=clientResponse,
authGSSClientStep=clientStep_continue): authGSSClientStep=clientStep_continue):
@ -422,7 +430,7 @@ class KerberosTestCase(unittest.TestCase):
def test_handle_response_401_rejected(self): def test_handle_response_401_rejected(self):
# Get a 401 from server, authenticate, and get another 401 back. # Get a 401 from server, authenticate, and get another 401 back.
# Ensure there is no infinite recursion. # Ensure there is no infinite recursion.
with patch.multiple('kerberos', with patch.multiple(kerberos_module_name,
authGSSClientInit=clientInit_complete, authGSSClientInit=clientInit_complete,
authGSSClientResponse=clientResponse, authGSSClientResponse=clientResponse,
authGSSClientStep=clientStep_continue): authGSSClientStep=clientStep_continue):
@ -465,7 +473,7 @@ class KerberosTestCase(unittest.TestCase):
clientResponse.assert_called_with("CTX") clientResponse.assert_called_with("CTX")
def test_generate_request_header_custom_service(self): def test_generate_request_header_custom_service(self):
with patch.multiple('kerberos', with patch.multiple(kerberos_module_name,
authGSSClientInit=clientInit_error, authGSSClientInit=clientInit_error,
authGSSClientResponse=clientResponse, authGSSClientResponse=clientResponse,
authGSSClientStep=clientStep_continue): authGSSClientStep=clientStep_continue):