Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Unwinding cases in Test_set_default_dataset_id and having better test separation #672

Merged
merged 5 commits into from
Feb 26, 2015
Merged
Changes from 1 commit
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Prev Previous commit
Removing most of Test_set_default_dataset_id.
Also adding the relative ordering parts to Test__determine_default_dataset_id.

Tests previously were mixing the logic of the 4 possible environments
while the simple ordering of them in set_default_dataset_id() and
now _determine_default_dataset_id().

This separates that logic and makes sure each environ helper is called
in a specific order.
  • Loading branch information
dhermes committed Feb 25, 2015
commit 1dd670973b86e9dfc0285a134f59a12806cf23e1
302 changes: 67 additions & 235 deletions gcloud/datastore/test__implicit_environ.py
Original file line number Diff line number Diff line change
Expand Up @@ -190,27 +190,28 @@ def _callFUT(self, dataset_id=None):
return _implicit_environ._determine_default_dataset_id(
dataset_id=dataset_id)

def test_it(self):
def _determine_default_helper(self, prod=None, gcd=None, gae=None,
gce=None, dataset_id=None):
from gcloud._testing import _Monkey
from gcloud.datastore import _implicit_environ

_callers = []

def prod_mock():
_callers.append('prod_mock')
return None
return prod

def gcd_mock():
_callers.append('gcd_mock')
return None
return gcd

def gae_mock():
_callers.append('gae_mock')
return None
return gae

def gce_mock():
_callers.append('gce_mock')
return None
return gce

patched_methods = {
'_get_production_dataset_id': prod_mock,
Expand All @@ -220,10 +221,46 @@ def gce_mock():
}

with _Monkey(_implicit_environ, **patched_methods):
dataset_id = self._callFUT()
self.assertEqual(dataset_id, None)
returned_dataset_id = self._callFUT(dataset_id)

return returned_dataset_id, _callers

def test_no_value(self):
dataset_id, callers = self._determine_default_helper()
self.assertEqual(dataset_id, None)
self.assertEqual(callers,
['prod_mock', 'gcd_mock', 'gae_mock', 'gce_mock'])

self.assertEqual(_callers,
def test_explicit(self):
DATASET_ID = object()
dataset_id, callers = self._determine_default_helper(
dataset_id=DATASET_ID)
self.assertEqual(dataset_id, DATASET_ID)
self.assertEqual(callers, [])

def test_prod(self):
DATASET_ID = object()
dataset_id, callers = self._determine_default_helper(prod=DATASET_ID)
self.assertEqual(dataset_id, DATASET_ID)
self.assertEqual(callers, ['prod_mock'])

def test_gcd(self):
DATASET_ID = object()
dataset_id, callers = self._determine_default_helper(gcd=DATASET_ID)
self.assertEqual(dataset_id, DATASET_ID)
self.assertEqual(callers, ['prod_mock', 'gcd_mock'])

def test_gae(self):
DATASET_ID = object()
dataset_id, callers = self._determine_default_helper(gae=DATASET_ID)
self.assertEqual(dataset_id, DATASET_ID)
self.assertEqual(callers, ['prod_mock', 'gcd_mock', 'gae_mock'])

def test_gce(self):
DATASET_ID = object()
dataset_id, callers = self._determine_default_helper(gce=DATASET_ID)
self.assertEqual(dataset_id, DATASET_ID)
self.assertEqual(callers,
['prod_mock', 'gcd_mock', 'gae_mock', 'gce_mock'])


Expand All @@ -241,246 +278,41 @@ def _callFUT(self, dataset_id=None):
from gcloud.datastore._implicit_environ import set_default_dataset_id
return set_default_dataset_id(dataset_id=dataset_id)

def _monkeyEnviron(self, implicit_dataset_id, environ=None):
import os
from gcloud._testing import _Monkey
from gcloud.datastore._implicit_environ import _DATASET_ENV_VAR_NAME
environ = environ or {_DATASET_ENV_VAR_NAME: implicit_dataset_id}
return _Monkey(os, getenv=environ.get)

def _monkeyImplicit(self, connection=None, app_identity=None):
def test_raises(self):
from gcloud._testing import _Monkey
from gcloud.datastore import _implicit_environ

if connection is None:
connection = _HTTPConnection(404, None)

def _factory(host, timeout):
connection.host = host
connection.timeout = timeout
return connection

return _Monkey(_implicit_environ,
HTTPConnection=_factory,
app_identity=app_identity)

def test_no_env_var_set(self):
from gcloud.datastore import _implicit_environ

with self._monkeyEnviron(None):
with self._monkeyImplicit():
self.assertRaises(EnvironmentError, self._callFUT)

self.assertEqual(_implicit_environ.get_default_dataset_id(), None)

def test_set_from_env_var(self):
from gcloud.datastore import _implicit_environ
IMPLICIT_DATASET_ID = 'IMPLICIT'

with self._monkeyEnviron(IMPLICIT_DATASET_ID):
with self._monkeyImplicit():
self._callFUT()

self.assertEqual(_implicit_environ.get_default_dataset_id(),
IMPLICIT_DATASET_ID)

def test_set_explicit_w_env_var_set(self):
from gcloud.datastore import _implicit_environ
EXPLICIT_DATASET_ID = 'EXPLICIT'

with self._monkeyEnviron(None):
with self._monkeyImplicit():
self._callFUT(EXPLICIT_DATASET_ID)

self.assertEqual(_implicit_environ.get_default_dataset_id(),
EXPLICIT_DATASET_ID)

def test_set_explicit_no_env_var_set(self):
from gcloud.datastore import _implicit_environ
IMPLICIT_DATASET_ID = 'IMPLICIT'
EXPLICIT_DATASET_ID = 'EXPLICIT'

with self._monkeyEnviron(IMPLICIT_DATASET_ID):
with self._monkeyImplicit():
self._callFUT(EXPLICIT_DATASET_ID)

self.assertEqual(_implicit_environ.get_default_dataset_id(),
EXPLICIT_DATASET_ID)

def test_set_explicit_None_wo_env_var_set(self):
from gcloud.datastore import _implicit_environ

with self._monkeyEnviron(None):
with self._monkeyImplicit():
self.assertRaises(EnvironmentError, self._callFUT, None)

self.assertEqual(_implicit_environ.get_default_dataset_id(), None)

def test_set_explicit_None_w_env_var_set(self):
from gcloud.datastore import _implicit_environ
IMPLICIT_DATASET_ID = 'IMPLICIT'

with self._monkeyEnviron(IMPLICIT_DATASET_ID):
with self._monkeyImplicit():
self._callFUT(None)
_called_dataset_id = []

self.assertEqual(_implicit_environ.get_default_dataset_id(),
IMPLICIT_DATASET_ID)

def test_set_from_gcd_env_var(self):
from gcloud.datastore import _implicit_environ

GCD_ENV = _implicit_environ._GCD_DATASET_ENV_VAR_NAME
GCD_DATASET_ID = 'GCD-IMPLICIT'
ENVIRON = {GCD_ENV: GCD_DATASET_ID}

with self._monkeyEnviron(None, environ=ENVIRON):
with self._monkeyImplicit():
self._callFUT()

self.assertEqual(_implicit_environ.get_default_dataset_id(),
GCD_DATASET_ID)

def test_set_gcd_and_production_env_vars(self):
from gcloud.datastore import _implicit_environ
from gcloud.datastore._implicit_environ import _DATASET_ENV_VAR_NAME

GCD_ENV = _implicit_environ._GCD_DATASET_ENV_VAR_NAME
IMPLICIT_DATASET_ID = 'IMPLICIT'
GCD_DATASET_ID = 'GCD-IMPLICIT'
ENVIRON = {
_DATASET_ENV_VAR_NAME: IMPLICIT_DATASET_ID,
GCD_ENV: GCD_DATASET_ID,
}

with self._monkeyEnviron(None, environ=ENVIRON):
with self._monkeyImplicit():
self._callFUT()

self.assertEqual(_implicit_environ.get_default_dataset_id(),
IMPLICIT_DATASET_ID)

def test_set_gcd_env_vars_and_appengine(self):
from gcloud.datastore import _implicit_environ

GCD_ENV = _implicit_environ._GCD_DATASET_ENV_VAR_NAME
GCD_DATASET_ID = 'GCD-IMPLICIT'
ENVIRON = {GCD_ENV: GCD_DATASET_ID}

APP_ENGINE_ID = 'GAE'
APP_IDENTITY = _AppIdentity(APP_ENGINE_ID)

with self._monkeyEnviron(None, environ=ENVIRON):
with self._monkeyImplicit(app_identity=APP_IDENTITY):
self._callFUT()

self.assertEqual(_implicit_environ.get_default_dataset_id(),
GCD_DATASET_ID)

def test_set_implicit_from_appengine(self):
from gcloud.datastore import _implicit_environ

APP_ENGINE_ID = 'GAE'
APP_IDENTITY = _AppIdentity(APP_ENGINE_ID)

with self._monkeyEnviron(None):
with self._monkeyImplicit(app_identity=APP_IDENTITY):
self._callFUT()

self.assertEqual(_implicit_environ.get_default_dataset_id(),
APP_ENGINE_ID)

def test_set_implicit_both_env_and_appengine(self):
from gcloud.datastore import _implicit_environ

IMPLICIT_DATASET_ID = 'IMPLICIT'
APP_IDENTITY = _AppIdentity('GAE')

with self._monkeyEnviron(IMPLICIT_DATASET_ID):
with self._monkeyImplicit(app_identity=APP_IDENTITY):
self._callFUT()

self.assertEqual(_implicit_environ.get_default_dataset_id(),
IMPLICIT_DATASET_ID)

def _implicit_compute_engine_helper(self, status):
from gcloud.datastore import _implicit_environ

COMPUTE_ENGINE_ID = 'GCE'
if status == 200:
EXPECTED_ID = COMPUTE_ENGINE_ID
else:
EXPECTED_ID = None

if status == 'RAISE':
connection = _TimeoutHTTPConnection()
else:
connection = _HTTPConnection(status, EXPECTED_ID)

with self._monkeyEnviron(None):
with self._monkeyImplicit(connection=connection):
if EXPECTED_ID is None:
self.assertRaises(EnvironmentError, self._callFUT)
else:
self._callFUT()

self.assertEqual(_implicit_environ.get_default_dataset_id(),
EXPECTED_ID)
self.assertEqual(connection.host, '169.254.169.254')
self.assertEqual(connection.timeout, 0.1)
self.assertEqual(
connection._called_args,
[('GET', '/computeMetadata/v1/project/project-id')])
expected_kwargs = {
'headers': {
'Metadata-Flavor': 'Google',
},
}
self.assertEqual(connection._called_kwargs, [expected_kwargs])
self.assertEqual(connection._close_count, 1)

def test_set_implicit_from_compute_engine(self):
self._implicit_compute_engine_helper(200)
def mock_determine(dataset_id):
_called_dataset_id.append(dataset_id)
return None

def test_set_implicit_from_compute_engine_bad_status(self):
self._implicit_compute_engine_helper(404)
with _Monkey(_implicit_environ,
_determine_default_dataset_id=mock_determine):
self.assertRaises(EnvironmentError, self._callFUT)

def test_set_implicit_from_compute_engine_raise_timeout(self):
self._implicit_compute_engine_helper('RAISE')
self.assertEqual(_called_dataset_id, [None])

def test_set_implicit_both_appengine_and_compute(self):
def test_set_correctly(self):
from gcloud._testing import _Monkey
from gcloud.datastore import _implicit_environ

APP_ENGINE_ID = 'GAE'
APP_IDENTITY = _AppIdentity(APP_ENGINE_ID)
connection = _HTTPConnection(200, 'GCE')
self.assertEqual(_implicit_environ._DEFAULTS.dataset_id, None)

with self._monkeyEnviron(None):
with self._monkeyImplicit(connection=connection,
app_identity=APP_IDENTITY):
self._callFUT()
DATASET_ID = object()
_called_dataset_id = []

self.assertEqual(_implicit_environ.get_default_dataset_id(),
APP_ENGINE_ID)
self.assertEqual(connection.host, None)
self.assertEqual(connection.timeout, None)
def mock_determine(dataset_id):
_called_dataset_id.append(dataset_id)
return DATASET_ID

def test_set_implicit_three_env_appengine_and_compute(self):
from gcloud.datastore import _implicit_environ

IMPLICIT_DATASET_ID = 'IMPLICIT'
APP_IDENTITY = _AppIdentity('GAE')
connection = _HTTPConnection(200, 'GCE')

with self._monkeyEnviron(IMPLICIT_DATASET_ID):
with self._monkeyImplicit(connection=connection,
app_identity=APP_IDENTITY):
self._callFUT()
with _Monkey(_implicit_environ,
_determine_default_dataset_id=mock_determine):
self._callFUT()

self.assertEqual(_implicit_environ.get_default_dataset_id(),
IMPLICIT_DATASET_ID)
self.assertEqual(connection.host, None)
self.assertEqual(connection.timeout, None)
self.assertEqual(_implicit_environ._DEFAULTS.dataset_id, DATASET_ID)
self.assertEqual(_called_dataset_id, [None])


class Test__lazy_property_deco(unittest2.TestCase):
Expand Down