tests/test_oauth2client_file.py

#!/usr/bin/python2.4
#
# Copyright 2010 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.file tests

Unit tests for oauth2client.file
"""

__author__ = 'jcgregorio@google.com (Joe Gregorio)'

import copy
import datetime
import httplib2
import os
import pickle
import stat
import tempfile
import unittest

from apiclient.http import HttpMockSequence
from oauth2client import multistore_file
from oauth2client.anyjson import simplejson
from oauth2client.client import AccessTokenCredentials
from oauth2client.client import AssertionCredentials
from oauth2client.client import OAuth2Credentials
from oauth2client.file import Storage


FILENAME = tempfile.mktemp('oauth2client_test.data')


class OAuth2ClientFileTests(unittest.TestCase):

  def tearDown(self):
    try:
      os.unlink(FILENAME)
    except OSError:
      pass

  def setUp(self):
    try:
      os.unlink(FILENAME)
    except OSError:
      pass

  def test_non_existent_file_storage(self):
    s = Storage(FILENAME)
    credentials = s.get()
    self.assertEquals(None, credentials)

  def test_pickle_and_json_interop(self):
    # Write a file with a pickled OAuth2Credentials.
    access_token = 'foo'
    client_id = 'some_client_id'
    client_secret = 'cOuDdkfjxxnv+'
    refresh_token = '1/0/a.df219fjls0'
    token_expiry = datetime.datetime.utcnow()
    token_uri = 'https://www.google.com/accounts/o8/oauth2/token'
    user_agent = 'refresh_checker/1.0'

    credentials = OAuth2Credentials(
        access_token, client_id, client_secret,
        refresh_token, token_expiry, token_uri,
        user_agent)

    f = open(FILENAME, 'w')
    pickle.dump(credentials, f)
    f.close()

    # Storage should be not be able to read that object, as the capability to
    # read and write credentials as pickled objects has been removed.
    s = Storage(FILENAME)
    read_credentials = s.get()
    self.assertEquals(None, read_credentials)

    # Now write it back out and confirm it has been rewritten as JSON
    s.put(credentials)
    f = file(FILENAME)
    data = simplejson.load(f)
    f.close()

    self.assertEquals(data['access_token'], 'foo')
    self.assertEquals(data['_class'], 'OAuth2Credentials')
    self.assertEquals(data['_module'], OAuth2Credentials.__module__)

  def test_token_refresh(self):
    access_token = 'foo'
    client_id = 'some_client_id'
    client_secret = 'cOuDdkfjxxnv+'
    refresh_token = '1/0/a.df219fjls0'
    token_expiry = datetime.datetime.utcnow()
    token_uri = 'https://www.google.com/accounts/o8/oauth2/token'
    user_agent = 'refresh_checker/1.0'

    credentials = OAuth2Credentials(
        access_token, client_id, client_secret,
        refresh_token, token_expiry, token_uri,
        user_agent)

    s = Storage(FILENAME)
    s.put(credentials)
    credentials = s.get()
    new_cred = copy.copy(credentials)
    new_cred.access_token = 'bar'
    s.put(new_cred)

    credentials._refresh(lambda x: x)
    self.assertEquals(credentials.access_token, 'bar')

  def test_credentials_delete(self):
    access_token = 'foo'
    client_id = 'some_client_id'
    client_secret = 'cOuDdkfjxxnv+'
    refresh_token = '1/0/a.df219fjls0'
    token_expiry = datetime.datetime.utcnow()
    token_uri = 'https://www.google.com/accounts/o8/oauth2/token'
    user_agent = 'refresh_checker/1.0'

    credentials = OAuth2Credentials(
        access_token, client_id, client_secret,
        refresh_token, token_expiry, token_uri,
        user_agent)

    s = Storage(FILENAME)
    s.put(credentials)
    credentials = s.get()
    self.assertNotEquals(None, credentials)
    s.delete()
    credentials = s.get()
    self.assertEquals(None, credentials)

  def test_access_token_credentials(self):
    access_token = 'foo'
    user_agent = 'refresh_checker/1.0'

    credentials = AccessTokenCredentials(access_token, user_agent)

    s = Storage(FILENAME)
    credentials = s.put(credentials)
    credentials = s.get()

    self.assertNotEquals(None, credentials)
    self.assertEquals('foo', credentials.access_token)
    mode = os.stat(FILENAME).st_mode

    if os.name == 'posix':
      self.assertEquals('0600', oct(stat.S_IMODE(os.stat(FILENAME).st_mode)))

  def test_read_only_file_fail_lock(self):
    access_token = 'foo'
    client_secret = 'cOuDdkfjxxnv+'
    refresh_token = '1/0/a.df219fjls0'
    token_expiry = datetime.datetime.utcnow()
    token_uri = 'https://www.google.com/accounts/o8/oauth2/token'
    user_agent = 'refresh_checker/1.0'
    client_id = 'some_client_id'

    credentials = OAuth2Credentials(
        access_token, client_id, client_secret,
        refresh_token, token_expiry, token_uri,
        user_agent)

    open(FILENAME, 'a+b').close()
    os.chmod(FILENAME, 0400)

    store = multistore_file.get_credential_storage(
        FILENAME,
        credentials.client_id,
        credentials.user_agent,
        ['some-scope', 'some-other-scope'])

    store.put(credentials)
    if os.name == 'posix':
      self.assertTrue(store._multistore._read_only)
    os.chmod(FILENAME, 0600)


  def test_multistore_non_existent_file(self):
    store = multistore_file.get_credential_storage(
        FILENAME,
        'some_client_id',
        'user-agent/1.0',
        ['some-scope', 'some-other-scope'])

    credentials = store.get()
    self.assertEquals(None, credentials)

  def test_multistore_file(self):
    access_token = 'foo'
    client_secret = 'cOuDdkfjxxnv+'
    refresh_token = '1/0/a.df219fjls0'
    token_expiry = datetime.datetime.utcnow()
    token_uri = 'https://www.google.com/accounts/o8/oauth2/token'
    user_agent = 'refresh_checker/1.0'
    client_id = 'some_client_id'

    credentials = OAuth2Credentials(
        access_token, client_id, client_secret,
        refresh_token, token_expiry, token_uri,
        user_agent)

    store = multistore_file.get_credential_storage(
        FILENAME,
        credentials.client_id,
        credentials.user_agent,
        ['some-scope', 'some-other-scope'])

    store.put(credentials)
    credentials = store.get()

    self.assertNotEquals(None, credentials)
    self.assertEquals('foo', credentials.access_token)

    store.delete()
    credentials = store.get()

    self.assertEquals(None, credentials)

    if os.name == 'posix':
      self.assertEquals('0600', oct(stat.S_IMODE(os.stat(FILENAME).st_mode)))

if __name__ == '__main__':
  unittest.main()