Add support for collection paging - mostly a matter of passing through the kwargs in a few places plus writing a bunch of new units for the feature.

Author: blovering

splunklib/client.py

@@ -221,8 +221,8 @@ def __init__(self, service, path, item=None, ctor=None, dtor=None):
         self.ctor = ctor # Item constructor
         self.dtor = dtor # Item desteructor
 
-    def __call__(self):
-        return self.list()
+    def __call__(self, **kwargs):
+        return self.list(**kwargs)
 
     def __contains__(self, name):
         return self.contains(name)
@@ -277,10 +277,9 @@ def itemmeta(self):
             'eai:attributes': content['eai:attributes']
         }
 
-    def list(self, **kwargs):
-        """Returns a list of collection members."""
-        count = kwargs.get('count', -1)
-        response = self.get(count=count)
+    # kwargs: count, offset, search, sort_dir, sort_key, sort_mode
+    def list(self, count=0, **kwargs):
+        response = self.get(count=count, **kwargs)
         return self._load_list(response)
 
 # kwargs: path, app, owner, sharing, state
@@ -535,7 +534,7 @@ def list(self, *args):
         for kind in kinds:
             response = None
             try:
-                response = self.service.get(self.kindpath(kind), count=-1)
+                response = self.service.get(self.kindpath(kind), count=0)
             except HTTPError as e:
                 if e.status == 404: 
                     continue # No inputs of this kind
@@ -643,17 +642,16 @@ class Jobs(Collection):
     def __init__(self, service):
         Collection.__init__(self, service, PATH_JOBS, item=Job)
 
+    def __iter__(self):
+        for item in self.list(): yield item
+
     def create(self, query, **kwargs):
         response = self.post(search=query, **kwargs)
         if kwargs.get("exec_mode", None) == "oneshot":
             return response.body
         sid = _load_atom(response).response.sid
         return Job(self.service, PATH_JOBS + sid)
 
-    def list(self, **kwargs):
-        """Returns a list of search Job entities."""
-        return self._load_list(self.get()) # No count argument allowed
-
 class Message(Entity):
     def __init__(self, service, name, **kwargs):
         Entity.__init__(self, service, _path(PATH_MESSAGES, name), **kwargs)

tests/runtests.py

@@ -24,6 +24,7 @@
 files = [
     "test_data.py",
     "test_binding.py",
+    "test_collection.py",
     "test_client.py",
     "test_examples.py",
 ]

tests/test_collection.py

@@ -0,0 +1,112 @@
+#!/usr/bin/env python
+#
+# Copyright 2011-2012 Splunk, Inc.
+#
+# Licensed under the Apache License, Version 2.0 (the "License"): you may
+# not use this file except in compliance with the License. You may obtain
+# a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+# License for the specific language governing permissions and limitations
+# under the License.
+
+import sys
+import unittest
+
+import splunklib.client as client
+from utils import parse
+
+opts = None # Command line options
+
+class TestCase(unittest.TestCase):
+    def setUp(self):
+        self.service = client.connect(**opts.kwargs)
+
+    # Verify that the given collections interface behaves as expected
+    def check_collection(self, collection):
+        # Check various collection options
+        collection.list() # Default
+        collection.list(search="title=*")
+        collection.list(sort_dir="asc")
+        collection.list(sort_dir="desc")
+        collection.list(sort_mode="auto")
+        collection.list(sort_mode="alpha")
+        collection.list(sort_mode="alpha_case")
+        collection.list(sort_mode="num")
+
+        # Retrieve the entire list
+        items = collection.list()
+        total = len(items)
+
+        # Make sure the default list method returns all items
+        items0 = collection.list(count=0)
+        total0 = len(items0)
+        self.assertEqual(total, total0)
+
+        self.check_iterable(collection, total)
+
+        # Page through contents one-at-a-time and check count
+        count = 0
+        for i in xrange(total):
+            item = collection.list(offset=i, count=1)
+            count += 1
+        self.assertEqual(count, total)
+
+        # Page through the collection using various page sizes and make sure
+        # the expected paging invariants hold.
+        page_size = int(total/2)
+        while page_size > 0:
+            offset = 0
+            while offset < total:
+                page = collection.list(offset=offset, count=page_size)
+                count = len(page)
+                offset += count
+                self.assertTrue(count == page_size or offset == total)
+            self.assertEqual(offset, total)
+            page_size = int(page_size/2) # Try half the page size
+
+    # Verify that the given collection's iterator works as expected.
+    def check_iterable(self, collection, count):
+        # Iterate contents and make sure we see the expected count.
+        seen = 0
+        for item in collection: 
+            seen += 1
+        self.assertEqual(seen, count)
+
+    def test_apps(self):
+        self.check_collection(self.service.apps)
+
+    def test_indexes(self):
+        self.check_collection(self.service.indexes)
+
+    def test_inputs(self):
+        # The Inputs collection is an aggregated view of the various REST API
+        # input endpoints, and does not support the paging interface.
+        count = len(self.service.inputs.list())
+        self.check_iterable(self.service.inputs, count)
+
+    def test_jobs(self):
+        # The Jobs REST API endpoint does not support the paging interface.
+        count = len(self.service.jobs.list())
+        self.check_iterable(self.service.jobs, count)
+
+    def test_loggers(self):
+        self.check_collection(self.service.loggers)
+
+    def test_messages(self):
+        self.check_collection(self.service.messages)
+
+    def test_roles(self):
+        self.check_collection(self.service.roles)
+
+    def test_users(self):
+        self.check_collection(self.service.users)
+
+if __name__ == "__main__":
+    opts = parse(sys.argv[1:], {}, ".splunkrc")
+    unittest.main(argv=sys.argv[:1])
+