Mercurial > universal-identity-relyer
changeset 7:c6aef586ab82
added success_delegate
author | Atul Varma <avarma@mozilla.com> |
---|---|
date | Sat, 12 Jun 2010 19:47:49 -0700 |
parents | f0273e301ee4 |
children | 51dfea268026 |
files | oauth_experiment.py test_twitter_client.py twitter_client.py |
diffstat | 3 files changed, 33 insertions(+), 22 deletions(-) [+] |
line wrap: on
line diff
--- a/oauth_experiment.py Sat Jun 12 19:12:31 2010 -0700 +++ b/oauth_experiment.py Sat Jun 12 19:47:49 2010 -0700 @@ -28,7 +28,14 @@ def __setitem__(self, name, value): open(self.__filename(name), 'w').write(json.dumps(value)) -app = twitter_client.TwitterOauthClientApp(consumer, oauth, - 'http://localhost:8000/callback', - Storage() - ) +def success_app(environ, start_response): + start_response('200 OK', [('Content-Type', 'text/plain')]) + return ['yay'] + +app = twitter_client.TwitterOauthClientApp( + consumer=consumer, + oauth=oauth, + oauth_callback='http://localhost:8000/callback', + request_tokens=Storage(), + success_delegate=success_app + )
--- a/test_twitter_client.py Sat Jun 12 19:12:31 2010 -0700 +++ b/test_twitter_client.py Sat Jun 12 19:47:49 2010 -0700 @@ -1,19 +1,24 @@ -from minimock import Mock +import minimock from twitter_client import TwitterOauthClientApp +class Mock(minimock.Mock): + def __repr__(self): + return "<Mock %s>" % self.mock_name + def app(request_tokens=None): if request_tokens is None: request_tokens = {} consumer = 'mock consumer' oauth = Mock('oauth') + success_delegate = Mock('success_delegate') toc = TwitterOauthClientApp(consumer, oauth, 'http://foo.com/oauth_callback', - request_tokens) - return (consumer, oauth, toc) + request_tokens, success_delegate) + return (consumer, oauth, success_delegate, toc) def test_404(): """ - >>> _, _, toc = app() + >>> _, _, _, toc = app() >>> environ = dict(PATH_INFO='/', QUERY_STRING='') >>> toc(environ, Mock('start_response')) Called start_response('404 Not Found', [('Content-Type', 'text/plain')]) @@ -24,7 +29,7 @@ def test_request_redirect(self): """ - >>> consumer, oauth, toc = app() + >>> consumer, oauth, _, toc = app() >>> client = Mock('client') >>> client.request.mock_returns = ( ... {'status': '200'}, @@ -46,14 +51,10 @@ pass -class MockToken(Mock): - def __repr__(self): - return "<Mock token>" - def test_callback(self): """ >>> storage = {'token': {'oauth_token': 'token', 'oauth_token_secret': 'secret'}} - >>> consumer, oauth, toc = app(storage) + >>> consumer, oauth, success_delegate, toc = app(storage) >>> client = Mock('client') >>> client.request.mock_returns = ( ... {'status': '200'}, @@ -61,18 +62,21 @@ ... 'user_id=userid&screen_name=bob' ... ) >>> oauth.Client.mock_returns = client - >>> token = MockToken('token') + >>> token = Mock('token') >>> oauth.Token.mock_returns = token >>> environ = dict( ... PATH_INFO='/callback', ... QUERY_STRING='oauth_token=token&oauth_verifier=verifier' ... ) + >>> success_delegate.mock_returns = ['success'] >>> toc(environ, Mock('start_response')) Called oauth.Token('token', 'secret') Called token.set_verifier('verifier') Called oauth.Client('mock consumer', <Mock token>) Called client.request('https://api.twitter.com/oauth/access_token', 'POST') - Called start_response('200 OK', [('Content-Type', 'text/plain')]) + Called success_delegate( + {'QUERY_STRING': 'oauth_token=token&oauth_verifier=verifier', 'PATH_INFO': '/callback', 'oauth.access_token': {'oauth_token_secret': 'secret', 'user_id': 'userid', 'oauth_token': 'token', 'screen_name': 'bob'}}, + <Mock start_response>) ['success'] """ @@ -81,4 +85,4 @@ if __name__ == '__main__': import doctest doctest.testmod() - print "all tests passed." + print "done running tests."
--- a/twitter_client.py Sat Jun 12 19:12:31 2010 -0700 +++ b/twitter_client.py Sat Jun 12 19:47:49 2010 -0700 @@ -7,10 +7,12 @@ access_token_url = 'https://api.twitter.com/oauth/access_token' authorize_url = 'https://api.twitter.com/oauth/authorize' - def __init__(self, consumer, oauth, oauth_callback, request_tokens): + def __init__(self, consumer, oauth, oauth_callback, request_tokens, + success_delegate): self.oauth_callback = oauth_callback self.oauth = oauth self.consumer = consumer + self.success_delegate = success_delegate self.request_tokens = request_tokens def __call__(self, environ, start_response): @@ -63,10 +65,8 @@ if resp['status'] != '200': raise Exception("Invalid response %s." % resp['status']) access_token = dict(urlparse.parse_qsl(content)) - start_response('200 OK', - [('Content-Type', 'text/plain')]) - # TODO: Stash the access token somewhere. - return ['success'] + environ['oauth.access_token'] = access_token + return self.success_delegate(environ, start_response) start_response('404 Not Found', [('Content-Type', 'text/plain')])