def test_message(self):

try:

raise oauth.Error

except oauth.Error as e:

self.assertEqual(e.message, 'OAuth error occurred.')

msg = 'OMG THINGS BROKE!!!!'

try:

raise oauth.Error(msg)

except oauth.Error as e:

self.assertEqual(e.message, msg)

def test_str(self):

try:

raise oauth.Error

except oauth.Error as e:

def test_build_auth_header(self):

header = oauth.build_authenticate_header()

self.assertEqual(header['WWW-Authenticate'], 'OAuth realm=""')

self.assertEqual(len(header), 1)

realm = 'http://example.myrealm.com/'

header = oauth.build_authenticate_header(realm)

self.assertEqual(header['WWW-Authenticate'], 'OAuth realm="%s"' %

realm)

self.assertEqual(len(header), 1)

def test_build_xoauth_string(self):

consumer = oauth.Consumer('consumer_token', 'consumer_secret')

token = oauth.Token('user_token', 'user_secret')

url = "https://mail.google.com/mail/b/[email protected]/imap/"

xoauth_string = oauth.build_xoauth_string(url, consumer, token)

method, oauth_url, oauth_string = xoauth_string.split(' ')

self.assertEqual("GET", method)

self.assertEqual(url, oauth_url)

returned = {}

parts = oauth_string.split(',')

for part in parts:

var, val = part.split('=')

returned[var] = val.strip('"')

self.assertEquals('HMAC-SHA1', returned['oauth_signature_method'])

self.assertEquals('user_token', returned['oauth_token'])

self.assertEquals('consumer_token', returned['oauth_consumer_key'])

self.assertTrue('oauth_signature' in returned, 'oauth_signature')

def test_escape(self):

string = 'http://whatever.com/~someuser/?test=test&other=other'

self.assert_('~' in oauth.escape(string))

string = '../../../../../../../etc/passwd'

self.assert_('../' not in oauth.escape(string))

def test_gen_nonce(self):

nonce = oauth.generate_nonce()

self.assertEqual(len(nonce), 8)

nonce = oauth.generate_nonce(20)

self.assertEqual(len(nonce), 20)

def test_gen_verifier(self):

verifier = oauth.generate_verifier()

self.assertEqual(len(verifier), 8)

verifier = oauth.generate_verifier(16)

self.assertEqual(len(verifier), 16)

def test_gen_timestamp(self):

exp = int(time.time())

now = oauth.generate_timestamp()

def setUp(self):

self.key = 'my-key'

self.secret = 'my-secret'

self.consumer = oauth.Consumer(key=self.key, secret=self.secret)

def test_init(self):

self.assertEqual(self.consumer.key, self.key)

self.assertEqual(self.consumer.secret, self.secret)

def test_basic(self):

self.assertRaises(ValueError, lambda: oauth.Consumer(None, None))

self.assertRaises(ValueError, lambda: oauth.Consumer('asf', None))

self.assertRaises(ValueError, lambda: oauth.Consumer(None, 'dasf'))

def test_str(self):

res = dict(parse_qsl(str(self.consumer)))

self.assertTrue('oauth_consumer_key' in res)

self.assertTrue('oauth_consumer_secret' in res)

self.assertEquals(res['oauth_consumer_key'], self.consumer.key)

def setUp(self):

self.key = 'my-key'

self.secret = 'my-secret'

self.token = oauth.Token(self.key, self.secret)

def test_basic(self):

self.assertRaises(ValueError, lambda: oauth.Token(None, None))

self.assertRaises(ValueError, lambda: oauth.Token('asf', None))

self.assertRaises(ValueError, lambda: oauth.Token(None, 'dasf'))

def test_init(self):

self.assertEqual(self.token.key, self.key)

self.assertEqual(self.token.secret, self.secret)

self.assertEqual(self.token.callback, None)

self.assertEqual(self.token.callback_confirmed, None)

self.assertEqual(self.token.verifier, None)

def test_set_callback(self):

self.assertEqual(self.token.callback, None)

self.assertEqual(self.token.callback_confirmed, None)

cb = 'http://www.example.com/my-callback'

self.token.set_callback(cb)

self.assertEqual(self.token.callback, cb)

self.assertEqual(self.token.callback_confirmed, 'true')

self.token.set_callback(None)

self.assertEqual(self.token.callback, None)

# TODO: The following test should probably not pass, but it does

# To fix this, check for None and unset 'true' in set_callback

# Additionally, should a confirmation truly be done of the callback?

self.assertEqual(self.token.callback_confirmed, 'true')

def test_set_verifier(self):

self.assertEqual(self.token.verifier, None)

v = oauth.generate_verifier()

self.token.set_verifier(v)

self.assertEqual(self.token.verifier, v)

self.token.set_verifier()

self.assertNotEqual(self.token.verifier, v)

self.token.set_verifier('')

self.assertEqual(self.token.verifier, '')

def test_get_callback_url(self):

self.assertEqual(self.token.get_callback_url(), None)

self.token.set_verifier()

self.assertEqual(self.token.get_callback_url(), None)

cb = 'http://www.example.com/my-callback?save=1&return=true'

v = oauth.generate_verifier()

self.token.set_callback(cb)

self.token.set_verifier(v)

url = self.token.get_callback_url()

verifier_str = '&oauth_verifier=%s' % v

self.assertEqual(url, '%s%s' % (cb, verifier_str))

cb = 'http://www.example.com/my-callback-no-query'

v = oauth.generate_verifier()

self.token.set_callback(cb)

self.token.set_verifier(v)

url = self.token.get_callback_url()

verifier_str = '?oauth_verifier=%s' % v

self.assertEqual(url, '%s%s' % (cb, verifier_str))

def test_to_string(self):

string = 'oauth_token_secret=%s&oauth_token=%s' % (self.secret,

self.key)

self.assertEqual(self.token.to_string(), string)

self.token.set_callback('http://www.example.com/my-callback')

string += '&oauth_callback_confirmed=true'

self.assertEqual(self.token.to_string(), string)

def _compare_tokens(self, new):

self.assertEqual(self.token.key, new.key)

self.assertEqual(self.token.secret, new.secret)

# TODO: What about copying the callback to the new token?

# self.assertEqual(self.token.callback, new.callback)

self.assertEqual(self.token.callback_confirmed,

new.callback_confirmed)

# TODO: What about copying the verifier to the new token?

# self.assertEqual(self.token.verifier, new.verifier)

def test_to_string(self):

tok = oauth.Token('tooken', 'seecret')

self.assertEqual(str(tok), 'oauth_token_secret=seecret&oauth_token=tooken')

def test_from_string(self):

self.assertRaises(ValueError, lambda: oauth.Token.from_string(''))

self.assertRaises(ValueError, lambda: oauth.Token.from_string('blahblahblah'))

self.assertRaises(ValueError, lambda: oauth.Token.from_string('blah=blah'))

self.assertRaises(ValueError, lambda: oauth.Token.from_string('oauth_token_secret=asfdasf'))

self.assertRaises(ValueError, lambda: oauth.Token.from_string('oauth_token_secret='))

self.assertRaises(ValueError, lambda: oauth.Token.from_string('oauth_token=asfdasf'))

self.assertRaises(ValueError, lambda: oauth.Token.from_string('oauth_token='))

self.assertRaises(ValueError, lambda: oauth.Token.from_string('oauth_token=&oauth_token_secret='))

self.assertRaises(ValueError, lambda: oauth.Token.from_string('oauth_token=tooken%26oauth_token_secret=seecret'))

string = self.token.to_string()

new = oauth.Token.from_string(string)

self._compare_tokens(new)

self.token.set_callback('http://www.example.com/my-callback')

string = self.token.to_string()

new = oauth.Token.from_string(string)

def test_to_unicode(self):

self.failUnlessRaises(TypeError, oauth.to_unicode, b'\xae')

self.failUnlessRaises(TypeError, oauth.to_unicode_optional_iterator, b'\xae')

self.failUnlessRaises(TypeError, oauth.to_unicode_optional_iterator, [b'\xae'])

self.failUnlessEqual(oauth.to_unicode(b':-)'), ':-)')

self.failUnlessEqual(oauth.to_unicode('\u00ae'), '\u00ae')

self.failUnlessEqual(oauth.to_unicode(b'\xc2\xae'), '\u00ae')

self.failUnlessEqual(oauth.to_unicode_optional_iterator([b':-)']), [':-)'])

def test_setter(self):

url = "http://example.com"

method = "GET"

req = oauth.Request(method)

self.assertTrue(not hasattr(req, 'url') or req.url is None)

self.assertTrue(not hasattr(req, 'normalized_url') or req.normalized_url is None)

def test_deleter(self):

url = "http://example.com"

method = "GET"

req = oauth.Request(method, url)

try:

del req.url

url = req.url

self.fail("AttributeError should have been raised on empty url.")

except AttributeError:

pass

except Exception as e:

self.fail(str(e))

def test_url(self):

url1 = "http://example.com:80/foo.php"

url2 = "https://example.com:443/foo.php"

exp1 = "http://example.com/foo.php"

exp2 = "https://example.com/foo.php"

method = "GET"

req = oauth.Request(method, url1)

self.assertEquals(req.normalized_url, exp1)

self.assertEquals(req.url, url1)

req = oauth.Request(method, url2)

self.assertEquals(req.normalized_url, exp2)

self.assertEquals(req.url, url2)

def test_bad_url(self):

request = oauth.Request()

try:

request.url = "ftp://example.com"

self.fail("Invalid URL scheme was accepted.")

except ValueError:

pass

def test_unset_consumer_and_token(self):

consumer = oauth.Consumer('my_consumer_key', 'my_consumer_secret')

token = oauth.Token('my_key', 'my_secret')

request = oauth.Request("GET", "http://example.com/fetch.php")

request.sign_request(oauth.SignatureMethod_HMAC_SHA1(), consumer,

token)

self.assertEquals(consumer.key, request['oauth_consumer_key'])

self.assertEquals(token.key, request['oauth_token'])

def test_no_url_set(self):

consumer = oauth.Consumer('my_consumer_key', 'my_consumer_secret')

token = oauth.Token('my_key', 'my_secret')

request = oauth.Request()

try:

try:

request.sign_request(oauth.SignatureMethod_HMAC_SHA1(),

consumer, token)

except TypeError:

self.fail("Signature method didn't check for a normalized URL.")

except ValueError:

pass

def test_url_query(self):

url = "https://www.google.com/m8/feeds/contacts/default/full/?alt=json&max-contacts=10"

normalized_url = urllib.parse.urlunparse(urllib.parse.urlparse(url)[:3] + (None, None, None))

method = "GET"

req = oauth.Request(method, url)

self.assertEquals(req.url, url)

self.assertEquals(req.normalized_url, normalized_url)

def test_get_parameter(self):

url = "http://example.com"

method = "GET"

params = {'oauth_consumer' : 'asdf'}

req = oauth.Request(method, url, parameters=params)

self.assertEquals(req.get_parameter('oauth_consumer'), 'asdf')

self.assertRaises(oauth.Error, req.get_parameter, 'blah')

def test_get_nonoauth_parameters(self):

oauth_params = {

'oauth_consumer': 'asdfasdfasdf'

}

other_params = {

'foo': 'baz',

'bar': 'foo',

'multi': ['FOO','BAR'],

'uni_utf8': '\xae',

'uni_unicode': '\u00ae',

'uni_unicode_2': 'åÅøØ',

}

params = oauth_params

params.update(other_params)

req = oauth.Request("GET", "http://example.com", params)

self.assertEquals(other_params, req.get_nonoauth_parameters())

def test_to_header(self):

realm = "http://sp.example.com/"

params = {

'oauth_version': "1.0",

'oauth_nonce': "4572616e48616d6d65724c61686176",

'oauth_timestamp': "137131200",

'oauth_consumer_key': "0685bd9184jfhq22",

'oauth_signature_method': "HMAC-SHA1",

'oauth_token': "ad180jjd733klru7",

'oauth_signature': "wOJIO9A2W5mFwDgiDvZbTSMK%2FPY%3D",

}

req = oauth.Request("GET", realm, params)

header, value = list(req.to_header(realm).items())[0]

parts = value.split('OAuth ')

vars = parts[1].split(', ')

self.assertTrue(len(vars), (len(params) + 1))

res = {}

for v in vars:

var, val = v.split('=')

res[var] = urllib.parse.unquote(val.strip('"'))

self.assertEquals(realm, res['realm'])

del res['realm']

self.assertTrue(len(res), len(params))

for key, val in list(res.items()):

self.assertEquals(val, params.get(key))

def test_to_postdata(self):

realm = "http://sp.example.com/"

params = {

'multi': ['FOO','BAR'],

'oauth_version': "1.0",

'oauth_nonce': "4572616e48616d6d65724c61686176",

'oauth_timestamp': "137131200",

'oauth_consumer_key': "0685bd9184jfhq22",

'oauth_signature_method': "HMAC-SHA1",

'oauth_token': "ad180jjd733klru7",

'oauth_signature': "wOJIO9A2W5mFwDgiDvZbTSMK%2FPY%3D",

}

req = oauth.Request("GET", realm, params)

flat = [('multi','FOO'),('multi','BAR')]

del params['multi']

flat.extend(list(params.items()))

kf = lambda x: x[0]

self.assertEquals(sorted(flat, key=kf), sorted(parse_qsl(req.to_postdata()), key=kf))

def test_to_url(self):

url = "http://sp.example.com/"

params = {

'oauth_version': "1.0",

'oauth_nonce': "4572616e48616d6d65724c61686176",

'oauth_timestamp': "137131200",

'oauth_consumer_key': "0685bd9184jfhq22",

'oauth_signature_method': "HMAC-SHA1",

'oauth_token': "ad180jjd733klru7",

'oauth_signature': "wOJIO9A2W5mFwDgiDvZbTSMK%2FPY%3D",

}

req = oauth.Request("GET", url, params)

exp = urllib.parse.urlparse("%s?%s" % (url, urllib.parse.urlencode(params)))

res = urllib.parse.urlparse(req.to_url())

self.assertEquals(exp.scheme, res.scheme)

self.assertEquals(exp.netloc, res.netloc)

self.assertEquals(exp.path, res.path)

a = parse_qs(exp.query)

b = parse_qs(res.query)

self.assertEquals(a, b)

def test_to_url_with_query(self):

url = "https://www.google.com/m8/feeds/contacts/default/full/?alt=json&max-contacts=10"

params = {

'oauth_version': "1.0",

'oauth_nonce': "4572616e48616d6d65724c61686176",

'oauth_timestamp': "137131200",

'oauth_consumer_key': "0685bd9184jfhq22",

'oauth_signature_method': "HMAC-SHA1",

'oauth_token': "ad180jjd733klru7",

'oauth_signature': "wOJIO9A2W5mFwDgiDvZbTSMK%2FPY%3D",

}

req = oauth.Request("GET", url, params)

# Note: the url above already has query parameters, so append new ones with &

exp = urllib.parse.urlparse("%s&%s" % (url, urllib.parse.urlencode(params)))

res = urllib.parse.urlparse(req.to_url())

self.assertEquals(exp.scheme, res.scheme)

self.assertEquals(exp.netloc, res.netloc)

self.assertEquals(exp.path, res.path)

a = parse_qs(exp.query)

b = parse_qs(res.query)

self.assertTrue('alt' in b)

self.assertTrue('max-contacts' in b)

self.assertEquals(b['alt'], ['json'])

self.assertEquals(b['max-contacts'], ['10'])

self.assertEquals(a, b)

def test_signature_base_string_nonascii_nonutf8(self):

consumer = oauth.Consumer('consumer_token', 'consumer_secret')

url = 'http://api.simplegeo.com:80/1.0/places/address.json?q=monkeys&category=animal&address=41+Decatur+St,+San+Francisc\u2766,+CA'

req = oauth.Request("GET", url)

self.failUnlessReallyEqual(req.normalized_url, 'http://api.simplegeo.com/1.0/places/address.json')

req.sign_request(oauth.SignatureMethod_HMAC_SHA1(), consumer, None)

self.failUnlessReallyEqual(req['oauth_signature'], b'WhufgeZKyYpKsI70GZaiDaYwl6g=')

url = b'http://api.simplegeo.com:80/1.0/places/address.json?q=monkeys&category=animal&address=41+Decatur+St,+San+Francisc\xe2\x9d\xa6,+CA'

req = oauth.Request("GET", url)

self.failUnlessReallyEqual(req.normalized_url, 'http://api.simplegeo.com/1.0/places/address.json')

req.sign_request(oauth.SignatureMethod_HMAC_SHA1(), consumer, None)

self.failUnlessReallyEqual(req['oauth_signature'], b'WhufgeZKyYpKsI70GZaiDaYwl6g=')

url = 'http://api.simplegeo.com:80/1.0/places/address.json?q=monkeys&category=animal&address=41+Decatur+St,+San+Francisc%E2%9D%A6,+CA'

req = oauth.Request("GET", url)

self.failUnlessReallyEqual(req.normalized_url, 'http://api.simplegeo.com/1.0/places/address.json')

req.sign_request(oauth.SignatureMethod_HMAC_SHA1(), consumer, None)

self.failUnlessReallyEqual(req['oauth_signature'], b'WhufgeZKyYpKsI70GZaiDaYwl6g=')

url = 'http://api.simplegeo.com:80/1.0/places/address.json?q=monkeys&category=animal&address=41+Decatur+St,+San+Francisc%E2%9D%A6,+CA'

req = oauth.Request("GET", url)

self.failUnlessReallyEqual(req.normalized_url, 'http://api.simplegeo.com/1.0/places/address.json')

req.sign_request(oauth.SignatureMethod_HMAC_SHA1(), consumer, None)

self.failUnlessReallyEqual(req['oauth_signature'], b'WhufgeZKyYpKsI70GZaiDaYwl6g=')

def test_signature_base_string_with_query(self):

url = "https://www.google.com/m8/feeds/contacts/default/full/?alt=json&max-contacts=10"

params = {

'oauth_version': "1.0",

'oauth_nonce': "4572616e48616d6d65724c61686176",

'oauth_timestamp': "137131200",

'oauth_consumer_key': "0685bd9184jfhq22",

'oauth_signature_method': "HMAC-SHA1",

'oauth_token': "ad180jjd733klru7",

'oauth_signature': "wOJIO9A2W5mFwDgiDvZbTSMK%2FPY%3D",

}

req = oauth.Request("GET", url, params)

self.assertEquals(req.normalized_url, 'https://www.google.com/m8/feeds/contacts/default/full/')

self.assertEquals(req.url, 'https://www.google.com/m8/feeds/contacts/default/full/?alt=json&max-contacts=10')

normalized_params = parse_qsl(req.get_normalized_parameters())

self.assertTrue(len(normalized_params), len(params) + 2)

normalized_params = dict(normalized_params)

for key, value in params.items():

if key == 'oauth_signature':

continue

self.assertEquals(value, normalized_params[key])

self.assertEquals(normalized_params['alt'], 'json')

self.assertEquals(normalized_params['max-contacts'], '10')

def test_get_normalized_parameters_empty(self):

url = "http://sp.example.com/?empty="

req = oauth.Request("GET", url)

res = req.get_normalized_parameters()

expected='empty='

self.assertEquals(expected, res)

def test_get_normalized_parameters_duplicate(self):

url = "http://example.com/v2/search/videos?oauth_nonce=79815175&oauth_timestamp=1295397962&oauth_consumer_key=mykey&oauth_signature_method=HMAC-SHA1&q=car&oauth_version=1.0&offset=10&oauth_signature=spWLI%2FGQjid7sQVd5%2FarahRxzJg%3D"

req = oauth.Request("GET", url)

res = req.get_normalized_parameters()

expected='oauth_consumer_key=mykey&oauth_nonce=79815175&oauth_signature_method=HMAC-SHA1&oauth_timestamp=1295397962&oauth_version=1.0&offset=10&q=car'

self.assertEquals(expected, res)

def test_get_normalized_parameters_from_url(self):

# example copied from

# https://github.com/ciaranj/node-oauth/blob/master/tests/oauth.js

# which in turns says that it was copied from

# http://oauth.net/core/1.0/#sig_base_example .

url = "http://photos.example.net/photos?file=vacation.jpg&oauth_consumer_key=dpf43f3p2l4k3l03&oauth_nonce=kllo9940pd9333jh&oauth_signature_method=HMAC-SHA1&oauth_timestamp=1191242096&oauth_token=nnch734d00sl2jdk&oauth_version=1.0&size=original"

req = oauth.Request("GET", url)

res = req.get_normalized_parameters()

expected = 'file=vacation.jpg&oauth_consumer_key=dpf43f3p2l4k3l03&oauth_nonce=kllo9940pd9333jh&oauth_signature_method=HMAC-SHA1&oauth_timestamp=1191242096&oauth_token=nnch734d00sl2jdk&oauth_version=1.0&size=original'

self.assertEquals(expected, res)

def test_signing_base(self):

# example copied from

# https://github.com/ciaranj/node-oauth/blob/master/tests/oauth.js

# which in turns says that it was copied from

# http://oauth.net/core/1.0/#sig_base_example .

url = "http://photos.example.net/photos?file=vacation.jpg&oauth_consumer_key=dpf43f3p2l4k3l03&oauth_nonce=kllo9940pd9333jh&oauth_signature_method=HMAC-SHA1&oauth_timestamp=1191242096&oauth_token=nnch734d00sl2jdk&oauth_version=1.0&size=original"

req = oauth.Request("GET", url)

sm = oauth.SignatureMethod_HMAC_SHA1()

consumer = oauth.Consumer('dpf43f3p2l4k3l03', 'foo')

key, raw = sm.signing_base(req, consumer, None)

expected = 'GET&http%3A%2F%2Fphotos.example.net%2Fphotos&file%3Dvacation.jpg%26oauth_consumer_key%3Ddpf43f3p2l4k3l03%26oauth_nonce%3Dkllo9940pd9333jh%26oauth_signature_method%3DHMAC-SHA1%26oauth_timestamp%3D1191242096%26oauth_token%3Dnnch734d00sl2jdk%26oauth_version%3D1.0%26size%3Doriginal'

self.assertEquals(expected, raw)

def test_get_normalized_parameters(self):

url = "http://sp.example.com/"

params = {

'oauth_version': "1.0",

'oauth_nonce': "4572616e48616d6d65724c61686176",

'oauth_timestamp': "137131200",

'oauth_consumer_key': "0685bd9184jfhq22",

'oauth_signature_method': "HMAC-SHA1",

'oauth_token': "ad180jjd733klru7",

'multi': ['FOO','BAR', '\u00ae', b'\xc2\xae'],

'multi_same': ['FOO','FOO'],

'uni_utf8_bytes': b'\xc2\xae',

'uni_unicode_object': '\u00ae'

}

req = oauth.Request("GET", url, params)

res = req.get_normalized_parameters()

expected='multi=BAR&multi=FOO&multi=%C2%AE&multi=%C2%AE&multi_same=FOO&multi_same=FOO&oauth_consumer_key=0685bd9184jfhq22&oauth_nonce=4572616e48616d6d65724c61686176&oauth_signature_method=HMAC-SHA1&oauth_timestamp=137131200&oauth_token=ad180jjd733klru7&oauth_version=1.0&uni_unicode_object=%C2%AE&uni_utf8_bytes=%C2%AE'

self.assertEquals(expected, res)

def test_get_normalized_parameters_ignores_auth_signature(self):

url = "http://sp.example.com/"

params = {

'oauth_version': "1.0",

'oauth_nonce': "4572616e48616d6d65724c61686176",

'oauth_timestamp': "137131200",

'oauth_consumer_key': "0685bd9184jfhq22",

'oauth_signature_method': "HMAC-SHA1",

'oauth_signature': "some-random-signature-%d" % random.randint(1000, 2000),

'oauth_token': "ad180jjd733klru7",

}

req = oauth.Request("GET", url, params)

res = req.get_normalized_parameters()

self.assertNotEquals(urllib.parse.urlencode(sorted(params.items())), res)

foo = params.copy()

del foo["oauth_signature"]

self.assertEqual(urllib.parse.urlencode(sorted(foo.items())), res)

def test_set_signature_method(self):

consumer = oauth.Consumer('key', 'secret')

client = oauth.Client(consumer)

class Blah:

pass

try:

client.set_signature_method(Blah())

self.fail("Client.set_signature_method() accepted invalid method.")

except ValueError:

pass

m = oauth.SignatureMethod_HMAC_SHA1()

client.set_signature_method(m)

self.assertEquals(m, client.method)

def test_get_normalized_string_escapes_spaces_properly(self):

url = "http://sp.example.com/"

params = {

"some_random_data": random.randint(100, 1000),

"data": "This data with a random number (%d) has spaces!" % random.randint(1000, 2000),

}

req = oauth.Request("GET", url, params)

res = req.get_normalized_parameters()

expected = urllib.parse.urlencode(sorted(params.items())).replace('+', '%20')

self.assertEqual(expected, res)

@mock.patch('oauth2.Request.make_timestamp')

@mock.patch('oauth2.Request.make_nonce')

def test_request_nonutf8_bytes(self, mock_make_nonce, mock_make_timestamp):

mock_make_nonce.return_value = 5

mock_make_timestamp.return_value = 6

tok = oauth.Token(key="tok-test-key", secret="tok-test-secret")

con = oauth.Consumer(key="con-test-key", secret="con-test-secret")

params = {

'oauth_version': "1.0",

'oauth_nonce': "4572616e48616d6d65724c61686176",

'oauth_timestamp': "137131200",

'oauth_token': tok.key,

'oauth_consumer_key': con.key

}

# If someone passes a sequence of bytes which is not ascii for

# url, we'll raise an exception as early as possible.

url = b"http://sp.example.com/\x92" # It's actually cp1252-encoding...

self.assertRaises(TypeError, oauth.Request, method="GET", url=url, parameters=params)

# And if they pass an unicode, then we'll use it.

url = 'http://sp.example.com/\u2019'

req = oauth.Request(method="GET", url=url, parameters=params)

req.sign_request(oauth.SignatureMethod_HMAC_SHA1(), con, None)

self.failUnlessReallyEqual(req['oauth_signature'], b'cMzvCkhvLL57+sTIxLITTHfkqZk=')

# And if it is a utf-8-encoded-then-percent-encoded non-ascii

# thing, we'll decode it and use it.

url = "http://sp.example.com/%E2%80%99"

req = oauth.Request(method="GET", url=url, parameters=params)

req.sign_request(oauth.SignatureMethod_HMAC_SHA1(), con, None)

self.failUnlessReallyEqual(req['oauth_signature'], b'yMLKOyNKC/DkyhUOb8DLSvceEWE=')

# Same thing with the params.

url = "http://sp.example.com/"

# If someone passes a sequence of bytes which is not ascii in

# params, we'll raise an exception as early as possible.

params['non_oauth_thing'] = b'\xae', # It's actually cp1252-encoding...

self.assertRaises(TypeError, oauth.Request, method="GET", url=url, parameters=params)

# And if they pass a unicode, then we'll use it.

params['non_oauth_thing'] = '\u2019'

req = oauth.Request(method="GET", url=url, parameters=params)

req.sign_request(oauth.SignatureMethod_HMAC_SHA1(), con, None)

self.failUnlessReallyEqual(req['oauth_signature'], b'0GU50m0v60CVDB5JnoBXnvvvKx4=')

# And if it is a utf-8-encoded non-ascii thing, we'll decode

# it and use it.

params['non_oauth_thing'] = b'\xc2\xae'

req = oauth.Request(method="GET", url=url, parameters=params)

req.sign_request(oauth.SignatureMethod_HMAC_SHA1(), con, None)

self.failUnlessReallyEqual(req['oauth_signature'], b'pqOCu4qvRTiGiXB8Z61Jsey0pMM=')

# Also if there are non-utf8 bytes in the query args.

url = b"http://sp.example.com/?q=\x92" # cp1252

self.assertRaises(TypeError, oauth.Request, method="GET", url=url, parameters=params)

def test_request_hash_of_body(self):

tok = oauth.Token(key="token", secret="tok-test-secret")

con = oauth.Consumer(key="consumer", secret="con-test-secret")

# Example 1a from Appendix A.1 of

# http://oauth.googlecode.com/svn/spec/ext/body_hash/1.0/oauth-bodyhash.html

# Except that we get a differetn result than they do.

params = {

'oauth_version': "1.0",

'oauth_token': tok.key,

'oauth_nonce': 10288510250934,

'oauth_timestamp': 1236874155,

'oauth_consumer_key': con.key

}

url = "http://www.example.com/resource"

req = oauth.Request(method="PUT", url=url, parameters=params, body="Hello World!", is_form_encoded=False)

req.sign_request(oauth.SignatureMethod_HMAC_SHA1(), con, None)

self.failUnlessReallyEqual(req['oauth_body_hash'], b'Lve95gjOVATpfV8EL5X4nxwjKHE=')

self.failUnlessReallyEqual(req['oauth_signature'], b't+MX8l/0S8hdbVQL99nD0X1fPnM=')

# oauth-bodyhash.html A.1 has

# '08bUFF%2Fjmp59mWB7cSgCYBUpJ0U%3D', but I don't see how that

# is possible.

# Example 1b

params = {

'oauth_version': "1.0",

'oauth_token': tok.key,

'oauth_nonce': 10369470270925,

'oauth_timestamp': 1236874236,

'oauth_consumer_key': con.key

}

req = oauth.Request(method="PUT", url=url, parameters=params, body="Hello World!", is_form_encoded=False)

req.sign_request(oauth.SignatureMethod_HMAC_SHA1(), con, None)

self.failUnlessReallyEqual(req['oauth_body_hash'], b'Lve95gjOVATpfV8EL5X4nxwjKHE=')

self.failUnlessReallyEqual(req['oauth_signature'], b'CTFmrqJIGT7NsWJ42OrujahTtTc=')

# Appendix A.2

params = {

'oauth_version': "1.0",

'oauth_token': tok.key,

'oauth_nonce': 8628868109991,

'oauth_timestamp': 1238395022,

'oauth_consumer_key': con.key

}

req = oauth.Request(method="GET", url=url, parameters=params, is_form_encoded=False)

req.sign_request(oauth.SignatureMethod_HMAC_SHA1(), con, None)

self.failUnlessReallyEqual(req['oauth_body_hash'], b'2jmj7l5rSw0yVb/vlWAYkK/YBwk=')

self.failUnlessReallyEqual(req['oauth_signature'], b'Zhl++aWSP0O3/hYQ0CuBc7jv38I=')

def test_sign_request(self):

url = "http://sp.example.com/"

params = {

'oauth_version': "1.0",

'oauth_nonce': "4572616e48616d6d65724c61686176",

'oauth_timestamp': "137131200"

}

tok = oauth.Token(key="tok-test-key", secret="tok-test-secret")

con = oauth.Consumer(key="con-test-key", secret="con-test-secret")

params['oauth_token'] = tok.key

params['oauth_consumer_key'] = con.key

req = oauth.Request(method="GET", url=url, parameters=params)

methods = {

b'DX01TdHws7OninCLK9VztNTH1M4=': oauth.SignatureMethod_HMAC_SHA1(),

'con-test-secret&tok-test-secret': oauth.SignatureMethod_PLAINTEXT()

}

for exp, method in list(methods.items()):

req.sign_request(method, con, tok)

self.assertEquals(req['oauth_signature_method'], method.name)

self.assertEquals(req['oauth_signature'], exp)

# Also if there are non-ascii chars in the URL.

url = b"http://sp.example.com/\xe2\x80\x99" # utf-8 bytes

req = oauth.Request(method="GET", url=url, parameters=params)

req.sign_request(oauth.SignatureMethod_HMAC_SHA1(), con, tok)

self.assertEquals(req['oauth_signature'], b'loFvp5xC7YbOgd9exIO6TxB7H4s=')

url = 'http://sp.example.com/\u2019' # Python unicode object

req = oauth.Request(method="GET", url=url, parameters=params)

req.sign_request(oauth.SignatureMethod_HMAC_SHA1(), con, tok)

self.assertEquals(req['oauth_signature'], b'loFvp5xC7YbOgd9exIO6TxB7H4s=')

# Also if there are non-ascii chars in the query args.

url = b"http://sp.example.com/?q=\xe2\x80\x99" # utf-8 bytes

req = oauth.Request(method="GET", url=url, parameters=params)

req.sign_request(oauth.SignatureMethod_HMAC_SHA1(), con, tok)

self.assertEquals(req['oauth_signature'], b'IBw5mfvoCsDjgpcsVKbyvsDqQaU=')

url = 'http://sp.example.com/?q=\u2019' # Python unicode object

req = oauth.Request(method="GET", url=url, parameters=params)

req.sign_request(oauth.SignatureMethod_HMAC_SHA1(), con, tok)

self.assertEquals(req['oauth_signature'], b'IBw5mfvoCsDjgpcsVKbyvsDqQaU=')

def test_from_request(self):

url = "http://sp.example.com/"

params = {

'oauth_version': "1.0",

'oauth_nonce': "4572616e48616d6d65724c61686176",

'oauth_timestamp': "137131200",

'oauth_consumer_key': "0685bd9184jfhq22",

'oauth_signature_method': "HMAC-SHA1",

'oauth_token': "ad180jjd733klru7",

'oauth_signature': "wOJIO9A2W5mFwDgiDvZbTSMK%2FPY%3D",

}

req = oauth.Request("GET", url, params)

headers = req.to_header()

# Test from the headers

req = oauth.Request.from_request("GET", url, headers)

self.assertEquals(req.method, "GET")

self.assertEquals(req.url, url)

self.assertEquals(params, req.copy())

# Test with bad OAuth headers

bad_headers = {

'Authorization' : 'OAuth this is a bad header'

}

self.assertRaises(oauth.Error, oauth.Request.from_request, "GET",

url, bad_headers)

# Test getting from query string

qs = urllib.parse.urlencode(params)

req = oauth.Request.from_request("GET", url, query_string=qs)

exp = parse_qs(qs, keep_blank_values=False)

for k, v in exp.items():

exp[k] = urllib.parse.unquote(v[0])

self.assertEquals(exp, req.copy())

# Test that a boned from_request() call returns None

req = oauth.Request.from_request("GET", url)

self.assertEquals(None, req)

def test_from_token_and_callback(self):

url = "http://sp.example.com/"

params = {

'oauth_version': "1.0",

'oauth_nonce': "4572616e48616d6d65724c61686176",

'oauth_timestamp': "137131200",

'oauth_consumer_key': "0685bd9184jfhq22",

'oauth_signature_method': "HMAC-SHA1",

'oauth_token': "ad180jjd733klru7",

'oauth_signature': "wOJIO9A2W5mFwDgiDvZbTSMK%2FPY%3D",

}

tok = oauth.Token(key="tok-test-key", secret="tok-test-secret")

req = oauth.Request.from_token_and_callback(tok)

self.assertFalse('oauth_callback' in req)

self.assertEquals(req['oauth_token'], tok.key)

req = oauth.Request.from_token_and_callback(tok, callback=url)

self.assertTrue('oauth_callback' in req)

self.assertEquals(req['oauth_callback'], url)

def test_from_consumer_and_token(self):

url = "http://sp.example.com/"

tok = oauth.Token(key="tok-test-key", secret="tok-test-secret")

tok.set_verifier('this_is_a_test_verifier')

con = oauth.Consumer(key="con-test-key", secret="con-test-secret")

req = oauth.Request.from_consumer_and_token(con, token=tok,

http_method="GET", http_url=url)

self.assertEquals(req['oauth_token'], tok.key)

self.assertEquals(req['oauth_consumer_key'], con.key)

name = "BAD"

def signing_base(self, request, consumer, token):

return ""

def sign(self, request, consumer, token):

def setUp(self):

url = "http://sp.example.com/"

params = {

'oauth_version': "1.0",

'oauth_nonce': "4572616e48616d6d65724c61686176",

'oauth_timestamp': int(time.time()),

'bar': 'blerg',

'multi': ['FOO','BAR'],

'foo': 59

}

self.consumer = oauth.Consumer(key="consumer-key",

secret="consumer-secret")

self.token = oauth.Token(key="token-key", secret="token-secret")

params['oauth_token'] = self.token.key

params['oauth_consumer_key'] = self.consumer.key

self.request = oauth.Request(method="GET", url=url, parameters=params)

signature_method = oauth.SignatureMethod_HMAC_SHA1()

self.request.sign_request(signature_method, self.consumer, self.token)

def test_init(self):

server = oauth.Server(signature_methods={'HMAC-SHA1' : oauth.SignatureMethod_HMAC_SHA1()})

self.assertTrue('HMAC-SHA1' in server.signature_methods)

self.assertTrue(isinstance(server.signature_methods['HMAC-SHA1'],

oauth.SignatureMethod_HMAC_SHA1))

server = oauth.Server()

self.assertEquals(server.signature_methods, {})

def test_add_signature_method(self):

server = oauth.Server()

res = server.add_signature_method(oauth.SignatureMethod_HMAC_SHA1())

self.assertTrue(len(res) == 1)

self.assertTrue('HMAC-SHA1' in res)

self.assertTrue(isinstance(res['HMAC-SHA1'],

oauth.SignatureMethod_HMAC_SHA1))

res = server.add_signature_method(oauth.SignatureMethod_PLAINTEXT())

self.assertTrue(len(res) == 2)

self.assertTrue('PLAINTEXT' in res)

self.assertTrue(isinstance(res['PLAINTEXT'],

oauth.SignatureMethod_PLAINTEXT))

def test_verify_request(self):

server = oauth.Server()

server.add_signature_method(oauth.SignatureMethod_HMAC_SHA1())

parameters = server.verify_request(self.request, self.consumer,

self.token)

self.assertTrue('bar' in parameters)

self.assertTrue('foo' in parameters)

self.assertTrue('multi' in parameters)

self.assertEquals(parameters['bar'], 'blerg')

self.assertEquals(parameters['foo'], 59)

self.assertEquals(parameters['multi'], ['FOO','BAR'])

def test_build_authenticate_header(self):

server = oauth.Server()

headers = server.build_authenticate_header('example.com')

self.assertTrue('WWW-Authenticate' in headers)

self.assertEquals('OAuth realm="example.com"',

headers['WWW-Authenticate'])

def test_no_version(self):

url = "http://sp.example.com/"

params = {

'oauth_nonce': "4572616e48616d6d65724c61686176",

'oauth_timestamp': int(time.time()),

'bar': 'blerg',

'multi': ['FOO','BAR'],

'foo': 59

}

self.consumer = oauth.Consumer(key="consumer-key",

secret="consumer-secret")

self.token = oauth.Token(key="token-key", secret="token-secret")