Add 'pubsub.api.list_subscriptions'.

Author: tseaver

gcloud/pubsub/api.py

@@ -52,3 +52,49 @@ def list_topics(page_size=None, page_token=None,
 
     path = '/projects/%s/topics' % project
     return connection.api_request(method='GET', path=path, query_params=params)
+
+
+def list_subscriptions(page_size=None, page_token=None, topic_name=None,
+                       project=None, connection=None):
+    """List subscriptions for a given project.
+
+    :type page_size: int
+    :param page_size: maximum number of topics to return, If not passed,
+                      defaults to a value set by the API.
+
+    :type page_token: string
+    :param page_token: opaque marker for the next "page" of topics. If not
+                       passed, the API will return the first page of topics.
+
+    :type topic_name: string
+    :param topic_name: limit results to subscriptions bound to the given topic.
+
+    :type project: string
+    :param project: project ID to query.  If not passed, defaults to the
+                    project ID inferred from the environment.
+
+    :type connection: :class:`gcloud.pubsub.connection.Connection`
+    :param connection: connection to use for the query.  If not passed,
+                       defaults to the connection inferred from the
+                       environment.
+
+    :rtype: dict
+    :returns: keys include ``subscriptions`` (a list of subscription mappings)
+              and ``nextPageToken`` (a string:  if non-empty, indicates that
+              more topics can be retrieved with another call (pass that
+              value as ``page_token``).
+    """
+    params = {}
+
+    if page_size is not None:
+        params['pageSize'] = page_size
+
+    if page_token is not None:
+        params['pageToken'] = page_token
+
+    if topic_name is None:
+        path = '/projects/%s/subscriptions' % project
+    else:
+        path = '/projects/%s/topics/%s/subscriptions' % (project, topic_name)
+
+    return connection.api_request(method='GET', path=path, query_params=params)

gcloud/pubsub/test_api.py

@@ -61,6 +61,86 @@ def test_w_explicit_connection_w_paging(self):
                          {'pageSize': SIZE, 'pageToken': TOKEN1})
 
 
+class Test_list_subscriptions(unittest2.TestCase):
+
+    def _callFUT(self, *args, **kw):
+        from gcloud.pubsub.api import list_subscriptions
+        return list_subscriptions(*args, **kw)
+
+    def test_w_explicit_connection_no_paging(self):
+        PROJECT = 'PROJECT'
+        SUB_NAME = 'topic_name'
+        SUB_PATH = 'projects/%s/subscriptions/%s' % (PROJECT, SUB_NAME)
+        TOPIC_NAME = 'topic_name'
+        TOPIC_PATH = 'projects/%s/topics/%s' % (PROJECT, TOPIC_NAME)
+        TOKEN = 'TOKEN'
+        returned = {'subscriptions': [{'name': SUB_PATH, 'topic': TOPIC_PATH}],
+                    'nextPageToken': TOKEN}
+        conn = _Connection(returned)
+        response = self._callFUT(project=PROJECT, connection=conn)
+        subscriptions = response['subscriptions']
+        self.assertEqual(len(subscriptions), 1)
+        self.assertEqual(subscriptions[0],
+                         {'name': SUB_PATH, 'topic': TOPIC_PATH})
+        self.assertEqual(response['nextPageToken'], TOKEN)
+        self.assertEqual(len(conn._requested), 1)
+        req = conn._requested[0]
+        self.assertEqual(req['method'], 'GET')
+        self.assertEqual(req['path'], '/projects/%s/subscriptions' % PROJECT)
+        self.assertEqual(req['query_params'], {})
+
+    def test_w_explicit_connection_w_paging(self):
+        PROJECT = 'PROJECT'
+        SUB_NAME = 'topic_name'
+        SUB_PATH = 'projects/%s/subscriptions/%s' % (PROJECT, SUB_NAME)
+        TOPIC_NAME = 'topic_name'
+        TOPIC_PATH = 'projects/%s/topics/%s' % (PROJECT, TOPIC_NAME)
+        TOKEN1 = 'TOKEN1'
+        TOKEN2 = 'TOKEN2'
+        SIZE = 1
+        returned = {'subscriptions': [{'name': SUB_PATH, 'topic': TOPIC_PATH}],
+                    'nextPageToken': TOKEN2}
+        conn = _Connection(returned)
+        response = self._callFUT(SIZE, TOKEN1,
+                                 project=PROJECT, connection=conn)
+        subscriptions = response['subscriptions']
+        self.assertEqual(len(subscriptions), 1)
+        self.assertEqual(subscriptions[0],
+                         {'name': SUB_PATH, 'topic': TOPIC_PATH})
+        self.assertEqual(response['nextPageToken'], TOKEN2)
+        self.assertEqual(len(conn._requested), 1)
+        req = conn._requested[0]
+        self.assertEqual(req['method'], 'GET')
+        self.assertEqual(req['path'], '/projects/%s/subscriptions' % PROJECT)
+        self.assertEqual(req['query_params'],
+                         {'pageSize': SIZE, 'pageToken': TOKEN1})
+
+    def test_w_topic_name(self):
+        PROJECT = 'PROJECT'
+        SUB_NAME = 'topic_name'
+        SUB_PATH = 'projects/%s/subscriptions/%s' % (PROJECT, SUB_NAME)
+        TOPIC_NAME = 'topic_name'
+        TOPIC_PATH = 'projects/%s/topics/%s' % (PROJECT, TOPIC_NAME)
+        TOKEN = 'TOKEN'
+        returned = {'subscriptions': [{'name': SUB_PATH, 'topic': TOPIC_PATH}],
+                    'nextPageToken': TOKEN}
+        conn = _Connection(returned)
+        response = self._callFUT(topic_name=TOPIC_NAME,
+                                 project=PROJECT, connection=conn)
+        subscriptions = response['subscriptions']
+        self.assertEqual(len(subscriptions), 1)
+        self.assertEqual(subscriptions[0],
+                         {'name': SUB_PATH, 'topic': TOPIC_PATH})
+        self.assertEqual(response['nextPageToken'], TOKEN)
+        self.assertEqual(len(conn._requested), 1)
+        req = conn._requested[0]
+        self.assertEqual(req['method'], 'GET')
+        self.assertEqual(req['path'],
+                         '/projects/%s/topics/%s/subscriptions'
+                         % (PROJECT, TOPIC_NAME))
+        self.assertEqual(req['query_params'], {})
+
+
 class _Connection(object):
 
     def __init__(self, *responses):