re-implement OAuth1.0 code

OAuth support for SmugMug needs some additional features
(auth-rebuild on redirect, query parameters in URL, ...)
and fixing this in the old code wouldn't work all that well.
This commit is contained in:
Mike Fährmann
2018-05-10 18:26:10 +02:00
parent 0e3883303f
commit 6a31ada9e3
7 changed files with 132 additions and 119 deletions

View File

@@ -8,10 +8,8 @@
# published by the Free Software Foundation.
import unittest
import requests
from gallery_dl import text
from gallery_dl.util import OAuthSession
from gallery_dl import oauth, text
TESTSERVER = "http://oauthbin.com"
CONSUMER_KEY = "key"
@@ -25,7 +23,7 @@ ACCESS_TOKEN_SECRET = "accesssecret"
class TestOAuthSession(unittest.TestCase):
def test_concat(self):
concat = OAuthSession.concat
concat = oauth.concat
self.assertEqual(concat(), "")
self.assertEqual(concat("str"), "str")
@@ -37,18 +35,18 @@ class TestOAuthSession(unittest.TestCase):
"GET&http%3A%2F%2Fexample.org%2F&foo%3Dbar%26baz%3Da"
)
def test_nonce(self, N=16):
nonce_values = set(OAuthSession.nonce(N) for _ in range(N))
def test_nonce(self, size=16):
nonce_values = set(oauth.nonce(size) for _ in range(size))
# uniqueness
self.assertEqual(len(nonce_values), N)
self.assertEqual(len(nonce_values), size)
# length
for nonce in nonce_values:
self.assertEqual(len(nonce), N)
self.assertEqual(len(nonce), size)
def test_quote(self):
quote = OAuthSession.quote
quote = oauth.quote
reserved = ",;:!\"§$%&/(){}[]=?`´+*'äöü"
unreserved = ("ABCDEFGHIJKLMNOPQRSTUVWXYZ"
@@ -65,33 +63,6 @@ class TestOAuthSession(unittest.TestCase):
self.assertTrue(len(quoted) >= 3)
self.assertEqual(quoted_hex.upper(), quoted_hex)
def test_urlencode(self):
urlencode = OAuthSession.urlencode
self.assertEqual(urlencode({}), "")
self.assertEqual(urlencode({"foo": "bar"}), "foo=bar")
self.assertEqual(
urlencode({"foo": "bar", "baz": "a", "a": "baz"}),
"a=baz&baz=a&foo=bar"
)
self.assertEqual(
urlencode({
"oauth_consumer_key": "0685bd9184jfhq22",
"oauth_token": "ad180jjd733klru7",
"oauth_signature_method": "HMAC-SHA1",
"oauth_timestamp": 137131200,
"oauth_nonce": "4572616e48616d6d65724c61686176",
"oauth_version": "1.0"
}),
"oauth_consumer_key=0685bd9184jfhq22&"
"oauth_nonce=4572616e48616d6d65724c61686176&"
"oauth_signature_method=HMAC-SHA1&"
"oauth_timestamp=137131200&"
"oauth_token=ad180jjd733klru7&"
"oauth_version=1.0"
)
def test_request_token(self):
response = self._oauth_request(
"/v1/request-token", {})
@@ -113,23 +84,20 @@ class TestOAuthSession(unittest.TestCase):
self.assertTrue(data["oauth_token_secret"], ACCESS_TOKEN_SECRET)
def test_authenticated_call(self):
params = {"method": "foo", "bar": "baz", "a": "äöüß/?&#"}
params = {"method": "foo", "a": "äöüß/?&#", "äöüß/?&#": "a"}
response = self._oauth_request(
"/v1/echo", params, ACCESS_TOKEN, ACCESS_TOKEN_SECRET)
expected = OAuthSession.urlencode(params)
self.assertEqual(response, expected, msg=response)
self.assertEqual(text.parse_query(response), params)
def _oauth_request(self, endpoint, params=None,
oauth_token=None, oauth_token_secret=None):
session = OAuthSession(
requests.session(),
session = oauth.OAuth1Session(
CONSUMER_KEY, CONSUMER_SECRET,
oauth_token, oauth_token_secret,
)
url = TESTSERVER + endpoint
return session.get(url, params.copy()).text
return session.get(url, params=params).text
if __name__ == "__main__":