Add tests for macros

tests/test_macro.py

@@ -0,0 +1,164 @@
+#!/usr/bin/env python
+#
+# Copyright 2011-2015 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.
+
+from __future__ import absolute_import
+from tests import testlib
+import logging
+
+import splunklib.client as client
+
+import pytest
+
+@pytest.mark.smoke
+class TestMacro(testlib.SDKTestCase):
+    def setUp(self):
+        super(TestMacro, self).setUp()
+        macros = self.service.macros
+        logging.debug("Macros namespace: %s", macros.service.namespace)
+        self.macro_name = testlib.tmpname()
+        definition = '| eval test="123"'
+        self.macro = macros.create(self.macro_name, definition)
+
+    def tearDown(self):
+        super(TestMacro, self).setUp()
+        for macro in self.service.macros:
+            if macro.name.startswith('delete-me'):
+                self.service.macros.delete(macro.name)
+
+    def check_macro(self, macro):
+        self.check_entity(macro)
+        expected_fields = ['definition',
+                           'iseval',
+                           'args',
+                           'validation',
+                           'errormsg']
+        for f in expected_fields:
+            macro[f]
+        is_eval = macro.iseval
+        self.assertTrue(is_eval == '1' or is_eval == '0')
+
+    def test_create(self):
+        self.assertTrue(self.macro_name in self.service.macros)
+        self.check_macro(self.macro)
+
+    def test_create_with_args(self):
+        macro_name = testlib.tmpname() + '(1)'
+        definition = '| eval value="$value$"'
+        kwargs = {
+            'args': 'value',
+            'validation': '$value$ > 10',
+            'errormsg': 'value must be greater than 10'
+        }
+        macro = self.service.macros.create(macro_name, definition=definition, **kwargs)
+        self.assertTrue(macro_name in self.service.macros)
+        self.check_macro(macro)
+        self.assertEqual(macro.iseval, '0')
+        self.assertEqual(macro.args, kwargs.get('args'))
+        self.assertEqual(macro.validation, kwargs.get('validation'))
+        self.assertEqual(macro.errormsg, kwargs.get('errormsg'))
+        self.service.macros.delete(macro_name)
+
+    def test_delete(self):
+        self.assertTrue(self.macro_name in self.service.macros)
+        self.service.macros.delete(self.macro_name)
+        self.assertFalse(self.macro_name in self.service.macros)
+        self.assertRaises(client.HTTPError,
+                          self.macro.refresh)
+
+    def test_update(self):
+        new_definition = '| eval updated="true"'
+        self.macro.update(definition=new_definition)
+        self.macro.refresh()
+        self.assertEqual(self.macro['definition'], new_definition)
+
+        is_eval = testlib.to_bool(self.macro['iseval'])
+        self.macro.update(iseval=not is_eval)
+        self.macro.refresh()
+        self.assertEqual(testlib.to_bool(self.macro['iseval']), not is_eval)
+
+    def test_cannot_update_name(self):
+        new_name = self.macro_name + '-alteration'
+        self.assertRaises(client.IllegalOperationException,
+                          self.macro.update, name=new_name)
+
+    def test_name_collision(self):
+        opts = self.opts.kwargs.copy()
+        opts['owner'] = '-'
+        opts['app'] = '-'
+        opts['sharing'] = 'user'
+        service = client.connect(**opts)
+        logging.debug("Namespace for collision testing: %s", service.namespace)
+        macros = service.macros
+        name = testlib.tmpname()
+
+        dispatch1 = '| eval macro_one="1"'
+        dispatch2 = '| eval macro_two="2"'
+        namespace1 = client.namespace(app='search', sharing='app')
+        namespace2 = client.namespace(owner='admin', app='search', sharing='user')
+        new_macro2 = macros.create(
+            name, dispatch2,
+            namespace=namespace1)
+        new_macro1 = macros.create(
+            name, dispatch1,
+            namespace=namespace2)
+
+        self.assertRaises(client.AmbiguousReferenceException,
+                          macros.__getitem__, name)
+        macro1 = macros[name, namespace1]
+        self.check_macro(macro1)
+        macro1.update(**{'definition': '| eval number=1'})
+        macro1.refresh()
+        self.assertEqual(macro1['definition'], '| eval number=1')
+        macro2 = macros[name, namespace2]
+        macro2.update(**{'definition': '| eval number=2'})
+        macro2.refresh()
+        self.assertEqual(macro2['definition'], '| eval number=2')
+        self.check_macro(macro2)
+
+    def test_no_equality(self):
+        self.assertRaises(client.IncomparableException,
+                          self.macro.__eq__, self.macro)
+
+    def test_acl(self):
+        self.assertEqual(self.macro.access["perms"], None)
+        self.macro.acl_update(sharing="app", owner="admin", **{"perms.read": "admin, nobody"})
+        self.assertEqual(self.macro.access["owner"], "admin")
+        self.assertEqual(self.macro.access["sharing"], "app")
+        self.assertEqual(self.macro.access["perms"]["read"], ['admin', 'nobody'])
+
+    def test_acl_fails_without_sharing(self):
+        self.assertRaisesRegex(
+            ValueError,
+            "Required argument 'sharing' is missing.",
+            self.macro.acl_update,
+            owner="admin", app="search", **{"perms.read": "admin, nobody"}
+        )
+
+    def test_acl_fails_without_owner(self):
+        self.assertRaisesRegex(
+            ValueError,
+            "Required argument 'owner' is missing.",
+            self.macro.acl_update,
+            sharing="app", app="search", **{"perms.read": "admin, nobody"}
+        )
+
+
+if __name__ == "__main__":
+    try:
+        import unittest2 as unittest
+    except ImportError:
+        import unittest
+    unittest.main()