Tres Seaver
2012-03-18 d853a748ac5babf102f85e48db6c0b3ffb0538bf
commit | author | age
f7efc0 1 import unittest
TS 2
afe561 3 class TestMiddleware(unittest.TestCase):
f7efc0 4
TS 5     def _getTargetClass(self):
6         from repoze.who.middleware import PluggableAuthenticationMiddleware
7         return PluggableAuthenticationMiddleware
8
9     def _makeOne(self,
10                  app=None,
11                  identifiers=None,
12                  authenticators=None,
13                  challengers=None,
cbc983 14                  request_classifier=None,
f7efc0 15                  mdproviders=None,
TS 16                  challenge_decider=None,
17                  log_stream=None,
18                  log_level=None,
b482a1 19                  remote_user_key='REMOTE_USER',
f7efc0 20                  ):
TS 21         if app is None:
22             app = DummyApp()
23         if identifiers is None:
24             identifiers = []
25         if authenticators is None:
26             authenticators = []
27         if challengers is None:
28             challengers = []
cbc983 29         if request_classifier is None:
TS 30             request_classifier = DummyRequestClassifier()
f7efc0 31         if mdproviders is None:
TS 32             mdproviders = []
33         if challenge_decider is None:
34             challenge_decider = DummyChallengeDecider()
35         if log_level is None:
36             import logging
37             log_level = logging.DEBUG
38         mw = self._getTargetClass()(app,
39                                     identifiers,
40                                     authenticators,
41                                     challengers,
42                                     mdproviders,
cbc983 43                                     request_classifier,
f7efc0 44                                     challenge_decider,
TS 45                                     log_stream,
b482a1 46                                     log_level=logging.DEBUG,
TS 47                                     remote_user_key=remote_user_key,
48                                    )
f7efc0 49         return mw
TS 50
afe561 51     def _makeEnviron(self, kw=None):
02a504 52         from wsgiref.util import setup_testing_defaults
afe561 53         environ = {}
02a504 54         setup_testing_defaults(environ)
afe561 55         if kw is not None:
TS 56             environ.update(kw)
57         return environ
58
cbc983 59     def test_ctor_positional_args(self):
TS 60         klass = self._getTargetClass()
61         app = DummyApp()
62         identifiers = []
63         authenticators = []
64         challengers = []
65         request_classifier = DummyRequestClassifier()
66         mdproviders = []
67         challenge_decider = DummyChallengeDecider()
68         mw = klass(app,
69                    identifiers,
70                    authenticators,
71                    challengers,
72                    mdproviders,
73                    request_classifier,
74                    challenge_decider,
75                   )
76         self.assertEqual(mw.app, app)
77         af = mw.api_factory
78         self.assertEqual(af.identifiers, identifiers)
79         self.assertEqual(af.authenticators, authenticators)
80         self.assertEqual(af.challengers, challengers)
81         self.assertEqual(af.mdproviders, mdproviders)
82         self.assertEqual(af.request_classifier, request_classifier)
83         self.assertEqual(af.challenge_decider, challenge_decider)
84
85     def test_ctor_wo_request_classifier_or_classifier_raises(self):
86         # BBB for old argument name
87         klass = self._getTargetClass()
88         app = DummyApp()
89         identifiers = []
90         authenticators = []
91         challengers = []
92         mdproviders = []
93         challenge_decider = DummyChallengeDecider()
94         self.assertRaises(ValueError,
95                           klass,
96                           app,
97                           identifiers,
98                           authenticators,
99                           challengers,
100                           mdproviders,
101                           challenge_decider = challenge_decider,
102                           )
103
104     def test_ctor_w_request_classifier_and_classifier_raises(self):
105         # BBB for old argument name
106         klass = self._getTargetClass()
107         app = DummyApp()
108         identifiers = []
109         authenticators = []
110         challengers = []
111         request_classifier = DummyRequestClassifier()
112         mdproviders = []
113         challenge_decider = DummyChallengeDecider()
114         self.assertRaises(ValueError,
115                           klass,
116                           app,
117                           identifiers,
118                           authenticators,
119                           challengers,
120                           mdproviders,
121                           request_classifier,
122                           challenge_decider,
123                           classifier = object()
124                           )
125
126     def test_ctor_wo_challenge_decider_raises(self):
127         # BBB for old argument name
128         klass = self._getTargetClass()
129         app = DummyApp()
130         identifiers = []
131         authenticators = []
132         challengers = []
133         request_classifier = DummyRequestClassifier()
134         mdproviders = []
135         self.assertRaises(ValueError,
136                           klass,
137                           app,
138                           identifiers,
139                           authenticators,
140                           challengers,
141                           mdproviders,
142                           classifier = request_classifier,
143                           )
144
145     def test_ctor_w_classifier(self):
146         # BBB for old argument name
147         klass = self._getTargetClass()
148         app = DummyApp()
149         identifiers = []
150         authenticators = []
151         challengers = []
152         request_classifier = DummyRequestClassifier()
153         mdproviders = []
154         challenge_decider = DummyChallengeDecider()
155         mw = klass(app,
156                    identifiers,
157                    authenticators,
158                    challengers,
159                    mdproviders,
160                    classifier = request_classifier,
161                    challenge_decider = challenge_decider,
162                   )
163         self.assertEqual(mw.app, app)
164         af = mw.api_factory
165         self.assertEqual(af.identifiers, identifiers)
166         self.assertEqual(af.authenticators, authenticators)
167         self.assertEqual(af.challengers, challengers)
168         self.assertEqual(af.mdproviders, mdproviders)
169         self.assertEqual(af.request_classifier, request_classifier)
170         self.assertEqual(af.challenge_decider, challenge_decider)
171
172     def test_ctor_accepts_logger(self):
f7efc0 173         import logging
140fb9 174         restore = logging.raiseExceptions
TS 175         logging.raiseExceptions = 0
176         try:
177             logger = logging.Logger('something')
178             logger.setLevel(logging.INFO)
179             mw = self._makeOne(log_stream=logger)
180             self.assertEqual(logger, mw.logger)
181         finally:
182             logging.raiseExceptions = restore
f7efc0 183
TS 184     def test_call_remoteuser_already_set(self):
185         environ = self._makeEnviron({'REMOTE_USER':'admin'})
186         mw = self._makeOne()
187         result = mw(environ, None)
188         self.assertEqual(mw.app.environ, environ)
189         self.assertEqual(result, [])
190
191     def test_call_200_no_plugins(self):
192         environ = self._makeEnviron()
193         headers = [('a', '1')]
194         app = DummyWorkingApp('200 OK', headers)
195         mw = self._makeOne(app=app)
196         start_response = DummyStartResponse()
197         result = mw(environ, start_response)
198         self.assertEqual(mw.app.environ, environ)
199         self.assertEqual(result, ['body'])
200         self.assertEqual(start_response.status, '200 OK')
201         self.assertEqual(start_response.headers, headers)
202
203     def test_call_401_no_challengers(self):
204         environ = self._makeEnviron()
205         headers = [('a', '1')]
206         app = DummyWorkingApp('401 Unauthorized', headers)
207         mw = self._makeOne(app=app)
208         start_response = DummyStartResponse()
209         self.assertRaises(RuntimeError, mw, environ, start_response)
210
211     def test_call_200_no_challengers(self):
212         environ = self._makeEnviron()
213         headers = [('a', '1')]
214         app = DummyWorkingApp('200 OK', headers)
215         credentials = {'login':'chris', 'password':'password'}
216         identifier = DummyIdentifier(credentials)
217         identifiers = [ ('identifier', identifier) ]
218         mw = self._makeOne(app=app, identifiers=identifiers)
219         start_response = DummyStartResponse()
220         result = mw(environ, start_response)
221         self.assertEqual(mw.app.environ, environ)
222         self.assertEqual(result, ['body'])
223         self.assertEqual(start_response.status, '200 OK')
224         self.assertEqual(start_response.headers, headers)
225
226     def test_call_401_no_identifiers(self):
02a504 227         from webob.exc import HTTPUnauthorized
f7efc0 228         environ = self._makeEnviron()
TS 229         headers = [('a', '1')]
230         app = DummyWorkingApp('401 Unauthorized', headers)
231         challenge_app = HTTPUnauthorized()
232         challenge = DummyChallenger(challenge_app)
233         challengers = [ ('challenge', challenge) ]
234         mw = self._makeOne(app=app, challengers=challengers)
235         start_response = DummyStartResponse()
236         result = mw(environ, start_response)
237         self.assertEqual(environ['challenged'], challenge_app)
02a504 238         self.failUnless(result[0].startswith('401 Unauthorized'))
f7efc0 239
TS 240     def test_call_401_challenger_and_identifier_no_authenticator(self):
02a504 241         from webob.exc import HTTPUnauthorized
f7efc0 242         environ = self._makeEnviron()
TS 243         headers = [('a', '1')]
244         app = DummyWorkingApp('401 Unauthorized', headers)
245         challenge_app = HTTPUnauthorized()
246         challenge = DummyChallenger(challenge_app)
247         challengers = [ ('challenge', challenge) ]
248         credentials = {'login':'a', 'password':'b'}
249         identifier = DummyIdentifier(credentials)
250         identifiers = [ ('identifier', identifier) ]
251         mw = self._makeOne(app=app, challengers=challengers,
252                            identifiers=identifiers)
253         start_response = DummyStartResponse()
254
255         result = mw(environ, start_response)
256         self.assertEqual(environ['challenged'], challenge_app)
02a504 257         self.failUnless(result[0].startswith('401 Unauthorized'))
f7efc0 258         self.assertEqual(identifier.forgotten, False)
TS 259         self.assertEqual(environ.get('REMOTE_USER'), None)
260
261     def test_call_401_challenger_and_identifier_and_authenticator(self):
02a504 262         from webob.exc import HTTPUnauthorized
f7efc0 263         environ = self._makeEnviron()
TS 264         headers = [('a', '1')]
265         app = DummyWorkingApp('401 Unauthorized', headers)
266         challenge_app = HTTPUnauthorized()
267         challenge = DummyChallenger(challenge_app)
268         challengers = [ ('challenge', challenge) ]
269         credentials = {'login':'chris', 'password':'password'}
270         identifier = DummyIdentifier(credentials)
271         identifiers = [ ('identifier', identifier) ]
272         authenticator = DummyAuthenticator()
273         authenticators = [ ('authenticator', authenticator) ]
274         mw = self._makeOne(app=app, challengers=challengers,
275                            identifiers=identifiers,
276                            authenticators=authenticators)
277         start_response = DummyStartResponse()
278         result = mw(environ, start_response)
279         self.assertEqual(environ['challenged'], challenge_app)
02a504 280         self.failUnless(result[0].startswith('401 Unauthorized'))
f7efc0 281         # @@ unfuck
TS 282 ##         self.assertEqual(identifier.forgotten, identifier.credentials)
283         self.assertEqual(environ['REMOTE_USER'], 'chris')
284 ##         self.assertEqual(environ['repoze.who.identity'], identifier.credentials)
285
286     def test_call_200_challenger_and_identifier_and_authenticator(self):
02a504 287         from webob.exc import HTTPUnauthorized
f7efc0 288         environ = self._makeEnviron()
TS 289         headers = [('a', '1')]
290         app = DummyWorkingApp('200 OK', headers)
291         challenge_app = HTTPUnauthorized()
292         challenge = DummyChallenger(challenge_app)
293         challengers = [ ('challenge', challenge) ]
294         credentials = {'login':'chris', 'password':'password'}
295         identifier = DummyIdentifier(credentials)
296         identifiers = [ ('identifier', identifier) ]
297         authenticator = DummyAuthenticator()
298         authenticators = [ ('authenticator', authenticator) ]
299         mw = self._makeOne(app=app, challengers=challengers,
300                            identifiers=identifiers,
301                            authenticators=authenticators)
302         start_response = DummyStartResponse()
303         result = mw(environ, start_response)
304         self.assertEqual(environ.get('challenged'), None)
305         self.assertEqual(identifier.forgotten, False)
306         # @@ figure out later
307 ##         self.assertEqual(dict(identifier.remembered)['login'], dict(identifier.credentials)['login'])
308 ##         self.assertEqual(dict(identifier.remembered)['password'], dict(identifier.credentials)['password'])
309         self.assertEqual(environ['REMOTE_USER'], 'chris')
310 ##         self.assertEqual(environ['repoze.who.identity'], identifier.credentials)
311
312
313     def test_call_200_identity_reset(self):
02a504 314         from webob.exc import HTTPUnauthorized
f7efc0 315         environ = self._makeEnviron()
TS 316         headers = [('a', '1')]
317         new_identity = {'user_id':'foo', 'password':'bar'}
318         app = DummyIdentityResetApp('200 OK', headers, new_identity)
319         challenge_app = HTTPUnauthorized()
320         challenge = DummyChallenger(challenge_app)
321         challengers = [ ('challenge', challenge) ]
322         credentials = {'login':'chris', 'password':'password'}
323         identifier = DummyIdentifier(credentials)
324         identifiers = [ ('identifier', identifier) ]
325         authenticator = DummyAuthenticator()
326         authenticators = [ ('authenticator', authenticator) ]
327         mw = self._makeOne(app=app, challengers=challengers,
328                            identifiers=identifiers,
329                            authenticators=authenticators)
330         start_response = DummyStartResponse()
331         result = mw(environ, start_response)
332         self.assertEqual(environ.get('challenged'), None)
333         self.assertEqual(identifier.forgotten, False)
334         new_credentials = identifier.credentials.copy()
335         new_credentials['login'] = 'fred'
336         new_credentials['password'] = 'schooled'
337         # @@ unfuck
338 ##         self.assertEqual(identifier.remembered, new_credentials)
339         self.assertEqual(environ['REMOTE_USER'], 'chris')
340 ##         self.assertEqual(environ['repoze.who.identity'], new_credentials)
341
342     def test_call_200_with_metadata(self):
02a504 343         from webob.exc import HTTPUnauthorized
f7efc0 344         environ = self._makeEnviron()
TS 345         headers = [('a', '1')]
346         app = DummyWorkingApp('200 OK', headers)
347         challenge_app = HTTPUnauthorized()
348         challenge = DummyChallenger(challenge_app)
349         challengers = [ ('challenge', challenge) ]
350         credentials = {'login':'chris', 'password':'password'}
351         identifier = DummyIdentifier(credentials)
352         identifiers = [ ('identifier', identifier) ]
353         authenticator = DummyAuthenticator()
354         authenticators = [ ('authenticator', authenticator) ]
355         mdprovider = DummyMDProvider({'foo':'bar'})
356         mdproviders = [ ('mdprovider', mdprovider) ]
357         mw = self._makeOne(app=app, challengers=challengers,
358                            identifiers=identifiers,
359                            authenticators=authenticators,
360                            mdproviders=mdproviders)
361         start_response = DummyStartResponse()
362         result = mw(environ, start_response)
363         # metadata
364         self.assertEqual(environ['repoze.who.identity']['foo'], 'bar')
365
366     def test_call_ingress_plugin_replaces_application(self):
02a504 367         from webob.exc import HTTPFound
f7efc0 368         environ = self._makeEnviron()
TS 369         headers = [('a', '1')]
370         app = DummyWorkingApp('200 OK', headers)
371         challengers = []
372         credentials = {'login':'chris', 'password':'password'}
373         identifier = DummyIdentifier(
374             credentials,
375             remember_headers=[('a', '1')],
376             replace_app = HTTPFound('http://example.com/redirect')
377             )
378         identifiers = [ ('identifier', identifier) ]
379         authenticator = DummyAuthenticator()
380         authenticators = [ ('authenticator', authenticator) ]
381         mdproviders = []
382         mw = self._makeOne(app=app,
383                            challengers=challengers,
384                            identifiers=identifiers,
385                            authenticators=authenticators,
386                            mdproviders=mdproviders)
387         start_response = DummyStartResponse()
388         result = ''.join(mw(environ, start_response))
389         self.failUnless(result.startswith('302 Found'))
390         self.assertEqual(start_response.status, '302 Found')
391         headers = start_response.headers
02a504 392         #self.assertEqual(len(headers), 3, headers)
AO 393         #self.assertEqual(headers[0],
394         #                 ('Location', 'http://example.com/redirect'))
f7efc0 395         self.assertEqual(headers[2],
02a504 396                          ('Content-Type', 'text/plain; charset=UTF-8'))
AO 397         self.assertEqual(headers[3],
f7efc0 398                          ('a', '1'))
TS 399         self.assertEqual(start_response.exc_info, None)
d853a7 400         self.failIf('repoze.who.application' in environ)
f7efc0 401
TS 402     def test_call_app_doesnt_call_start_response(self):
02a504 403         from webob.exc import HTTPUnauthorized
f7efc0 404         environ = self._makeEnviron()
TS 405         headers = [('a', '1')]
406         app = DummyGeneratorApp('200 OK', headers)
407         challenge_app = HTTPUnauthorized()
408         challenge = DummyChallenger(challenge_app)
409         challengers = [ ('challenge', challenge) ]
410         credentials = {'login':'chris', 'password':'password'}
411         identifier = DummyIdentifier(credentials)
412         identifiers = [ ('identifier', identifier) ]
413         authenticator = DummyAuthenticator()
414         authenticators = [ ('authenticator', authenticator) ]
415         mdprovider = DummyMDProvider({'foo':'bar'})
416         mdproviders = [ ('mdprovider', mdprovider) ]
417         mw = self._makeOne(app=app, challengers=challengers,
418                            identifiers=identifiers,
419                            authenticators=authenticators,
420                            mdproviders=mdproviders)
421         start_response = DummyStartResponse()
422         result = mw(environ, start_response)
423         # metadata
424         self.assertEqual(environ['repoze.who.identity']['foo'], 'bar')
425
b01f44 426     def test_call_w_challenge_closes_iterable(self):
02a504 427         from webob.exc import HTTPUnauthorized
b01f44 428         environ = self._makeEnviron()
TS 429         headers = [('a', '1')]
430         app = DummyIterableWithCloseApp('401 Unauthorized', headers)
431         challenge_app = HTTPUnauthorized()
432         challenge = DummyChallenger(challenge_app)
433         challengers = [ ('challenge', challenge) ]
434         credentials = {'login':'chris', 'password':'password'}
435         identifier = DummyIdentifier(credentials)
436         identifiers = [ ('identifier', identifier) ]
437         authenticator = DummyAuthenticator()
438         authenticators = [ ('authenticator', authenticator) ]
439         mdprovider = DummyMDProvider({'foo':'bar'})
440         mdproviders = [ ('mdprovider', mdprovider) ]
441         mw = self._makeOne(app=app, challengers=challengers,
442                            identifiers=identifiers,
443                            authenticators=authenticators,
444                            mdproviders=mdproviders)
445         start_response = DummyStartResponse()
446         result = mw(environ, start_response)
02a504 447         self.failUnless(result[0].startswith('401 Unauthorized'))
b01f44 448         self.failUnless(app._iterable._closed)
TS 449
493726 450     def test_call_w_challenge_but_no_challenger_still_closes_iterable(self):
TS 451         environ = self._makeEnviron()
452         headers = [('a', '1')]
453         app = DummyIterableWithCloseApp('401 Unauthorized', headers)
454         challengers = []
455         credentials = {'login':'chris', 'password':'password'}
456         identifier = DummyIdentifier(credentials)
457         identifiers = [ ('identifier', identifier) ]
458         authenticator = DummyAuthenticator()
459         authenticators = [ ('authenticator', authenticator) ]
460         mdprovider = DummyMDProvider({'foo':'bar'})
461         mdproviders = [ ('mdprovider', mdprovider) ]
462         mw = self._makeOne(app=app, challengers=challengers,
463                            identifiers=identifiers,
464                            authenticators=authenticators,
465                            mdproviders=mdproviders)
466         start_response = DummyStartResponse()
467         self.assertRaises(RuntimeError, mw, environ, start_response)
468         self.failUnless(app._iterable._closed)
469
f7efc0 470     # XXX need more call tests:
TS 471     #  - auth_id sorting
472
473 class TestStartResponseWrapper(unittest.TestCase):
474
475     def _getTargetClass(self):
476         from repoze.who.middleware import StartResponseWrapper
477         return StartResponseWrapper
478
479     def _makeOne(self, *arg, **kw):
480         plugin = self._getTargetClass()(*arg, **kw)
481         return plugin
482
483     def test_ctor(self):
484         wrapper = self._makeOne(None)
485         self.assertEqual(wrapper.start_response, None)
486         self.assertEqual(wrapper.headers, [])
487         self.failUnless(wrapper.buffer)
488
489     def test_finish_response(self):
b482a1 490         from StringIO import StringIO
f7efc0 491         statuses = []
TS 492         headerses = []
493         datases = []
494         closededs = []
495         def write(data):
496             datases.append(data)
497         def close():
498             closededs.append(True)
499         write.close = close
500
501         def start_response(status, headers, exc_info=None):
502             statuses.append(status)
503             headerses.append(headers)
504             return write
505
506         wrapper = self._makeOne(start_response)
507         wrapper.status = '401 Unauthorized'
508         wrapper.headers = [('a', '1')]
509         wrapper.buffer = StringIO('written')
510         extra_headers = [('b', '2')]
511         result = wrapper.finish_response(extra_headers)
512         self.assertEqual(result, None)
513         self.assertEqual(headerses[0], wrapper.headers + extra_headers)
514         self.assertEqual(statuses[0], wrapper.status)
515         self.assertEqual(datases[0], 'written')
516         self.assertEqual(closededs[0], True)
517
518 class WrapGeneratorTests(unittest.TestCase):
519
b01f44 520     def _callFUT(self, iterable):
f7efc0 521         from repoze.who.middleware import wrap_generator
b01f44 522         return wrap_generator(iterable)
f7efc0 523
b01f44 524     def test_w_generator(self):
f7efc0 525         L = []
TS 526         def gen(L=L):
527             L.append('yo!')
528             yield 'a'
529             yield 'b'
b01f44 530         newgen = self._callFUT(gen())
f7efc0 531         self.assertEqual(L, ['yo!'])
TS 532         self.assertEqual(list(newgen), ['a', 'b'])
b01f44 533
TS 534     def test_w_iterator_having_close(self):
535         def gen():
536             yield 'a'
537             yield 'b'
538         iterable = DummyIterableWithClose(gen())
539         newgen = self._callFUT(iterable)
540         self.failIf(iterable._closed)
541         self.assertEqual(list(newgen), ['a', 'b'])
542         self.failUnless(iterable._closed)
f7efc0 543
d32c12 544 class TestMakeTestMiddleware(unittest.TestCase):
TS 545
546     def setUp(self):
547         import os
548         self._old_WHO_LOG = os.environ.get('WHO_LOG')
549
550     def tearDown(self):
551         import os
552         if self._old_WHO_LOG is not None:
553             os.environ['WHO_LOG'] = self._old_WHO_LOG
554         else:
555             if 'WHO_LOG' in os.environ:
556                 del os.environ['WHO_LOG']
557
558     def _getFactory(self):
559         from repoze.who.middleware import make_test_middleware
560         return make_test_middleware
561
562     def test_it_no_WHO_LOG_in_environ(self):
563         app = DummyApp()
564         factory = self._getFactory()
565         global_conf = {'here': '/'}
566         middleware = factory(app, global_conf)
993216 567         api_factory = middleware.api_factory
d7f613 568         self.assertEqual(len(api_factory.identifiers), 2)
993216 569         self.assertEqual(len(api_factory.authenticators), 1)
TS 570         self.assertEqual(len(api_factory.challengers), 2)
571         self.assertEqual(len(api_factory.mdproviders), 0)
d32c12 572         self.assertEqual(middleware.logger, None)
TS 573
574     def test_it_w_WHO_LOG_in_environ(self):
575         import logging
576         import os
577         os.environ['WHO_LOG'] = '1'
578         app = DummyApp()
579         factory = self._getFactory()
580         global_conf = {'here': '/'}
581         middleware = factory(app, global_conf)
582         self.assertEqual(middleware.logger.getEffectiveLevel(), logging.DEBUG)
583
f7efc0 584 class DummyApp:
TS 585     environ = None
586     def __call__(self, environ, start_response):
587         self.environ = environ
588         return []
589
590 class DummyWorkingApp:
591     def __init__(self, status, headers):
592         self.status = status
593         self.headers = headers
594
595     def __call__(self, environ, start_response):
596         self.environ = environ
597         start_response(self.status, self.headers)
598         return ['body']
599
600 class DummyGeneratorApp:
601     def __init__(self, status, headers):
602         self.status = status
603         self.headers = headers
604
605     def __call__(self, environ, start_response):
606         def gen(self=self, start_response=start_response):
607             self.environ = environ
608             start_response(self.status, self.headers)
609             yield 'body'
610         return gen()
611
b01f44 612 class DummyIterableWithClose:
TS 613     _closed = False
614     def __init__(self, iterable):
615         self._iterable = iterable
616     def __iter__(self):
617         return iter(self._iterable)
618     def close(self):
619         self._closed = True
620
621 class DummyIterableWithCloseApp:
622     def __init__(self, status, headers):
623         self.status = status
624         self.headers = headers
625         self._iterable = DummyIterableWithClose(['body'])
626
627     def __call__(self, environ, start_response):
628         self.environ = environ
629         start_response(self.status, self.headers)
630         return self._iterable
631
f7efc0 632 class DummyIdentityResetApp:
TS 633     def __init__(self, status, headers, new_identity):
634         self.status = status
635         self.headers = headers
636         self.new_identity = new_identity
637
638     def __call__(self, environ, start_response):
639         self.environ = environ
640         environ['repoze.who.identity']['login'] = 'fred'
641         environ['repoze.who.identity']['password'] = 'schooled'
642         start_response(self.status, self.headers)
643         return ['body']
644
645 class DummyChallenger:
646     def __init__(self, app=None):
647         self.app = app
648
649     def challenge(self, environ, status, app_headers, forget_headers):
650         environ['challenged'] = self.app
651         return self.app
652
653 class DummyIdentifier:
654     forgotten = False
655     remembered = False
656
657     def __init__(self, credentials=None, remember_headers=None,
658                  forget_headers=None, replace_app=None):
659         self.credentials = credentials
660         self.remember_headers = remember_headers
661         self.forget_headers = forget_headers
662         self.replace_app = replace_app
663
664     def identify(self, environ):
665         if self.replace_app:
666             environ['repoze.who.application'] = self.replace_app
667         return self.credentials
668
669     def forget(self, environ, identity):
670         self.forgotten = identity
671         return self.forget_headers
672
673     def remember(self, environ, identity):
674         self.remembered = identity
675         return self.remember_headers
676
677 class DummyAuthenticator:
678     def __init__(self, userid=None):
679         self.userid = userid
680
681     def authenticate(self, environ, credentials):
682         if self.userid is None:
683             return credentials['login']
684         return self.userid
685
686 class DummyFailAuthenticator:
687     def authenticate(self, environ, credentials):
688         return None
689
690 class DummyRequestClassifier:
691     def __call__(self, environ):
692         return 'browser'
693
694 class DummyChallengeDecider:
695     def __call__(self, environ, status, headers):
696         if status.startswith('401 '):
697             return True
698
699 class DummyNoResultsIdentifier:
700     def identify(self, environ):
701         return None
702
703     def remember(self, *arg, **kw):
704         pass
705
706     def forget(self, *arg, **kw):
707         pass
708
709 class DummyStartResponse:
710     def __call__(self, status, headers, exc_info=None):
711         self.status = status
712         self.headers = headers
713         self.exc_info = exc_info
714         return []
715
716 class DummyMDProvider:
717     def __init__(self, metadata=None):
718         self._metadata = metadata
719
720     def add_metadata(self, environ, identity):
721         return identity.update(self._metadata)
722
723 class DummyMultiPlugin:
724     pass