#741: hoist 'get_scoped_connection' into '_helpers'.

Implement 'datastore._implicit_environ.get_connection' and
'storage.get_connection' in terms of it.

gcloud/_helpers.py

@@ -31,6 +31,8 @@ class Local(object):
 except ImportError:
     app_identity = None
 
+from gcloud import credentials
+
 
 class _LocalStack(Local):
     """Manage a thread-local LIFO stack of resources.
@@ -236,3 +238,20 @@ def __init__(self, project=None, implicit=False):
 
 
 _DEFAULTS = _DefaultsContainer(implicit=True)
+
+
+def get_scoped_connection(klass, scopes):
+    """Create a scoped connection to GCloud.
+
+    :type klass: type
+    :param klass: the specific ``Connection`` class to instantiate.
+
+    :type scopes: list of URLs
+    :param scopes: the effective service auth scopes for the connection.
+
+    :rtype: instance of ``klass``
+    :returns: A connection defined with the proper credentials.
+    """
+    implicit_credentials = credentials.get_credentials()
+    scoped_credentials = implicit_credentials.create_scoped(scopes)
+    return klass(credentials=scoped_credentials)

gcloud/datastore/_implicit_environ.py

@@ -23,7 +23,7 @@
 from gcloud._helpers import _app_engine_id
 from gcloud._helpers import _compute_engine_id
 from gcloud._helpers import _lazy_property_deco
-from gcloud import credentials
+from gcloud._helpers import get_scoped_connection
 from gcloud.datastore.connection import Connection
 
 
@@ -126,9 +126,7 @@ def get_connection():
     :rtype: :class:`gcloud.datastore.connection.Connection`
     :returns: A connection defined with the proper credentials.
     """
-    implicit_credentials = credentials.get_credentials()
-    scoped_credentials = implicit_credentials.create_scoped(SCOPE)
-    return Connection(credentials=scoped_credentials)
+    return get_scoped_connection(Connection, SCOPE)
 
 
 def set_default_connection(connection=None):

gcloud/datastore/test__implicit_environ.py

@@ -305,6 +305,7 @@ def _callFUT(self):
 
     def test_it(self):
         from gcloud import credentials
+        from gcloud.datastore._implicit_environ import SCOPE
         from gcloud.datastore.connection import Connection
         from gcloud.test_credentials import _Client
         from gcloud._testing import _Monkey
@@ -314,6 +315,7 @@ def test_it(self):
             found = self._callFUT()
         self.assertTrue(isinstance(found, Connection))
         self.assertTrue(found._credentials is client._signed)
+        self.assertEqual(found._credentials._scopes, SCOPE)
         self.assertTrue(client._get_app_default_called)
 
 

gcloud/storage/__init__.py

@@ -41,6 +41,7 @@
 import os
 
 from gcloud import credentials
+from gcloud._helpers import get_scoped_connection
 from gcloud._helpers import get_default_project
 from gcloud._helpers import set_default_project
 from gcloud.storage import _implicit_environ
@@ -133,6 +134,4 @@ def get_connection():
     :rtype: :class:`gcloud.storage.connection.Connection`
     :returns: A connection defined with the proper credentials.
     """
-    implicit_credentials = credentials.get_credentials()
-    scoped_credentials = implicit_credentials.create_scoped(SCOPE)
-    return Connection(credentials=scoped_credentials)
+    return get_scoped_connection(Connection, SCOPE)

gcloud/storage/test___init__.py

@@ -23,6 +23,7 @@ def _callFUT(self, *args, **kw):
 
     def test_it(self):
         from gcloud import credentials
+        from gcloud.storage import SCOPE
         from gcloud.storage.connection import Connection
         from gcloud.test_credentials import _Client
         from gcloud._testing import _Monkey
@@ -31,6 +32,7 @@ def test_it(self):
             found = self._callFUT()
         self.assertTrue(isinstance(found, Connection))
         self.assertTrue(found._credentials is client._signed)
+        self.assertEqual(found._credentials._scopes, SCOPE)
         self.assertTrue(client._get_app_default_called)
 
 

gcloud/test__helpers.py

@@ -302,6 +302,34 @@ def test_descriptor_for_project(self):
         self.assertTrue('project' in _helpers._DEFAULTS.__dict__)
 
 
+class Test_get_scoped_connection(unittest2.TestCase):
+
+    def _callFUT(self, klass, scopes):
+        from gcloud._helpers import get_scoped_connection
+        return get_scoped_connection(klass, scopes)
+
+    def test_it(self):
+        from gcloud import credentials
+        from gcloud.test_credentials import _Client
+        from gcloud._testing import _Monkey
+
+        class _Connection(object):
+            def __init__(self, credentials):
+                self._credentials = credentials
+
+        SCOPES = ('https://www.googleapis.com/auth/example',
+                  'https://www.googleapis.com/auth/userinfo.email')
+
+        client = _Client()
+        with _Monkey(credentials, client=client):
+            found = self._callFUT(_Connection, SCOPES)
+
+        self.assertTrue(isinstance(found, _Connection))
+        self.assertTrue(found._credentials is client._signed)
+        self.assertEqual(found._credentials._scopes, SCOPES)
+        self.assertTrue(client._get_app_default_called)
+
+
 class _AppIdentity(object):
 
     def __init__(self, app_id):