Fix set_priority tests on job. Refactor all tests to use assertEventuallyEqual and assertEventuallyTrue instead of previous forms.

Add better control of restarts in the suite and checking if restarts are required.

Author: Frederick Ross

splunklib/client.py

@@ -446,6 +446,19 @@ def restart(self, timeout=None):
                 sleep(2)
         raise Exception, "Operation timed out."
 
+    @property
+    def restart_required(self):
+        """Is splunkd in a state that requires a restart?"""
+        response = self.get("messages").body.read()
+        messages = data.load(response)['feed']
+        if 'entry' not in messages:
+            titles = []
+        elif isinstance(messages['entry'], dict):
+            titles = [messages['entry']['title']]
+        else:
+            titles = [x['title'] for x in messages['entry']]
+        return 'restart_required' in titles
+
     @property
     def roles(self):
         """Returns a collection of user roles."""
@@ -2040,7 +2053,7 @@ def refresh(self, state=None):
                 raw_state = self._load_state(response)
                 raw_state['links'] = dict([(k, urllib.unquote(v)) for k,v in raw_state['links'].iteritems()])
                 self._state = raw_state
-                return self
+        return self
 
     def results(self, timeout=None, wait_time=1, **query_params):
         """Fetch search results as an InputStream IO handle.
@@ -2151,6 +2164,10 @@ def searchlog(self, **kwargs):
     def set_priority(self, value):
         """Sets this job's search priority in the range of 0-10.
 
+        Higher numbers indicate higher priority. Unless splunkd is
+        running as root, you can only decrease the priority of
+        a running job.
+
         :param `value`: The search priority.
         """
         self.post('control', action="setpriority", priority=value)

tests/test_app.py

@@ -23,6 +23,7 @@
 class TestApp(testlib.SDKTestCase):
     app = None
     app_name = None
+
     def setUp(self):
         super(TestApp, self).setUp()
         if self.app is None:
@@ -39,13 +40,17 @@ def setUp(self):
         else:
             logging.debug("App %s already exists. Skipping creation.", self.app_name)
 
-    @classmethod
-    def tearDownClass(cls):
-        import splunklib.client as client
-        service = client.connect(**cls.opts.kwargs)
-        for app in service.apps:
-            if app.name.startswith('delete-me'):
-                service.apps.delete(app.name)
+    def tearDown(self):
+        super(TestApp, self).tearDown()
+        # The rest of this will leave Splunk in a state requiring a restart.
+        # It doesn't actually matter, though.
+        self.service = client.connect(**self.opts.kwargs)
+        for app in self.service.apps:
+            app_name = app.name
+            if app_name.startswith('delete-me'):
+                self.service.apps.delete(app_name)
+                self.assertEventuallyTrue(lambda: app_name not in self.service.apps)
+        self.clearRestartMessage()
 
     def test_app_integrity(self):
         self.check_entity(self.app)
@@ -81,6 +86,7 @@ def test_delete(self):
         self.assertTrue(name in self.service.apps)
         self.service.apps.delete(name)
         self.assertFalse(name in self.service.apps)
+        self.clearRestartMessage() # We don't actually have to restart here.
 
     def test_package(self):
         p = self.app.package()

tests/test_conf.py

@@ -79,7 +79,7 @@ def test_confs(self):
         key = testlib.tmpname()
         val = testlib.tmpname()
         stanza.update(**{key: val})
-        testlib.retry(stanza, lambda s: len(s), 1, step=0.2)
+        self.assertEventuallyEqual(1, lambda: stanza.refresh() and len(stanza), pause_time=0.2)
         self.assertEqual(len(stanza), 1)
         self.assertTrue(key in stanza)
 

tests/test_examples.py

@@ -283,7 +283,7 @@ def test_analytics(self):
         tracker.track("test_event", distinct_id="123abc", abc="12345")
 
         # Wait until the events get indexed
-        testlib.wait(index, lambda index: index['totalEventCount'] == '2')
+        self.assertEventuallyEqual('2', lambda: index.refresh()['totalEventCount'])
 
         # Now, we create a retriever to retrieve the events
         retriever = analytics.output.AnalyticsRetriever(

tests/test_fired_alert.py

@@ -59,18 +59,22 @@ def test_new_search_is_empty(self):
     def test_alerts_on_events(self):
         self.assertEqual(self.saved_search.alert_count, 0)
         self.assertEqual(len(self.saved_search.fired_alerts), 0)
+
         self.index.enable()
-        testlib.retry(self.index, 'disabled', '0', step=0.5, times=50)
+        self.assertEventuallyEqual('0', lambda: self.index.refresh() and self.index['disabled'], timeout=25)
+
         eventCount = int(self.index['totalEventCount'])
         self.assertEqual(self.index['sync'], '0')
         self.assertEqual(self.index['disabled'], '0')
         self.index.refresh()
         self.index.submit('This is a test ' + testlib.tmpname(),
                           sourcetype='sdk_use', host='boris')
-        testlib.retry(self.index, 'totalEventCount', str(eventCount+1), step=1, times=50)
-        self.assertEqual(self.index['totalEventCount'], str(eventCount+1))
-        testlib.retry(self.saved_search, lambda s: s.alert_count, 1, step=2, times=100)
-        self.assertEqual(self.saved_search.alert_count, 1)
+        self.assertEventuallyEqual(str(eventCount+1), lambda: self.index.refresh() and self.index['totalEventCount'], timeout=50)
+        self.assertEventuallyEqual(
+            1,
+            lambda: self.saved_search.refresh() and self.saved_search.alert_count,
+            timeout=200
+        )
         alerts = self.saved_search.fired_alerts
         self.assertEqual(len(alerts), 1)
 

tests/test_index.py

@@ -27,6 +27,7 @@ def setUp(self):
         super(IndexTest, self).setUp()
         self.index_name = testlib.tmpname()
         self.index = self.service.indexes.create(self.index_name)
+        self.assertEventuallyEqual('0', lambda: self.index.refresh()['disabled'])
 
     def tearDown(self):
         super(IndexTest, self).tearDown()
@@ -36,12 +37,17 @@ def tearDown(self):
         # clashes, though.
         if self.service.splunk_version >= (5,):
             self.service.indexes.delete(self.index_name)
+            self.assertEventuallyTrue(lambda: self.index_name not in self.service.indexes)
         else:
             logging.warning("test_index.py:TestDeleteIndex: Skipped: cannot "
                             "delete indexes via the REST API in Splunk 4.x")
 
 
 class IndexWithoutRestart(IndexTest):
+    def totalEventCount(self):
+        self.index.refresh()
+        return int(self.index['totalEventCount'])
+
     def test_integrity(self):
         self.check_entity(self.index)
 
@@ -62,55 +68,41 @@ def test_submit_and_clean(self):
 
         originalCount = int(self.index['totalEventCount'])
         self.index.submit("Hello again!", sourcetype="Boris", host="meep")
-        testlib.retry(self.index, 'totalEventCount', str(originalCount+1), step=1)
-        self.assertEqual(self.index['totalEventCount'], str(originalCount+1))
-
+        self.assertEventuallyEqual(originalCount+1, self.totalEventCount)
         self.index.clean(timeout=500)
         self.assertEqual(self.index['totalEventCount'], '0')
 
-class IndexWithRestartTest(IndexTest):
-    def setUp(self):
-        super(IndexWithRestartTest, self).setUp()
-        self.service.restart(timeout=300)
-        self.index = self.service.indexes[self.index_name]
-
     def test_prefresh(self):
-        index = self.service.indexes[self.index_name]
         self.assertEqual(self.index['disabled'], '0') # Index is prefreshed
 
     def test_submit(self):
         eventCount = int(self.index['totalEventCount'])
         self.assertEqual(self.index['sync'], '0')
         self.assertEqual(self.index['disabled'], '0')
         self.index.submit("Hello again!", sourcetype="Boris", host="meep")
-        testlib.retry(self.index, 'totalEventCount', str(eventCount+1), step=1)
-        self.assertEqual(self.index['totalEventCount'], str(eventCount+1))
+        self.assertEventuallyEqual(eventCount+1, self.totalEventCount)
 
     def test_submit_via_attach(self):
         eventCount = int(self.index['totalEventCount'])
         cn = self.index.attach()
         cn.send("Hello Boris!\r\n")
         cn.close()
-        testlib.retry(self.index, 'totalEventCount', str(eventCount+1), step=1)
-        self.index.refresh()
-        self.assertEqual(self.index['totalEventCount'], str(eventCount+1))
+        self.assertEventuallyEqual(eventCount+1, self.totalEventCount)
 
     def test_submit_via_attached_socket(self):
         eventCount = int(self.index['totalEventCount'])
         f = self.index.attached_socket
         with f() as sock:
             sock.send('Hello world!\r\n')
-        testlib.retry(self.index, 'totalEventCount', str(eventCount+1), step=1)
-        self.assertEqual(self.index['totalEventCount'], str(eventCount+1))
+        self.assertEventuallyEqual(eventCount+1, self.totalEventCount)
 
     def test_upload(self):
         # The following test must run on machine where splunkd runs,
         # otherwise a failure is expected
         eventCount = int(self.index['totalEventCount'])
         testpath = path.dirname(path.abspath(__file__))
         self.index.upload(path.join(testpath, "testfile.txt"))
-        testlib.retry(self.index, 'totalEventCount', str(eventCount+1), step=1)
-        self.assertEqual(self.index['totalEventCount'], str(eventCount+1))
+        self.assertEventuallyEqual(eventCount+1, self.totalEventCount)
 
 if __name__ == "__main__":
     testlib.main()

tests/test_input.py

@@ -61,14 +61,16 @@ def test_inputs_list_on_one_kind_with_search(self):
     def test_oneshot(self):
         index_name = testlib.tmpname()
         index = self.service.indexes.create(index_name)
-        self.service.restart(timeout=120)
+        self.restartSplunk(timeout=120)
         index = self.service.indexes[index_name]
         eventCount = int(index['totalEventCount'])
         from os import path
         testpath = path.dirname(path.abspath(__file__))
         self.service.inputs.oneshot(path.join(testpath, 'testfile.txt'), index=index_name)
-        testlib.retry(index, 'totalEventCount', str(eventCount+1), step=1)
-        self.assertEqual(index['totalEventCount'], str(eventCount+1))
+        self.assertEventuallyEqual(
+            str(eventCount+1),
+            lambda: index.refresh()['totalEventCount']
+        )
 
     def test_oneshot_on_nonexistant_file(self):
         name = testlib.tmpname()

tests/test_job.py

@@ -88,34 +88,6 @@ def test_cancel_is_idempotent(self):
         job.cancel()
         job.cancel() # Second call should be nop
 
-    def test_enable_preview(self):
-        query = "search index=_internal"
-        job = self.service.jobs.create(
-            query=query,
-            earliest_time="-1m",
-            latest_time="now")
-        self.assertEqual(job['isPreviewEnabled'], '0')
-        job.enable_preview()
-        tries = 10
-        while tries > 0:
-            if job.is_done():
-                self.fail('Job finished before preview enabled.')
-            if job['isPreviewEnabled'] == '1':
-                break
-            tries -= 1
-        self.assertEqual(job['isPreviewEnabled'], '1')
-        job.disable_preview()
-        tries = 10
-        while tries > 0:
-            if job.is_done():
-                self.fail('Job finished before preview disabled.')
-            if job['isPreviewEnabled'] == '0':
-                break
-            tries -= 1
-        self.assertEqual(job['isPreviewEnabled'], '0')
-
-
-
     def check_job(self, job):
         self.check_entity(job)
         keys = ['cursorTime', 'delegate', 'diskUsage', 'dispatchState',
@@ -139,11 +111,53 @@ def test_read_jobs(self):
             job.refresh()
             self.check_job(job)
 
+class TestJobWithDelayedDone(testlib.SDKTestCase):
+    def setUp(self):
+        super(TestJobWithDelayedDone, self).setUp()
+        self.installAppFromCollection("sleep_command")
+        self.query = "search index=_internal | sleep done=100"
+        self.job = self.service.jobs.create(
+            query=self.query,
+            earliest_time="-1m",
+            priority=5,
+            latest_time="now")
+
+    def tearDown(self):
+        super(TestJobWithDelayedDone, self).tearDown()
+        self.job.cancel()
+        self.assertEventuallyTrue(lambda: self.job.sid not in self.service.jobs)
+
+    def test_enable_preview(self):
+        self.assertEqual(self.job['isPreviewEnabled'], '0')
+        self.job.enable_preview()
+        def is_preview():
+            if self.job.is_done():
+                self.fail('Job finished before preview enabled.')
+            return self.job['isPreviewEnabled'] == '1'
+        self.assertEventuallyTrue(is_preview)
+
+    def test_setpriority(self):
+        # Note that you can only *decrease* the priority (i.e., 5 decreased to 3)
+        # of a job unless Splunk is running as root. This is because Splunk jobs
+        # are tied up with operating system processes and their priorities.
+        self.assertEqual(5, int(self.job['priority']))
+        new_priority = 3
+        self.job.set_priority(new_priority)
+        def priority():
+            if self.job.is_done():
+                self.fail("Job already done before priority was set.")
+            self.job.refresh()
+            return int(self.job['priority'])
+        self.assertEventuallyEqual(
+            new_priority,
+            priority,
+            timeout=120
+        )
 
 class TestJob(testlib.SDKTestCase):
     def setUp(self):
         super(TestJob, self).setUp()
-        self.query = "search index=_internal earliest=-1m | head 3"
+        self.query = "search index=_internal | head 3"
         self.job = self.service.jobs.create(
             query=self.query, 
             earliest_time="-1m", 
@@ -155,7 +169,7 @@ def tearDown(self):
 
     @log_duration
     def test_get_preview_and_events(self):
-        testlib.retry(self.job, 'isDone', '1')
+        self.assertEventuallyTrue(self.job.is_done)
         self.assertLessEqual(int(self.job['eventCount']), 3)
 
         preview_stream = self.job.preview()
@@ -176,34 +190,28 @@ def test_pause(self):
             self.assertEqual(self.job['isPaused'], '0')
 
         self.job.pause()
-        testlib.retry(self.job, 'isPaused', '1')
-        self.assertEqual(self.job['isPaused'], '1')
+        self.assertEventuallyEqual(
+            '1',
+            lambda: self.job.refresh()['isPaused']
+        )
 
     def test_unpause(self):
         if self.job['isPaused'] == '0':
             self.job.pause()
             self.job.refresh()
             self.assertEqual(self.job['isPaused'], '1')
         self.job.unpause()
-        testlib.retry(self.job, 'isPaused', '0')
-        self.assertEqual(self.job['isPaused'], '0')
+        self.assertEventuallyEqual(
+            '0',
+            lambda: self.job.refresh()['isPaused']
+        )
 
     def test_finalize(self):
         if self.job['isFinalized'] == '1':
             self.fail("Job is already finalized; can't test .finalize() method.")
         else:
             self.job.finalize()
-            testlib.retry(self.job, 'isFinalized', '1')
-            self.assertEqual(self.job['isFinalized'], '1')
-
-    def test_setpriority(self):
-        old_priority = int(self.job['priority'])
-        new_priority = old_priority%10 + 1
-        self.assertNotEqual(old_priority, new_priority)
-
-        self.job.set_priority(new_priority)
-        testlib.retry(self.job, 'priority', str(new_priority))
-        self.assertEqual(int(self.job['priority']), new_priority)
+            self.assertEventuallyEqual('1', lambda: self.job.refresh()['isFinalized'])
 
     def test_setttl(self):
         old_ttl = int(self.job['ttl'])

tests/test_modular_input_kinds.py

@@ -18,6 +18,10 @@
 import splunklib.client as client
 
 class ModularInputKindTestCase(testlib.SDKTestCase):
+    def setUp(self):
+        super(ModularInputKindTestCase, self).setUp()
+        self.installAppFromCollection("modular-inputs")
+
     def test_list_arguments(self):
         if self.service.splunk_version[0] < 5:
             # Not implemented before 5.0

tests/test_service.py

@@ -128,6 +128,7 @@ def test_update_settings(self):
         settings.refresh()
         updated = settings['sessionTimeout']
         self.assertEqual(updated, original)
+        self.restartSplunk()
 
 class TestTrailing(unittest.TestCase):
     template = '/servicesNS/boris/search/another/path/segment/that runs on'