<s> import unittest <EOL> class TestMiddleware ( unittest . TestCase ) : <EOL> def _getTargetClass ( self ) : <EOL> from repoze . who . middleware import PluggableAuthenticationMiddleware <EOL> return PluggableAuthenticationMiddleware <EOL> def _makeOne ( self , <EOL> app = None , <EOL> identifiers = None , <EOL> authenticators = None , <EOL> challengers = None , <EOL> request_classifier = None , <EOL> mdproviders = None , <EOL> challenge_decider = None , <EOL> log_stream = None , <EOL> log_level = None , <EOL> remote_user_key = '<STR_LIT>' , <EOL> ) : <EOL> if app is None : <EOL> app = DummyApp ( ) <EOL> if identifiers is None : <EOL> identifiers = [ ] <EOL> if authenticators is None : <EOL> authenticators = [ ] <EOL> if challengers is None : <EOL> challengers = [ ] <EOL> if request_classifier is None : <EOL> request_classifier = DummyRequestClassifier ( ) <EOL> if mdproviders is None : <EOL> mdproviders = [ ] <EOL> if challenge_decider is None : <EOL> challenge_decider = DummyChallengeDecider ( ) <EOL> if log_level is None : <EOL> import logging <EOL> log_level = logging . DEBUG <EOL> mw = self . _getTargetClass ( ) ( app , <EOL> identifiers , <EOL> authenticators , <EOL> challengers , <EOL> mdproviders , <EOL> request_classifier , <EOL> challenge_decider , <EOL> log_stream , <EOL> log_level = logging . DEBUG , <EOL> remote_user_key = remote_user_key , <EOL> ) <EOL> return mw <EOL> def _makeEnviron ( self , kw = None ) : <EOL> from wsgiref . util import setup_testing_defaults <EOL> environ = { } <EOL> setup_testing_defaults ( environ ) <EOL> if kw is not None : <EOL> environ . update ( kw ) <EOL> return environ <EOL> def test_ctor_positional_args ( self ) : <EOL> klass = self . _getTargetClass ( ) <EOL> app = DummyApp ( ) <EOL> identifiers = [ ] <EOL> authenticators = [ ] <EOL> challengers = [ ] <EOL> request_classifier = DummyRequestClassifier ( ) <EOL> mdproviders = [ ] <EOL> challenge_decider = DummyChallengeDecider ( ) <EOL> mw = klass ( app , <EOL> identifiers , <EOL> authenticators , <EOL> challengers , <EOL> mdproviders , <EOL> request_classifier , <EOL> challenge_decider , <EOL> ) <EOL> self . assertEqual ( mw . app , app ) <EOL> af = mw . api_factory <EOL> self . assertEqual ( af . identifiers , identifiers ) <EOL> self . assertEqual ( af . authenticators , authenticators ) <EOL> self . assertEqual ( af . challengers , challengers ) <EOL> self . assertEqual ( af . mdproviders , mdproviders ) <EOL> self . assertEqual ( af . request_classifier , request_classifier ) <EOL> self . assertEqual ( af . challenge_decider , challenge_decider ) <EOL> def test_ctor_wo_request_classifier_or_classifier_raises ( self ) : <EOL> klass = self . _getTargetClass ( ) <EOL> app = DummyApp ( ) <EOL> identifiers = [ ] <EOL> authenticators = [ ] <EOL> challengers = [ ] <EOL> mdproviders = [ ] <EOL> challenge_decider = DummyChallengeDecider ( ) <EOL> self . assertRaises ( ValueError , <EOL> klass , <EOL> app , <EOL> identifiers , <EOL> authenticators , <EOL> challengers , <EOL> mdproviders , <EOL> challenge_decider = challenge_decider , <EOL> ) <EOL> def test_ctor_w_request_classifier_and_classifier_raises ( self ) : <EOL> klass = self . _getTargetClass ( ) <EOL> app = DummyApp ( ) <EOL> identifiers = [ ] <EOL> authenticators = [ ] <EOL> challengers = [ ] <EOL> request_classifier = DummyRequestClassifier ( ) <EOL> mdproviders = [ ] <EOL> challenge_decider = DummyChallengeDecider ( ) <EOL> self . assertRaises ( ValueError , <EOL> klass , <EOL> app , <EOL> identifiers , <EOL> authenticators , <EOL> challengers , <EOL> mdproviders , <EOL> request_classifier , <EOL> challenge_decider , <EOL> classifier = object ( ) <EOL> ) <EOL> def test_ctor_wo_challenge_decider_raises ( self ) : <EOL> klass = self . _getTargetClass ( ) <EOL> app = DummyApp ( ) <EOL> identifiers = [ ] <EOL> authenticators = [ ] <EOL> challengers = [ ] <EOL> request_classifier = DummyRequestClassifier ( ) <EOL> mdproviders = [ ] <EOL> self . assertRaises ( ValueError , <EOL> klass , <EOL> app , <EOL> identifiers , <EOL> authenticators , <EOL> challengers , <EOL> mdproviders , <EOL> classifier = request_classifier , <EOL> ) <EOL> def test_ctor_w_classifier ( self ) : <EOL> klass = self . _getTargetClass ( ) <EOL> app = DummyApp ( ) <EOL> identifiers = [ ] <EOL> authenticators = [ ] <EOL> challengers = [ ] <EOL> request_classifier = DummyRequestClassifier ( ) <EOL> mdproviders = [ ] <EOL> challenge_decider = DummyChallengeDecider ( ) <EOL> mw = klass ( app , <EOL> identifiers , <EOL> authenticators , <EOL> challengers , <EOL> mdproviders , <EOL> classifier = request_classifier , <EOL> challenge_decider = challenge_decider , <EOL> ) <EOL> self . assertEqual ( mw . app , app ) <EOL> af = mw . api_factory <EOL> self . assertEqual ( af . identifiers , identifiers ) <EOL> self . assertEqual ( af . authenticators , authenticators ) <EOL> self . assertEqual ( af . challengers , challengers ) <EOL> self . assertEqual ( af . mdproviders , mdproviders ) <EOL> self . assertEqual ( af . request_classifier , request_classifier ) <EOL> self . assertEqual ( af . challenge_decider , challenge_decider ) <EOL> def test_ctor_accepts_logger ( self ) : <EOL> import logging <EOL> restore = logging . raiseExceptions <EOL> logging . raiseExceptions = <NUM_LIT:0> <EOL> try : <EOL> logger = logging . Logger ( '<STR_LIT>' ) <EOL> logger . setLevel ( logging . INFO ) <EOL> mw = self . _makeOne ( log_stream = logger ) <EOL> self . assertEqual ( logger , mw . logger ) <EOL> finally : <EOL> logging . raiseExceptions = restore <EOL> def test_call_remoteuser_already_set ( self ) : <EOL> environ = self . _makeEnviron ( { '<STR_LIT>' : '<STR_LIT>' } ) <EOL> mw = self . _makeOne ( ) <EOL> result = mw ( environ , None ) <EOL> self . assertEqual ( mw . app . environ , environ ) <EOL> self . assertEqual ( result , [ ] ) <EOL> def test_call_200_no_plugins ( self ) : <EOL> environ = self . _makeEnviron ( ) <EOL> headers = [ ( '<STR_LIT:a>' , '<STR_LIT:1>' ) ] <EOL> app = DummyWorkingApp ( '<STR_LIT>' , headers ) <EOL> mw = self . _makeOne ( app = app ) <EOL> start_response = DummyStartResponse ( ) <EOL> result = mw ( environ , start_response ) <EOL> self . assertEqual ( mw . app . environ , environ ) <EOL> self . assertEqual ( result , [ '<STR_LIT:body>' ] ) <EOL> self . assertEqual ( start_response . status , '<STR_LIT>' ) <EOL> self . assertEqual ( start_response . headers , headers ) <EOL> def test_call_401_no_challengers ( self ) : <EOL> environ = self . _makeEnviron ( ) <EOL> headers = [ ( '<STR_LIT:a>' , '<STR_LIT:1>' ) ] <EOL> app = DummyWorkingApp ( '<STR_LIT>' , headers ) <EOL> mw = self . _makeOne ( app = app ) <EOL> start_response = DummyStartResponse ( ) <EOL> self . assertRaises ( RuntimeError , mw , environ , start_response ) <EOL> def test_call_200_no_challengers ( self ) : <EOL> environ = self . _makeEnviron ( ) <EOL> headers = [ ( '<STR_LIT:a>' , '<STR_LIT:1>' ) ] <EOL> app = DummyWorkingApp ( '<STR_LIT>' , headers ) <EOL> credentials = { '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT:password>' : '<STR_LIT:password>' } <EOL> identifier = DummyIdentifier ( credentials ) <EOL> identifiers = [ ( '<STR_LIT>' , identifier ) ] <EOL> mw = self . _makeOne ( app = app , identifiers = identifiers ) <EOL> start_response = DummyStartResponse ( ) <EOL> result = mw ( environ , start_response ) <EOL> self . assertEqual ( mw . app . environ , environ ) <EOL> self . assertEqual ( result , [ '<STR_LIT:body>' ] ) <EOL> self . assertEqual ( start_response . status , '<STR_LIT>' ) <EOL> self . assertEqual ( start_response . headers , headers ) <EOL> def test_call_401_no_identifiers ( self ) : <EOL> from webob . exc import HTTPUnauthorized <EOL> environ = self . _makeEnviron ( ) <EOL> headers = [ ( '<STR_LIT:a>' , '<STR_LIT:1>' ) ] <EOL> app = DummyWorkingApp ( '<STR_LIT>' , headers ) <EOL> challenge_app = HTTPUnauthorized ( ) <EOL> challenge = DummyChallenger ( challenge_app ) <EOL> challengers = [ ( '<STR_LIT>' , challenge ) ] <EOL> mw = self . _makeOne ( app = app , challengers = challengers ) <EOL> start_response = DummyStartResponse ( ) <EOL> result = b'<STR_LIT>' . join ( mw ( environ , start_response ) ) . decode ( '<STR_LIT:ascii>' ) <EOL> self . assertEqual ( environ [ '<STR_LIT>' ] , challenge_app ) <EOL> self . assertTrue ( result . startswith ( '<STR_LIT>' ) ) <EOL> def test_call_401_challenger_and_identifier_no_authenticator ( self ) : <EOL> from webob . exc import HTTPUnauthorized <EOL> environ = self . _makeEnviron ( ) <EOL> headers = [ ( '<STR_LIT:a>' , '<STR_LIT:1>' ) ] <EOL> app = DummyWorkingApp ( '<STR_LIT>' , headers ) <EOL> challenge_app = HTTPUnauthorized ( ) <EOL> challenge = DummyChallenger ( challenge_app ) <EOL> challengers = [ ( '<STR_LIT>' , challenge ) ] <EOL> credentials = { '<STR_LIT>' : '<STR_LIT:a>' , '<STR_LIT:password>' : '<STR_LIT:b>' } <EOL> identifier = DummyIdentifier ( credentials ) <EOL> identifiers = [ ( '<STR_LIT>' , identifier ) ] <EOL> mw = self . _makeOne ( app = app , challengers = challengers , <EOL> identifiers = identifiers ) <EOL> start_response = DummyStartResponse ( ) <EOL> result = b'<STR_LIT>' . join ( mw ( environ , start_response ) ) . decode ( '<STR_LIT:ascii>' ) <EOL> self . assertEqual ( environ [ '<STR_LIT>' ] , challenge_app ) <EOL> self . assertTrue ( result . startswith ( '<STR_LIT>' ) ) <EOL> self . assertEqual ( identifier . forgotten , False ) <EOL> self . assertEqual ( environ . get ( '<STR_LIT>' ) , None ) <EOL> def test_call_401_challenger_and_identifier_and_authenticator ( self ) : <EOL> from webob . exc import HTTPUnauthorized <EOL> environ = self . _makeEnviron ( ) <EOL> headers = [ ( '<STR_LIT:a>' , '<STR_LIT:1>' ) ] <EOL> app = DummyWorkingApp ( '<STR_LIT>' , headers ) <EOL> challenge_app = HTTPUnauthorized ( ) <EOL> challenge = DummyChallenger ( challenge_app ) <EOL> challengers = [ ( '<STR_LIT>' , challenge ) ] <EOL> credentials = { '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT:password>' : '<STR_LIT:password>' } <EOL> identifier = DummyIdentifier ( credentials ) <EOL> identifiers = [ ( '<STR_LIT>' , identifier ) ] <EOL> authenticator = DummyAuthenticator ( ) <EOL> authenticators = [ ( '<STR_LIT>' , authenticator ) ] <EOL> mw = self . _makeOne ( app = app , challengers = challengers , <EOL> identifiers = identifiers , <EOL> authenticators = authenticators ) <EOL> start_response = DummyStartResponse ( ) <EOL> result = b'<STR_LIT>' . join ( mw ( environ , start_response ) ) . decode ( '<STR_LIT:ascii>' ) <EOL> self . assertEqual ( environ [ '<STR_LIT>' ] , challenge_app ) <EOL> self . assertTrue ( result . startswith ( '<STR_LIT>' ) ) <EOL> self . assertEqual ( environ [ '<STR_LIT>' ] , '<STR_LIT>' ) <EOL> def test_call_200_challenger_and_identifier_and_authenticator ( self ) : <EOL> from webob . exc import HTTPUnauthorized <EOL> environ = self . _makeEnviron ( ) <EOL> headers = [ ( '<STR_LIT:a>' , '<STR_LIT:1>' ) ] <EOL> app = DummyWorkingApp ( '<STR_LIT>' , headers ) <EOL> challenge_app = HTTPUnauthorized ( ) <EOL> challenge = DummyChallenger ( challenge_app ) <EOL> challengers = [ ( '<STR_LIT>' , challenge ) ] <EOL> credentials = { '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT:password>' : '<STR_LIT:password>' } <EOL> identifier = DummyIdentifier ( credentials ) <EOL> identifiers = [ ( '<STR_LIT>' , identifier ) ] <EOL> authenticator = DummyAuthenticator ( ) <EOL> authenticators = [ ( '<STR_LIT>' , authenticator ) ] <EOL> mw = self . _makeOne ( app = app , challengers = challengers , <EOL> identifiers = identifiers , <EOL> authenticators = authenticators ) <EOL> start_response = DummyStartResponse ( ) <EOL> result = mw ( environ , start_response ) <EOL> self . assertEqual ( environ . get ( '<STR_LIT>' ) , None ) <EOL> self . assertEqual ( identifier . forgotten , False ) <EOL> self . assertEqual ( environ [ '<STR_LIT>' ] , '<STR_LIT>' ) <EOL> def test_call_200_identity_reset ( self ) : <EOL> from webob . exc import HTTPUnauthorized <EOL> environ = self . _makeEnviron ( ) <EOL> headers = [ ( '<STR_LIT:a>' , '<STR_LIT:1>' ) ] <EOL> new_identity = { '<STR_LIT>' : '<STR_LIT:foo>' , '<STR_LIT:password>' : '<STR_LIT:bar>' } <EOL> app = DummyIdentityResetApp ( '<STR_LIT>' , headers , new_identity ) <EOL> challenge_app = HTTPUnauthorized ( ) <EOL> challenge = DummyChallenger ( challenge_app ) <EOL> challengers = [ ( '<STR_LIT>' , challenge ) ] <EOL> credentials = { '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT:password>' : '<STR_LIT:password>' } <EOL> identifier = DummyIdentifier ( credentials ) <EOL> identifiers = [ ( '<STR_LIT>' , identifier ) ] <EOL> authenticator = DummyAuthenticator ( ) <EOL> authenticators = [ ( '<STR_LIT>' , authenticator ) ] <EOL> mw = self . _makeOne ( app = app , challengers = challengers , <EOL> identifiers = identifiers , <EOL> authenticators = authenticators ) <EOL> start_response = DummyStartResponse ( ) <EOL> result = mw ( environ , start_response ) <EOL> self . assertEqual ( environ . get ( '<STR_LIT>' ) , None ) <EOL> self . assertEqual ( identifier . forgotten , False ) <EOL> new_credentials = identifier . credentials . copy ( ) <EOL> new_credentials [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> new_credentials [ '<STR_LIT:password>' ] = '<STR_LIT>' <EOL> self . assertEqual ( environ [ '<STR_LIT>' ] , '<STR_LIT>' ) <EOL> def test_call_200_with_metadata ( self ) : <EOL> from webob . exc import HTTPUnauthorized <EOL> environ = self . _makeEnviron ( ) <EOL> headers = [ ( '<STR_LIT:a>' , '<STR_LIT:1>' ) ] <EOL> app = DummyWorkingApp ( '<STR_LIT>' , headers ) <EOL> challenge_app = HTTPUnauthorized ( ) <EOL> challenge = DummyChallenger ( challenge_app ) <EOL> challengers = [ ( '<STR_LIT>' , challenge ) ] <EOL> credentials = { '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT:password>' : '<STR_LIT:password>' } <EOL> identifier = DummyIdentifier ( credentials ) <EOL> identifiers = [ ( '<STR_LIT>' , identifier ) ] <EOL> authenticator = DummyAuthenticator ( ) <EOL> authenticators = [ ( '<STR_LIT>' , authenticator ) ] <EOL> mdprovider = DummyMDProvider ( { '<STR_LIT:foo>' : '<STR_LIT:bar>' } ) <EOL> mdproviders = [ ( '<STR_LIT>' , mdprovider ) ] <EOL> mw = self . _makeOne ( app = app , challengers = challengers , <EOL> identifiers = identifiers , <EOL> authenticators = authenticators , <EOL> mdproviders = mdproviders ) <EOL> start_response = DummyStartResponse ( ) <EOL> result = mw ( environ , start_response ) <EOL> self . assertEqual ( environ [ '<STR_LIT>' ] [ '<STR_LIT:foo>' ] , '<STR_LIT:bar>' ) <EOL> def test_call_ingress_plugin_replaces_application ( self ) : <EOL> from webob . exc import HTTPFound <EOL> environ = self . _makeEnviron ( ) <EOL> headers = [ ( '<STR_LIT:a>' , '<STR_LIT:1>' ) ] <EOL> app = DummyWorkingApp ( '<STR_LIT>' , headers ) <EOL> challengers = [ ] <EOL> credentials = { '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT:password>' : '<STR_LIT:password>' } <EOL> identifier = DummyIdentifier ( <EOL> credentials , <EOL> remember_headers = [ ( '<STR_LIT:a>' , '<STR_LIT:1>' ) ] , <EOL> replace_app = HTTPFound ( '<STR_LIT>' ) <EOL> ) <EOL> identifiers = [ ( '<STR_LIT>' , identifier ) ] <EOL> authenticator = DummyAuthenticator ( ) <EOL> authenticators = [ ( '<STR_LIT>' , authenticator ) ] <EOL> mdproviders = [ ] <EOL> mw = self . _makeOne ( app = app , <EOL> challengers = challengers , <EOL> identifiers = identifiers , <EOL> authenticators = authenticators , <EOL> mdproviders = mdproviders ) <EOL> start_response = DummyStartResponse ( ) <EOL> result = b'<STR_LIT>' . join ( mw ( environ , start_response ) ) . decode ( '<STR_LIT:ascii>' ) <EOL> self . assertTrue ( result . startswith ( '<STR_LIT>' ) ) <EOL> self . assertEqual ( start_response . status , '<STR_LIT>' ) <EOL> headers = start_response . headers <EOL> self . assertEqual ( headers [ <NUM_LIT:2> ] , <EOL> ( '<STR_LIT:Content-Type>' , '<STR_LIT>' ) ) <EOL> self . assertEqual ( headers [ <NUM_LIT:3> ] , <EOL> ( '<STR_LIT:a>' , '<STR_LIT:1>' ) ) <EOL> self . assertEqual ( start_response . exc_info , None ) <EOL> self . assertFalse ( '<STR_LIT>' in environ ) <EOL> def test_call_app_doesnt_call_start_response ( self ) : <EOL> from webob . exc import HTTPUnauthorized <EOL> environ = self . _makeEnviron ( ) <EOL> headers = [ ( '<STR_LIT:a>' , '<STR_LIT:1>' ) ] <EOL> app = DummyGeneratorApp ( '<STR_LIT>' , headers ) <EOL> challenge_app = HTTPUnauthorized ( ) <EOL> challenge = DummyChallenger ( challenge_app ) <EOL> challengers = [ ( '<STR_LIT>' , challenge ) ] <EOL> credentials = { '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT:password>' : '<STR_LIT:password>' } <EOL> identifier = DummyIdentifier ( credentials ) <EOL> identifiers = [ ( '<STR_LIT>' , identifier ) ] <EOL> authenticator = DummyAuthenticator ( ) <EOL> authenticators = [ ( '<STR_LIT>' , authenticator ) ] <EOL> mdprovider = DummyMDProvider ( { '<STR_LIT:foo>' : '<STR_LIT:bar>' } ) <EOL> mdproviders = [ ( '<STR_LIT>' , mdprovider ) ] <EOL> mw = self . _makeOne ( app = app , challengers = challengers , <EOL> identifiers = identifiers , <EOL> authenticators = authenticators , <EOL> mdproviders = mdproviders ) <EOL> start_response = DummyStartResponse ( ) <EOL> result = mw ( environ , start_response ) <EOL> self . assertEqual ( environ [ '<STR_LIT>' ] [ '<STR_LIT:foo>' ] , '<STR_LIT:bar>' ) <EOL> def test_call_w_challenge_closes_iterable ( self ) : <EOL> from webob . exc import HTTPUnauthorized <EOL> environ = self . _makeEnviron ( ) <EOL> headers = [ ( '<STR_LIT:a>' , '<STR_LIT:1>' ) ] <EOL> app = DummyIterableWithCloseApp ( '<STR_LIT>' , headers ) <EOL> challenge_app = HTTPUnauthorized ( ) <EOL> challenge = DummyChallenger ( challenge_app ) <EOL> challengers = [ ( '<STR_LIT>' , challenge ) ] <EOL> credentials = { '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT:password>' : '<STR_LIT:password>' } <EOL> identifier = DummyIdentifier ( credentials ) <EOL> identifiers = [ ( '<STR_LIT>' , identifier ) ] <EOL> authenticator = DummyAuthenticator ( ) <EOL> authenticators = [ ( '<STR_LIT>' , authenticator ) ] <EOL> mdprovider = DummyMDProvider ( { '<STR_LIT:foo>' : '<STR_LIT:bar>' } ) <EOL> mdproviders = [ ( '<STR_LIT>' , mdprovider ) ] <EOL> mw = self . _makeOne ( app = app , challengers = challengers , <EOL> identifiers = identifiers , <EOL> authenticators = authenticators , <EOL> mdproviders = mdproviders ) <EOL> start_response = DummyStartResponse ( ) <EOL> result = b'<STR_LIT>' . join ( mw ( environ , start_response ) ) . decode ( '<STR_LIT:ascii>' ) <EOL> self . assertTrue ( result . startswith ( '<STR_LIT>' ) ) <EOL> self . assertTrue ( app . _iterable . _closed ) <EOL> def test_call_w_challenge_but_no_challenger_still_closes_iterable ( self ) : <EOL> environ = self . _makeEnviron ( ) <EOL> headers = [ ( '<STR_LIT:a>' , '<STR_LIT:1>' ) ] <EOL> app = DummyIterableWithCloseApp ( '<STR_LIT>' , headers ) <EOL> challengers = [ ] <EOL> credentials = { '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT:password>' : '<STR_LIT:password>' } <EOL> identifier = DummyIdentifier ( credentials ) <EOL> identifiers = [ ( '<STR_LIT>' , identifier ) ] <EOL> authenticator = DummyAuthenticator ( ) <EOL> authenticators = [ ( '<STR_LIT>' , authenticator ) ] <EOL> mdprovider = DummyMDProvider ( { '<STR_LIT:foo>' : '<STR_LIT:bar>' } ) <EOL> mdproviders = [ ( '<STR_LIT>' , mdprovider ) ] <EOL> mw = self . _makeOne ( app = app , challengers = challengers , <EOL> identifiers = identifiers , <EOL> authenticators = authenticators , <EOL> mdproviders = mdproviders ) <EOL> start_response = DummyStartResponse ( ) <EOL> self . assertRaises ( RuntimeError , mw , environ , start_response ) <EOL> self . assertTrue ( app . _iterable . _closed ) <EOL> class TestStartResponseWrapper ( unittest . TestCase ) : <EOL> def _getTargetClass ( self ) : <EOL> from repoze . who . middleware import StartResponseWrapper <EOL> return StartResponseWrapper <EOL> def _makeOne ( self , * arg , ** kw ) : <EOL> plugin = self . _getTargetClass ( ) ( * arg , ** kw ) <EOL> return plugin <EOL> def test_ctor ( self ) : <EOL> wrapper = self . _makeOne ( None ) <EOL> self . assertEqual ( wrapper . start_response , None ) <EOL> self . assertEqual ( wrapper . headers , [ ] ) <EOL> self . assertTrue ( wrapper . buffer ) <EOL> def test_finish_response ( self ) : <EOL> from repoze . who . _compat import StringIO <EOL> statuses = [ ] <EOL> headerses = [ ] <EOL> datases = [ ] <EOL> closededs = [ ] <EOL> def write ( data ) : <EOL> datases . append ( data ) <EOL> def close ( ) : <EOL> closededs . append ( True ) <EOL> write . close = close <EOL> def start_response ( status , headers , exc_info = None ) : <EOL> statuses . append ( status ) <EOL> headerses . append ( headers ) <EOL> return write <EOL> wrapper = self . _makeOne ( start_response ) <EOL> wrapper . status = '<STR_LIT>' <EOL> wrapper . headers = [ ( '<STR_LIT:a>' , '<STR_LIT:1>' ) ] <EOL> wrapper . buffer = StringIO ( '<STR_LIT>' ) <EOL> extra_headers = [ ( '<STR_LIT:b>' , '<STR_LIT:2>' ) ] <EOL> result = wrapper . finish_response ( extra_headers ) <EOL> self . assertEqual ( result , None ) <EOL> self . assertEqual ( headerses [ <NUM_LIT:0> ] , wrapper . headers + extra_headers ) <EOL> self . assertEqual ( statuses [ <NUM_LIT:0> ] , wrapper . status ) <EOL> self . assertEqual ( datases [ <NUM_LIT:0> ] , '<STR_LIT>' ) <EOL> self . assertEqual ( closededs [ <NUM_LIT:0> ] , True ) <EOL> class WrapGeneratorTests ( unittest . TestCase ) : <EOL> def _callFUT ( self , iterable ) : <EOL> from repoze . who . middleware import wrap_generator <EOL> return wrap_generator ( iterable ) <EOL> def test_w_generator ( self ) : <EOL> L = [ ] <EOL> def gen ( L = L ) : <EOL> L . append ( '<STR_LIT>' ) <EOL> yield '<STR_LIT:a>' <EOL> yield '<STR_LIT:b>' <EOL> newgen = self . _callFUT ( gen ( ) ) <EOL> self . assertEqual ( L , [ '<STR_LIT>' ] ) <EOL> self . assertEqual ( list ( newgen ) , [ '<STR_LIT:a>' , '<STR_LIT:b>' ] ) <EOL> def test_w_empty_generator ( self ) : <EOL> def gen ( ) : <EOL> if False : <EOL> yield '<STR_LIT:a>' <EOL> newgen = self . _callFUT ( gen ( ) ) <EOL> self . assertEqual ( list ( newgen ) , [ ] ) <EOL> def test_w_iterator_having_close ( self ) : <EOL> def gen ( ) : <EOL> yield '<STR_LIT:a>' <EOL> yield '<STR_LIT:b>' <EOL> iterable = DummyIterableWithClose ( gen ( ) ) <EOL> newgen = self . _callFUT ( iterable ) <EOL> self . assertFalse ( iterable . _closed ) <EOL> self . assertEqual ( list ( newgen ) , [ '<STR_LIT:a>' , '<STR_LIT:b>' ] ) <EOL> self . assertTrue ( iterable . _closed ) <EOL> class TestMakeTestMiddleware ( unittest . TestCase ) : <EOL> def setUp ( self ) : <EOL> import os <EOL> try : <EOL> del os . environ [ '<STR_LIT>' ] <EOL> except KeyError : <EOL> pass <EOL> def tearDown ( self ) : <EOL> import os <EOL> try : <EOL> del os . environ [ '<STR_LIT>' ] <EOL> except KeyError : <EOL> pass <EOL> def _getFactory ( self ) : <EOL> from repoze . who . middleware import make_test_middleware <EOL> return make_test_middleware <EOL> def test_it_no_WHO_LOG_in_environ ( self ) : <EOL> app = DummyApp ( ) <EOL> factory = self . _getFactory ( ) <EOL> global_conf = { '<STR_LIT>' : '<STR_LIT:/>' } <EOL> middleware = factory ( app , global_conf ) <EOL> api_factory = middleware . api_factory <EOL> self . assertEqual ( len ( api_factory . identifiers ) , <NUM_LIT:2> ) <EOL> self . assertEqual ( len ( api_factory . authenticators ) , <NUM_LIT:1> ) <EOL> self . assertEqual ( len ( api_factory . challengers ) , <NUM_LIT:2> ) <EOL> self . assertEqual ( len ( api_factory . mdproviders ) , <NUM_LIT:0> ) <EOL> self . assertEqual ( middleware . logger , None ) <EOL> def test_it_w_WHO_LOG_in_environ ( self ) : <EOL> import logging <EOL> import os <EOL> os . environ [ '<STR_LIT>' ] = '<STR_LIT:1>' <EOL> app = DummyApp ( ) <EOL> factory = self . _getFactory ( ) <EOL> global_conf = { '<STR_LIT>' : '<STR_LIT:/>' } <EOL> middleware = factory ( app , global_conf ) <EOL> self . assertEqual ( middleware . logger . getEffectiveLevel ( ) , logging . DEBUG ) <EOL> class DummyApp ( object ) : <EOL> environ = None <EOL> def __call__ ( self , environ , start_response ) : <EOL> self . environ = environ <EOL> return [ ] <EOL> class DummyWorkingApp ( object ) : <EOL> def __init__ ( self , status , headers ) : <EOL> self . status = status <EOL> self . headers = headers <EOL> def __call__ ( self , environ , start_response ) : <EOL> self . environ = environ <EOL> start_response ( self . status , self . headers ) <EOL> return [ '<STR_LIT:body>' ] <EOL> class DummyGeneratorApp ( object ) : <EOL> def __init__ ( self , status , headers ) : <EOL> self . status = status <EOL> self . headers = headers <EOL> def __call__ ( self , environ , start_response ) : <EOL> def gen ( self = self , start_response = start_response ) : <EOL> self . environ = environ <EOL> start_response ( self . status , self . headers ) <EOL> yield '<STR_LIT:body>' <EOL> return gen ( ) <EOL> class DummyIterableWithClose ( object ) : <EOL> _closed = False <EOL> def __init__ ( self , iterable ) : <EOL> self . _iterable = iterable <EOL> def __iter__ ( self ) : <EOL> return iter ( self . _iterable ) <EOL> def close ( self ) : <EOL> self . _closed = True <EOL> class DummyIterableWithCloseApp ( object ) : <EOL> def __init__ ( self , status , headers ) : <EOL> self . status = status <EOL> self . headers = headers <EOL> self . _iterable = DummyIterableWithClose ( [ '<STR_LIT:body>' ] ) <EOL> def __call__ ( self , environ , start_response ) : <EOL> self . environ = environ <EOL> start_response ( self . status , self . headers ) <EOL> return self . _iterable <EOL> class DummyIdentityResetApp ( object ) : <EOL> def __init__ ( self , status , headers , new_identity ) : <EOL> self . status = status <EOL> self . headers = headers <EOL> self . new_identity = new_identity <EOL> def __call__ ( self , environ , start_response ) : <EOL> self . environ = environ <EOL> environ [ '<STR_LIT>' ] [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> environ [ '<STR_LIT>' ] [ '<STR_LIT:password>' ] = '<STR_LIT>' <EOL> start_response ( self . status , self . headers ) <EOL> return [ '<STR_LIT:body>' ] <EOL> class DummyChallenger ( object ) : <EOL> def __init__ ( self , app = None ) : <EOL> self . app = app <EOL> def challenge ( self , environ , status , app_headers , forget_headers ) : <EOL> environ [ '<STR_LIT>' ] = self . app <EOL> return self . app <EOL> class DummyIdentifier ( object ) : <EOL> forgotten = False <EOL> remembered = False <EOL> def __init__ ( self , credentials = None , remember_headers = None , <EOL> forget_headers = None , replace_app = None ) : <EOL> self . credentials = credentials <EOL> self . remember_headers = remember_headers <EOL> self . forget_headers = forget_headers <EOL> self . replace_app = replace_app <EOL> def identify ( self , environ ) : <EOL> if self . replace_app : <EOL> environ [ '<STR_LIT>' ] = self . replace_app <EOL> return self . credentials <EOL> def forget ( self , environ , identity ) : <EOL> self . forgotten = identity <EOL> return self . forget_headers <EOL> def remember ( self , environ , identity ) : <EOL> self . remembered = identity <EOL> return self . remember_headers <EOL> class DummyAuthenticator ( object ) : <EOL> def authenticate ( self , environ , credentials ) : <EOL> return credentials [ '<STR_LIT>' ] <EOL> class DummyRequestClassifier ( object ) : <EOL> def __call__ ( self , environ ) : <EOL> return '<STR_LIT>' <EOL> class DummyChallengeDecider ( object ) : <EOL> def __call__ ( self , environ , status , headers ) : <EOL> if status . startswith ( '<STR_LIT>' ) : <EOL> return True <EOL> class DummyStartResponse ( object ) : <EOL> def __call__ ( self , status , headers , exc_info = None ) : <EOL> self . status = status <EOL> self . headers = headers <EOL> self . exc_info = exc_info <EOL> return [ ] <EOL> class DummyMDProvider ( object ) : <EOL> def __init__ ( self , metadata = None ) : <EOL> self . _metadata = metadata <EOL> def add_metadata ( self , environ , identity ) : <EOL> return identity . update ( self . _metadata ) </s>
