python类GoogleCredentials()的实例源码

def test_get_adc_from_env_var_malformed_file(self, *stubs):
        # Set up stubs.
        get_well_known, get_env_file, in_gae, in_gce = stubs
        get_env_file.return_value = datafile(
            os.path.join('gcloud',
                         'application_default_credentials_malformed_3.json'))

        expected_err = client.ApplicationDefaultCredentialsError
        with self.assertRaises(expected_err) as exc_manager:
            client.GoogleCredentials.get_application_default()

        self.assertTrue(str(exc_manager.exception).startswith(
            'An error was encountered while reading json file: ' +
            get_env_file.return_value + ' (pointed to by ' +
            client.GOOGLE_APPLICATION_CREDENTIALS + ' environment variable):'))

        get_env_file.assert_called_once_with()
        get_well_known.assert_not_called()
        in_gae.assert_not_called()
        in_gce.assert_not_called()

def test_get_adc_env_from_well_known(self, *stubs):
        # Unpack stubs.
        get_well_known, get_env_file, in_gae, in_gce = stubs
        # Make sure the well-known file is an actual file.
        get_well_known.return_value = __file__
        # Make sure the well-known file actually doesn't exist.
        self.assertTrue(os.path.exists(get_well_known.return_value))

        method_name = (
            'oauth2client.client.'
            '_get_application_default_credential_from_file')
        result_creds = object()
        with mock.patch(method_name,
                        return_value=result_creds) as get_from_file:
            result = client.GoogleCredentials.get_application_default()
            self.assertEqual(result, result_creds)
            get_from_file.assert_called_once_with(__file__)

        get_env_file.assert_called_once_with()
        get_well_known.assert_called_once_with()
        in_gae.assert_not_called()
        in_gce.assert_not_called()

def test_get_adc_from_env_var_malformed_file(self, *stubs):
        # Set up stubs.
        get_well_known, get_env_file, in_gae, in_gce = stubs
        get_env_file.return_value = datafile(
            os.path.join('gcloud',
                         'application_default_credentials_malformed_3.json'))

        expected_err = client.ApplicationDefaultCredentialsError
        with self.assertRaises(expected_err) as exc_manager:
            client.GoogleCredentials.get_application_default()

        self.assertTrue(str(exc_manager.exception).startswith(
            'An error was encountered while reading json file: ' +
            get_env_file.return_value + ' (pointed to by ' +
            client.GOOGLE_APPLICATION_CREDENTIALS + ' environment variable):'))

        get_env_file.assert_called_once_with()
        get_well_known.assert_not_called()
        in_gae.assert_not_called()
        in_gce.assert_not_called()

def test_get_adc_env_from_well_known(self, *stubs):
        # Unpack stubs.
        get_well_known, get_env_file, in_gae, in_gce = stubs
        # Make sure the well-known file is an actual file.
        get_well_known.return_value = __file__
        # Make sure the well-known file actually doesn't exist.
        self.assertTrue(os.path.exists(get_well_known.return_value))

        method_name = (
            'oauth2client.client.'
            '_get_application_default_credential_from_file')
        result_creds = object()
        with mock.patch(method_name,
                        return_value=result_creds) as get_from_file:
            result = client.GoogleCredentials.get_application_default()
            self.assertEqual(result, result_creds)
            get_from_file.assert_called_once_with(__file__)

        get_env_file.assert_called_once_with()
        get_well_known.assert_called_once_with()
        in_gae.assert_not_called()
        in_gce.assert_not_called()

def _set_ua_and_scopes(credentials):
    """Set custom Forseti user agent and add cloud scopes on credential object.

    Args:
        credentials (client.OAuth2Credentials): The credentials object used to
            authenticate all http requests.

    Returns:
        client.OAuth2Credentials: The credentials object with the user agent
            attribute set or updated.
    """
    if isinstance(credentials, client.OAuth2Credentials):
        user_agent = credentials.user_agent
        if (not user_agent or
                forseti_security.__package_name__ not in user_agent):

            credentials.user_agent = (
                'Python-httplib2/{} (gzip), {}/{}'.format(
                    httplib2.__version__,
                    forseti_security.__package_name__,
                    forseti_security.__version__))
        if (isinstance(credentials, client.GoogleCredentials) and
                credentials.create_scoped_required()):
            credentials = credentials.create_scoped(list(CLOUD_SCOPES))
    return credentials

def run_user_json():
    with open(USER_KEY_PATH, 'r') as file_object:
        client_credentials = json.load(file_object)

    credentials = client.GoogleCredentials(
        access_token=None,
        client_id=client_credentials['client_id'],
        client_secret=client_credentials['client_secret'],
        refresh_token=client_credentials['refresh_token'],
        token_expiry=None,
        token_uri=oauth2client.GOOGLE_TOKEN_URI,
        user_agent='Python client library',
    )

    _check_user_info(credentials, USER_KEY_EMAIL)

def validate_google_credentials(self, credentials):
        self.assertIsInstance(credentials, client.GoogleCredentials)
        self.assertEqual(None, credentials.access_token)
        self.assertEqual('123', credentials.client_id)
        self.assertEqual('secret', credentials.client_secret)
        self.assertEqual('alabalaportocala', credentials.refresh_token)
        self.assertEqual(None, credentials.token_expiry)
        self.assertEqual(oauth2client.GOOGLE_TOKEN_URI, credentials.token_uri)
        self.assertEqual('Python client library', credentials.user_agent)

def get_a_google_credentials_object(self):
        return client.GoogleCredentials(None, None, None, None,
                                        None, None, None, None)

def test_get_application_default_in_gae(self, gae_adc, in_gae,
                                            from_gce, from_files):
        credentials = client.GoogleCredentials.get_application_default()
        self.assertEqual(credentials, gae_adc.return_value)
        from_files.assert_called_once_with()
        in_gae.assert_called_once_with()
        from_gce.assert_not_called()

def test_get_application_default_in_gce(self, gce_adc, in_gce,
                                            from_files, from_gae):
        credentials = client.GoogleCredentials.get_application_default()
        self.assertEqual(credentials, gce_adc.return_value)
        in_gce.assert_called_once_with()
        from_gae.assert_called_once_with()
        from_files.assert_called_once_with()

def test_get_adc_from_env_var_authorized_user(self, *stubs):
        # Set up stubs.
        get_well_known, get_env_file, in_gae, in_gce = stubs
        get_env_file.return_value = datafile(os.path.join(
            'gcloud',
            'application_default_credentials_authorized_user.json'))

        credentials = client.GoogleCredentials.get_application_default()
        self.validate_google_credentials(credentials)

        get_env_file.assert_called_once_with()
        get_well_known.assert_not_called()
        in_gae.assert_not_called()
        in_gce.assert_not_called()

def test_get_adc_env_not_set_up(self, *stubs):
        # Unpack stubs.
        get_well_known, get_env_file, in_gae, in_gce = stubs
        # Make sure the well-known file actually doesn't exist.
        self.assertFalse(os.path.exists(get_well_known.return_value))

        expected_err = client.ApplicationDefaultCredentialsError
        with self.assertRaises(expected_err) as exc_manager:
            client.GoogleCredentials.get_application_default()

        self.assertEqual(client.ADC_HELP_MSG, str(exc_manager.exception))
        get_env_file.assert_called_once_with()
        get_well_known.assert_called_once_with()
        in_gae.assert_called_once_with()
        in_gce.assert_called_once_with()

def test_to_from_json_authorized_user(self):
        filename = 'application_default_credentials_authorized_user.json'
        credentials_file = datafile(os.path.join('gcloud', filename))
        creds = client.GoogleCredentials.from_stream(credentials_file)
        json = creds.to_json()
        creds2 = client.GoogleCredentials.from_json(json)

        self.assertEqual(creds.__dict__, creds2.__dict__)

def test_to_from_json_service_account_scoped(self):
        credentials_file = datafile(
            os.path.join('gcloud', client._WELL_KNOWN_CREDENTIALS_FILE))
        creds1 = client.GoogleCredentials.from_stream(credentials_file)
        creds1 = creds1.create_scoped(['dummy_scope'])
        # Convert to and then back from json.
        creds2 = client.GoogleCredentials.from_json(creds1.to_json())

        creds1_vals = creds1.__dict__
        creds1_vals.pop('_signer')
        creds2_vals = creds2.__dict__
        creds2_vals.pop('_signer')
        self.assertEqual(creds1_vals, creds2_vals)

def validate_google_credentials(self, credentials):
        self.assertIsInstance(credentials, client.GoogleCredentials)
        self.assertEqual(None, credentials.access_token)
        self.assertEqual('123', credentials.client_id)
        self.assertEqual('secret', credentials.client_secret)
        self.assertEqual('alabalaportocala', credentials.refresh_token)
        self.assertEqual(None, credentials.token_expiry)
        self.assertEqual(oauth2client.GOOGLE_TOKEN_URI, credentials.token_uri)
        self.assertEqual('Python client library', credentials.user_agent)

def get_a_google_credentials_object(self):
        return client.GoogleCredentials(None, None, None, None,
                                        None, None, None, None)

def test_get_application_default_in_gae(self, gae_adc, in_gae,
                                            from_gce, from_files):
        credentials = client.GoogleCredentials.get_application_default()
        self.assertEqual(credentials, gae_adc.return_value)
        from_files.assert_called_once_with()
        in_gae.assert_called_once_with()
        from_gce.assert_not_called()

def test_get_application_default_in_gce(self, gce_adc, in_gce,
                                            from_files, from_gae):
        credentials = client.GoogleCredentials.get_application_default()
        self.assertEqual(credentials, gce_adc.return_value)
        in_gce.assert_called_once_with()
        from_gae.assert_called_once_with()
        from_files.assert_called_once_with()

def test_get_adc_from_env_var_service_account(self, *stubs):
        # Set up stubs.
        get_well_known, get_env_file, in_gae, in_gce = stubs
        get_env_file.return_value = datafile(
            os.path.join('gcloud', client._WELL_KNOWN_CREDENTIALS_FILE))

        credentials = client.GoogleCredentials.get_application_default()
        self.validate_service_account_credentials(credentials)

        get_env_file.assert_called_once_with()
        get_well_known.assert_not_called()
        in_gae.assert_not_called()
        in_gce.assert_not_called()

def test_get_adc_env_not_set_up(self, *stubs):
        # Unpack stubs.
        get_well_known, get_env_file, in_gae, in_gce = stubs
        # Make sure the well-known file actually doesn't exist.
        self.assertFalse(os.path.exists(get_well_known.return_value))

        expected_err = client.ApplicationDefaultCredentialsError
        with self.assertRaises(expected_err) as exc_manager:
            client.GoogleCredentials.get_application_default()

        self.assertEqual(client.ADC_HELP_MSG, str(exc_manager.exception))
        get_env_file.assert_called_once_with()
        get_well_known.assert_called_once_with()
        in_gae.assert_called_once_with()
        in_gce.assert_called_once_with()

def test_to_from_json_authorized_user(self):
        filename = 'application_default_credentials_authorized_user.json'
        credentials_file = datafile(os.path.join('gcloud', filename))
        creds = client.GoogleCredentials.from_stream(credentials_file)
        json = creds.to_json()
        creds2 = client.GoogleCredentials.from_json(json)

        self.assertEqual(creds.__dict__, creds2.__dict__)

def test_to_from_json_service_account(self):
        credentials_file = datafile(
            os.path.join('gcloud', client._WELL_KNOWN_CREDENTIALS_FILE))
        creds1 = client.GoogleCredentials.from_stream(credentials_file)
        # Convert to and then back from json.
        creds2 = client.GoogleCredentials.from_json(creds1.to_json())

        creds1_vals = creds1.__dict__
        creds1_vals.pop('_signer')
        creds2_vals = creds2.__dict__
        creds2_vals.pop('_signer')
        self.assertEqual(creds1_vals, creds2_vals)