| """Tests for oauth2l.""" |
| |
| import json |
| import os |
| import sys |
| |
| import mock |
| import oauth2client.client |
| import six |
| from six.moves import http_client |
| import unittest2 |
| |
| import apitools.base.py as apitools_base |
| |
| _OAUTH2L_MAIN_RUN = False |
| |
| if six.PY2: |
| import gflags as flags |
| from google.apputils import appcommands |
| from apitools.scripts import oauth2l |
| FLAGS = flags.FLAGS |
| |
| |
| class _FakeResponse(object): |
| |
| def __init__(self, status_code, scopes=None): |
| self.status_code = status_code |
| if self.status_code == http_client.OK: |
| self.content = json.dumps({'scope': ' '.join(scopes or [])}) |
| else: |
| self.content = 'Error' |
| self.info = str(http_client.responses[self.status_code]) |
| self.request_url = 'some-url' |
| |
| |
| def _GetCommandOutput(t, command_name, command_argv): |
| global _OAUTH2L_MAIN_RUN # pylint: disable=global-statement |
| if not _OAUTH2L_MAIN_RUN: |
| oauth2l.main(None) |
| _OAUTH2L_MAIN_RUN = True |
| command = appcommands.GetCommandByName(command_name) |
| if command is None: |
| t.fail('Unknown command: %s' % command_name) |
| orig_stdout = sys.stdout |
| new_stdout = six.StringIO() |
| try: |
| sys.stdout = new_stdout |
| command.CommandRun([command_name] + command_argv) |
| finally: |
| sys.stdout = orig_stdout |
| FLAGS.Reset() |
| new_stdout.seek(0) |
| return new_stdout.getvalue().rstrip() |
| |
| |
| @unittest2.skipIf(six.PY3, 'oauth2l unsupported in python3') |
| class TestTest(unittest2.TestCase): |
| |
| def testOutput(self): |
| self.assertRaises(AssertionError, |
| _GetCommandOutput, self, 'foo', []) |
| |
| |
| @unittest2.skipIf(six.PY3, 'oauth2l unsupported in python3') |
| class Oauth2lFormattingTest(unittest2.TestCase): |
| |
| def setUp(self): |
| # Set up an access token to use |
| self.access_token = 'ya29.abdefghijklmnopqrstuvwxyz' |
| self.user_agent = 'oauth2l/1.0' |
| self.credentials = oauth2client.client.AccessTokenCredentials( |
| self.access_token, self.user_agent) |
| |
| def _Args(self, credentials_format): |
| return ['--credentials_format=' + credentials_format, 'userinfo.email'] |
| |
| def testFormatBare(self): |
| with mock.patch.object(oauth2l, 'FetchCredentials', |
| return_value=self.credentials, |
| autospec=True) as mock_credentials: |
| output = _GetCommandOutput(self, 'fetch', self._Args('bare')) |
| self.assertEqual(self.access_token, output) |
| self.assertEqual(1, mock_credentials.call_count) |
| |
| def testFormatHeader(self): |
| with mock.patch.object(oauth2l, 'FetchCredentials', |
| return_value=self.credentials, |
| autospec=True) as mock_credentials: |
| output = _GetCommandOutput(self, 'fetch', self._Args('header')) |
| header = 'Authorization: Bearer %s' % self.access_token |
| self.assertEqual(header, output) |
| self.assertEqual(1, mock_credentials.call_count) |
| |
| def testHeaderCommand(self): |
| with mock.patch.object(oauth2l, 'FetchCredentials', |
| return_value=self.credentials, |
| autospec=True) as mock_credentials: |
| output = _GetCommandOutput(self, 'header', ['userinfo.email']) |
| header = 'Authorization: Bearer %s' % self.access_token |
| self.assertEqual(header, output) |
| self.assertEqual(1, mock_credentials.call_count) |
| |
| def testFormatJson(self): |
| with mock.patch.object(oauth2l, 'FetchCredentials', |
| return_value=self.credentials, |
| autospec=True) as mock_credentials: |
| output = _GetCommandOutput(self, 'fetch', self._Args('json')) |
| output_lines = [l.strip() for l in output.splitlines()] |
| expected_lines = [ |
| '"_class": "AccessTokenCredentials",', |
| '"access_token": "%s",' % self.access_token, |
| ] |
| for line in expected_lines: |
| self.assertIn(line, output_lines) |
| self.assertEqual(1, mock_credentials.call_count) |
| |
| def testFormatJsonCompact(self): |
| with mock.patch.object(oauth2l, 'FetchCredentials', |
| return_value=self.credentials, |
| autospec=True) as mock_credentials: |
| output = _GetCommandOutput(self, 'fetch', |
| self._Args('json_compact')) |
| expected_clauses = [ |
| '"_class":"AccessTokenCredentials",', |
| '"access_token":"%s",' % self.access_token, |
| ] |
| for clause in expected_clauses: |
| self.assertIn(clause, output) |
| self.assertEqual(1, len(output.splitlines())) |
| self.assertEqual(1, mock_credentials.call_count) |
| |
| def testFormatPretty(self): |
| with mock.patch.object(oauth2l, 'FetchCredentials', |
| return_value=self.credentials, |
| autospec=True) as mock_credentials: |
| output = _GetCommandOutput(self, 'fetch', self._Args('pretty')) |
| expecteds = ['oauth2client.client.AccessTokenCredentials', |
| self.access_token] |
| for expected in expecteds: |
| self.assertIn(expected, output) |
| self.assertEqual(1, mock_credentials.call_count) |
| |
| def testFakeFormat(self): |
| self.assertRaises(ValueError, |
| oauth2l._Format, 'xml', self.credentials) |
| |
| |
| @unittest2.skipIf(six.PY3, 'oauth2l unsupported in python3') |
| class TestFetch(unittest2.TestCase): |
| |
| def setUp(self): |
| # Set up an access token to use |
| self.access_token = 'ya29.abdefghijklmnopqrstuvwxyz' |
| self.user_agent = 'oauth2l/1.0' |
| self.credentials = oauth2client.client.AccessTokenCredentials( |
| self.access_token, self.user_agent) |
| |
| def testNoScopes(self): |
| output = _GetCommandOutput(self, 'fetch', []) |
| self.assertEqual( |
| 'Exception raised in fetch operation: No scopes provided', |
| output) |
| |
| def testScopes(self): |
| expected_scopes = [ |
| 'https://www.googleapis.com/auth/userinfo.email', |
| 'https://www.googleapis.com/auth/cloud-platform', |
| ] |
| with mock.patch.object(apitools_base, 'GetCredentials', |
| return_value=self.credentials, |
| autospec=True) as mock_fetch: |
| with mock.patch.object(oauth2l, '_GetTokenScopes', |
| return_value=expected_scopes, |
| autospec=True) as mock_get_scopes: |
| output = _GetCommandOutput( |
| self, 'fetch', ['userinfo.email', 'cloud-platform']) |
| self.assertIn(self.access_token, output) |
| self.assertEqual(1, mock_fetch.call_count) |
| args, _ = mock_fetch.call_args |
| self.assertEqual(expected_scopes, args[-1]) |
| self.assertEqual(1, mock_get_scopes.call_count) |
| self.assertEqual((self.access_token,), |
| mock_get_scopes.call_args[0]) |
| |
| def testCredentialsRefreshed(self): |
| with mock.patch.object(apitools_base, 'GetCredentials', |
| return_value=self.credentials, |
| autospec=True) as mock_fetch: |
| with mock.patch.object(oauth2l, '_ValidateToken', |
| return_value=False, |
| autospec=True) as mock_validate: |
| with mock.patch.object(self.credentials, 'refresh', |
| return_value=None, |
| autospec=True) as mock_refresh: |
| output = _GetCommandOutput(self, 'fetch', |
| ['userinfo.email']) |
| self.assertIn(self.access_token, output) |
| self.assertEqual(1, mock_fetch.call_count) |
| self.assertEqual(1, mock_validate.call_count) |
| self.assertEqual(1, mock_refresh.call_count) |
| |
| def testDefaultClientInfo(self): |
| with mock.patch.object(apitools_base, 'GetCredentials', |
| return_value=self.credentials, |
| autospec=True) as mock_fetch: |
| with mock.patch.object(oauth2l, '_ValidateToken', |
| return_value=True, |
| autospec=True) as mock_validate: |
| output = _GetCommandOutput(self, 'fetch', ['userinfo.email']) |
| self.assertIn(self.access_token, output) |
| self.assertEqual(1, mock_fetch.call_count) |
| _, kwargs = mock_fetch.call_args |
| self.assertEqual( |
| '1042881264118.apps.googleusercontent.com', |
| kwargs['client_id']) |
| self.assertEqual(1, mock_validate.call_count) |
| |
| def testMissingClientSecrets(self): |
| try: |
| FLAGS.client_secrets = '/non/existent/file' |
| self.assertRaises( |
| ValueError, |
| oauth2l.GetClientInfoFromFlags) |
| finally: |
| FLAGS.Reset() |
| |
| def testWrongClientSecretsFormat(self): |
| client_secrets_path = os.path.join( |
| os.path.dirname(__file__), |
| 'testdata/noninstalled_client_secrets.json') |
| try: |
| FLAGS.client_secrets = client_secrets_path |
| self.assertRaises( |
| ValueError, |
| oauth2l.GetClientInfoFromFlags) |
| finally: |
| FLAGS.Reset() |
| |
| def testCustomClientInfo(self): |
| client_secrets_path = os.path.join( |
| os.path.dirname(__file__), 'testdata/fake_client_secrets.json') |
| with mock.patch.object(apitools_base, 'GetCredentials', |
| return_value=self.credentials, |
| autospec=True) as mock_fetch: |
| with mock.patch.object(oauth2l, '_ValidateToken', |
| return_value=True, |
| autospec=True) as mock_validate: |
| fetch_args = [ |
| '--client_secrets=' + client_secrets_path, |
| 'userinfo.email'] |
| output = _GetCommandOutput(self, 'fetch', fetch_args) |
| self.assertIn(self.access_token, output) |
| self.assertEqual(1, mock_fetch.call_count) |
| _, kwargs = mock_fetch.call_args |
| self.assertEqual('144169.apps.googleusercontent.com', |
| kwargs['client_id']) |
| self.assertEqual('awesomesecret', |
| kwargs['client_secret']) |
| self.assertEqual(1, mock_validate.call_count) |
| |
| |
| @unittest2.skipIf(six.PY3, 'oauth2l unsupported in python3') |
| class TestOtherCommands(unittest2.TestCase): |
| |
| def setUp(self): |
| # Set up an access token to use |
| self.access_token = 'ya29.abdefghijklmnopqrstuvwxyz' |
| self.user_agent = 'oauth2l/1.0' |
| self.credentials = oauth2client.client.AccessTokenCredentials( |
| self.access_token, self.user_agent) |
| |
| def testEmail(self): |
| user_info = {'email': 'foo@example.com'} |
| with mock.patch.object(apitools_base, 'GetUserinfo', |
| return_value=user_info, |
| autospec=True) as mock_get_userinfo: |
| output = _GetCommandOutput(self, 'email', [self.access_token]) |
| self.assertEqual(user_info['email'], output) |
| self.assertEqual(1, mock_get_userinfo.call_count) |
| self.assertEqual(self.access_token, |
| mock_get_userinfo.call_args[0][0].access_token) |
| |
| def testNoEmail(self): |
| with mock.patch.object(apitools_base, 'GetUserinfo', |
| return_value={}, |
| autospec=True) as mock_get_userinfo: |
| output = _GetCommandOutput(self, 'email', [self.access_token]) |
| self.assertEqual('', output) |
| self.assertEqual(1, mock_get_userinfo.call_count) |
| |
| def testUserinfo(self): |
| user_info = {'email': 'foo@example.com'} |
| with mock.patch.object(apitools_base, 'GetUserinfo', |
| return_value=user_info, |
| autospec=True) as mock_get_userinfo: |
| output = _GetCommandOutput(self, 'userinfo', [self.access_token]) |
| self.assertEqual(json.dumps(user_info, indent=4), output) |
| self.assertEqual(1, mock_get_userinfo.call_count) |
| self.assertEqual(self.access_token, |
| mock_get_userinfo.call_args[0][0].access_token) |
| |
| def testUserinfoCompact(self): |
| user_info = {'email': 'foo@example.com'} |
| with mock.patch.object(apitools_base, 'GetUserinfo', |
| return_value=user_info, |
| autospec=True) as mock_get_userinfo: |
| output = _GetCommandOutput( |
| self, 'userinfo', ['--format=json_compact', self.access_token]) |
| self.assertEqual(json.dumps(user_info, separators=(',', ':')), |
| output) |
| self.assertEqual(1, mock_get_userinfo.call_count) |
| self.assertEqual(self.access_token, |
| mock_get_userinfo.call_args[0][0].access_token) |
| |
| def testScopes(self): |
| scopes = [u'https://www.googleapis.com/auth/userinfo.email', |
| u'https://www.googleapis.com/auth/cloud-platform'] |
| response = _FakeResponse(http_client.OK, scopes=scopes) |
| with mock.patch.object(apitools_base, 'MakeRequest', |
| return_value=response, |
| autospec=True) as mock_make_request: |
| output = _GetCommandOutput(self, 'scopes', [self.access_token]) |
| self.assertEqual(sorted(scopes), output.splitlines()) |
| self.assertEqual(1, mock_make_request.call_count) |
| |
| def testValidate(self): |
| scopes = [u'https://www.googleapis.com/auth/userinfo.email', |
| u'https://www.googleapis.com/auth/cloud-platform'] |
| response = _FakeResponse(http_client.OK, scopes=scopes) |
| with mock.patch.object(apitools_base, 'MakeRequest', |
| return_value=response, |
| autospec=True) as mock_make_request: |
| output = _GetCommandOutput(self, 'validate', [self.access_token]) |
| self.assertEqual('', output) |
| self.assertEqual(1, mock_make_request.call_count) |
| |
| def testBadResponseCode(self): |
| response = _FakeResponse(http_client.BAD_REQUEST) |
| with mock.patch.object(apitools_base, 'MakeRequest', |
| return_value=response, |
| autospec=True) as mock_make_request: |
| output = _GetCommandOutput(self, 'scopes', [self.access_token]) |
| self.assertEqual('', output) |
| self.assertEqual(1, mock_make_request.call_count) |
| |
| def testUnexpectedResponseCode(self): |
| response = _FakeResponse(http_client.INTERNAL_SERVER_ERROR) |
| with mock.patch.object(apitools_base, 'MakeRequest', |
| return_value=response, |
| autospec=True) as mock_make_request: |
| output = _GetCommandOutput(self, 'scopes', [self.access_token]) |
| self.assertIn(str(http_client.responses[response.status_code]), |
| output) |
| self.assertIn('Exception raised in scopes operation: HttpError', |
| output) |
| self.assertEqual(1, mock_make_request.call_count) |
