From 89a5a9cb3d756da74019977d4fbe6eec680e4710 Mon Sep 17 00:00:00 2001 From: Matt Behrens Date: Tue, 9 Oct 2012 01:32:13 -0700 Subject: [PATCH] add working OAuth2 POST functionality + test --- libgreader/auth.py | 21 ++++++++++++++------- tests/test_auth.py | 22 ++++++++++++++++++++++ 2 files changed, 36 insertions(+), 7 deletions(-) diff --git a/libgreader/auth.py b/libgreader/auth.py index 3088281..002aa51 100644 --- a/libgreader/auth.py +++ b/libgreader/auth.py @@ -332,17 +332,24 @@ def get(self, url, parameters=None): except (ValueError, KeyError, IOError) as e: return None - def post(self, url, postParameters=None, urlParameters={}): + def post(self, url, postParameters=None, urlParameters=None): + """ + Convenience method for requesting to google with proper cookies/params. + """ if not self.access_token: raise IOError("No authorized client available.") - urlParameters.update({'access_token': self.access_token, 'alt': 'json'}) - getString = self.getParameters(urlParameters) - request = urllib2.Request(url + '?' + self.getParameters(parameters)) + if not self.action_token: + raise IOError("Need to generate action token.") + if urlParameters is None: + urlParameters = {} + headers = {'Authorization': 'Bearer ' + self.access_token} + postParameters.update({'T':self.action_token}) postString = self.postParameters(postParameters) + request = urllib2.Request(url + '?' + self.getParameters(urlParameters), data=postString, headers=headers) try: - response = urllib2.urlopen(request, data=postString) - toUnicode(response.read()) - except (ValueError, KeyError, IOError): + response = urllib2.urlopen(request) + return toUnicode(response.read()) + except (ValueError, KeyError, IOError) as e: return None class GAPDecoratorAuthMethod(AuthenticationMethod): diff --git a/tests/test_auth.py b/tests/test_auth.py index 86204b2..ba33ca5 100644 --- a/tests/test_auth.py +++ b/tests/test_auth.py @@ -154,6 +154,28 @@ def test_full_auth_and_access_userdata(self): self.assertEqual(dict, type(info)) self.assertEqual(firstname, info['userName']) + def test_oauth_subscribe(self): + auth = OAuth2Method(client_id, client_secret) + auth.setRedirectUri(redirect_url) + url = auth.buildAuthUrl() + token = automated_oauth2_approval(url) + auth.code = token + auth.setAccessToken() + auth.setActionToken() + + reader = GoogleReader(auth) + + slashdot = 'feed/http://rss.slashdot.org/Slashdot/slashdot' + #unsubscribe always return true; revert feedlist state + self.assertTrue(reader.unsubscribe(slashdot)) + # now subscribe + self.assertTrue(reader.subscribe(slashdot)) + # wait for server to update + import time + time.sleep(1) + reader.buildSubscriptionList() + # test subscribe successful + self.assertIn(slashdot, [x.id for x in reader.getSubscriptionList()]) if __name__ == '__main__': unittest.main()