| #!/usr/bin/python |
| # -*- coding: utf-8 -*- |
| |
| __author__ = 'samuel.cyprian@gmail.com (Samuel Cyprian)' |
| |
| import unittest |
| from gdata import oauth, test_config |
| |
| HTTP_METHOD_POST = 'POST' |
| VERSION = '1.0' |
| |
| |
| class OauthUtilsTest(unittest.TestCase): |
| def test_build_authenticate_header(self): |
| self.assertEqual(oauth.build_authenticate_header(), |
| {'WWW-Authenticate' :'OAuth realm=""'}) |
| self.assertEqual(oauth.build_authenticate_header('foo'), |
| {'WWW-Authenticate': 'OAuth realm="foo"'}) |
| |
| def test_escape(self): |
| #Special cases |
| self.assertEqual(oauth.escape('~'), '~') |
| self.assertEqual(oauth.escape('/'), '%2F') |
| self.assertEqual(oauth.escape('+'), '%2B') |
| self.assertEqual(oauth.escape(' '), '%20') |
| self.assertEqual(oauth.escape('Peter Strömberg'), |
| 'Peter%20Str%C3%B6mberg') |
| |
| def test_generate_timestamp(self): |
| self.assertTrue(oauth.generate_timestamp()>0) |
| self.assertTrue(type(oauth.generate_timestamp()) is type(0)) |
| |
| def test_generate_nonce(self): |
| DEFAULT_NONCE_LENGTH = 8 |
| self.assertTrue(len(oauth.generate_nonce()) is DEFAULT_NONCE_LENGTH) |
| self.assertTrue(type(oauth.generate_nonce()) is type('')) |
| |
| class OAuthConsumerTest(unittest.TestCase): |
| def setUp(self): |
| self.key = 'key' |
| self.secret = 'secret' |
| self.consumer = oauth.OAuthConsumer(self.key, self.secret) |
| |
| def test_OAuthConsumer_attr_key(self): |
| self.assertEqual(self.consumer.key, self.key) |
| |
| def test_OAuthConsumer_attr_secret(self): |
| self.assertEqual(self.consumer.secret, self.secret) |
| |
| class OAuthTokenTest(unittest.TestCase): |
| def setUp(self): |
| self.key = 'key' |
| self.secret = 'secret' |
| self.token = oauth.OAuthToken(self.key, self.secret) |
| |
| def test_OAuthToken_attr_key(self): |
| self.assertEqual(self.token.key, self.key) |
| |
| def test_OAuthToken_attr_secret(self): |
| self.assertEqual(self.token.secret, self.secret) |
| |
| def test_to_string(self): |
| self.assertEqual(self.token.to_string(), |
| 'oauth_token_secret=secret&oauth_token=key') |
| t = oauth.OAuthToken('+', '%') |
| self.assertEqual(t.to_string(), |
| 'oauth_token_secret=%25&oauth_token=%2B') |
| |
| def test_from_string(self): |
| s = 'oauth_token_secret=secret&oauth_token=key' |
| t = oauth.OAuthToken.from_string(s) |
| self.assertEqual(t.key, 'key') |
| self.assertEqual(t.secret, 'secret') |
| t = oauth.OAuthToken.from_string('oauth_token_secret=%25&oauth_token=%2B') |
| self.assertEqual(t.key, '+') |
| self.assertEqual(t.secret, '%') |
| |
| def test___str__(self): |
| self.assertEqual(str(self.token), |
| 'oauth_token_secret=secret&oauth_token=key') |
| t = oauth.OAuthToken('+', '%') |
| self.assertEqual(str(t), 'oauth_token_secret=%25&oauth_token=%2B') |
| |
| class OAuthParameters(object): |
| CONSUMER_KEY = 'oauth_consumer_key' |
| TOKEN = 'oauth_token' |
| SIGNATURE_METHOD = 'oauth_signature_method' |
| SIGNATURE = 'oauth_signature' |
| TIMESTAMP = 'oauth_timestamp' |
| NONCE = 'oauth_nonce' |
| VERSION = 'oauth_version' |
| CALLBACK = 'oauth_callback' |
| |
| ALL_PARAMETERS = (CONSUMER_KEY, |
| TOKEN, |
| SIGNATURE_METHOD, |
| SIGNATURE, |
| TIMESTAMP, |
| NONCE, |
| VERSION) |
| |
| class OAuthTest(unittest.TestCase): |
| def setUp(self): |
| self.consumer = oauth.OAuthConsumer('a56b5ff0a637ab283d1d8e32ced37a9c', |
| '9a3248210c84b264b56b98c0b872bc8a') |
| self.token = oauth.OAuthToken('5b2cafbf20b11bace53b29e37d8a673d', |
| '3f71254637df2002d8819458ae4f6c51') |
| self.http_url = 'http://dev.alicehub.com/server/api/newsfeed/update/' |
| |
| self.http_method = HTTP_METHOD_POST |
| |
| class OAuthRequestTest(OAuthTest): |
| def setUp(self): |
| super(OAuthRequestTest, self).setUp() |
| self.signature_method = oauth.OAuthSignatureMethod_HMAC_SHA1() |
| |
| self.non_oauth_param_message = 'message' |
| self.non_oauth_param_context_id = 'context_id' |
| self.parameters = {OAuthParameters.CONSUMER_KEY:self.consumer.key, |
| OAuthParameters.TOKEN: self.token.key, |
| OAuthParameters.SIGNATURE_METHOD: 'HMAC-SHA1', |
| OAuthParameters.SIGNATURE: |
| '947ysBZiMn6FGZ11AW06Ioco4mo=', |
| OAuthParameters.TIMESTAMP: '1278573584', |
| OAuthParameters.NONCE: '1770704051', |
| OAuthParameters.VERSION: VERSION, |
| self.non_oauth_param_message:'hey', |
| self.non_oauth_param_context_id:'',} |
| oauth_params_string = """ |
| oauth_nonce="1770704051", |
| oauth_timestamp="1278573584", |
| oauth_consumer_key="a56b5ff0a637ab283d1d8e32ced37a9c", |
| oauth_signature_method="HMAC-SHA1", |
| oauth_version="1.0", |
| oauth_token="5b2cafbf20b11bace53b29e37d8a673d", |
| oauth_signature="947ysBZiMn6FGZ11AW06Ioco4mo%3D" |
| """ |
| self.oauth_header_with_realm = {'Authorization': """OAuth |
| realm="http://example.com", %s """ % oauth_params_string} |
| self.oauth_header_without_realm = {'Authorization': 'OAuth %s' |
| % oauth_params_string} |
| |
| |
| self.additional_param = 'foo' |
| self.additional_value = 'bar' |
| |
| self.oauth_request = oauth.OAuthRequest(self.http_method, |
| self.http_url, |
| self.parameters) |
| |
| def test_set_parameter(self): |
| self.oauth_request.set_parameter(self.additional_param, |
| self.additional_value) |
| self.assertEqual(self.oauth_request.get_parameter(self.additional_param), |
| self.additional_value) |
| |
| def test_get_parameter(self): |
| self.assertRaises(oauth.OAuthError, |
| self.oauth_request.get_parameter, |
| self.additional_param) |
| self.oauth_request.set_parameter(self.additional_param, |
| self.additional_value) |
| self.assertEqual(self.oauth_request.get_parameter(self.additional_param), |
| self.additional_value) |
| |
| def test__get_timestamp_nonce(self): |
| self.assertEqual(self.oauth_request._get_timestamp_nonce(), |
| (self.parameters[OAuthParameters.TIMESTAMP], |
| self.parameters[OAuthParameters.NONCE])) |
| |
| def test_get_nonoauth_parameters(self): |
| non_oauth_params = self.oauth_request.get_nonoauth_parameters() |
| self.assertTrue(non_oauth_params.has_key(self.non_oauth_param_message)) |
| self.assertFalse(non_oauth_params.has_key(OAuthParameters.CONSUMER_KEY)) |
| |
| def test_to_header(self): |
| realm = 'google' |
| header_without_realm = self.oauth_request.to_header()\ |
| .get('Authorization') |
| header_with_realm = self.oauth_request.to_header(realm)\ |
| .get('Authorization') |
| self.assertTrue(header_with_realm.find(realm)) |
| for k in OAuthParameters.ALL_PARAMETERS: |
| self.assertTrue(header_without_realm.find(k) > -1) |
| self.assertTrue(header_with_realm.find(k) > -1) |
| |
| def check_for_params_in_string(self, params, s): |
| for k, v in params.iteritems(): |
| self.assertTrue(s.find(oauth.escape(k)) > -1) |
| self.assertTrue(s.find(oauth.escape(v)) > -1) |
| |
| def test_to_postdata(self): |
| post_data = self.oauth_request.to_postdata() |
| self.check_for_params_in_string(self.parameters, post_data) |
| |
| def test_to_url(self): |
| GET_url = self.oauth_request.to_url() |
| self.assertTrue(GET_url\ |
| .find(self.oauth_request.get_normalized_http_url()) > -1) |
| self.assertTrue(GET_url.find('?') > -1) |
| self.check_for_params_in_string(self.parameters, GET_url) |
| |
| def test_get_normalized_parameters(self): |
| _params = self.parameters.copy() |
| normalized_params = self.oauth_request.get_normalized_parameters() |
| self.assertFalse(normalized_params\ |
| .find(OAuthParameters.SIGNATURE + '=') > -1) |
| self.assertTrue(self.parameters.get(OAuthParameters.SIGNATURE) is None) |
| |
| key_values = [tuple(kv.split('=')) for kv in normalized_params.split('&')] |
| del _params[OAuthParameters.SIGNATURE] |
| expected_key_values = _params.items() |
| expected_key_values.sort() |
| |
| for k, v in expected_key_values: |
| self.assertTrue(expected_key_values.index((k,v))\ |
| is key_values.index((oauth.escape(k), oauth.escape(v)))) |
| |
| def test_get_normalized_http_method(self): |
| lower_case_http_method = HTTP_METHOD_POST.lower() |
| self.oauth_request.http_method = lower_case_http_method |
| self.assertEqual(self.oauth_request.get_normalized_http_method(), |
| lower_case_http_method.upper()) |
| |
| def test_get_normalized_http_url(self): |
| url1 = 'HTTP://Example.com:80/resource?id=123' |
| expected_url1 = "http://example.com/resource" |
| self.oauth_request.http_url = url1 |
| self.assertEqual(self.oauth_request.get_normalized_http_url(), |
| expected_url1) |
| |
| url2 = 'HTTPS://Example.com:443/resource?id=123' |
| expected_url2 = "https://example.com/resource" |
| self.oauth_request.http_url = url2 |
| self.assertEqual(self.oauth_request.get_normalized_http_url(), |
| expected_url2) |
| |
| url3 = 'HTTP://Example.com:8080/resource?id=123' |
| expected_url3 = "http://example.com:8080/resource" |
| self.oauth_request.http_url = url3 |
| self.assertEqual(self.oauth_request.get_normalized_http_url(), |
| expected_url3) |
| |
| def test_sign_request(self): |
| expected_signature = self.oauth_request.parameters\ |
| .get(OAuthParameters.SIGNATURE) |
| del self.oauth_request.parameters[OAuthParameters.SIGNATURE] |
| self.oauth_request.sign_request(self.signature_method, |
| self.consumer, |
| self.token) |
| self.assertEqual(self.oauth_request.parameters\ |
| .get(OAuthParameters.SIGNATURE), expected_signature) |
| |
| def test_build_signature(self): |
| expected_signature = self.oauth_request.parameters\ |
| .get(OAuthParameters.SIGNATURE) |
| self.assertEqual(self.oauth_request.build_signature(self.signature_method, |
| self.consumer, |
| self.token), |
| expected_signature) |
| |
| def test_from_request(self): |
| request = oauth.OAuthRequest.from_request(self.http_method, self.http_url, |
| self.oauth_header_with_realm, |
| {}, |
| "message=hey&context_id=") |
| self.assertEqual(request.__dict__, self.oauth_request.__dict__) |
| self.assertTrue(isinstance(request, oauth.OAuthRequest)) |
| |
| def test_from_consumer_and_token(self): |
| request = oauth.OAuthRequest.from_consumer_and_token(self.consumer, |
| self.token, |
| self.http_method, |
| self.http_url) |
| self.assertTrue(isinstance(request, oauth.OAuthRequest)) |
| |
| def test_from_token_and_callback(self): |
| callback = 'http://example.com' |
| request = oauth.OAuthRequest.from_token_and_callback(self.token, |
| callback, |
| self.http_method, |
| self.http_url) |
| self.assertTrue(isinstance(request, oauth.OAuthRequest)) |
| self.assertEqual(request.get_parameter(OAuthParameters.CALLBACK), callback) |
| |
| def test__split_header(self): |
| del self.parameters[self.non_oauth_param_message] |
| del self.parameters[self.non_oauth_param_context_id] |
| self.assertEqual(oauth.OAuthRequest._split_header(self\ |
| .oauth_header_with_realm['Authorization']), self.parameters) |
| self.assertEqual(oauth.OAuthRequest._split_header(self\ |
| .oauth_header_without_realm['Authorization']), self.parameters) |
| |
| def test_split_url_string(self): |
| qs = "a=1&c=hi%20there&empty=" |
| expected_result = {'a': '1', |
| 'c': 'hi there', |
| 'empty': ''} |
| self.assertEqual(oauth.OAuthRequest._split_url_string(qs), expected_result) |
| |
| class OAuthServerTest(OAuthTest): |
| def setUp(self): |
| super(OAuthServerTest, self).setUp() |
| self.signature_method = oauth.OAuthSignatureMethod_HMAC_SHA1() |
| self.data_store = MockOAuthDataStore() |
| self.user = MockUser('Foo Bar') |
| |
| self.request_token_url = "http://example.com/oauth/request_token" |
| self.access_token_url = "http://example.com/oauth/access_token" |
| |
| self.oauth_server = oauth.OAuthServer(self.data_store, |
| {self.signature_method.get_name():self.signature_method}) |
| |
| def _prepare_request(self, request, token = None): |
| request.set_parameter(OAuthParameters.SIGNATURE_METHOD, |
| self.signature_method.get_name()) |
| request.set_parameter(OAuthParameters.NONCE, oauth.generate_nonce()) |
| request.set_parameter(OAuthParameters.TIMESTAMP, |
| oauth.generate_timestamp()) |
| request.sign_request(self.signature_method, self.consumer, token) |
| |
| def _get_token(self, request): |
| self._prepare_request(request) |
| return self.oauth_server.fetch_request_token(request) |
| |
| def _get_authorized_token(self, request): |
| req_token = self._get_token(request) |
| return self.oauth_server.authorize_token(req_token, self.user) |
| |
| def test_set_data_store(self): |
| self.oauth_server.data_store = None |
| self.assertTrue(self.oauth_server.data_store is None) |
| self.oauth_server.set_data_store(self.data_store) |
| self.assertTrue(self.oauth_server.data_store is not None) |
| self.assertTrue(isinstance(self.oauth_server.data_store, |
| oauth.OAuthDataStore)) |
| |
| def test_get_data_store(self): |
| self.assertEqual(self.oauth_server.data_store, self.data_store) |
| |
| def test_add_signature_method(self): |
| signature_method = oauth.OAuthSignatureMethod_PLAINTEXT() |
| self.oauth_server.add_signature_method(signature_method) |
| self.assertTrue(isinstance(self.oauth_server.signature_methods\ |
| .get(signature_method.get_name()), |
| oauth.OAuthSignatureMethod_PLAINTEXT)) |
| |
| def test_fetch_request_token(self): |
| initial_request = oauth.OAuthRequest.from_consumer_and_token( |
| self.consumer, |
| http_method=self.http_method, |
| http_url=self.request_token_url |
| ) |
| req_token_1 = self._get_token(initial_request) |
| authorization_request = oauth.OAuthRequest.from_consumer_and_token( |
| self.consumer, |
| req_token_1, |
| http_method=self.http_method, |
| http_url=self.http_url |
| ) |
| req_token_2 = self._get_token(authorization_request) |
| self.assertEqual(req_token_1.key, req_token_2.key) |
| self.assertEqual(req_token_1.secret, req_token_2.secret) |
| |
| def _get_token_for_authorization(self): |
| request = oauth.OAuthRequest.from_consumer_and_token( |
| self.consumer, |
| http_method=self.http_method, |
| http_url=self.request_token_url |
| ) |
| request_token = self._get_token(request) |
| authorization_request = oauth.OAuthRequest.from_consumer_and_token( |
| self.consumer, |
| request_token, |
| http_method=self.http_method, |
| http_url=self.http_url |
| ) |
| return self._get_authorized_token(authorization_request) |
| |
| def test_authorize_token(self): |
| authorized_token = self._get_token_for_authorization() |
| self.assertTrue(authorized_token is not None) |
| |
| def _get_access_token_request(self, authorized_token): |
| access_token_request = oauth.OAuthRequest.from_consumer_and_token( |
| self.consumer, |
| authorized_token, |
| http_method=self.http_method, |
| http_url=self.access_token_url |
| ) |
| self._prepare_request(access_token_request, authorized_token) |
| return access_token_request |
| |
| def test_fetch_access_token(self): |
| authorized_token = self._get_token_for_authorization() |
| access_token_request = self._get_access_token_request(authorized_token) |
| access_token = self.oauth_server.fetch_access_token(access_token_request) |
| self.assertTrue(access_token is not None) |
| self.assertNotEqual(str(authorized_token), str(access_token)) |
| # Try to fetch access_token with used request token |
| self.assertRaises(oauth.OAuthError, self.oauth_server.fetch_access_token, |
| access_token_request) |
| |
| def test_verify_request(self): |
| authorized_token = self._get_token_for_authorization() |
| access_token_request = self._get_access_token_request(authorized_token) |
| access_token = self.oauth_server.fetch_access_token(access_token_request) |
| param1 = 'p1' |
| value1 = 'v1' |
| api_request = oauth.OAuthRequest.from_consumer_and_token( |
| self.consumer, |
| access_token, |
| http_method=self.http_method, |
| http_url=self.http_url, |
| parameters={param1:value1} |
| ) |
| self._prepare_request(api_request, access_token) |
| result = self.oauth_server.verify_request(api_request) |
| self.assertTrue(result is not None) |
| consumer, token, parameters = result |
| self.assertEqual(parameters.get(param1), value1) |
| |
| def test_get_callback(self): |
| request = oauth.OAuthRequest.from_consumer_and_token( |
| self.consumer, |
| None, |
| http_method=self.http_method, |
| http_url=self.http_url |
| ) |
| self._prepare_request(request) |
| cb_url = 'http://example.com/cb' |
| request.set_parameter(OAuthParameters.CALLBACK, cb_url) |
| self.assertEqual(self.oauth_server.get_callback(request), cb_url) |
| |
| def test_build_authenticate_header(self): |
| self.assertEqual(oauth.build_authenticate_header(), {'WWW-Authenticate': |
| 'OAuth realm=""'}) |
| self.assertEqual(oauth.build_authenticate_header('foo'), |
| {'WWW-Authenticate': 'OAuth realm="foo"'}) |
| |
| class OAuthClientTest(OAuthTest): |
| def setUp(self): |
| super(OAuthClientTest, self).setUp() |
| self.oauth_client = oauth.OAuthClient(self.consumer, self.token) |
| |
| def test_get_consumer(self): |
| consumer = self.oauth_client.get_consumer() |
| self.assertTrue(isinstance(consumer, oauth.OAuthConsumer)) |
| self.assertEqual(consumer.__dict__, self.consumer.__dict__) |
| |
| def test_get_token(self): |
| token = self.oauth_client.get_token() |
| self.assertTrue(isinstance(token, oauth.OAuthToken)) |
| self.assertEqual(token.__dict__, self.token.__dict__) |
| |
| #Mockup OAuthDataStore |
| TOKEN_TYPE_REQUEST = 'request' |
| TOKEN_TYPE_ACCESS = 'access' |
| class MockOAuthDataStore(oauth.OAuthDataStore): |
| def __init__(self): |
| self.consumer = oauth.OAuthConsumer('a56b5ff0a637ab283d1d8e32ced37a9c', |
| '9a3248210c84b264b56b98c0b872bc8a') |
| |
| self.consumer_db = {self.consumer.key: self.consumer} |
| self.request_token_db = {} |
| self.access_token_db = {} |
| self.nonce = None |
| |
| def lookup_consumer(self, key): |
| return self.consumer_db.get(key) |
| |
| def lookup_token(self, oauth_consumer, token_type, token_field): |
| data = None |
| if token_type == TOKEN_TYPE_REQUEST: |
| data = self.request_token_db.get(token_field) |
| elif token_type == TOKEN_TYPE_ACCESS: |
| data = self.access_token_db.get(token_field) |
| |
| if data: |
| token, consumer, authenticated_user = data |
| if consumer.key == oauth_consumer.key: |
| return token |
| return None |
| |
| def lookup_nonce(self, oauth_consumer, oauth_token, nonce): |
| is_used = self.nonce == nonce |
| self.nonce = nonce |
| return is_used |
| |
| def fetch_request_token(self, oauth_consumer): |
| token = oauth.OAuthToken("5b2cafbf20b11bace53b29e37d8a673dRT", |
| "3f71254637df2002d8819458ae4f6c51RT") |
| self.request_token_db[token.key] = (token, oauth_consumer, None) |
| return token |
| |
| def fetch_access_token(self, oauth_consumer, oauth_token): |
| data = self.request_token_db.get(oauth_token.key) |
| if data: |
| del self.request_token_db[oauth_token.key] |
| request_token, consumer, authenticated_user = data |
| access_token = oauth.OAuthToken("5b2cafbf20b11bace53b29e37d8a673dAT", |
| "3f71254637df2002d8819458ae4f6c51AT") |
| self.access_token_db[access_token.key] = (access_token, |
| consumer, |
| authenticated_user) |
| return access_token |
| else: |
| return None |
| |
| def authorize_request_token(self, oauth_token, user): |
| data = self.request_token_db.get(oauth_token.key) |
| if data and data[2] == None: |
| request_token, consumer, authenticated_user = data |
| authenticated_user = user |
| self.request_token_db[request_token.key] = (request_token, |
| consumer, |
| authenticated_user) |
| return request_token |
| else: |
| return None |
| |
| #Mock user |
| class MockUser(object): |
| def __init__(self, name): |
| self.name = name |
| |
| def suite(): |
| return test_config.build_suite([OauthUtilsTest, |
| OAuthConsumerTest, |
| OAuthTokenTest, |
| OAuthRequestTest, |
| OAuthServerTest, |
| OAuthClientTest]) |
| |
| if __name__ == '__main__': |
| unittest.main() |
