From d08463f7a398e1a6c49e5d0d58a7e3430c9077ed Mon Sep 17 00:00:00 2001 From: Scott Crunkleton Date: Tue, 1 Dec 2015 23:17:34 -0800 Subject: [PATCH] Fix unittests on Python 2.6. Tests were being run with a Python 2.6 test runner instead of a 2.7 test runner. Using a conditional import in setup.py, tests are now run with the standard 2.7 unittest library (known as unittest2 to Python 2.6). Modified/cleaned up Google tests. Tests were not hermetic, but now proper mocking has been put in place to make them hermetic. Tests: python2.7 setup.py test python2.6 setup.py test --- libcloud/common/google.py | 126 +++++++------ libcloud/test/common/test_google.py | 178 ++++++++++++++----- libcloud/test/compute/test_gce.py | 11 +- libcloud/test/dns/test_google.py | 13 +- libcloud/test/loadbalancer/test_gce.py | 13 +- libcloud/test/storage/test_google_storage.py | 30 +--- setup.py | 11 +- 7 files changed, 225 insertions(+), 157 deletions(-) diff --git a/libcloud/common/google.py b/libcloud/common/google.py index 1977546fca..e29451fcb8 100644 --- a/libcloud/common/google.py +++ b/libcloud/common/google.py @@ -101,35 +101,30 @@ RSA = None PKCS1_v1_5 = None -TIMESTAMP_FORMAT = '%Y-%m-%dT%H:%M:%SZ' +UTC_TIMESTAMP_FORMAT = '%Y-%m-%dT%H:%M:%SZ' -def _now(): +def _utcnow(): + """ + Mocked in libcloud.test.common.google.GoogleTestCase. + """ return datetime.datetime.utcnow() -def _is_gce(): - http_code, http_reason, body = _get_gce_metadata() - if http_code == httplib.OK and body: - return True - return False - - -def _is_gcs_s3(user_id): - """Checks S3 key format: 20 alphanumeric chars starting with GOOG.""" - return len(user_id) == 20 and user_id.startswith('GOOG') +def _utc_timestamp(datetime_obj): + return datetime_obj.strftime(UTC_TIMESTAMP_FORMAT) -def _is_sa(user_id): - return user_id.endswith('@developer.gserviceaccount.com') +def _from_utc_timestamp(timestamp): + return datetime.datetime.strptime(timestamp, UTC_TIMESTAMP_FORMAT) def _get_gce_metadata(path=''): try: - url = "http://metadata/computeMetadata/v1/" + path.lstrip('/') + url = 'http://metadata/computeMetadata/v1/' + path.lstrip('/') headers = {'Metadata-Flavor': 'Google'} response = get_response_object(url, headers=headers) - return response.status, "", response.body + return response.status, '', response.body except Exception as e: return -1, str(e), None @@ -358,7 +353,6 @@ def _token_request(self, request_body): :rtype: ``dict`` """ data = urlencode(request_body) - now = _now() try: response = self.request('/o/oauth2/token', method='POST', data=data) @@ -367,9 +361,9 @@ def _token_request(self, request_body): 'check your credentials and time drift.') token_info = response.object if 'expires_in' in token_info: - expire_time = now + datetime.timedelta( + expire_time = _utcnow() + datetime.timedelta( seconds=token_info['expires_in']) - token_info['expire_time'] = expire_time.strftime(TIMESTAMP_FORMAT) + token_info['expire_time'] = _utc_timestamp(expire_time) return token_info def refresh_token(self, token_info): @@ -395,6 +389,8 @@ def get_code(self): Give the user a URL that they can visit to authenticate and obtain a code. This method will ask for that code that the user can paste in. + Mocked in libcloud.test.common.google.GoogleTestCase. + :return: Code supplied by the user after authenticating :rtype: ``str`` """ @@ -555,9 +551,9 @@ def get_new_token(self): "'%s'" % str(http_reason)) token_info = json.loads(token_info) if 'expires_in' in token_info: - expire_time = _now() + datetime.timedelta( + expire_time = _utcnow() + datetime.timedelta( seconds=token_info['expires_in']) - token_info['expire_time'] = expire_time.strftime(TIMESTAMP_FORMAT) + token_info['expire_time'] = _utc_timestamp(expire_time) return token_info @@ -578,11 +574,11 @@ class GoogleAuthType(object): @classmethod def guess_type(cls, user_id): - if _is_sa(user_id): + if cls._is_sa(user_id): return cls.SA - elif _is_gce(): + elif cls._is_gce(): return cls.GCE - elif _is_gcs_s3(user_id): + elif cls._is_gcs_s3(user_id): return cls.GCS_S3 else: return cls.IA @@ -591,6 +587,28 @@ def guess_type(cls, user_id): def is_oauth2(cls, auth_type): return auth_type in cls.OAUTH2_TYPES + @staticmethod + def _is_gce(): + """ + Checks if we can access the GCE metadata server. + Mocked in libcloud.test.common.google.GoogleTestCase. + """ + http_code, http_reason, body = _get_gce_metadata() + if http_code == httplib.OK and body: + return True + return False + + @staticmethod + def _is_gcs_s3(user_id): + """ + Checks S3 key format: 20 alphanumeric chars starting with GOOG. + """ + return len(user_id) == 20 and user_id.startswith('GOOG') + + @staticmethod + def _is_sa(user_id): + return user_id.endswith('@developer.gserviceaccount.com') + class GoogleBaseConnection(ConnectionUserAndKey, PollingConnection): """Base connection class for interacting with Google APIs.""" @@ -639,16 +657,15 @@ def __init__(self, user_id, key=None, auth_type=None, # OAuth2 stuff and placeholders self.scopes = scopes - self.auth_conn = None - self.token_expire_time = None - self.token_info = None + self.oauth2_conn = None + self.oauth2_token = None if credential_file: self.credential_file = credential_file elif self.auth_type == GoogleAuthType.SA: self.credential_file += '.' + user_id if GoogleAuthType.is_oauth2(self.auth_type): - self._setup_oauth2(**kwargs) + self._init_oauth2(**kwargs) super(GoogleBaseConnection, self).__init__(user_id, key, **kwargs) @@ -657,6 +674,10 @@ def __init__(self, user_id, key=None, auth_type=None, ver_platform = 'Python %s/%s' % (python_ver, sys.platform) self.user_agent_append(ver_platform) + @property + def token_expire_utc_datetime(self): + return _from_utc_timestamp(self.oauth2_token['expire_time']) + def add_default_headers(self, headers): """ @inherits: :class:`Connection.add_default_headers` @@ -672,14 +693,10 @@ def pre_connect_hook(self, params, headers): @inherits: :class:`Connection.pre_connect_hook` """ - now = _now() - if self.token_expire_time < now: - self.token_info = self.auth_conn.refresh_token(self.token_info) - self.token_expire_time = datetime.datetime.strptime( - self.token_info['expire_time'], TIMESTAMP_FORMAT) - self._write_token_info_to_file() + if self.token_expire_utc_datetime < _utcnow(): + self._refresh_oauth2_token() headers['Authorization'] = 'Bearer %s' % ( - self.token_info['access_token']) + self.oauth2_token['access_token']) return params, headers @@ -751,7 +768,11 @@ def morph_action_hook(self, action): request = self.request_path + action return request - def _setup_oauth2(self, **kwargs): + def _refresh_oauth2_token(self): + self.oauth2_token = self.oauth2_conn.refresh_token(self.oauth2_token) + self._write_token_to_file() + + def _init_oauth2(self, **kwargs): # Default scopes to read/write for compute, storage, and dns. Can # override this when calling get_driver() or setting in secrets.py if not self.scopes: @@ -760,51 +781,50 @@ def _setup_oauth2(self, **kwargs): 'https://www.googleapis.com/auth/devstorage.full_control', 'https://www.googleapis.com/auth/ndev.clouddns.readwrite', ] - self.token_info = self._get_token_info_from_file() + self.oauth2_token = self._get_token_from_file() if self.auth_type == GoogleAuthType.GCE: - self.auth_conn = GoogleGCEServiceAcctAuthConnection( + self.oauth2_conn = GoogleGCEServiceAcctAuthConnection( self.user_id, self.scopes, **kwargs) elif self.auth_type == GoogleAuthType.SA: - self.auth_conn = GoogleServiceAcctAuthConnection( + self.oauth2_conn = GoogleServiceAcctAuthConnection( self.user_id, self.key, self.scopes, **kwargs) elif self.auth_type == GoogleAuthType.IA: - self.auth_conn = GoogleInstalledAppAuthConnection( + self.oauth2_conn = GoogleInstalledAppAuthConnection( self.user_id, self.key, self.scopes, **kwargs) else: raise GoogleAuthError('Invalid auth_type: %s' % str(self.auth_type)) - if self.token_info is None: - self.token_info = self.auth_conn.get_new_token() - self._write_token_info_to_file() + if self.oauth2_token is None: + self.oauth2_token = self.oauth2_conn.get_new_token() + self._write_token_to_file() - self.token_expire_time = datetime.datetime.strptime( - self.token_info['expire_time'], TIMESTAMP_FORMAT) - - def _get_token_info_from_file(self): + def _get_token_from_file(self): """ Read credential file and return token information. + Mocked in libcloud.test.common.google.GoogleTestCase. :return: Token information dictionary, or None :rtype: ``dict`` or ``None`` """ - token_info = None + token = None filename = os.path.realpath(os.path.expanduser(self.credential_file)) try: with open(filename, 'r') as f: data = f.read() - token_info = json.loads(data) + token = json.loads(data) except IOError: pass - return token_info + return token - def _write_token_info_to_file(self): + def _write_token_to_file(self): """ - Write token_info to credential file. + Write token to credential file. + Mocked in libcloud.test.common.google.GoogleTestCase. """ filename = os.path.realpath(os.path.expanduser(self.credential_file)) - data = json.dumps(self.token_info) + data = json.dumps(self.oauth2_token) with open(filename, 'w') as f: f.write(data) diff --git a/libcloud/test/common/test_google.py b/libcloud/test/common/test_google.py index b16488be73..d463147282 100644 --- a/libcloud/test/common/test_google.py +++ b/libcloud/test/common/test_google.py @@ -26,16 +26,17 @@ except ImportError: import json -from libcloud.utils.py3 import httplib - -from libcloud.test import MockHttp, LibcloudTestCase from libcloud.common.google import (GoogleAuthError, GoogleAuthType, GoogleBaseAuthConnection, GoogleInstalledAppAuthConnection, GoogleServiceAcctAuthConnection, GoogleGCEServiceAcctAuthConnection, - GoogleBaseConnection) + GoogleBaseConnection, + _utcnow, + _utc_timestamp) +from libcloud.test import MockHttp, LibcloudTestCase +from libcloud.utils.py3 import httplib # Skip some tests if PyCrypto is unavailable @@ -61,13 +62,92 @@ GCS_S3_PARAMS = ('GOOG0123456789ABCXYZ', # GOOG + 16 alphanumeric chars '0102030405060708091011121314151617181920') # 40 base64 chars +STUB_UTCNOW = _utcnow() + +STUB_TOKEN = { + 'access_token': 'tokentoken', + 'token_type': 'Bearer', + 'expires_in': 3600 +} + +STUB_IA_TOKEN = { + 'access_token': 'installedapp', + 'token_type': 'Bearer', + 'expires_in': 3600, + 'refresh_token': 'refreshrefresh' +} + +STUB_REFRESH_TOKEN = { + 'access_token': 'refreshrefresh', + 'token_type': 'Bearer', + 'expires_in': 3600 +} + +STUB_TOKEN_FROM_FILE = { + 'access_token': 'token_from_file', + 'token_type': 'Bearer', + 'expire_time': _utc_timestamp(STUB_UTCNOW + + datetime.timedelta(seconds=3600)), + 'expires_in': 3600 +} + class MockJsonResponse(object): def __init__(self, body): self.object = body -class GoogleBaseAuthConnectionTest(LibcloudTestCase): +class GoogleTestCase(LibcloudTestCase): + """ + Assists in making Google tests hermetic and deterministic. + + Add anything that needs to be mocked here. Create a patcher with the + suffix '_patcher'. + + e.g. + _foo_patcher = mock.patch('module.submodule.class.foo', ...) + + Patchers are started at setUpClass and stopped at tearDownClass. + + Ideally, you should make a note in the thing being mocked, for clarity. + """ + PATCHER_SUFFIX = '_patcher' + + _utcnow_patcher = mock.patch( + 'libcloud.common.google._utcnow', return_value=STUB_UTCNOW) + + _authtype_is_gce_patcher = mock.patch( + 'libcloud.common.google.GoogleAuthType._is_gce', return_value=False) + + _read_token_file_patcher = mock.patch( + 'libcloud.common.google.GoogleBaseConnection._get_token_from_file', + return_value=STUB_TOKEN_FROM_FILE + ) + + _write_token_file_patcher = mock.patch( + 'libcloud.common.google.GoogleBaseConnection._write_token_to_file') + + _ia_get_code_patcher = mock.patch( + 'libcloud.common.google.GoogleInstalledAppAuthConnection.get_code', + return_value=1234 + ) + + @classmethod + def setUpClass(cls): + super(GoogleTestCase, cls).setUpClass() + + for patcher in [a for a in dir(cls) if a.endswith(cls.PATCHER_SUFFIX)]: + getattr(cls, patcher).start() + + @classmethod + def tearDownClass(cls): + super(GoogleTestCase, cls).tearDownClass() + + for patcher in [a for a in dir(cls) if a.endswith(cls.PATCHER_SUFFIX)]: + getattr(cls, patcher).stop() + + +class GoogleBaseAuthConnectionTest(GoogleTestCase): """ Tests for GoogleBaseAuthConnection """ @@ -91,23 +171,23 @@ def test_add_default_headers(self): new_headers = self.conn.add_default_headers(old_headers) self.assertEqual(new_headers, expected_headers) - @mock.patch('libcloud.common.google._now') - def test_token_request(self, mock_now): - mock_now.return_value = datetime.datetime(2013, 6, 26, 19, 0, 0) + def test_token_request(self): request_body = {'code': 'asdf', 'client_id': self.conn.user_id, 'client_secret': self.conn.key, 'redirect_uri': self.conn.redirect_uri, 'grant_type': 'authorization_code'} new_token = self.conn._token_request(request_body) - self.assertEqual(new_token['access_token'], 'installedapp') - self.assertEqual(new_token['expire_time'], '2013-06-26T20:00:00Z') + self.assertEqual(new_token['access_token'], + STUB_IA_TOKEN['access_token']) + exp = STUB_UTCNOW + datetime.timedelta( + seconds=STUB_IA_TOKEN['expires_in']) + self.assertEqual(new_token['expire_time'], _utc_timestamp(exp)) -class GoogleInstalledAppAuthConnectionTest(LibcloudTestCase): +class GoogleInstalledAppAuthConnectionTest(GoogleTestCase): """ Tests for GoogleInstalledAppAuthConnection """ - GoogleInstalledAppAuthConnection.get_code = lambda x: '1234' def setUp(self): GoogleInstalledAppAuthConnection.conn_classes = (GoogleAuthMockHttp, @@ -123,21 +203,23 @@ def test_refresh_token(self): token_info1 = {'access_token': 'tokentoken', 'token_type': 'Bearer', 'expires_in': 3600} new_token1 = self.conn.refresh_token(token_info1) - self.assertEqual(new_token1['access_token'], 'installedapp') + self.assertEqual(new_token1['access_token'], + STUB_IA_TOKEN['access_token']) # This token info has a refresh token, so it will be able to be # refreshed. token_info2 = {'access_token': 'tokentoken', 'token_type': 'Bearer', 'expires_in': 3600, 'refresh_token': 'refreshrefresh'} new_token2 = self.conn.refresh_token(token_info2) - self.assertEqual(new_token2['access_token'], 'refreshrefresh') + self.assertEqual(new_token2['access_token'], + STUB_REFRESH_TOKEN['access_token']) # Both sets should have refresh info self.assertTrue('refresh_token' in new_token1) self.assertTrue('refresh_token' in new_token2) -class GoogleAuthTypeTest(LibcloudTestCase): +class GoogleAuthTypeTest(GoogleTestCase): def test_guess(self): self.assertEqual( @@ -146,7 +228,7 @@ def test_guess(self): self.assertEqual( GoogleAuthType.guess_type(GCE_PARAMS_IA[0]), GoogleAuthType.IA) - with mock.patch('libcloud.common.google._is_gce', return_value=True): + with mock.patch.object(GoogleAuthType, '_is_gce', return_value=True): self.assertEqual( GoogleAuthType.guess_type(GCE_PARAMS_GCE[0]), GoogleAuthType.GCE) @@ -155,25 +237,17 @@ def test_guess(self): GoogleAuthType.GCS_S3) -class GoogleBaseConnectionTest(LibcloudTestCase): +class GoogleBaseConnectionTest(GoogleTestCase): """ Tests for GoogleBaseConnection """ - GoogleBaseConnection._get_token_info_from_file = lambda x: None - GoogleBaseConnection._write_token_info_to_file = lambda x: None - GoogleInstalledAppAuthConnection.get_code = lambda x: '1234' - GoogleServiceAcctAuthConnection.get_new_token = \ - lambda x: x._token_request({}) - GoogleGCEServiceAcctAuthConnection.get_new_token = \ - lambda x: x._token_request({}) - GoogleBaseConnection._now = lambda x: datetime.datetime(2013, 6, 26, - 19, 0, 0) def setUp(self): GoogleBaseAuthConnection.conn_classes = (GoogleAuthMockHttp, GoogleAuthMockHttp) self.mock_scopes = ['https://www.googleapis.com/auth/foo'] - kwargs = {'scopes': self.mock_scopes, 'auth_type': 'IA'} + kwargs = {'scopes': self.mock_scopes, + 'auth_type': GoogleAuthType.IA} self.conn = GoogleBaseConnection(*GCE_PARAMS, **kwargs) def test_auth_type(self): @@ -185,30 +259,30 @@ def test_auth_type(self): if SHA256: kwargs['auth_type'] = GoogleAuthType.SA conn1 = GoogleBaseConnection(*GCE_PARAMS_PEM_KEY, **kwargs) - self.assertTrue(isinstance(conn1.auth_conn, + self.assertTrue(isinstance(conn1.oauth2_conn, GoogleServiceAcctAuthConnection)) conn1 = GoogleBaseConnection(*GCE_PARAMS_JSON_KEY, **kwargs) - self.assertTrue(isinstance(conn1.auth_conn, + self.assertTrue(isinstance(conn1.oauth2_conn, GoogleServiceAcctAuthConnection)) conn1 = GoogleBaseConnection(*GCE_PARAMS_KEY, **kwargs) - self.assertTrue(isinstance(conn1.auth_conn, + self.assertTrue(isinstance(conn1.oauth2_conn, GoogleServiceAcctAuthConnection)) kwargs['auth_type'] = GoogleAuthType.IA conn2 = GoogleBaseConnection(*GCE_PARAMS_IA, **kwargs) - self.assertTrue(isinstance(conn2.auth_conn, + self.assertTrue(isinstance(conn2.oauth2_conn, GoogleInstalledAppAuthConnection)) kwargs['auth_type'] = GoogleAuthType.GCE conn3 = GoogleBaseConnection(*GCE_PARAMS_GCE, **kwargs) - self.assertTrue(isinstance(conn3.auth_conn, + self.assertTrue(isinstance(conn3.oauth2_conn, GoogleGCEServiceAcctAuthConnection)) kwargs['auth_type'] = GoogleAuthType.GCS_S3 conn4 = GoogleBaseConnection(*GCS_S3_PARAMS, **kwargs) - self.assertIsNone(conn4.auth_conn) + self.assertIsNone(conn4.oauth2_conn) def test_add_default_headers(self): old_headers = {} @@ -220,8 +294,10 @@ def test_add_default_headers(self): def test_pre_connect_hook(self): old_params = {} old_headers = {} + auth_str = '%s %s' % (STUB_TOKEN_FROM_FILE['token_type'], + STUB_TOKEN_FROM_FILE['access_token']) new_expected_params = {} - new_expected_headers = {'Authorization': 'Bearer installedapp'} + new_expected_headers = {'Authorization': auth_str} new_params, new_headers = self.conn.pre_connect_hook(old_params, old_headers) self.assertEqual(new_params, new_expected_params) @@ -271,6 +347,24 @@ def test_morph_action_hook(self): self.assertEqual(request1, expected_request) self.assertEqual(request2, expected_request) + def test_init_oauth2(self): + mock_scopes = ['https://www.googleapis.com/auth/foo'] + kwargs = {'scopes': mock_scopes, + 'auth_type': GoogleAuthType.IA} + conn = GoogleBaseConnection(*GCE_PARAMS, **kwargs) + + # If there is a viable token file, this gets used first + self.assertEqual(conn.oauth2_token, STUB_TOKEN_FROM_FILE) + + # No token file, get a new token. Check that it gets written to file. + with mock.patch.object(GoogleBaseConnection, + '_get_token_from_file', return_value=None): + conn = GoogleBaseConnection(*GCE_PARAMS, **kwargs) + expected = STUB_IA_TOKEN + expected['expire_time'] = conn.oauth2_token['expire_time'] + self.assertEqual(conn.oauth2_token, expected) + conn._write_token_to_file.assert_called_once_with() + class GoogleAuthMockHttp(MockHttp): """ @@ -279,22 +373,12 @@ class GoogleAuthMockHttp(MockHttp): json_hdr = {'content-type': 'application/json; charset=UTF-8'} def _o_oauth2_token(self, method, url, body, headers): - token_info = {'access_token': 'tokentoken', - 'token_type': 'Bearer', - 'expires_in': 3600} - refresh_token = {'access_token': 'refreshrefresh', - 'token_type': 'Bearer', - 'expires_in': 3600} - ia_token = {'access_token': 'installedapp', - 'token_type': 'Bearer', - 'expires_in': 3600, - 'refresh_token': 'refreshrefresh'} if 'code' in body: - body = json.dumps(ia_token) + body = json.dumps(STUB_IA_TOKEN) elif 'refresh_token' in body: - body = json.dumps(refresh_token) + body = json.dumps(STUB_REFRESH_TOKEN) else: - body = json.dumps(token_info) + body = json.dumps(STUB_TOKEN) return (httplib.OK, body, self.json_hdr, httplib.responses[httplib.OK]) diff --git a/libcloud/test/compute/test_gce.py b/libcloud/test/compute/test_gce.py index 8e1866f694..72e41b6f11 100644 --- a/libcloud/test/compute/test_gce.py +++ b/libcloud/test/compute/test_gce.py @@ -29,29 +29,24 @@ GCETargetHttpProxy, GCEUrlMap, GCEZone) from libcloud.common.google import (GoogleBaseAuthConnection, - GoogleInstalledAppAuthConnection, - GoogleBaseConnection, ResourceNotFoundError, ResourceExistsError) -from libcloud.test.common.test_google import GoogleAuthMockHttp +from libcloud.test.common.test_google import GoogleAuthMockHttp, GoogleTestCase from libcloud.compute.base import Node, StorageVolume -from libcloud.test import MockHttpTestCase, LibcloudTestCase +from libcloud.test import MockHttpTestCase from libcloud.test.compute import TestCaseMixin from libcloud.test.file_fixtures import ComputeFileFixtures from libcloud.test.secrets import GCE_PARAMS, GCE_KEYWORD_PARAMS -class GCENodeDriverTest(LibcloudTestCase, TestCaseMixin): +class GCENodeDriverTest(GoogleTestCase, TestCaseMixin): """ Google Compute Engine Test Class. """ # Mock out a few specific calls that interact with the user, system or # environment. - GoogleBaseConnection._get_token_info_from_file = lambda x: None - GoogleBaseConnection._write_token_info_to_file = lambda x: None - GoogleInstalledAppAuthConnection.get_code = lambda x: '1234' GCEZone._now = lambda x: datetime.datetime(2013, 6, 26, 19, 0, 0) datacenter = 'us-central1-a' diff --git a/libcloud/test/dns/test_google.py b/libcloud/test/dns/test_google.py index ddb3b91407..5c8adca3b0 100644 --- a/libcloud/test/dns/test_google.py +++ b/libcloud/test/dns/test_google.py @@ -20,20 +20,15 @@ from libcloud.dns.types import ZoneDoesNotExistError from libcloud.dns.types import RecordDoesNotExistError from libcloud.dns.drivers.google import GoogleDNSDriver -from libcloud.common.google import (GoogleBaseAuthConnection, - GoogleInstalledAppAuthConnection, - GoogleBaseConnection) +from libcloud.common.google import GoogleBaseAuthConnection -from libcloud.test.common.test_google import GoogleAuthMockHttp -from libcloud.test import MockHttpTestCase, LibcloudTestCase +from libcloud.test.common.test_google import GoogleAuthMockHttp, GoogleTestCase +from libcloud.test import MockHttpTestCase from libcloud.test.file_fixtures import DNSFileFixtures from libcloud.test.secrets import DNS_PARAMS_GOOGLE, DNS_KEYWORD_PARAMS_GOOGLE -class GoogleTests(LibcloudTestCase): - GoogleBaseConnection._get_token_info_from_file = lambda x: None - GoogleBaseConnection._write_token_info_to_file = lambda x: None - GoogleInstalledAppAuthConnection.get_code = lambda x: '1234' +class GoogleTests(GoogleTestCase): def setUp(self): GoogleDNSMockHttp.test = self diff --git a/libcloud/test/loadbalancer/test_gce.py b/libcloud/test/loadbalancer/test_gce.py index 9b55a2a89f..41d0594c25 100644 --- a/libcloud/test/loadbalancer/test_gce.py +++ b/libcloud/test/loadbalancer/test_gce.py @@ -18,23 +18,16 @@ import sys import unittest -from libcloud.common.google import (GoogleBaseAuthConnection, - GoogleInstalledAppAuthConnection, - GoogleBaseConnection) +from libcloud.common.google import GoogleBaseAuthConnection from libcloud.compute.drivers.gce import (GCENodeDriver) from libcloud.loadbalancer.drivers.gce import (GCELBDriver) -from libcloud.test.common.test_google import GoogleAuthMockHttp +from libcloud.test.common.test_google import GoogleAuthMockHttp, GoogleTestCase from libcloud.test.compute.test_gce import GCEMockHttp -from libcloud.test import LibcloudTestCase - from libcloud.test.secrets import GCE_PARAMS, GCE_KEYWORD_PARAMS -class GCELoadBalancerTest(LibcloudTestCase): - GoogleBaseConnection._get_token_info_from_file = lambda x: None - GoogleBaseConnection._write_token_info_to_file = lambda x: None - GoogleInstalledAppAuthConnection.get_code = lambda x: '1234' +class GCELoadBalancerTest(GoogleTestCase): datacenter = 'us-central1-a' def setUp(self): diff --git a/libcloud/test/storage/test_google_storage.py b/libcloud/test/storage/test_google_storage.py index b1d23b0872..001a9f3a5e 100644 --- a/libcloud/test/storage/test_google_storage.py +++ b/libcloud/test/storage/test_google_storage.py @@ -22,7 +22,7 @@ from libcloud.common.google import GoogleAuthType from libcloud.storage.drivers import google_storage -from libcloud.test import LibcloudTestCase +from libcloud.test.common.test_google import GoogleTestCase from libcloud.test.file_fixtures import StorageFileFixtures from libcloud.test.secrets import STORAGE_GOOGLE_STORAGE_PARAMS from libcloud.test.storage.test_s3 import S3Tests, S3MockHttp @@ -33,9 +33,6 @@ TODAY = email.utils.formatdate(usegmt=True) -OAUTH2_MOCK = mock.patch( - 'libcloud.common.google.GoogleBaseConnection._setup_oauth2', spec=True) - class GoogleStorageMockHttp(S3MockHttp): fixtures = StorageFileFixtures('google_storage') @@ -60,17 +57,7 @@ def _test2_test_get_object(self, method, url, body, headers): ) -class GoogleStorageConnectionTest(LibcloudTestCase): - - @classmethod - def setUpClass(cls): - super(LibcloudTestCase, cls).setUpClass() - OAUTH2_MOCK.start() - - @classmethod - def tearDownClass(cls): - super(LibcloudTestCase, cls).tearDownClass() - OAUTH2_MOCK.stop() +class GoogleStorageConnectionTest(GoogleTestCase): @mock.patch('email.utils.formatdate') @mock.patch('libcloud.common.google.' @@ -254,20 +241,11 @@ def fake_hmac_method(params, headers): ) -class GoogleStorageTests(S3Tests): +class GoogleStorageTests(S3Tests, GoogleTestCase): driver_type = STORAGE_CLS driver_args = STORAGE_GOOGLE_STORAGE_PARAMS mock_response_klass = GoogleStorageMockHttp - - @classmethod - def setUpClass(cls): - super(S3Tests, cls).setUpClass() - OAUTH2_MOCK.start() - - @classmethod - def tearDownClass(cls): - super(S3Tests, cls).tearDownClass() - OAUTH2_MOCK.stop() + driver = google_storage.GoogleStorageDriver def test_billing_not_enabled(self): # TODO diff --git a/setup.py b/setup.py index 9a29e3bfa9..a8ca61da10 100644 --- a/setup.py +++ b/setup.py @@ -97,6 +97,8 @@ def forbid_publish(): class TestCommand(Command): description = "run test suite" user_options = [] + unittest_TestLoader = TestLoader + unittest_TextTestRunner = TextTestRunner def initialize_options(self): THIS_DIR = os.path.abspath(os.path.split(__file__)[0]) @@ -121,8 +123,9 @@ def run(self): if unittest2_required: try: - import unittest2 - unittest2 + from unittest2 import TextTestRunner, TestLoader + self.unittest_TestLoader = TestLoader + self.unittest_TextTestRunner = TextTestRunner except ImportError: print('Python version: %s' % (sys.version)) print('Missing "unittest2" library. unittest2 is library is ' @@ -188,12 +191,12 @@ def _run_tests(self): str(e))) raise e - tests = TestLoader().loadTestsFromNames(testfiles) + tests = self.unittest_TestLoader().loadTestsFromNames(testfiles) for test_module in DOC_TEST_MODULES: tests.addTests(doctest.DocTestSuite(test_module)) - t = TextTestRunner(verbosity=2) + t = self.unittest_TextTestRunner(verbosity=2) res = t.run(tests) return not res.wasSuccessful()