text: #!/usr/bin/python2.4
  #
  # Copyright 2014 Google Inc.
  #
  # Licensed under the Apache License, Version 2.0 (the "License");
  # you may not use this file except in compliance with the License.
  # You may obtain a copy of the License at
  #
  #      http://www.apache.org/licenses/LICENSE-2.0
  #
  # Unless required by applicable law or agreed to in writing, software
  # distributed under the License is distributed on an "AS IS" BASIS,
  # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  # See the License for the specific language governing permissions and
  # limitations under the License.
  
  
  """Oauth2client tests.
  
  Unit tests for service account credentials implemented using RSA.
  """
  
  import os
  import rsa
  import time
  import unittest
  
  from http_mock import HttpMockSequence
  from oauth2client.anyjson import simplejson
  from oauth2client.service_account import _ServiceAccountCredentials
  
  
  def datafile(filename):
    # TODO(orestica): Refactor this using pkgutil.get_data
    f = open(os.path.join(os.path.dirname(__file__), 'data', filename), 'r')
    data = f.read()
    f.close()
    return data
  
  
  class ServiceAccountCredentialsTests(unittest.TestCase):
    def setUp(self):
      self.service_account_id = '123'
      self.service_account_email = 'dummy@google.com'
      self.private_key_id = 'ABCDEF'
      self.private_key = datafile('pem_from_pkcs12.pem')
      self.scopes = ['dummy_scope']
      self.credentials = _ServiceAccountCredentials(self.service_account_id,
                                                    self.service_account_email,
                                                    self.private_key_id,
                                                    self.private_key,
                                                    [])
  
    def test_sign_blob(self):
      private_key_id, signature = self.credentials.sign_blob('Google')
      self.assertEqual( self.private_key_id, private_key_id)
  
      pub_key = rsa.PublicKey.load_pkcs1_openssl_pem(
          datafile('publickey_openssl.pem'))
  
      self.assertTrue(rsa.pkcs1.verify('Google', signature, pub_key))
  
      try:
        rsa.pkcs1.verify('Orest', signature, pub_key)
        self.fail('Verification should have failed!')
      except rsa.pkcs1.VerificationError:
        pass  # Expected
  
      try:
        rsa.pkcs1.verify('Google', 'bad signature', pub_key)
        self.fail('Verification should have failed!')
      except rsa.pkcs1.VerificationError:
        pass  # Expected
  
    def test_service_account_email(self):
      self.assertEqual(self.service_account_email,
                       self.credentials.service_account_email)
  
    def test_create_scoped_required_without_scopes(self):
      self.assertTrue(self.credentials.create_scoped_required())
  
    def test_create_scoped_required_with_scopes(self):
      self.credentials = _ServiceAccountCredentials(self.service_account_id,
                                                    self.service_account_email,
                                                    self.private_key_id,
                                                    self.private_key,
                                                    self.scopes)
      self.assertFalse(self.credentials.create_scoped_required())
  
    def test_create_scoped(self):
      new_credentials = self.credentials.create_scoped(self.scopes)
      self.assertNotEqual(self.credentials, new_credentials)
      self.assertTrue(isinstance(new_credentials, _ServiceAccountCredentials))
      self.assertEqual('dummy_scope', new_credentials._scopes)
  
    def test_access_token(self):
      S = 2  # number of seconds in which the token expires
      token_response_first = {'access_token': 'first_token', 'expires_in': S}
      token_response_second = {'access_token': 'second_token', 'expires_in': S}
      http = HttpMockSequence([
          ({'status': '200'}, simplejson.dumps(token_response_first)),
          ({'status': '200'}, simplejson.dumps(token_response_second)),
      ])
  
      token = self.credentials.get_access_token(http=http)
      self.assertEqual('first_token', token.access_token)
      self.assertEqual(S - 1, token.expires_in)  
      self.assertFalse(self.credentials.access_token_expired)
      self.assertEqual(token_response_first, self.credentials.token_response)
  
      token = self.credentials.get_access_token(http=http)
      self.assertEqual('first_token', token.access_token)
      self.assertEqual(S - 1, token.expires_in)
      self.assertFalse(self.credentials.access_token_expired)
      self.assertEqual(token_response_first, self.credentials.token_response)
  
      time.sleep(S)
      self.assertTrue(self.credentials.access_token_expired)
  
      token = self.credentials.get_access_token(http=http)
      self.assertEqual('second_token', token.access_token)
      self.assertEqual(S - 1, token.expires_in)
      self.assertFalse(self.credentials.access_token_expired)
      self.assertEqual(token_response_second, self.credentials.token_response)
  