Source Code for Module repoze.who.tests.test_middleware

   1  import unittest 
   2   
   3 -class TestMiddleware(unittest.TestCase): 
   4   
   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, 
  14                   classifier=None, 
  15                   mdproviders=None, 
  16                   challenge_decider=None, 
  17                   log_stream=None, 
  18                   log_level=None, 
  19                   ): 
  20          if app is None: 
  21              app = DummyApp() 
  22          if identifiers is None: 
  23              identifiers = [] 
  24          if authenticators is None: 
  25              authenticators = [] 
  26          if challengers is None: 
  27              challengers = [] 
  28          if classifier is None: 
  29              classifier = DummyRequestClassifier() 
  30          if mdproviders is None: 
  31              mdproviders = [] 
  32          if challenge_decider is None: 
  33              challenge_decider = DummyChallengeDecider() 
  34          if log_level is None: 
  35              import logging 
  36              log_level = logging.DEBUG 
  37          mw = self._getTargetClass()(app, 
  38                                      identifiers, 
  39                                      authenticators, 
  40                                      challengers, 
  41                                      mdproviders, 
  42                                      classifier, 
  43                                      challenge_decider, 
  44                                      log_stream, 
  45                                      log_level=logging.DEBUG) 
  46          return mw 
  47   
  48 -    def _makeEnviron(self, kw=None): 
  49          environ = {} 
  50          environ['wsgi.version'] = (1,0) 
  51          if kw is not None: 
  52              environ.update(kw) 
  53          return environ 
  54   
  55 -    def test_accepts_logger(self): 
  56          import logging 
  57          logger = logging.Logger('something') 
  58          logger.setLevel(logging.INFO) 
  59          mw = self._makeOne(log_stream=logger) 
  60          self.assertEqual(logger, mw.logger) 
  61   
  62 -    def test_identify_success(self): 
  63          environ = self._makeEnviron() 
  64          credentials = {'login':'chris', 'password':'password'} 
  65          identifier = DummyIdentifier(credentials) 
  66          identifiers = [ ('i', identifier) ] 
  67          mw = self._makeOne(identifiers=identifiers) 
  68          results = mw.identify(environ, None) 
  69          self.assertEqual(len(results), 1) 
  70          new_identifier, identity = results[0] 
  71          self.assertEqual(new_identifier, identifier) 
  72          self.assertEqual(identity['login'], 'chris') 
  73          self.assertEqual(identity['password'], 'password') 
  74   
  75 -    def test_identify_success_empty_identity(self): 
  76          environ = self._makeEnviron() 
  77          identifier = DummyIdentifier({}) 
  78          identifiers = [ ('i', identifier) ] 
  79          mw = self._makeOne(identifiers=identifiers) 
  80          results = mw.identify(environ, None) 
  81          self.assertEqual(len(results), 1) 
  82          new_identifier, identity = results[0] 
  83          self.assertEqual(new_identifier, identifier) 
  84          self.assertEqual(identity, {}) 
  85   
  86 -    def test_identify_fail(self): 
  87          environ = self._makeEnviron() 
  88          plugin = DummyNoResultsIdentifier() 
  89          plugins = [ ('dummy', plugin) ] 
  90          mw = self._makeOne(identifiers=plugins) 
  91          results = mw.identify(environ, None) 
  92          self.assertEqual(len(results), 0) 
  93   
  94 -    def test_identify_success_skip_noresults(self): 
  95          environ = self._makeEnviron() 
  96          mw = self._makeOne() 
  97          plugin1 = DummyNoResultsIdentifier() 
  98          credentials = {'login':'chris', 'password':'password'} 
  99          plugin2 = DummyIdentifier(credentials) 
 100          plugins = [ ('identifier1', plugin1), ('identifier2', plugin2) ] 
 101          mw = self._makeOne(identifiers=plugins) 
 102          results = mw.identify(environ, None) 
 103          self.assertEqual(len(results), 1) 
 104          new_identifier, identity = results[0] 
 105          self.assertEqual(new_identifier, plugin2) 
 106          self.assertEqual(identity['login'], 'chris') 
 107          self.assertEqual(identity['password'], 'password') 
 108   
 109 -    def test_identify_success_multiresults(self): 
 110          environ = self._makeEnviron() 
 111          mw = self._makeOne() 
 112          plugin1 = DummyIdentifier({'login':'fred','password':'fred'}) 
 113          plugin2 = DummyIdentifier({'login':'bob','password':'bob'}) 
 114          plugins = [ ('identifier1', plugin1), ('identifier2', plugin2) ] 
 115          mw = self._makeOne(identifiers=plugins) 
 116          results = mw.identify(environ, None) 
 117          self.assertEqual(len(results), 2) 
 118          new_identifier, identity = results[0] 
 119          self.assertEqual(new_identifier, plugin1) 
 120          self.assertEqual(identity['login'], 'fred') 
 121          self.assertEqual(identity['password'], 'fred') 
 122          new_identifier, identity = results[1] 
 123          self.assertEqual(new_identifier, plugin2) 
 124          self.assertEqual(identity['login'], 'bob') 
 125          self.assertEqual(identity['password'], 'bob') 
 126   
 127 -    def test_identify_find_implicit_classifier(self): 
 128          environ = self._makeEnviron() 
 129          mw = self._makeOne() 
 130          plugin1 = DummyIdentifier({'login':'fred','password':'fred'}) 
 131          from repoze.who.interfaces import IIdentifier 
 132          plugin1.classifications = {IIdentifier:['nomatch']} 
 133          plugin2 = DummyIdentifier({'login':'bob','password':'bob'}) 
 134          plugins = [ ('identifier1', plugin1),  ('identifier2', plugin2) ] 
 135          mw = self._makeOne(identifiers=plugins) 
 136          results = mw.identify(environ, 'match') 
 137          self.assertEqual(len(results), 1) 
 138          plugin, creds = results[0] 
 139          self.assertEqual(creds['login'], 'bob') 
 140          self.assertEqual(creds['password'], 'bob') 
 141          self.assertEqual(plugin, plugin2) 
 142   
 143 -    def test_identify_find_explicit_classifier(self): 
 144          environ = self._makeEnviron() 
 145          from repoze.who.interfaces import IIdentifier 
 146          plugin1 = DummyIdentifier({'login':'fred','password':'fred'}) 
 147          plugin1.classifications = {IIdentifier:['nomatch']} 
 148          plugin2 = DummyIdentifier({'login':'bob','password':'bob'}) 
 149          plugin2.classifications = {IIdentifier:['match']} 
 150          plugins= [ ('identifier1', plugin1), ('identifier2', plugin2) ] 
 151          mw = self._makeOne(identifiers=plugins) 
 152          results = mw.identify(environ, 'match') 
 153          self.assertEqual(len(results), 1) 
 154          plugin, creds = results[0] 
 155          self.assertEqual(creds['login'], 'bob') 
 156          self.assertEqual(creds['password'], 'bob') 
 157          self.assertEqual(plugin, plugin2) 
 158   
 159 -    def test_authenticate_success(self): 
 160          environ = self._makeEnviron() 
 161          plugin1 = DummyAuthenticator('a') 
 162          plugins = [ ('identifier1', plugin1) ] 
 163          mw = self._makeOne(authenticators=plugins) 
 164          identities = [ (None, {'login':'chris', 'password':'password'}) ] 
 165          results = mw.authenticate(environ, None, identities) 
 166          self.assertEqual(len(results), 1) 
 167          result = results[0] 
 168          rank, authenticator, identifier, creds, userid = result 
 169          self.assertEqual(rank, (0,0)) 
 170          self.assertEqual(authenticator, plugin1) 
 171          self.assertEqual(identifier, None) 
 172          self.assertEqual(creds['login'], 'chris') 
 173          self.assertEqual(creds['password'], 'password') 
 174          self.assertEqual(userid, 'a') 
 175   
 176 -    def test_authenticate_fail(self): 
 177          environ = self._makeEnviron() 
 178          mw = self._makeOne() # no authenticators 
 179          identities = [ (None, {'login':'chris', 'password':'password'}) ] 
 180          result = mw.authenticate(environ, None, identities) 
 181          self.assertEqual(len(result), 0) 
 182   
 183 -    def test_authenticate_success_skip_fail(self): 
 184          environ = self._makeEnviron() 
 185          mw = self._makeOne() 
 186          plugin1 = DummyFailAuthenticator() 
 187          plugin2 = DummyAuthenticator() 
 188          plugins = [ ('dummy1', plugin1), ('dummy2', plugin2) ] 
 189          mw = self._makeOne(authenticators=plugins) 
 190          creds = {'login':'chris', 'password':'password'} 
 191          identities = [ (None, {'login':'chris', 'password':'password'}) ] 
 192          results = mw.authenticate(environ, None, identities) 
 193          self.assertEqual(len(results), 1) 
 194          result = results[0] 
 195          rank, authenticator, identifier, creds, userid = result 
 196          self.assertEqual(rank, (1,0)) 
 197          self.assertEqual(authenticator, plugin2) 
 198          self.assertEqual(identifier, None) 
 199          self.assertEqual(creds['login'], 'chris') 
 200          self.assertEqual(creds['password'], 'password') 
 201          self.assertEqual(userid, 'chris') 
 202   
 203 -    def test_authenticate_success_multiresult(self): 
 204          environ = self._makeEnviron() 
 205          mw = self._makeOne() 
 206          plugin1 = DummyAuthenticator('chris_id1') 
 207          plugin2 = DummyAuthenticator('chris_id2') 
 208          plugins = [ ('dummy1',plugin1), ('dummy2',plugin2) ] 
 209          mw = self._makeOne(authenticators=plugins) 
 210          creds = {'login':'chris', 'password':'password'} 
 211          identities = [ (None, {'login':'chris', 'password':'password'}) ] 
 212          results = mw.authenticate(environ, None, identities) 
 213          self.assertEqual(len(results), 2) 
 214          result = results[0] 
 215          rank, authenticator, identifier, creds, userid = result 
 216          self.assertEqual(rank, (0,0,)) 
 217          self.assertEqual(authenticator, plugin1) 
 218          self.assertEqual(identifier, None) 
 219          self.assertEqual(creds['login'], 'chris') 
 220          self.assertEqual(creds['password'], 'password') 
 221          self.assertEqual(userid, 'chris_id1') 
 222          result = results[1] 
 223          rank, authenticator, identifier, creds, userid = result 
 224          self.assertEqual(rank, (1,0)) 
 225          self.assertEqual(authenticator, plugin2) 
 226          self.assertEqual(identifier, None) 
 227          self.assertEqual(creds['login'], 'chris') 
 228          self.assertEqual(creds['password'], 'password') 
 229          self.assertEqual(userid, 'chris_id2') 
 230   
 231 -    def test_authenticate_find_implicit_classifier(self): 
 232          environ = self._makeEnviron() 
 233          mw = self._makeOne() 
 234          plugin1 = DummyAuthenticator('chris_id1') 
 235          from repoze.who.interfaces import IAuthenticator 
 236          plugin1.classifications = {IAuthenticator:['nomatch']} 
 237          plugin2 = DummyAuthenticator('chris_id2') 
 238          plugins = [ ('auth1', plugin1), ('auth2', plugin2) ] 
 239          mw = self._makeOne(authenticators = plugins) 
 240          identities = [ (None, {'login':'chris', 'password':'password'}) ] 
 241          results = mw.authenticate(environ, 'match', identities) 
 242          self.assertEqual(len(results), 1) 
 243          result = results[0] 
 244          rank, authenticator, identifier, creds, userid = result 
 245          self.assertEqual(rank, (0,0)) 
 246          self.assertEqual(authenticator, plugin2) 
 247          self.assertEqual(identifier, None) 
 248          self.assertEqual(creds['login'], 'chris') 
 249          self.assertEqual(creds['password'], 'password') 
 250          self.assertEqual(userid, 'chris_id2') 
 251   
 252 -    def test_authenticate_find_explicit_classifier(self): 
 253          environ = self._makeEnviron() 
 254          mw = self._makeOne() 
 255          from repoze.who.interfaces import IAuthenticator 
 256          plugin1 = DummyAuthenticator('chris_id1') 
 257          plugin1.classifications = {IAuthenticator:['nomatch']} 
 258          plugin2 = DummyAuthenticator('chris_id2') 
 259          plugin2.classifications = {IAuthenticator:['match']} 
 260          plugins = [ ('auth1', plugin1), ('auth2', plugin2) ] 
 261          mw = self._makeOne(authenticators = plugins) 
 262          identities = [ (None, {'login':'chris', 'password':'password'}) ] 
 263          results = mw.authenticate(environ, 'match', identities) 
 264          self.assertEqual(len(results), 1) 
 265          result = results[0] 
 266          rank, authenticator, identifier, creds, userid = result 
 267          self.assertEqual(rank, (0, 0)) 
 268          self.assertEqual(authenticator, plugin2) 
 269          self.assertEqual(identifier, None) 
 270          self.assertEqual(creds['login'], 'chris') 
 271          self.assertEqual(creds['password'], 'password') 
 272          self.assertEqual(userid, 'chris_id2') 
 273   
 274 -    def test_authenticate_user_null_but_not_none(self): 
 275          environ = self._makeEnviron() 
 276          plugin1 = DummyAuthenticator(0) 
 277          plugins = [ ('identifier1', plugin1) ] 
 278          mw = self._makeOne(authenticators=plugins) 
 279          identities = [ (None, {'login':'chris', 'password':'password'}) ] 
 280          results = mw.authenticate(environ, None, identities) 
 281          self.assertEqual(len(results), 1) 
 282          result = results[0] 
 283          rank, authenticator, identifier, creds, userid = result 
 284          self.assertEqual(rank, (0,0)) 
 285          self.assertEqual(authenticator, plugin1) 
 286          self.assertEqual(identifier, None) 
 287          self.assertEqual(creds['login'], 'chris') 
 288          self.assertEqual(creds['password'], 'password') 
 289          self.assertEqual(userid, 0) 
 290   
 291 -    def test_authenticate_success_multiresult_one_preauthenticated(self): 
 292          environ = self._makeEnviron() 
 293          mw = self._makeOne() 
 294          preauth = DummyIdentifier({'repoze.who.userid':'preauthenticated'}) 
 295          plugin1 = DummyAuthenticator('chris_id1') 
 296          plugin2 = DummyAuthenticator('chris_id2') 
 297          plugins = [ ('dummy1',plugin1), ('dummy2',plugin2) ] 
 298          mw = self._makeOne(authenticators=plugins) 
 299          creds = {'login':'chris', 'password':'password'} 
 300          identities = [ (None, {'login':'chris', 'password':'password'}), 
 301                         (preauth, preauth.credentials) ] 
 302          results = mw.authenticate(environ, None, identities) 
 303          self.assertEqual(len(results), 3) 
 304          result = results[0] 
 305          rank, authenticator, identifier, creds, userid = result 
 306          self.assertEqual(rank, (0,0,)) 
 307          self.assertEqual(authenticator, None) 
 308          self.assertEqual(identifier, preauth) 
 309          self.assertEqual(creds['repoze.who.userid'], 'preauthenticated') 
 310          self.assertEqual(userid, 'preauthenticated') 
 311          result = results[1] 
 312          rank, authenticator, identifier, creds, userid = result 
 313          self.assertEqual(rank, (0,1)) 
 314          self.assertEqual(authenticator, plugin1) 
 315          self.assertEqual(identifier, None) 
 316          self.assertEqual(creds['login'], 'chris') 
 317          self.assertEqual(creds['password'], 'password') 
 318          self.assertEqual(userid, 'chris_id1') 
 319          result = results[2] 
 320          rank, authenticator, identifier, creds, userid = result 
 321          self.assertEqual(rank, (1,1)) 
 322          self.assertEqual(authenticator, plugin2) 
 323          self.assertEqual(identifier, None) 
 324          self.assertEqual(creds['login'], 'chris') 
 325          self.assertEqual(creds['password'], 'password') 
 326          self.assertEqual(userid, 'chris_id2') 
 327   
 328 -    def test_challenge_noidentifier_noapp(self): 
 329          environ = self._makeEnviron() 
 330          challenger = DummyChallenger() 
 331          plugins = [ ('challenge', challenger) ] 
 332          mw = self._makeOne(challengers = plugins) 
 333          identity = {'login':'chris', 'password':'password'} 
 334          app = mw.challenge(environ, 'match', '401 Unauthorized', 
 335                             [], None, identity) 
 336          self.assertEqual(app, None) 
 337          self.assertEqual(environ['challenged'], app) 
 338   
 339 -    def test_challenge_noidentifier_withapp(self): 
 340          environ = self._makeEnviron() 
 341          app = DummyApp() 
 342          challenger = DummyChallenger(app) 
 343          plugins = [ ('challenge', challenger) ] 
 344          mw = self._makeOne(challengers = plugins) 
 345          identity = {'login':'chris', 'password':'password'} 
 346          result = mw.challenge(environ, 'match', '401 Unauthorized', 
 347                                 [], None, identity) 
 348          self.assertEqual(result, app) 
 349          self.assertEqual(environ['challenged'], app) 
 350   
 351 -    def test_challenge_identifier_noapp(self): 
 352          environ = self._makeEnviron() 
 353          challenger = DummyChallenger() 
 354          credentials = {'login':'chris', 'password':'password'} 
 355          identifier = DummyIdentifier(credentials) 
 356          plugins = [ ('challenge', challenger) ] 
 357          mw = self._makeOne(challengers = plugins) 
 358          identity = {'login':'chris', 'password':'password'} 
 359          result = mw.challenge(environ, 'match', '401 Unauthorized', 
 360                                [], identifier, identity) 
 361          self.assertEqual(result, None) 
 362          self.assertEqual(environ['challenged'], None) 
 363          self.assertEqual(identifier.forgotten, identity) 
 364   
 365 -    def test_challenge_identifier_app(self): 
 366          environ = self._makeEnviron() 
 367          app = DummyApp() 
 368          challenger = DummyChallenger(app) 
 369          credentials = {'login':'chris', 'password':'password'} 
 370          identifier = DummyIdentifier(credentials) 
 371          plugins = [ ('challenge', challenger) ] 
 372          mw = self._makeOne(challengers = plugins) 
 373          identity = {'login':'chris', 'password':'password'} 
 374          result = mw.challenge(environ, 'match', '401 Unauthorized', 
 375                                 [], identifier, identity) 
 376          self.assertEqual(result, app) 
 377          self.assertEqual(environ['challenged'], app) 
 378          self.assertEqual(identifier.forgotten, identity) 
 379   
 380 -    def test_challenge_identifier_forget_headers(self): 
 381          FORGET_HEADERS = [('X-testing-forget', 'Oubliez!')] 
 382          environ = self._makeEnviron() 
 383          app = DummyApp() 
 384          challenger = DummyChallenger(app) 
 385          credentials = {'login':'chris', 'password':'password'} 
 386          identifier = DummyIdentifier(credentials, 
 387                                       forget_headers=FORGET_HEADERS) 
 388          plugins = [ ('challenge', challenger) ] 
 389          mw = self._makeOne(challengers = plugins) 
 390          identity = {'login':'chris', 'password':'password'} 
 391          result = mw.challenge(environ, 'match', '401 Unauthorized', 
 392                                 [], identifier, identity) 
 393   
 394 -    def test_multi_challenge_firstwins(self): 
 395          environ = self._makeEnviron() 
 396          app1 = DummyApp() 
 397          app2 = DummyApp() 
 398          challenger1 = DummyChallenger(app1) 
 399          challenger2 = DummyChallenger(app2) 
 400          credentials = {'login':'chris', 'password':'password'} 
 401          identifier = DummyIdentifier(credentials) 
 402          plugins = [ ('challenge1', challenger1), ('challenge2', challenger2) ] 
 403          mw = self._makeOne(challengers = plugins) 
 404          identity = {'login':'chris', 'password':'password'} 
 405          result = mw.challenge(environ, 'match', '401 Unauthorized', 
 406                                [], identifier, identity) 
 407          self.assertEqual(result, app1) 
 408          self.assertEqual(environ['challenged'], app1) 
 409          self.assertEqual(identifier.forgotten, identity) 
 410   
 411 -    def test_multi_challenge_skipnomatch_findimplicit(self): 
 412          environ = self._makeEnviron() 
 413          app1 = DummyApp() 
 414          app2 = DummyApp() 
 415          from repoze.who.interfaces import IChallenger 
 416          challenger1 = DummyChallenger(app1) 
 417          challenger1.classifications = {IChallenger:['nomatch']} 
 418          challenger2 = DummyChallenger(app2) 
 419          challenger2.classifications = {IChallenger:None} 
 420          credentials = {'login':'chris', 'password':'password'} 
 421          identifier = DummyIdentifier(credentials) 
 422          plugins = [ ('challenge1', challenger1), ('challenge2', challenger2) ] 
 423          mw = self._makeOne(challengers = plugins) 
 424          identity = {'login':'chris', 'password':'password'} 
 425          result = mw.challenge(environ, 'match', '401 Unauthorized', 
 426                                 [], identifier, identity) 
 427          self.assertEqual(result, app2) 
 428          self.assertEqual(environ['challenged'], app2) 
 429          self.assertEqual(identifier.forgotten, identity) 
 430   
 431 -    def test_multi_challenge_skipnomatch_findexplicit(self): 
 432          environ = self._makeEnviron() 
 433          app1 = DummyApp() 
 434          app2 = DummyApp() 
 435          from repoze.who.interfaces import IChallenger 
 436          challenger1 = DummyChallenger(app1) 
 437          challenger1.classifications = {IChallenger:['nomatch']} 
 438          challenger2 = DummyChallenger(app2) 
 439          challenger2.classifications = {IChallenger:['match']} 
 440          credentials = {'login':'chris', 'password':'password'} 
 441          identifier = DummyIdentifier(credentials) 
 442          plugins = [ ('challenge1', challenger1), ('challenge2', challenger2) ] 
 443          mw = self._makeOne(challengers = plugins) 
 444          identity = {'login':'chris', 'password':'password'} 
 445          result = mw.challenge(environ, 'match', '401 Unauthorized', 
 446                                 [], identifier, identity) 
 447          self.assertEqual(result, app2) 
 448          self.assertEqual(environ['challenged'], app2) 
 449          self.assertEqual(identifier.forgotten, identity) 
 450   
 451 -    def test_add_metadata(self): 
 452          environ = self._makeEnviron() 
 453          plugin1 = DummyMDProvider({'foo':'bar'}) 
 454          plugin2 = DummyMDProvider({'fuz':'baz'}) 
 455          plugins = [ ('meta1', plugin1), ('meta2', plugin2) ] 
 456          mw = self._makeOne(mdproviders=plugins) 
 457          classification = '' 
 458          identity = {} 
 459          results = mw.add_metadata(environ, classification, identity) 
 460          self.assertEqual(identity['foo'], 'bar') 
 461          self.assertEqual(identity['fuz'], 'baz') 
 462   
 463 -    def test_add_metadata_w_classification(self): 
 464          environ = self._makeEnviron() 
 465          plugin1 = DummyMDProvider({'foo':'bar'}) 
 466          plugin2 = DummyMDProvider({'fuz':'baz'}) 
 467          from repoze.who.interfaces import IMetadataProvider 
 468          plugin2.classifications = {IMetadataProvider:['foo']} 
 469          plugins = [ ('meta1', plugin1), ('meta2', plugin2) ] 
 470          mw = self._makeOne(mdproviders=plugins) 
 471          classification = 'monkey' 
 472          identity = {} 
 473          mw.add_metadata(environ, classification, identity) 
 474          self.assertEqual(identity['foo'], 'bar') 
 475          self.assertEqual(identity.get('fuz'), None) 
 476   
 477 -    def test_call_remoteuser_already_set(self): 
 478          environ = self._makeEnviron({'REMOTE_USER':'admin'}) 
 479          mw = self._makeOne() 
 480          result = mw(environ, None) 
 481          self.assertEqual(mw.app.environ, environ) 
 482          self.assertEqual(result, []) 
 483   
 484 -    def test_call_200_no_plugins(self): 
 485          environ = self._makeEnviron() 
 486          headers = [('a', '1')] 
 487          app = DummyWorkingApp('200 OK', headers) 
 488          mw = self._makeOne(app=app) 
 489          start_response = DummyStartResponse() 
 490          result = mw(environ, start_response) 
 491          self.assertEqual(mw.app.environ, environ) 
 492          self.assertEqual(result, ['body']) 
 493          self.assertEqual(start_response.status, '200 OK') 
 494          self.assertEqual(start_response.headers, headers) 
 495   
 496 -    def test_call_401_no_challengers(self): 
 497          environ = self._makeEnviron() 
 498          headers = [('a', '1')] 
 499          app = DummyWorkingApp('401 Unauthorized', headers) 
 500          mw = self._makeOne(app=app) 
 501          start_response = DummyStartResponse() 
 502          self.assertRaises(RuntimeError, mw, environ, start_response) 
 503   
 504 -    def test_call_200_no_challengers(self): 
 505          environ = self._makeEnviron() 
 506          headers = [('a', '1')] 
 507          app = DummyWorkingApp('200 OK', headers) 
 508          credentials = {'login':'chris', 'password':'password'} 
 509          identifier = DummyIdentifier(credentials) 
 510          identifiers = [ ('identifier', identifier) ] 
 511          mw = self._makeOne(app=app, identifiers=identifiers) 
 512          start_response = DummyStartResponse() 
 513          result = mw(environ, start_response) 
 514          self.assertEqual(mw.app.environ, environ) 
 515          self.assertEqual(result, ['body']) 
 516          self.assertEqual(start_response.status, '200 OK') 
 517          self.assertEqual(start_response.headers, headers) 
 518   
 519 -    def test_call_401_no_identifiers(self): 
 520          environ = self._makeEnviron() 
 521          headers = [('a', '1')] 
 522          app = DummyWorkingApp('401 Unauthorized', headers) 
 523          from paste.httpexceptions import HTTPUnauthorized 
 524          challenge_app = HTTPUnauthorized() 
 525          challenge = DummyChallenger(challenge_app) 
 526          challengers = [ ('challenge', challenge) ] 
 527          mw = self._makeOne(app=app, challengers=challengers) 
 528          start_response = DummyStartResponse() 
 529          result = mw(environ, start_response) 
 530          self.assertEqual(environ['challenged'], challenge_app) 
 531          self.failUnless(result[0].startswith('401 Unauthorized\r\n')) 
 532   
 533 -    def test_call_401_challenger_and_identifier_no_authenticator(self): 
 534          environ = self._makeEnviron() 
 535          headers = [('a', '1')] 
 536          app = DummyWorkingApp('401 Unauthorized', headers) 
 537          from paste.httpexceptions import HTTPUnauthorized 
 538          challenge_app = HTTPUnauthorized() 
 539          challenge = DummyChallenger(challenge_app) 
 540          challengers = [ ('challenge', challenge) ] 
 541          credentials = {'login':'a', 'password':'b'} 
 542          identifier = DummyIdentifier(credentials) 
 543          identifiers = [ ('identifier', identifier) ] 
 544          mw = self._makeOne(app=app, challengers=challengers, 
 545                             identifiers=identifiers) 
 546          start_response = DummyStartResponse() 
 547   
 548          result = mw(environ, start_response) 
 549          self.assertEqual(environ['challenged'], challenge_app) 
 550          self.failUnless(result[0].startswith('401 Unauthorized\r\n')) 
 551          self.assertEqual(identifier.forgotten, False) 
 552          self.assertEqual(environ.get('REMOTE_USER'), None) 
 553   
 554 -    def test_call_401_challenger_and_identifier_and_authenticator(self): 
 555          environ = self._makeEnviron() 
 556          headers = [('a', '1')] 
 557          app = DummyWorkingApp('401 Unauthorized', headers) 
 558          from paste.httpexceptions import HTTPUnauthorized 
 559          challenge_app = HTTPUnauthorized() 
 560          challenge = DummyChallenger(challenge_app) 
 561          challengers = [ ('challenge', challenge) ] 
 562          credentials = {'login':'chris', 'password':'password'} 
 563          identifier = DummyIdentifier(credentials) 
 564          identifiers = [ ('identifier', identifier) ] 
 565          authenticator = DummyAuthenticator() 
 566          authenticators = [ ('authenticator', authenticator) ] 
 567          mw = self._makeOne(app=app, challengers=challengers, 
 568                             identifiers=identifiers, 
 569                             authenticators=authenticators) 
 570          start_response = DummyStartResponse() 
 571          result = mw(environ, start_response) 
 572          self.assertEqual(environ['challenged'], challenge_app) 
 573          self.failUnless(result[0].startswith('401 Unauthorized\r\n')) 
 574          # @@ unfuck 
 575  ##         self.assertEqual(identifier.forgotten, identifier.credentials) 
 576          self.assertEqual(environ['REMOTE_USER'], 'chris') 
 577  ##         self.assertEqual(environ['repoze.who.identity'], identifier.credentials) 
 578   
 579 -    def test_call_200_challenger_and_identifier_and_authenticator(self): 
 580          environ = self._makeEnviron() 
 581          headers = [('a', '1')] 
 582          app = DummyWorkingApp('200 OK', headers) 
 583          from paste.httpexceptions import HTTPUnauthorized 
 584          challenge_app = HTTPUnauthorized() 
 585          challenge = DummyChallenger(challenge_app) 
 586          challengers = [ ('challenge', challenge) ] 
 587          credentials = {'login':'chris', 'password':'password'} 
 588          identifier = DummyIdentifier(credentials) 
 589          identifiers = [ ('identifier', identifier) ] 
 590          authenticator = DummyAuthenticator() 
 591          authenticators = [ ('authenticator', authenticator) ] 
 592          mw = self._makeOne(app=app, challengers=challengers, 
 593                             identifiers=identifiers, 
 594                             authenticators=authenticators) 
 595          start_response = DummyStartResponse() 
 596          result = mw(environ, start_response) 
 597          self.assertEqual(environ.get('challenged'), None) 
 598          self.assertEqual(identifier.forgotten, False) 
 599          # @@ figure out later 
 600  ##         self.assertEqual(dict(identifier.remembered)['login'], dict(identifier.credentials)['login']) 
 601  ##         self.assertEqual(dict(identifier.remembered)['password'], dict(identifier.credentials)['password']) 
 602          self.assertEqual(environ['REMOTE_USER'], 'chris') 
 603  ##         self.assertEqual(environ['repoze.who.identity'], identifier.credentials) 
 604   
 605   
 606 -    def test_call_200_identity_reset(self): 
 607          environ = self._makeEnviron() 
 608          headers = [('a', '1')] 
 609          new_identity = {'user_id':'foo', 'password':'bar'} 
 610          app = DummyIdentityResetApp('200 OK', headers, new_identity) 
 611          from paste.httpexceptions import HTTPUnauthorized 
 612          challenge_app = HTTPUnauthorized() 
 613          challenge = DummyChallenger(challenge_app) 
 614          challengers = [ ('challenge', challenge) ] 
 615          credentials = {'login':'chris', 'password':'password'} 
 616          identifier = DummyIdentifier(credentials) 
 617          identifiers = [ ('identifier', identifier) ] 
 618          authenticator = DummyAuthenticator() 
 619          authenticators = [ ('authenticator', authenticator) ] 
 620          mw = self._makeOne(app=app, challengers=challengers, 
 621                             identifiers=identifiers, 
 622                             authenticators=authenticators) 
 623          start_response = DummyStartResponse() 
 624          result = mw(environ, start_response) 
 625          self.assertEqual(environ.get('challenged'), None) 
 626          self.assertEqual(identifier.forgotten, False) 
 627          new_credentials = identifier.credentials.copy() 
 628          new_credentials['login'] = 'fred' 
 629          new_credentials['password'] = 'schooled' 
 630          # @@ unfuck 
 631  ##         self.assertEqual(identifier.remembered, new_credentials) 
 632          self.assertEqual(environ['REMOTE_USER'], 'chris') 
 633  ##         self.assertEqual(environ['repoze.who.identity'], new_credentials) 
 634   
 635 -    def test_call_200_with_metadata(self): 
 636          environ = self._makeEnviron() 
 637          headers = [('a', '1')] 
 638          app = DummyWorkingApp('200 OK', headers) 
 639          from paste.httpexceptions import HTTPUnauthorized 
 640          challenge_app = HTTPUnauthorized() 
 641          challenge = DummyChallenger(challenge_app) 
 642          challengers = [ ('challenge', challenge) ] 
 643          credentials = {'login':'chris', 'password':'password'} 
 644          identifier = DummyIdentifier(credentials) 
 645          identifiers = [ ('identifier', identifier) ] 
 646          authenticator = DummyAuthenticator() 
 647          authenticators = [ ('authenticator', authenticator) ] 
 648          mdprovider = DummyMDProvider({'foo':'bar'}) 
 649          mdproviders = [ ('mdprovider', mdprovider) ] 
 650          mw = self._makeOne(app=app, challengers=challengers, 
 651                             identifiers=identifiers, 
 652                             authenticators=authenticators, 
 653                             mdproviders=mdproviders) 
 654          start_response = DummyStartResponse() 
 655          result = mw(environ, start_response) 
 656          # metadata 
 657          self.assertEqual(environ['repoze.who.identity']['foo'], 'bar') 
 658   
 659 -    def test_call_ingress_plugin_replaces_application(self): 
 660          environ = self._makeEnviron() 
 661          headers = [('a', '1')] 
 662          app = DummyWorkingApp('200 OK', headers) 
 663          challengers = [] 
 664          credentials = {'login':'chris', 'password':'password'} 
 665          from paste.httpexceptions import HTTPFound 
 666          identifier = DummyIdentifier( 
 667              credentials, 
 668              remember_headers=[('a', '1')], 
 669              replace_app = HTTPFound('http://example.com/redirect') 
 670              ) 
 671          identifiers = [ ('identifier', identifier) ] 
 672          authenticator = DummyAuthenticator() 
 673          authenticators = [ ('authenticator', authenticator) ] 
 674          mdproviders = [] 
 675          mw = self._makeOne(app=app, 
 676                             challengers=challengers, 
 677                             identifiers=identifiers, 
 678                             authenticators=authenticators, 
 679                             mdproviders=mdproviders) 
 680          start_response = DummyStartResponse() 
 681          result = ''.join(mw(environ, start_response)) 
 682          self.failUnless(result.startswith('302 Found')) 
 683          self.assertEqual(start_response.status, '302 Found') 
 684          headers = start_response.headers 
 685          self.assertEqual(len(headers), 3) 
 686          self.assertEqual(headers[0], 
 687                           ('location', 'http://example.com/redirect')) 
 688          self.assertEqual(headers[1], 
 689                           ('content-type', 'text/plain; charset=utf8')) 
 690          self.assertEqual(headers[2], 
 691                           ('a', '1')) 
 692          self.assertEqual(start_response.exc_info, None) 
 693          self.failIf(environ.has_key('repoze.who.application')) 
 694   
 695 -    def test_call_app_doesnt_call_start_response(self): 
 696          environ = self._makeEnviron() 
 697          headers = [('a', '1')] 
 698          app = DummyGeneratorApp('200 OK', headers) 
 699          from paste.httpexceptions import HTTPUnauthorized 
 700          challenge_app = HTTPUnauthorized() 
 701          challenge = DummyChallenger(challenge_app) 
 702          challengers = [ ('challenge', challenge) ] 
 703          credentials = {'login':'chris', 'password':'password'} 
 704          identifier = DummyIdentifier(credentials) 
 705          identifiers = [ ('identifier', identifier) ] 
 706          authenticator = DummyAuthenticator() 
 707          authenticators = [ ('authenticator', authenticator) ] 
 708          mdprovider = DummyMDProvider({'foo':'bar'}) 
 709          mdproviders = [ ('mdprovider', mdprovider) ] 
 710          mw = self._makeOne(app=app, challengers=challengers, 
 711                             identifiers=identifiers, 
 712                             authenticators=authenticators, 
 713                             mdproviders=mdproviders) 
 714          start_response = DummyStartResponse() 
 715          result = mw(environ, start_response) 
 716          # metadata 
 717          self.assertEqual(environ['repoze.who.identity']['foo'], 'bar') 
 718   
 719      # XXX need more call tests: 
 720      #  - auth_id sorting 
 721   
 722 -class TestMatchClassification(unittest.TestCase): 
 723   
 724 -    def _getFUT(self): 
 725          from repoze.who.middleware import match_classification 
 726          return match_classification 
 727   
 728 -    def test_match_classification(self): 
 729          f = self._getFUT() 
 730          from repoze.who.interfaces import IIdentifier 
 731          from repoze.who.interfaces import IChallenger 
 732          from repoze.who.interfaces import IAuthenticator 
 733          multi1 = DummyMultiPlugin() 
 734          multi2 = DummyMultiPlugin() 
 735          multi1.classifications = {IIdentifier:('foo', 'bar'), 
 736                                    IChallenger:('buz',), 
 737                                    IAuthenticator:None} 
 738          multi2.classifications = {IIdentifier:('foo', 'baz', 'biz')} 
 739          plugins = (multi1, multi2) 
 740          # specific 
 741          self.assertEqual(f(IIdentifier, plugins, 'foo'), [multi1, multi2]) 
 742          self.assertEqual(f(IIdentifier, plugins, 'bar'), [multi1]) 
 743          self.assertEqual(f(IIdentifier, plugins, 'biz'), [multi2]) 
 744          # any for multi2 
 745          self.assertEqual(f(IChallenger, plugins, 'buz'), [multi1, multi2]) 
 746          # any for either 
 747          self.assertEqual(f(IAuthenticator, plugins, 'buz'), [multi1, multi2]) 
 748   
 749 -class TestStartResponseWrapper(unittest.TestCase): 
 750   
 751 -    def _getTargetClass(self): 
 752          from repoze.who.middleware import StartResponseWrapper 
 753          return StartResponseWrapper 
 754   
 755 -    def _makeOne(self, *arg, **kw): 
 756          plugin = self._getTargetClass()(*arg, **kw) 
 757          return plugin 
 758   
 759 -    def test_ctor(self): 
 760          wrapper = self._makeOne(None) 
 761          self.assertEqual(wrapper.start_response, None) 
 762          self.assertEqual(wrapper.headers, []) 
 763          self.failUnless(wrapper.buffer) 
 764   
 765 -    def test_finish_response(self): 
 766          statuses = [] 
 767          headerses = [] 
 768          datases = [] 
 769          closededs = [] 
 770          from StringIO import StringIO 
 771          def write(data): 
 772              datases.append(data) 
 773          def close(): 
 774              closededs.append(True) 
 775          write.close = close 
 776   
 777          def start_response(status, headers, exc_info=None): 
 778              statuses.append(status) 
 779              headerses.append(headers) 
 780              return write 
 781   
 782          wrapper = self._makeOne(start_response) 
 783          wrapper.status = '401 Unauthorized' 
 784          wrapper.headers = [('a', '1')] 
 785          wrapper.buffer = StringIO('written') 
 786          extra_headers = [('b', '2')] 
 787          result = wrapper.finish_response(extra_headers) 
 788          self.assertEqual(result, None) 
 789          self.assertEqual(headerses[0], wrapper.headers + extra_headers) 
 790          self.assertEqual(statuses[0], wrapper.status) 
 791          self.assertEqual(datases[0], 'written') 
 792          self.assertEqual(closededs[0], True) 
 793   
 794 -class TestMakeRegistries(unittest.TestCase): 
 795   
 796 -    def _getFUT(self): 
 797          from repoze.who.middleware import make_registries 
 798          return make_registries 
 799   
 800 -    def test_empty(self): 
 801          fn = self._getFUT() 
 802          iface_reg, name_reg = fn([], [], [], []) 
 803          self.assertEqual(iface_reg, {}) 
 804          self.assertEqual(name_reg, {}) 
 805   
 806 -    def test_brokenimpl(self): 
 807          fn = self._getFUT() 
 808          self.assertRaises(ValueError, fn, [(None, DummyApp())], [], [], []) 
 809   
 810 -    def test_ok(self): 
 811          fn = self._getFUT() 
 812          credentials1 = {'login':'chris', 'password':'password'} 
 813          dummy_id1 = DummyIdentifier(credentials1) 
 814          credentials2 = {'login':'chris', 'password':'password'} 
 815          dummy_id2 = DummyIdentifier(credentials2) 
 816          identifiers = [ ('id1', dummy_id1), ('id2', dummy_id2) ] 
 817          dummy_auth = DummyAuthenticator(None) 
 818          authenticators = [ ('auth', dummy_auth) ] 
 819          dummy_challenger = DummyChallenger(None) 
 820          challengers = [ ('challenger', dummy_challenger) ] 
 821          dummy_mdprovider = DummyMDProvider() 
 822          mdproviders = [ ('mdproviders', dummy_mdprovider) ] 
 823          iface_reg, name_reg = fn(identifiers, authenticators, challengers, 
 824                                   mdproviders) 
 825          from repoze.who.interfaces import IIdentifier 
 826          from repoze.who.interfaces import IAuthenticator 
 827          from repoze.who.interfaces import IChallenger 
 828          self.assertEqual(iface_reg[IIdentifier], [dummy_id1, dummy_id2]) 
 829          self.assertEqual(iface_reg[IAuthenticator], [dummy_auth]) 
 830          self.assertEqual(iface_reg[IChallenger], [dummy_challenger]) 
 831          self.assertEqual(name_reg['id1'], dummy_id1) 
 832          self.assertEqual(name_reg['id2'], dummy_id2) 
 833          self.assertEqual(name_reg['auth'], dummy_auth) 
 834          self.assertEqual(name_reg['challenger'], dummy_challenger) 
 835   
 836 -class TestIdentityDict(unittest.TestCase): 
 837   
 838 -    def _getTargetClass(self): 
 839          from repoze.who.middleware import Identity 
 840          return Identity 
 841   
 842 -    def _makeOne(self, **kw): 
 843          klass = self._getTargetClass() 
 844          return klass(**kw) 
 845   
 846 -    def test_str(self): 
 847          identity = self._makeOne(foo=1) 
 848          self.failUnless(str(identity).startswith('<repoze.who identity')) 
 849          self.assertEqual(identity['foo'], 1) 
 850   
 851 -    def test_repr(self): 
 852          identity = self._makeOne(foo=1) 
 853          self.failUnless(str(identity).startswith('<repoze.who identity')) 
 854          self.assertEqual(identity['foo'], 1) 
 855   
 856 -class WrapGeneratorTests(unittest.TestCase): 
 857   
 858 -    def _getFUT(self): 
 859          from repoze.who.middleware import wrap_generator 
 860          return wrap_generator 
 861   
 862 -    def test_it(self): 
 863          L = [] 
 864          def gen(L=L): 
 865              L.append('yo!') 
 866              yield 'a' 
 867              yield 'b' 
 868          wrap_generator = self._getFUT() 
 869          newgen = wrap_generator(gen()) 
 870          self.assertEqual(L, ['yo!']) 
 871          self.assertEqual(list(newgen), ['a', 'b']) 
 872   
 873 -class TestMakeTestMiddleware(unittest.TestCase): 
 874   
 875 -    def setUp(self): 
 876          import os 
 877          self._old_WHO_LOG = os.environ.get('WHO_LOG') 
 878   
 879 -    def tearDown(self): 
 880          import os 
 881          if self._old_WHO_LOG is not None: 
 882              os.environ['WHO_LOG'] = self._old_WHO_LOG 
 883          else: 
 884              if 'WHO_LOG' in os.environ: 
 885                  del os.environ['WHO_LOG'] 
 886   
 887 -    def _getFactory(self): 
 888          from repoze.who.middleware import make_test_middleware 
 889          return make_test_middleware 
 890   
 891 -    def test_it_no_WHO_LOG_in_environ(self): 
 892          from repoze.who.interfaces import IIdentifier 
 893          from repoze.who.interfaces import IAuthenticator 
 894          from repoze.who.interfaces import IChallenger 
 895          app = DummyApp() 
 896          factory = self._getFactory() 
 897          global_conf = {'here': '/'} 
 898          middleware = factory(app, global_conf) 
 899          self.assertEqual(len(middleware.registry[IIdentifier]), 3) 
 900          self.assertEqual(len(middleware.registry[IAuthenticator]), 1) 
 901          self.assertEqual(len(middleware.registry[IChallenger]), 2) 
 902          self.assertEqual(middleware.logger, None) 
 903   
 904 -    def test_it_w_WHO_LOG_in_environ(self): 
 905          import logging 
 906          import os 
 907          os.environ['WHO_LOG'] = '1' 
 908          app = DummyApp() 
 909          factory = self._getFactory() 
 910          global_conf = {'here': '/'} 
 911          middleware = factory(app, global_conf) 
 912          self.assertEqual(middleware.logger.getEffectiveLevel(), logging.DEBUG) 
 913   
 914 -class DummyApp: 
 915      environ = None 
 916 -    def __call__(self, environ, start_response): 
 917          self.environ = environ 
 918          return [] 
 919   
 920 -class DummyWorkingApp: 
 921 -    def __init__(self, status, headers): 
 922          self.status = status 
 923          self.headers = headers 
 924   
 925 -    def __call__(self, environ, start_response): 
 926          self.environ = environ 
 927          start_response(self.status, self.headers) 
 928          return ['body'] 
 929   
 930 -class DummyGeneratorApp: 
 931 -    def __init__(self, status, headers): 
 932          self.status = status 
 933          self.headers = headers 
 934   
 935 -    def __call__(self, environ, start_response): 
 936          def gen(self=self, start_response=start_response): 
 937              self.environ = environ 
 938              start_response(self.status, self.headers) 
 939              yield 'body' 
 940          return gen() 
 941   
 942 -class DummyIdentityResetApp: 
 943 -    def __init__(self, status, headers, new_identity): 
 944          self.status = status 
 945          self.headers = headers 
 946          self.new_identity = new_identity 
 947   
 948 -    def __call__(self, environ, start_response): 
 949          self.environ = environ 
 950          environ['repoze.who.identity']['login'] = 'fred' 
 951          environ['repoze.who.identity']['password'] = 'schooled' 
 952          start_response(self.status, self.headers) 
 953          return ['body'] 
 954   
 955 -class DummyChallenger: 
 956 -    def __init__(self, app=None): 
 957          self.app = app 
 958   
 959 -    def challenge(self, environ, status, app_headers, forget_headers): 
 960          environ['challenged'] = self.app 
 961          return self.app 
 962   
 963 -class DummyIdentifier: 
 964      forgotten = False 
 965      remembered = False 
 966   
 967 -    def __init__(self, credentials=None, remember_headers=None, 
 968                   forget_headers=None, replace_app=None): 
 969          self.credentials = credentials 
 970          self.remember_headers = remember_headers 
 971          self.forget_headers = forget_headers 
 972          self.replace_app = replace_app 
 973   
 974 -    def identify(self, environ): 
 975          if self.replace_app: 
 976              environ['repoze.who.application'] = self.replace_app 
 977          return self.credentials 
 978   
 979 -    def forget(self, environ, identity): 
 980          self.forgotten = identity 
 981          return self.forget_headers 
 982   
 983 -    def remember(self, environ, identity): 
 984          self.remembered = identity 
 985          return self.remember_headers 
 986   
 987 -class DummyAuthenticator: 
 988 -    def __init__(self, userid=None): 
 989          self.userid = userid 
 990   
 991 -    def authenticate(self, environ, credentials): 
 992          if self.userid is None: 
 993              return credentials['login'] 
 994          return self.userid 
 995   
 996 -class DummyFailAuthenticator: 
 997 -    def authenticate(self, environ, credentials): 
 998          return None 
 999   
1000 -class DummyRequestClassifier: 
1001 -    def __call__(self, environ): 
1002          return 'browser' 
1003   
1004 -class DummyChallengeDecider: 
1005 -    def __call__(self, environ, status, headers): 
1006          if status.startswith('401 '): 
1007              return True 
1008   
1009 -class DummyNoResultsIdentifier: 
1010 -    def identify(self, environ): 
1011          return None 
1012   
1013 -    def remember(self, *arg, **kw): 
1014          pass 
1015   
1016 -    def forget(self, *arg, **kw): 
1017          pass 
1018   
1019 -class DummyStartResponse: 
1020 -    def __call__(self, status, headers, exc_info=None): 
1021          self.status = status 
1022          self.headers = headers 
1023          self.exc_info = exc_info 
1024          return [] 
1025   
1026 -class DummyMDProvider: 
1027 -    def __init__(self, metadata=None): 
1028          self._metadata = metadata 
1029   
1030 -    def add_metadata(self, environ, identity): 
1031          return identity.update(self._metadata) 
1032   
1033 -class DummyMultiPlugin: 
1034      pass 
1035