Add 'Topic.test_iam_permissions' API wrapper.

docs/pubsub-usage.rst

@@ -94,6 +94,17 @@ Update the IAM policy for a topic:
    >>> policy.writers.add(policy.group('[email protected]'))
    >>> topic.set_iam_policy(policy)  # API request
 
+Test permissions allowed by the current IAM policy on a topic:
+
+.. doctest::
+
+   >>> from gcloud import pubsub
+   >>> client = pubsub.Client()
+   >>> topic = client.topic('topic_name')
+   >>> topic.test_iam_permissions(
+   ...     ['roles/reader', 'roles/writer', 'roles/owner'])  # API request
+   ['roles/reader', 'roles/writer']
+
 
 Publish messages to a topic
 ---------------------------

gcloud/pubsub/test_topic.py

@@ -601,6 +601,57 @@ 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):
+        TOPIC_NAME = 'topic_name'
+        PROJECT = 'PROJECT'
+        PATH = 'projects/%s/topics/%s:testIamPermissions' % (
+            PROJECT, TOPIC_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 = self._makeOne(TOPIC_NAME, client=CLIENT)
+
+        allowed = topic.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):
+        TOPIC_NAME = 'topic_name'
+        PROJECT = 'PROJECT'
+        PATH = 'projects/%s/topics/%s:testIamPermissions' % (
+            PROJECT, TOPIC_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 = self._makeOne(TOPIC_NAME, client=CLIENT1)
+
+        allowed = topic.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 TestBatch(unittest2.TestCase):
 

gcloud/pubsub/topic.py

@@ -303,6 +303,31 @@ def set_iam_policy(self, policy, client=None):
             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.topics/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 batch.
+
+        :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', ())
+
 
 class Batch(object):
     """Context manager:  collect messages to publish via a single API call.