# 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,
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):
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):
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):
('/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',
'_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',
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',