def test_override_flow_via_kwargs(self):
"""Passing kwargs to override defaults."""
flow = client.OAuth2WebServerFlow(
client_id='client_id+1',
client_secret='secret+1',
scope='foo',
redirect_uri=client.OOB_CALLBACK_URN,
user_agent='unittest-sample/1.0',
access_type='online',
response_type='token'
)
authorize_url = flow.step1_get_authorize_url()
parsed = urllib.parse.urlparse(authorize_url)
q = urllib.parse.parse_qs(parsed[4])
self.assertEqual('client_id+1', q['client_id'][0])
self.assertEqual('token', q['response_type'][0])
self.assertEqual('foo', q['scope'][0])
self.assertEqual(client.OOB_CALLBACK_URN, q['redirect_uri'][0])
self.assertEqual('online', q['access_type'][0])
python类OOB_CALLBACK_URN的实例源码
def test_step1_get_authorize_url_redirect_override(self, logger):
flow = client.OAuth2WebServerFlow('client_id+1', scope='foo',
redirect_uri=client.OOB_CALLBACK_URN)
alt_redirect = 'foo:bar'
self.assertEqual(flow.redirect_uri, client.OOB_CALLBACK_URN)
result = flow.step1_get_authorize_url(redirect_uri=alt_redirect)
# Make sure the redirect value was updated.
self.assertEqual(flow.redirect_uri, alt_redirect)
query_params = {
'client_id': flow.client_id,
'redirect_uri': alt_redirect,
'scope': flow.scope,
'access_type': 'offline',
'response_type': 'code',
}
expected = _helpers.update_query_params(flow.auth_uri, query_params)
assertUrisEqual(self, expected, result)
# Check stubs.
self.assertEqual(logger.warning.call_count, 1)
def test_step1_get_authorize_url_without_login_hint(self):
login_hint = 'There are wascally wabbits nearby'
flow = client.OAuth2WebServerFlow('client_id+1', scope='foo',
redirect_uri=client.OOB_CALLBACK_URN,
login_hint=login_hint)
result = flow.step1_get_authorize_url()
query_params = {
'client_id': flow.client_id,
'login_hint': login_hint,
'redirect_uri': client.OOB_CALLBACK_URN,
'scope': flow.scope,
'access_type': 'offline',
'response_type': 'code',
}
expected = _helpers.update_query_params(flow.auth_uri, query_params)
assertUrisEqual(self, expected, result)
def test_run_flow_webserver_fallback(
self, input_mock, server_ctor_mock, logging_mock):
server_ctor_mock.side_effect = socket.error()
input_mock.return_value = 'auth_code'
# It should catch the socket error and proceed as if
# noauth_local_webserver was specified.
returned_credentials = tools.run_flow(
self.flow, self.storage, flags=self.server_flags)
self.assertEqual(self.credentials, returned_credentials)
self.assertEqual(self.flow.redirect_uri, client.OOB_CALLBACK_URN)
self.flow.step2_exchange.assert_called_once_with(
'auth_code', http=None)
self.assertTrue(server_ctor_mock.called)
self.assertFalse(self.server.handle_request.called)
def test_override_flow_via_kwargs(self):
"""Passing kwargs to override defaults."""
flow = client.OAuth2WebServerFlow(
client_id='client_id+1',
client_secret='secret+1',
scope='foo',
redirect_uri=client.OOB_CALLBACK_URN,
user_agent='unittest-sample/1.0',
access_type='online',
response_type='token'
)
authorize_url = flow.step1_get_authorize_url()
parsed = urllib.parse.urlparse(authorize_url)
q = urllib.parse.parse_qs(parsed[4])
self.assertEqual('client_id+1', q['client_id'][0])
self.assertEqual('token', q['response_type'][0])
self.assertEqual('foo', q['scope'][0])
self.assertEqual(client.OOB_CALLBACK_URN, q['redirect_uri'][0])
self.assertEqual('online', q['access_type'][0])
def test_step1_get_authorize_url_redirect_override(self, logger):
flow = client.OAuth2WebServerFlow('client_id+1', scope='foo',
redirect_uri=client.OOB_CALLBACK_URN)
alt_redirect = 'foo:bar'
self.assertEqual(flow.redirect_uri, client.OOB_CALLBACK_URN)
result = flow.step1_get_authorize_url(redirect_uri=alt_redirect)
# Make sure the redirect value was updated.
self.assertEqual(flow.redirect_uri, alt_redirect)
query_params = {
'client_id': flow.client_id,
'redirect_uri': alt_redirect,
'scope': flow.scope,
'access_type': 'offline',
'response_type': 'code',
}
expected = _helpers.update_query_params(flow.auth_uri, query_params)
assertUrisEqual(self, expected, result)
# Check stubs.
self.assertEqual(logger.warning.call_count, 1)
def test_step1_get_authorize_url_without_login_hint(self):
login_hint = 'There are wascally wabbits nearby'
flow = client.OAuth2WebServerFlow('client_id+1', scope='foo',
redirect_uri=client.OOB_CALLBACK_URN,
login_hint=login_hint)
result = flow.step1_get_authorize_url()
query_params = {
'client_id': flow.client_id,
'login_hint': login_hint,
'redirect_uri': client.OOB_CALLBACK_URN,
'scope': flow.scope,
'access_type': 'offline',
'response_type': 'code',
}
expected = _helpers.update_query_params(flow.auth_uri, query_params)
assertUrisEqual(self, expected, result)
def test_run_flow_webserver_fallback(
self, input_mock, server_ctor_mock, logging_mock):
server_ctor_mock.side_effect = socket.error()
input_mock.return_value = 'auth_code'
# It should catch the socket error and proceed as if
# noauth_local_webserver was specified.
returned_credentials = tools.run_flow(
self.flow, self.storage, flags=self.server_flags)
self.assertEqual(self.credentials, returned_credentials)
self.assertEqual(self.flow.redirect_uri, client.OOB_CALLBACK_URN)
self.flow.step2_exchange.assert_called_once_with(
'auth_code', http=None)
self.assertTrue(server_ctor_mock.called)
self.assertFalse(self.server.handle_request.called)
def add_login_begin(self):
flow = self._flow = client.flow_from_clientsecrets('googledrive.json',
'https://www.googleapis.com/auth/drive.file')
flow.user_agent = const.APP_USERAGENT
flow.redirect_uri = client.OOB_CALLBACK_URN
authorize_url = flow.step1_get_authorize_url()
return {'Login URL': authorize_url}
def setUp(self):
self.flow = client.OAuth2WebServerFlow(
client_id='client_id+1',
client_secret='secret+1',
scope='foo',
redirect_uri=client.OOB_CALLBACK_URN,
user_agent='unittest-sample/1.0',
revoke_uri='dummy_revoke_uri',
)
self.bad_verifier = b'__NOT_THE_VERIFIER_YOURE_LOOKING_FOR__'
self.good_verifier = b'__TEST_VERIFIER__'
self.good_challenger = b'__TEST_CHALLENGE__'
def test_construct_authorize_url(self):
authorize_url = self.flow.step1_get_authorize_url(state='state+1')
parsed = urllib.parse.urlparse(authorize_url)
q = urllib.parse.parse_qs(parsed[4])
self.assertEqual('client_id+1', q['client_id'][0])
self.assertEqual('code', q['response_type'][0])
self.assertEqual('foo', q['scope'][0])
self.assertEqual(client.OOB_CALLBACK_URN, q['redirect_uri'][0])
self.assertEqual('offline', q['access_type'][0])
self.assertEqual('state+1', q['state'][0])
def test_run_flow_no_webserver(self, input_mock, logging_mock):
input_mock.return_value = 'auth_code'
# Successful exchange.
returned_credentials = tools.run_flow(self.flow, self.storage)
self.assertEqual(self.credentials, returned_credentials)
self.assertEqual(self.flow.redirect_uri, client.OOB_CALLBACK_URN)
self.flow.step2_exchange.assert_called_once_with(
'auth_code', http=None)
self.storage.put.assert_called_once_with(self.credentials)
self.credentials.set_store.assert_called_once_with(self.storage)
def test_run_flow_no_webserver_explicit_flags(
self, input_mock, logging_mock):
input_mock.return_value = 'auth_code'
# Successful exchange.
returned_credentials = tools.run_flow(
self.flow, self.storage, flags=self.flags)
self.assertEqual(self.credentials, returned_credentials)
self.assertEqual(self.flow.redirect_uri, client.OOB_CALLBACK_URN)
self.flow.step2_exchange.assert_called_once_with(
'auth_code', http=None)
def setUp(self):
self.flow = client.OAuth2WebServerFlow(
client_id='client_id+1',
client_secret='secret+1',
scope='foo',
redirect_uri=client.OOB_CALLBACK_URN,
user_agent='unittest-sample/1.0',
revoke_uri='dummy_revoke_uri',
)
self.bad_verifier = b'__NOT_THE_VERIFIER_YOURE_LOOKING_FOR__'
self.good_verifier = b'__TEST_VERIFIER__'
self.good_challenger = b'__TEST_CHALLENGE__'
def test_construct_authorize_url(self):
authorize_url = self.flow.step1_get_authorize_url(state='state+1')
parsed = urllib.parse.urlparse(authorize_url)
q = urllib.parse.parse_qs(parsed[4])
self.assertEqual('client_id+1', q['client_id'][0])
self.assertEqual('code', q['response_type'][0])
self.assertEqual('foo', q['scope'][0])
self.assertEqual(client.OOB_CALLBACK_URN, q['redirect_uri'][0])
self.assertEqual('offline', q['access_type'][0])
self.assertEqual('state+1', q['state'][0])
def test_run_flow_no_webserver(self, input_mock, logging_mock):
input_mock.return_value = 'auth_code'
# Successful exchange.
returned_credentials = tools.run_flow(self.flow, self.storage)
self.assertEqual(self.credentials, returned_credentials)
self.assertEqual(self.flow.redirect_uri, client.OOB_CALLBACK_URN)
self.flow.step2_exchange.assert_called_once_with(
'auth_code', http=None)
self.storage.put.assert_called_once_with(self.credentials)
self.credentials.set_store.assert_called_once_with(self.storage)
def test_run_flow_no_webserver_explicit_flags(
self, input_mock, logging_mock):
input_mock.return_value = 'auth_code'
# Successful exchange.
returned_credentials = tools.run_flow(
self.flow, self.storage, flags=self.flags)
self.assertEqual(self.credentials, returned_credentials)
self.assertEqual(self.flow.redirect_uri, client.OOB_CALLBACK_URN)
self.flow.step2_exchange.assert_called_once_with(
'auth_code', http=None)
