Add 'Subscription.test_iam_permissions' API wrapper.

docs/pubsub-usage.rst

@@ -341,3 +341,15 @@ Update the IAM policy for a subscription:
    >>> policy = subscription.get_iam_policy()  # API request
    >>> policy.writers.add(policy.group('editors-list@example.com'))
    >>> subscription.set_iam_policy(policy)  # API request
+
+Test permissions allowed by the current IAM policy on a subscription:
+
+.. doctest::
+
+   >>> from gcloud import pubsub
+   >>> client = pubsub.Client()
+   >>> topic = client.topic('topic_name')
+   >>> subscription = topic.subscription('subscription_name')
+   >>> subscription.test_iam_permissions(
+   ...     ['roles/reader', 'roles/writer', 'roles/owner'])  # API request
+   ['roles/reader', 'roles/writer']

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_subscription.py

@@ -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):