Add 'Subscription.test_iam_permissions' API wrapper.

docs/pubsub-iam.rst

@@ -3,6 +3,7 @@ IAM Policy
 
 .. automodule:: gcloud.pubsub.iam
   :members:
+  :member-order: bysource
   :undoc-members:
   :show-inheritance:
 

docs/pubsub-usage.rst

@@ -99,11 +99,13 @@ Test permissions allowed by the current IAM policy on a topic:
 .. doctest::
 
    >>> from gcloud import pubsub
+   >>> from gcloud.pubsub.iam import OWNER_ROLE, WRITER_ROLE, READER_ROLE
    >>> client = pubsub.Client()
    >>> topic = client.topic('topic_name')
-   >>> topic.test_iam_permissions(
-   ...     ['roles/reader', 'roles/writer', 'roles/owner'])  # API request
-   ['roles/reader', 'roles/writer']
+   >>> allowed = topic.test_iam_permissions(
+   ...     [READER_ROLE, WRITER_ROLE, OWNER_ROLE])  # API request
+   >>> allowed == [READER_ROLE, WRITER_ROLE]
+   True
 
 
 Publish messages to a topic
@@ -341,3 +343,17 @@ Update the IAM policy for a subscription:
    >>> policy = subscription.get_iam_policy()  # API request
    >>> policy.writers.add(policy.group('[email protected]'))
    >>> subscription.set_iam_policy(policy)  # API request
+
+Test permissions allowed by the current IAM policy on a subscription:
+
+.. doctest::
+
+   >>> from gcloud import pubsub
+   >>> from gcloud.pubsub.iam import OWNER_ROLE, WRITER_ROLE, READER_ROLE
+   >>> client = pubsub.Client()
+   >>> topic = client.topic('topic_name')
+   >>> subscription = topic.subscription('subscription_name')
+   >>> allowed = subscription.test_iam_permissions(
+   ...     [READER_ROLE, WRITER_ROLE, OWNER_ROLE])  # API request
+   >>> allowed == [READER_ROLE, WRITER_ROLE]
+   True

gcloud/pubsub/iam.py

@@ -13,9 +13,14 @@
 # limitations under the License.
 """PubSub API IAM policy definitions"""
 
-_OWNER_ROLE = 'roles/owner'
-_WRITER_ROLE = 'roles/writer'
-_READER_ROLE = 'roles/reader'
+OWNER_ROLE = 'roles/owner'
+"""IAM permission implying all rights to an object."""
+
+WRITER_ROLE = 'roles/writer'
+"""IAM permission implying rights to modify an object."""
+
+READER_ROLE = 'roles/reader'
+"""IAM permission implying rights to access an object without modifying it."""
 
 
 class Policy(object):
@@ -120,11 +125,11 @@ def from_api_repr(cls, resource):
         for binding in resource.get('bindings', ()):
             role = binding['role']
             members = set(binding['members'])
-            if role == _OWNER_ROLE:
+            if role == OWNER_ROLE:
                 policy.owners = members
-            elif role == _WRITER_ROLE:
+            elif role == WRITER_ROLE:
                 policy.writers = members
-            elif role == _READER_ROLE:
+            elif role == READER_ROLE:
                 policy.readers = members
             else:
                 raise ValueError('Unknown role: %s' % (role,))
@@ -148,15 +153,15 @@ def to_api_repr(self):
 
         if self.owners:
             bindings.append(
-                {'role': _OWNER_ROLE, 'members': sorted(self.owners)})
+                {'role': OWNER_ROLE, 'members': sorted(self.owners)})
 
         if self.writers:
             bindings.append(
-                {'role': _WRITER_ROLE, 'members': sorted(self.writers)})
+                {'role': WRITER_ROLE, 'members': sorted(self.writers)})
 
         if self.readers:
             bindings.append(
-                {'role': _READER_ROLE, 'members': sorted(self.readers)})
+                {'role': READER_ROLE, 'members': sorted(self.readers)})
 
         if bindings:
             resource['bindings'] = bindings

gcloud/pubsub/subscription.py

@@ -308,3 +308,28 @@ def set_iam_policy(self, policy, client=None):
         resp = client.connection.api_request(
             method='POST', path=path, data=resource)
         return Policy.from_api_repr(resp)
+
+    def test_iam_permissions(self, permissions, client=None):
+        """Permissions allowed for the current user by the effective IAM policy.
+
+        See:
+        https://cloud.google.com/pubsub/reference/rest/v1/projects.subscriptions/testIamPermissions
+
+        :type permissions: list of string
+        :param permissions: list of permissions to be tested
+
+        :type client: :class:`gcloud.pubsub.client.Client` or ``NoneType``
+        :param client: the client to use.  If not passed, falls back to the
+                       ``client`` stored on the current subscription's topic.
+
+        :rtype: sequence of string
+        :returns: subset of ``permissions`` allowed by current IAM policy.
+        """
+        client = self._require_client(client)
+        path = '%s:testIamPermissions' % (self.path,)
+        data = {
+            'permissions': list(permissions),
+        }
+        resp = client.connection.api_request(
+            method='POST', path=path, data=data)
+        return resp.get('permissions', ())

gcloud/pubsub/test_iam.py

@@ -87,7 +87,7 @@ def test_from_api_repr_only_etag(self):
         self.assertEqual(list(policy.readers), [])
 
     def test_from_api_repr_complete(self):
-        from gcloud.pubsub.iam import _OWNER_ROLE, _WRITER_ROLE, _READER_ROLE
+        from gcloud.pubsub.iam import OWNER_ROLE, WRITER_ROLE, READER_ROLE
         OWNER1 = 'user:[email protected]'
         OWNER2 = 'group:[email protected]'
         WRITER1 = 'domain:google.com'
@@ -98,9 +98,9 @@ def test_from_api_repr_complete(self):
             'etag': 'DEADBEEF',
             'version': 17,
             'bindings': [
-                {'role': _OWNER_ROLE, 'members': [OWNER1, OWNER2]},
-                {'role': _WRITER_ROLE, 'members': [WRITER1, WRITER2]},
-                {'role': _READER_ROLE, 'members': [READER1, READER2]},
+                {'role': OWNER_ROLE, 'members': [OWNER1, OWNER2]},
+                {'role': WRITER_ROLE, 'members': [WRITER1, WRITER2]},
+                {'role': READER_ROLE, 'members': [READER1, READER2]},
             ],
         }
         klass = self._getTargetClass()
@@ -134,7 +134,7 @@ def test_to_api_repr_only_etag(self):
         self.assertEqual(policy.to_api_repr(), {'etag': 'DEADBEEF'})
 
     def test_to_api_repr_full(self):
-        from gcloud.pubsub.iam import _OWNER_ROLE, _WRITER_ROLE, _READER_ROLE
+        from gcloud.pubsub.iam import OWNER_ROLE, WRITER_ROLE, READER_ROLE
         OWNER1 = 'group:[email protected]'
         OWNER2 = 'user:[email protected]'
         WRITER1 = 'domain:google.com'
@@ -145,9 +145,9 @@ def test_to_api_repr_full(self):
             'etag': 'DEADBEEF',
             'version': 17,
             'bindings': [
-                {'role': _OWNER_ROLE, 'members': [OWNER1, OWNER2]},
-                {'role': _WRITER_ROLE, 'members': [WRITER1, WRITER2]},
-                {'role': _READER_ROLE, 'members': [READER1, READER2]},
+                {'role': OWNER_ROLE, 'members': [OWNER1, OWNER2]},
+                {'role': WRITER_ROLE, 'members': [WRITER1, WRITER2]},
+                {'role': READER_ROLE, 'members': [READER1, READER2]},
             ],
         }
         policy = self._makeOne('DEADBEEF', 17)

gcloud/pubsub/test_subscription.py

@@ -485,7 +485,7 @@ def test_delete_w_alternate_client(self):
         self.assertEqual(req['path'], '/%s' % SUB_PATH)
 
     def test_get_iam_policy_w_bound_client(self):
-        from gcloud.pubsub.iam import _OWNER_ROLE, _WRITER_ROLE, _READER_ROLE
+        from gcloud.pubsub.iam import OWNER_ROLE, WRITER_ROLE, READER_ROLE
         OWNER1 = 'user:[email protected]'
         OWNER2 = 'group:[email protected]'
         WRITER1 = 'domain:google.com'
@@ -496,9 +496,9 @@ def test_get_iam_policy_w_bound_client(self):
             'etag': 'DEADBEEF',
             'version': 17,
             'bindings': [
-                {'role': _OWNER_ROLE, 'members': [OWNER1, OWNER2]},
-                {'role': _WRITER_ROLE, 'members': [WRITER1, WRITER2]},
-                {'role': _READER_ROLE, 'members': [READER1, READER2]},
+                {'role': OWNER_ROLE, 'members': [OWNER1, OWNER2]},
+                {'role': WRITER_ROLE, 'members': [WRITER1, WRITER2]},
+                {'role': READER_ROLE, 'members': [READER1, READER2]},
             ],
         }
         PROJECT = 'PROJECT'
@@ -557,7 +557,7 @@ def test_get_iam_policy_w_alternate_client(self):
         self.assertEqual(req['path'], '/%s' % PATH)
 
     def test_set_iam_policy_w_bound_client(self):
-        from gcloud.pubsub.iam import _OWNER_ROLE, _WRITER_ROLE, _READER_ROLE
+        from gcloud.pubsub.iam import OWNER_ROLE, WRITER_ROLE, READER_ROLE
         from gcloud.pubsub.iam import Policy
         OWNER1 = 'group:[email protected]'
         OWNER2 = 'user:[email protected]'
@@ -569,9 +569,9 @@ def test_set_iam_policy_w_bound_client(self):
             'etag': 'DEADBEEF',
             'version': 17,
             'bindings': [
-                {'role': _OWNER_ROLE, 'members': [OWNER1, OWNER2]},
-                {'role': _WRITER_ROLE, 'members': [WRITER1, WRITER2]},
-                {'role': _READER_ROLE, 'members': [READER1, READER2]},
+                {'role': OWNER_ROLE, 'members': [OWNER1, OWNER2]},
+                {'role': WRITER_ROLE, 'members': [WRITER1, WRITER2]},
+                {'role': READER_ROLE, 'members': [READER1, READER2]},
             ],
         }
         RESPONSE = POLICY.copy()
@@ -641,6 +641,61 @@ def test_set_iam_policy_w_alternate_client(self):
         self.assertEqual(req['path'], '/%s' % PATH)
         self.assertEqual(req['data'], {})
 
+    def test_test_iam_permissions_w_bound_client(self):
+        PROJECT = 'PROJECT'
+        TOPIC_NAME = 'topic_name'
+        SUB_NAME = 'sub_name'
+        PATH = 'projects/%s/subscriptions/%s:testIamPermissions' % (
+            PROJECT, SUB_NAME)
+        ROLES = ['roles/reader', 'roles/writer', 'roles/owner']
+        REQUESTED = {
+            'permissions': ROLES,
+        }
+        RESPONSE = {
+            'permissions': ROLES[:-1],
+        }
+        conn = _Connection(RESPONSE)
+        CLIENT = _Client(project=PROJECT, connection=conn)
+        topic = _Topic(TOPIC_NAME, client=CLIENT)
+        subscription = self._makeOne(SUB_NAME, topic)
+
+        allowed = subscription.test_iam_permissions(ROLES)
+
+        self.assertEqual(allowed, ROLES[:-1])
+        self.assertEqual(len(conn._requested), 1)
+        req = conn._requested[0]
+        self.assertEqual(req['method'], 'POST')
+        self.assertEqual(req['path'], '/%s' % PATH)
+        self.assertEqual(req['data'], REQUESTED)
+
+    def test_test_iam_permissions_w_alternate_client(self):
+        PROJECT = 'PROJECT'
+        TOPIC_NAME = 'topic_name'
+        SUB_NAME = 'sub_name'
+        PATH = 'projects/%s/subscriptions/%s:testIamPermissions' % (
+            PROJECT, SUB_NAME)
+        ROLES = ['roles/reader', 'roles/writer', 'roles/owner']
+        REQUESTED = {
+            'permissions': ROLES,
+        }
+        RESPONSE = {}
+        conn1 = _Connection()
+        CLIENT1 = _Client(project=PROJECT, connection=conn1)
+        conn2 = _Connection(RESPONSE)
+        CLIENT2 = _Client(project=PROJECT, connection=conn2)
+        topic = _Topic(TOPIC_NAME, client=CLIENT1)
+        subscription = self._makeOne(SUB_NAME, topic)
+
+        allowed = subscription.test_iam_permissions(ROLES, client=CLIENT2)
+
+        self.assertEqual(len(allowed), 0)
+        self.assertEqual(len(conn1._requested), 0)
+        self.assertEqual(len(conn2._requested), 1)
+        req = conn2._requested[0]
+        self.assertEqual(req['method'], 'POST')
+        self.assertEqual(req['path'], '/%s' % PATH)
+        self.assertEqual(req['data'], REQUESTED)
+
 
 class _Connection(object):
 

gcloud/pubsub/test_topic.py

@@ -453,7 +453,7 @@ def test_list_subscriptions_missing_key(self):
         self.assertEqual(req['query_params'], {})
 
     def test_get_iam_policy_w_bound_client(self):
-        from gcloud.pubsub.iam import _OWNER_ROLE, _WRITER_ROLE, _READER_ROLE
+        from gcloud.pubsub.iam import OWNER_ROLE, WRITER_ROLE, READER_ROLE
         OWNER1 = 'user:[email protected]'
         OWNER2 = 'group:[email protected]'
         WRITER1 = 'domain:google.com'
@@ -464,9 +464,9 @@ def test_get_iam_policy_w_bound_client(self):
             'etag': 'DEADBEEF',
             'version': 17,
             'bindings': [
-                {'role': _OWNER_ROLE, 'members': [OWNER1, OWNER2]},
-                {'role': _WRITER_ROLE, 'members': [WRITER1, WRITER2]},
-                {'role': _READER_ROLE, 'members': [READER1, READER2]},
+                {'role': OWNER_ROLE, 'members': [OWNER1, OWNER2]},
+                {'role': WRITER_ROLE, 'members': [WRITER1, WRITER2]},
+                {'role': READER_ROLE, 'members': [READER1, READER2]},
             ],
         }
         TOPIC_NAME = 'topic_name'
@@ -522,7 +522,7 @@ def test_get_iam_policy_w_alternate_client(self):
 
     def test_set_iam_policy_w_bound_client(self):
         from gcloud.pubsub.iam import Policy
-        from gcloud.pubsub.iam import _OWNER_ROLE, _WRITER_ROLE, _READER_ROLE
+        from gcloud.pubsub.iam import OWNER_ROLE, WRITER_ROLE, READER_ROLE
         OWNER1 = 'group:[email protected]'
         OWNER2 = 'user:[email protected]'
         WRITER1 = 'domain:google.com'
@@ -533,9 +533,9 @@ def test_set_iam_policy_w_bound_client(self):
             'etag': 'DEADBEEF',
             'version': 17,
             'bindings': [
-                {'role': _OWNER_ROLE, 'members': [OWNER1, OWNER2]},
-                {'role': _WRITER_ROLE, 'members': [WRITER1, WRITER2]},
-                {'role': _READER_ROLE, 'members': [READER1, READER2]},
+                {'role': OWNER_ROLE, 'members': [OWNER1, OWNER2]},
+                {'role': WRITER_ROLE, 'members': [WRITER1, WRITER2]},
+                {'role': READER_ROLE, 'members': [READER1, READER2]},
             ],
         }
         RESPONSE = POLICY.copy()
@@ -602,12 +602,12 @@ def test_set_iam_policy_w_alternate_client(self):
         self.assertEqual(req['data'], {})
 
     def test_test_iam_permissions_w_bound_client(self):
-        from gcloud.pubsub.iam import _OWNER_ROLE, _WRITER_ROLE, _READER_ROLE
+        from gcloud.pubsub.iam import OWNER_ROLE, WRITER_ROLE, READER_ROLE
         TOPIC_NAME = 'topic_name'
         PROJECT = 'PROJECT'
         PATH = 'projects/%s/topics/%s:testIamPermissions' % (
             PROJECT, TOPIC_NAME)
-        ROLES = [_READER_ROLE, _WRITER_ROLE, _OWNER_ROLE]
+        ROLES = [READER_ROLE, WRITER_ROLE, OWNER_ROLE]
         REQUESTED = {
             'permissions': ROLES,
         }
@@ -628,12 +628,12 @@ def test_test_iam_permissions_w_bound_client(self):
         self.assertEqual(req['data'], REQUESTED)
 
     def test_test_iam_permissions_w_alternate_client(self):
-        from gcloud.pubsub.iam import _OWNER_ROLE, _WRITER_ROLE, _READER_ROLE
+        from gcloud.pubsub.iam import OWNER_ROLE, WRITER_ROLE, READER_ROLE
         TOPIC_NAME = 'topic_name'
         PROJECT = 'PROJECT'
         PATH = 'projects/%s/topics/%s:testIamPermissions' % (
             PROJECT, TOPIC_NAME)
-        ROLES = [_READER_ROLE, _WRITER_ROLE, _OWNER_ROLE]
+        ROLES = [READER_ROLE, WRITER_ROLE, OWNER_ROLE]
         REQUESTED = {
             'permissions': ROLES,
         }