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

Test macros pr

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