From 5ecd916feecff1a03072d40ab352cdcc112a0f26 Mon Sep 17 00:00:00 2001 From: Mateusz Poliwczak Date: Tue, 2 Sep 2025 15:23:48 +0200 Subject: [PATCH] Change macros endpoint from configs/conf-macros to data/macros --- splunklib/client.py | 2 +- tests/integration/test_macro.py | 68 ++++++++++++++++++++++++++++++++- 2 files changed, 67 insertions(+), 3 deletions(-) diff --git a/splunklib/client.py b/splunklib/client.py index 2c4b3ea8b..1ef32edb7 100644 --- a/splunklib/client.py +++ b/splunklib/client.py @@ -108,7 +108,7 @@ PATH_JOBS = "search/jobs/" PATH_JOBS_V2 = "search/v2/jobs/" PATH_LOGGER = "/services/server/logger/" -PATH_MACROS = "configs/conf-macros/" +PATH_MACROS = "data/macros/" PATH_MESSAGES = "messages/" PATH_MODULAR_INPUTS = "data/modular-inputs" PATH_ROLES = "authorization/roles/" diff --git a/tests/integration/test_macro.py b/tests/integration/test_macro.py index 580613176..a556c3bcb 100755 --- a/tests/integration/test_macro.py +++ b/tests/integration/test_macro.py @@ -15,7 +15,7 @@ # under the License. from __future__ import absolute_import -from splunklib.binding import HTTPError +from splunklib.binding import HTTPError, namespace from tests import testlib import logging @@ -264,7 +264,7 @@ def query(): self.assertRaisesRegex(HTTPError, "value must be smaller that value2", query) -# This test makes sure that the endpoint we use for macros (configs/conf-macros) +# This test makes sure that the endpoint we use for macros (data/macros/) # does not require admin privileges and can be used by normal users. class TestPrivileges(testlib.SDKTestCase): macro_name = "SDKTestMacro" @@ -318,6 +318,70 @@ def test_create_macro_no_admin(self): self.assertEqual(out[0]["test"], "123") +# This test makes sure that the endpoint we use for macros (data/macros/) +# does not require admin privileges and can be used by normal users. +class TestPrivilegesWithNamespace(testlib.SDKTestCase): + macro_name = "SDKTestMacro" + username = "SDKTestMacroUser".lower() + password = "SDKTestMacroUserPassword!" + + # TODO: Would be nice to create app on-demand here and control the + # permissions, so that we are not dependent on default settings. + # namespace = namespace(owner="nobody", app="launcher") + namespace = namespace(owner="nobody", app="search") + + def setUp(self): + testlib.SDKTestCase.setUp(self) + self.cleanUsers() + + self.service.users.create( + username=self.username, password=self.password, roles=["power"] + ) + + self.service.logout() + kwargs = self.opts.kwargs.copy() + kwargs["username"] = self.username + kwargs["password"] = self.password + self.service = client.connect(**kwargs) + + self.cleanMacros() + + def tearDown(self): + testlib.SDKTestCase.tearDown(self) + self.cleanMacros() + self.service = client.connect(**self.opts.kwargs) + self.cleanUsers() + + def cleanUsers(self): + for user in self.service.users: + if user.name == self.username: + self.service.users.delete(self.username) + + def cleanMacros(self): + try: + m = self.service.macros[self.macro_name, self.namespace] + except KeyError: + return # macro does not exist + m.delete() + + def test_create_macro_no_admin(self): + self.service.macros.create( + self.macro_name, 'eval test="123"', namespace=self.namespace + ) + + stream = self.service.jobs.oneshot( + f"| makeresults count=1 | `{self.macro_name}`", + output_mode="json", + namespace=self.namespace, + ) + + result = results.JSONResultsReader(stream) + out = list(result) + + self.assertTrue(len(out) == 1) + self.assertEqual(out[0]["test"], "123") + + if __name__ == "__main__": try: import unittest2 as unittest