Mark tests as app

Author: amysutedja

pytest.ini

@@ -1,5 +1,6 @@
 [pytest]
 markers =
+    app: requires sdk-app-collection
     smoke: essential smoke tests
 
 junit_family =

tests/test_index.py

@@ -170,10 +170,8 @@ def test_submit_via_attach_with_multiple_cookie_headers(self):
         cn.close()
         self.assertEventuallyTrue(lambda: self.totalEventCount() == event_count+1, timeout=60)
 
+    @pytest.mark.app
     def test_upload(self):
-        if not self.app_collection_installed():
-            print("Test requires sdk-app-collection. Skipping.")
-            return
         self.install_app_from_collection("file_to_upload")
 
         event_count = int(self.index['totalEventCount'])

tests/test_input.py

@@ -28,6 +28,8 @@
 
 import splunklib.client as client
 
+import pytest
+
 
 def highest_port(service, base_port, *kinds):
     """Find the first port >= base_port not in use by any input in kinds."""
@@ -164,10 +166,8 @@ def test_inputs_list_on_one_kind_with_search(self):
         found = [x.name for x in self.service.inputs.list('monitor', search=search)]
         self.assertEqual(expected, found)
 
+    @pytest.mark.app
     def test_oneshot(self):
-        if not self.app_collection_installed():
-            print("Test requires sdk-app-collection. Skipping.")
-            return
         self.install_app_from_collection('file_to_upload')
 
         index_name = testlib.tmpname()
@@ -225,25 +225,19 @@ def test_list(self):
         for input in input_list:
             self.assertTrue(input.name is not None)
 
+    @pytest.mark.app
     def test_lists_modular_inputs(self):
-        if self.service.splunk_version[0] < 5:
-            print("Modular inputs don't exist prior to Splunk 5.0. Skipping.")
-            return
-        elif not self.app_collection_installed():
-            print("Test requires sdk-app-collection. Skipping.")
-            return
-        else:
-            # Install modular inputs to list, and restart
-            # so they'll show up.
-            self.install_app_from_collection("modular-inputs")
-            self.uncheckedRestartSplunk()
-
-            inputs = self.service.inputs
-            if ('abcd','test2') not in inputs:
-                inputs.create('abcd', 'test2', field1='boris')
-
-            input = inputs['abcd', 'test2']
-            self.assertEqual(input.field1, 'boris')
+        # Install modular inputs to list, and restart
+        # so they'll show up.
+        self.install_app_from_collection("modular-inputs")
+        self.uncheckedRestartSplunk()
+
+        inputs = self.service.inputs
+        if ('abcd','test2') not in inputs:
+            inputs.create('abcd', 'test2', field1='boris')
+
+        input = inputs['abcd', 'test2']
+        self.assertEqual(input.field1, 'boris')
 
     def test_create(self):
         inputs = self.service.inputs

tests/test_job.py

@@ -34,6 +34,8 @@
 
 from splunklib.binding import _log_duration, HTTPError
 
+import pytest
+
 # TODO: Determine if we should be importing ExpatError if ParseError is not avaialble (e.g., on Python 2.6)
 # There's code below that now catches SyntaxError instead of ParseError. Should we be catching ExpathError instead?
 
@@ -214,10 +216,8 @@ def tearDown(self):
             self.job.cancel()
             self.assertEventuallyTrue(lambda: self.job.sid not in self.service.jobs)
 
+    @pytest.mark.app
     def test_enable_preview(self):
-        if not self.app_collection_installed():
-            print("Test requires sdk-app-collection. Skipping.")
-            return
         self.install_app_from_collection("sleep_command")
         sleep_duration = 100
         self.query = "search index=_internal | sleep %d" % sleep_duration
@@ -240,10 +240,8 @@ def is_preview_enabled():
         self.assertEventuallyTrue(is_preview_enabled)
         return
 
+    @pytest.mark.app
     def test_setpriority(self):
-        if not self.app_collection_installed():
-            print("Test requires sdk-app-collection. Skipping.")
-            return
         self.install_app_from_collection("sleep_command")
         sleep_duration = 100
         self.query = "search index=_internal | sleep %s" % sleep_duration

tests/test_modular_input.py

@@ -30,27 +30,21 @@ def setUp(self):
         super(ModularInputKindTestCase, self).setUp()
         self.uncheckedRestartSplunk()
 
+    @pytest.mark.app
     def test_lists_modular_inputs(self):
-        if self.service.splunk_version[0] < 5:
-            print("Modular inputs don't exist prior to Splunk 5.0. Skipping.")
-            return
-        elif not self.app_collection_installed():
-            print("Test requires sdk-app-collection. Skipping.")
-            return
-        else:
-            # Install modular inputs to list, and restart
-            # so they'll show up.
-            self.install_app_from_collection("modular-inputs")
-            self.uncheckedRestartSplunk()
+        # Install modular inputs to list, and restart
+        # so they'll show up.
+        self.install_app_from_collection("modular-inputs")
+        self.uncheckedRestartSplunk()
 
-            inputs = self.service.inputs
-            if ('abcd','test2') not in inputs:
-                inputs.create('abcd', 'test2', field1='boris')
+        inputs = self.service.inputs
+        if ('abcd','test2') not in inputs:
+            inputs.create('abcd', 'test2', field1='boris')
 
-            input = inputs['abcd', 'test2']
-            self.assertEqual(input.field1, 'boris')
-            for m in self.service.modular_input_kinds:
-                self.check_modular_input_kind(m)
+        input = inputs['abcd', 'test2']
+        self.assertEqual(input.field1, 'boris')
+        for m in self.service.modular_input_kinds:
+            self.check_modular_input_kind(m)
 
     def check_modular_input_kind(self, m):
         print(m.name)

tests/test_modular_input_kinds.py

@@ -23,15 +23,15 @@
     import unittest2 as unittest
 import splunklib.client as client
 
+import pytest
+
 class ModularInputKindTestCase(testlib.SDKTestCase):
     def setUp(self):
         super(ModularInputKindTestCase, self).setUp()
         self.uncheckedRestartSplunk()
 
+    @pytest.mark.app
     def test_list_arguments(self):
-        if not self.app_collection_installed():
-            print("Test requires sdk-app-collection. Skipping.")
-            return
         self.install_app_from_collection("modular-inputs")
 
         if self.service.splunk_version[0] < 5:
@@ -47,10 +47,8 @@ def test_list_arguments(self):
 
         self.assertEqual(expected_args, found_args)
 
+    @pytest.mark.app
     def test_update_raises_exception(self):
-        if not self.app_collection_installed():
-            print("Test requires sdk-app-collection. Skipping.")
-            return
         self.install_app_from_collection("modular-inputs")
 
         if self.service.splunk_version[0] < 5:
@@ -68,10 +66,8 @@ def check_modular_input_kind(self, m):
             self.assertEqual('test2', m['title'])
             self.assertEqual('simple', m['streaming_mode'])
 
+    @pytest.mark.app
     def test_list_modular_inputs(self):
-        if not self.app_collection_installed():
-            print("Test requires sdk-app-collection. Skipping.")
-            return
         self.install_app_from_collection("modular-inputs")
 
         if self.service.splunk_version[0] < 5:

tests/testlib.py

@@ -97,7 +97,7 @@ def assertEventuallyTrue(self, predicate, timeout=30, pause_time=0.5,
             logging.debug("wait finished after %s seconds", datetime.now()-start)
 
     def check_content(self, entity, **kwargs):
-        for k, v in six.iteritems(kwargs): 
+        for k, v in six.iteritems(kwargs):
             self.assertEqual(entity[k], str(v))
 
     def check_entity(self, entity):