]> git.ipfire.org Git - thirdparty/tornado.git/commitdiff
auth: Forward-port tests from branch5.1 2497/head
authorBen Darnell <ben@bendarnell.com>
Sun, 16 Sep 2018 16:04:52 +0000 (12:04 -0400)
committerBen Darnell <ben@bendarnell.com>
Sun, 16 Sep 2018 17:59:33 +0000 (13:59 -0400)
The bugs fixed in #2496 were already fixed in the 6.0 branch, but the
tests added there are still useful

tornado/test/auth_test.py

index eba46a6d4a71fc464735784eadcc3d64858c7989..6fee29fcf51b77fe57154602c02831ddd9f6c31c 100644 (file)
@@ -3,6 +3,8 @@
 # and ensure that it doesn't blow up (e.g. with unicode/bytes issues in
 # python 3)
 
+import unittest
+
 from tornado.auth import (
     OpenIdMixin, OAuthMixin, OAuth2Mixin,
     GoogleOAuth2Mixin, FacebookGraphMixin, TwitterMixin,
@@ -11,9 +13,15 @@ from tornado.escape import json_decode
 from tornado import gen
 from tornado.httpclient import HTTPClientError
 from tornado.httputil import url_concat
-from tornado.testing import AsyncHTTPTestCase
+from tornado.log import app_log
+from tornado.testing import AsyncHTTPTestCase, ExpectLog
 from tornado.web import RequestHandler, Application, HTTPError
 
+try:
+    from unittest import mock
+except ImportError:
+    mock = None  # type: ignore
+
 
 class OpenIdClientLoginHandler(RequestHandler, OpenIdMixin):
     def initialize(self, test):
@@ -154,6 +162,7 @@ class TwitterClientHandler(RequestHandler, TwitterMixin):
         self._OAUTH_REQUEST_TOKEN_URL = test.get_url('/oauth1/server/request_token')
         self._OAUTH_ACCESS_TOKEN_URL = test.get_url('/twitter/server/access_token')
         self._OAUTH_AUTHORIZE_URL = test.get_url('/oauth1/server/authorize')
+        self._OAUTH_AUTHENTICATE_URL = test.get_url('/twitter/server/authenticate')
         self._TWITTER_BASE_URL = test.get_url('/twitter/api')
 
     def get_auth_http_client(self):
@@ -172,6 +181,20 @@ class TwitterClientLoginHandler(TwitterClientHandler):
         yield self.authorize_redirect()
 
 
+class TwitterClientAuthenticateHandler(TwitterClientHandler):
+    # Like TwitterClientLoginHandler, but uses authenticate_redirect
+    # instead of authorize_redirect.
+    @gen.coroutine
+    def get(self):
+        if self.get_argument("oauth_token", None):
+            user = yield self.get_authenticated_user()
+            if user is None:
+                raise Exception("user is None")
+            self.finish(user)
+            return
+        yield self.authenticate_redirect()
+
+
 class TwitterClientLoginGenCoroutineHandler(TwitterClientHandler):
     @gen.coroutine
     def get(self):
@@ -257,6 +280,7 @@ class AuthTest(AsyncHTTPTestCase):
                 ('/facebook/client/login', FacebookClientLoginHandler, dict(test=self)),
 
                 ('/twitter/client/login', TwitterClientLoginHandler, dict(test=self)),
+                ('/twitter/client/authenticate', TwitterClientAuthenticateHandler, dict(test=self)),
                 ('/twitter/client/login_gen_coroutine',
                  TwitterClientLoginGenCoroutineHandler, dict(test=self)),
                 ('/twitter/client/show_user',
@@ -333,6 +357,14 @@ class AuthTest(AsyncHTTPTestCase):
             '_oauth_request_token="enhjdg==|MTIzNA=="' in response.headers['Set-Cookie'],
             response.headers['Set-Cookie'])
 
+    @unittest.skipIf(mock is None, 'mock package not present')
+    def test_oauth10a_redirect_error(self):
+        with mock.patch.object(OAuth1ServerRequestTokenHandler, 'get') as get:
+            get.side_effect = Exception("boom")
+            with ExpectLog(app_log, "Uncaught exception"):
+                response = self.fetch('/oauth10a/client/login', follow_redirects=False)
+            self.assertEqual(response.code, 500)
+
     def test_oauth10a_get_user(self):
         response = self.fetch(
             '/oauth10a/client/login?oauth_token=zxcv',
@@ -389,6 +421,16 @@ class AuthTest(AsyncHTTPTestCase):
     def test_twitter_redirect_gen_coroutine(self):
         self.base_twitter_redirect('/twitter/client/login_gen_coroutine')
 
+    def test_twitter_authenticate_redirect(self):
+        response = self.fetch('/twitter/client/authenticate', follow_redirects=False)
+        self.assertEqual(response.code, 302)
+        self.assertTrue(response.headers['Location'].endswith(
+            '/twitter/server/authenticate?oauth_token=zxcv'), response.headers['Location'])
+        # the cookie is base64('zxcv')|base64('1234')
+        self.assertTrue(
+            '_oauth_request_token="enhjdg==|MTIzNA=="' in response.headers['Set-Cookie'],
+            response.headers['Set-Cookie'])
+
     def test_twitter_get_user(self):
         response = self.fetch(
             '/twitter/client/login?oauth_token=zxcv',