Made modular inputs and oneshot input testing use the app collection.

Author: Frederick Ross

tests/test_input.py

@@ -59,23 +59,23 @@ def test_inputs_list_on_one_kind_with_search(self):
         self.assertEqual(expected, found)
 
     def test_oneshot(self):
+        self.installAppFromCollection('file_to_upload')
+
         index_name = testlib.tmpname()
         index = self.service.indexes.create(index_name)
-        self.restartSplunk(timeout=120)
-        index = self.service.indexes[index_name]
+        self.assertEventuallyTrue(lambda: index.refresh() and index['disabled'] == '0')
+
         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)
+
+        path = self.pathInApp("file_to_upload", ["log.txt"])
+        self.service.inputs.oneshot(path, index=index_name)
         self.assertEventuallyEqual(
-            str(eventCount+1),
+            str(eventCount+4),
             lambda: index.refresh()['totalEventCount']
         )
 
     def test_oneshot_on_nonexistant_file(self):
         name = testlib.tmpname()
-        from os import path
-        self.assertFalse(path.exists(name))
         self.assertRaises(client.OperationFailedException,
             self.service.inputs.oneshot, name)
 
@@ -121,9 +121,15 @@ def test_lists_modular_inputs(self):
         if self.service.splunk_version[0] < 5:
             return # Modular inputs don't exist prior to 5.0
         else:
+            # Install modular inputs to list, and restart
+            # so they'll show up.
+            self.installAppFromCollection("modular-inputs")
+            self.uncheckedRestartSplunk()
+
             inputs = self.service.inputs
             if ('test2','abcd') not in inputs:
                 inputs.create('test2', 'abcd', field1='boris')
+
             input = inputs['test2', 'abcd']
             self.assertEqual(input.field1, 'boris')
 

tests/test_modular_input_kinds.py

@@ -21,6 +21,7 @@ class ModularInputKindTestCase(testlib.SDKTestCase):
     def setUp(self):
         super(ModularInputKindTestCase, self).setUp()
         self.installAppFromCollection("modular-inputs")
+        self.uncheckedRestartSplunk()
 
     def test_list_arguments(self):
         if self.service.splunk_version[0] < 5:

tests/testlib.py

@@ -147,6 +147,25 @@ def installAppFromCollection(self, name):
         collectionName = 'sdk-app-collection'
         if collectionName not in self.service.apps:
             raise ValueError("sdk-test-application not installed in splunkd")
+        appPath = self.pathInApp(collectionName, ["build", name+".tar"])
+        kwargs = {"update": 1, "name": appPath}
+        try:
+            self.service.post("apps/appinstall", **kwargs)
+        except client.HTTPError as he:
+            if he.status == 400:
+                raise IOError("App %s not found in app collection" % name)
+        self.installedApps.append(name)
+
+    def pathInApp(self, appName, pathComponents):
+        """Return a path to *pathComponents* in *appName*.
+
+        *pathComponents* shold be a list of strings giving the components.
+        This function will try to figure out the correct separator (/ or \)
+        for the platform that splunkd is running on and construct the path
+        as needed.
+
+        :return: A string giving the path.
+        """
         splunkHome = self.service.settings['SPLUNK_HOME']
         if "/" in splunkHome and "\\" in splunkHome:
             raise ValueError("There are both forward and back slashes in $SPLUNK_HOME. What system are you on?!?")
@@ -156,15 +175,11 @@ def installAppFromCollection(self, name):
             separator = "\\"
         else:
             raise ValueError("No separators in $SPLUNK_HOME. Can't determine what file separator to use.")
-        appPath = separator.join([splunkHome, "etc", "apps", collectionName,
-                                  "build", name + ".tar"])
-        kwargs = {"update": 1, "name": appPath}
-        try:
-            self.service.post("apps/appinstall", **kwargs)
-        except client.HTTPError as he:
-            if he.status == 400:
-                raise IOError("App %s not found in app collection" % name)
-        self.installedApps.append(name)
+        appPath = separator.join([splunkHome, "etc", "apps", appName] + pathComponents)
+        return appPath
+
+    def uncheckedRestartSplunk(self, timeout=120):
+        self.service.restart(timeout)
 
     def restartSplunk(self, timeout=120):
         if self.service.restart_required:
@@ -178,19 +193,20 @@ def setUpClass(cls):
 
     def setUp(self):
         unittest.TestCase.setUp(self)
-        self.caused_restart_required = False
         self.service = client.connect(**self.opts.kwargs)
         if self.service.restart_required:
-            self.restart_already_required = True
+            self.restartSplunk()
         logging.debug("Connected to splunkd version %s", '.'.join(str(x) for x in self.service.splunk_version))
 
     def tearDown(self):
-        if self.service.restart_required and not self.restart_already_required:
+        if self.service.restart_required:
             self.fail("Test left Splunk in a state requiring a restart.")
         for appName in self.installedApps:
             if appName in self.service.apps:
                 self.service.apps.delete(appName)
                 wait(lambda: appName not in self.service.apps)
+        if self.service.restart_required:
+            self.clearRestartMessage()
 
 def main():
     unittest.main()