| | |
| | | import unittest |
| | | |
| | | class _Base(unittest.TestCase): |
| | | |
| | | def failUnless(self, predicate, message=''): |
| | | self.assertTrue(predicate, message) # Nannies go home! |
| | | |
| | | class TestMiddleware(_Base): |
| | | class TestMiddleware(unittest.TestCase): |
| | | |
| | | def _getTargetClass(self): |
| | | from repoze.who.middleware import PluggableAuthenticationMiddleware |
| | |
| | | self.assertEqual(start_response.status, '200 OK') |
| | | self.assertEqual(start_response.headers, headers) |
| | | |
| | | def test_call_200_no_challengers_app_calls_forget(self): |
| | | # See https://github.com/repoze/repoze.who/issues/21 |
| | | environ = self._makeEnviron() |
| | | remember_headers = [('remember', '1')] |
| | | forget_headers = [('forget', '1')] |
| | | app = DummyLogoutApp('200 OK') |
| | | credentials = {'login':'chris', 'password':'password'} |
| | | identifier = DummyIdentifier( |
| | | credentials, |
| | | remember_headers=remember_headers, |
| | | forget_headers=forget_headers) |
| | | identifiers = [ ('identifier', identifier) ] |
| | | authenticator = DummyAuthenticator() |
| | | authenticators = [ ('authenticator', authenticator) ] |
| | | mw = self._makeOne( |
| | | app=app, identifiers=identifiers, authenticators=authenticators) |
| | | start_response = DummyStartResponse() |
| | | result = mw(environ, start_response) |
| | | self.assertEqual(mw.app.environ, environ) |
| | | self.assertEqual(result, ['body']) |
| | | self.assertEqual(start_response.status, '200 OK') |
| | | self.assertEqual(start_response.headers, forget_headers) |
| | | |
| | | def test_call_401_no_identifiers(self): |
| | | from webob.exc import HTTPUnauthorized |
| | | environ = self._makeEnviron() |
| | |
| | | start_response = DummyStartResponse() |
| | | result = b''.join(mw(environ, start_response)).decode('ascii') |
| | | self.assertEqual(environ['challenged'], challenge_app) |
| | | self.failUnless(result.startswith('401 Unauthorized')) |
| | | self.assertTrue(result.startswith('401 Unauthorized')) |
| | | |
| | | def test_call_401_challenger_and_identifier_no_authenticator(self): |
| | | from webob.exc import HTTPUnauthorized |
| | |
| | | |
| | | result = b''.join(mw(environ, start_response)).decode('ascii') |
| | | self.assertEqual(environ['challenged'], challenge_app) |
| | | self.failUnless(result.startswith('401 Unauthorized')) |
| | | self.assertTrue(result.startswith('401 Unauthorized')) |
| | | self.assertEqual(identifier.forgotten, False) |
| | | self.assertEqual(environ.get('REMOTE_USER'), None) |
| | | |
| | |
| | | start_response = DummyStartResponse() |
| | | result = b''.join(mw(environ, start_response)).decode('ascii') |
| | | self.assertEqual(environ['challenged'], challenge_app) |
| | | self.failUnless(result.startswith('401 Unauthorized')) |
| | | self.assertTrue(result.startswith('401 Unauthorized')) |
| | | # @@ unfuck |
| | | ## self.assertEqual(identifier.forgotten, identifier.credentials) |
| | | self.assertEqual(environ['REMOTE_USER'], 'chris') |
| | |
| | | mdproviders=mdproviders) |
| | | start_response = DummyStartResponse() |
| | | result = b''.join(mw(environ, start_response)).decode('ascii') |
| | | self.failUnless(result.startswith('302 Found')) |
| | | self.assertTrue(result.startswith('302 Found')) |
| | | self.assertEqual(start_response.status, '302 Found') |
| | | headers = start_response.headers |
| | | #self.assertEqual(len(headers), 3, headers) |
| | |
| | | self.assertEqual(headers[3], |
| | | ('a', '1')) |
| | | self.assertEqual(start_response.exc_info, None) |
| | | self.failIf('repoze.who.application' in environ) |
| | | self.assertFalse('repoze.who.application' in environ) |
| | | |
| | | def test_call_app_doesnt_call_start_response(self): |
| | | from webob.exc import HTTPUnauthorized |
| | |
| | | mdproviders=mdproviders) |
| | | start_response = DummyStartResponse() |
| | | result = b''.join(mw(environ, start_response)).decode('ascii') |
| | | self.failUnless(result.startswith('401 Unauthorized')) |
| | | self.failUnless(app._iterable._closed) |
| | | self.assertTrue(result.startswith('401 Unauthorized')) |
| | | self.assertTrue(app._iterable._closed) |
| | | |
| | | def test_call_w_challenge_but_no_challenger_still_closes_iterable(self): |
| | | environ = self._makeEnviron() |
| | |
| | | mdproviders=mdproviders) |
| | | start_response = DummyStartResponse() |
| | | self.assertRaises(RuntimeError, mw, environ, start_response) |
| | | self.failUnless(app._iterable._closed) |
| | | self.assertTrue(app._iterable._closed) |
| | | |
| | | # XXX need more call tests: |
| | | # - auth_id sorting |
| | | |
| | | class TestStartResponseWrapper(_Base): |
| | | class TestStartResponseWrapper(unittest.TestCase): |
| | | |
| | | def _getTargetClass(self): |
| | | from repoze.who.middleware import StartResponseWrapper |
| | |
| | | wrapper = self._makeOne(None) |
| | | self.assertEqual(wrapper.start_response, None) |
| | | self.assertEqual(wrapper.headers, []) |
| | | self.failUnless(wrapper.buffer) |
| | | self.assertTrue(wrapper.buffer) |
| | | |
| | | def test_finish_response(self): |
| | | from repoze.who._compat import StringIO |
| | |
| | | self.assertEqual(datases[0], 'written') |
| | | self.assertEqual(closededs[0], True) |
| | | |
| | | class WrapGeneratorTests(_Base): |
| | | class WrapGeneratorTests(unittest.TestCase): |
| | | |
| | | def _callFUT(self, iterable): |
| | | from repoze.who.middleware import wrap_generator |
| | |
| | | yield 'b' |
| | | iterable = DummyIterableWithClose(gen()) |
| | | newgen = self._callFUT(iterable) |
| | | self.failIf(iterable._closed) |
| | | self.assertFalse(iterable._closed) |
| | | self.assertEqual(list(newgen), ['a', 'b']) |
| | | self.failUnless(iterable._closed) |
| | | self.assertTrue(iterable._closed) |
| | | |
| | | class TestMakeTestMiddleware(_Base): |
| | | class TestMakeTestMiddleware(unittest.TestCase): |
| | | |
| | | def setUp(self): |
| | | import os |
| | |
| | | start_response(self.status, self.headers) |
| | | return ['body'] |
| | | |
| | | class DummyLogoutApp(object): |
| | | def __init__(self, status): |
| | | self.status = status |
| | | |
| | | def __call__(self, environ, start_response): |
| | | self.environ = environ |
| | | api = environ['repoze.who.api'] |
| | | headers = api.logout() |
| | | start_response(self.status, headers) |
| | | return ['body'] |
| | | |
| | | class DummyGeneratorApp(object): |
| | | def __init__(self, status, headers): |
| | | self.status = status |