Merge pull request splunk#589 from splunk/test-macros-pr

Test macros pr

Author: maszyk99

splunklib/client.py

@@ -101,6 +101,7 @@
 PATH_JOBS = "search/jobs/"
 PATH_JOBS_V2 = "search/v2/jobs/"
 PATH_LOGGER = "/services/server/logger/"
+PATH_MACROS = "configs/conf-macros/"
 PATH_MESSAGES = "messages/"
 PATH_MODULAR_INPUTS = "data/modular-inputs"
 PATH_ROLES = "authorization/roles/"
@@ -667,6 +668,15 @@ def saved_searches(self):
         """
         return SavedSearches(self)
 
+    @property
+    def macros(self):
+        """Returns the collection of macros.
+
+        :return: A :class:`Macros` collection of :class:`Macro`
+            entities.
+        """
+        return Macros(self)
+
     @property
     def settings(self):
         """Returns the configuration settings for this instance of Splunk.
@@ -3440,6 +3450,90 @@ def create(self, name, search, **kwargs):
         return Collection.create(self, name, search=search, **kwargs)
 
 
+class Macro(Entity):
+    """This class represents a search macro."""
+    def __init__(self, service, path, **kwargs):
+        Entity.__init__(self, service, path, **kwargs)
+
+    @property
+    def args(self):
+        """Returns the macro arguments.
+        :return: The macro arguments.
+        :rtype: ``string``
+        """
+        return self._state.content.get('args', '')
+
+    @property
+    def definition(self):
+        """Returns the macro definition.
+        :return: The macro definition.
+        :rtype: ``string``
+        """
+        return self._state.content.get('definition', '')
+
+    @property
+    def errormsg(self):
+        """Returns the validation error message for the macro.
+        :return: The validation error message for the macro.
+        :rtype: ``string``
+        """
+        return self._state.content.get('errormsg', '')
+
+    @property
+    def iseval(self):
+        """Returns the eval-based definition status of the macro.
+        :return: The iseval value for the macro.
+        :rtype: ``string``
+        """
+        return self._state.content.get('iseval', '0')
+
+    def update(self, definition=None, **kwargs):
+        """Updates the server with any changes you've made to the current macro
+        along with any additional arguments you specify.
+        :param `definition`: The macro definition (optional).
+        :type definition: ``string``
+        :param `kwargs`: Additional arguments (optional). Available parameters are:
+            'disabled', 'iseval', 'validation', and 'errormsg'.
+        :type kwargs: ``dict``
+        :return: The :class:`Macro`.
+        """
+        # Updates to a macro *require* that the definition be
+        # passed, so we pass the current definition if a value wasn't
+        # provided by the caller.
+        if definition is None: definition = self.content.definition
+        Entity.update(self, definition=definition, **kwargs)
+        return self
+
+    @property
+    def validation(self):
+        """Returns the validation expression for the macro.
+        :return: The validation expression for the macro.
+        :rtype: ``string``
+        """
+        return self._state.content.get('validation', '')
+
+
+class Macros(Collection):
+    """This class represents a collection of macros. Retrieve this
+    collection using :meth:`Service.macros`."""
+    def __init__(self, service):
+        Collection.__init__(
+            self, service, PATH_MACROS, item=Macro)
+
+    def create(self, name, definition, **kwargs):
+        """ Creates a macro.
+        :param name: The name for the macro.
+        :type name: ``string``
+        :param definition: The macro definition.
+        :type definition: ``string``
+        :param kwargs: Additional arguments (optional). Available parameters are:
+            'disabled', 'iseval', 'validation', and 'errormsg'.
+        :type kwargs: ``dict``
+        :return: The :class:`Macros` collection.
+        """
+        return Collection.create(self, name, definition=definition, **kwargs)
+
+
 class Settings(Entity):
     """This class represents configuration settings for a Splunk service.
     Retrieve this collection using :meth:`Service.settings`."""
@@ -3905,4 +3999,4 @@ def batch_save(self, *documents):
         data = json.dumps(documents)
 
         return json.loads(
-            self._post('batch_save', headers=KVStoreCollectionData.JSON_HEADER, body=data).body.read().decode('utf-8'))
+            self._post('batch_save', headers=KVStoreCollectionData.JSON_HEADER, body=data).body.read().decode('utf-8'))

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()