Change macros endpoint from configs/conf-macros to data/macros

Author: mateusz834

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/"

test-reports/junit-py39.xml

@@ -0,0 +1,61 @@
+<?xml version="1.0" encoding="utf-8"?><testsuites name="pytest tests"><testsuite name="pytest" errors="0" failures="1" skipped="0" tests="1" time="0.377" timestamp="2025-09-03T12:33:38.683850+02:00" hostname="MPOLIWCZ-M-3NJ6"><testcase classname="tests.integration.test_macro.TestPrivilegesWithNamespace" name="test_create_macro_no_admin" time="0.176"><failure message="splunklib.binding.HTTPError: HTTP 409 Conflict -- An object with name=SDKTestMacro already exists">self = &lt;test_macro.TestPrivilegesWithNamespace testMethod=test_create_macro_no_admin&gt;
+
+    def test_create_macro_no_admin(self):
+&gt;       self.service.macros.create(
+            self.macro_name, 'eval test="123"', namespace=self.namespace
+        )
+
+tests/integration/test_macro.py:363: 
+_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
+splunklib/client.py:3644: in create
+    return Collection.create(self, name, definition=definition, **kwargs)
+splunklib/client.py:1755: in create
+    response = self.post(name=name, **params)
+splunklib/client.py:981: in post
+    return self.service.post(path, owner=owner, app=app, sharing=sharing, **query)
+splunklib/binding.py:335: in wrapper
+    return request_fun(self, *args, **kwargs)
+splunklib/binding.py:90: in new_f
+    val = f(*args, **kwargs)
+splunklib/binding.py:856: in post
+    response = self.http.post(path, all_headers, **query)
+splunklib/binding.py:1382: in post
+    return self.request(url, message)
+_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
+
+self = &lt;splunklib.binding.HttpLib object at 0x101d280a0&gt;, url = UrlEncoded('https://localhost:8089/servicesNS/nobody/launcher/data/macros/')
+message = {'body': b'name=SDKTestMacro&amp;definition=eval+test%3D%22123%22', 'headers': [('Cookie', 'splunkd_8089=KgRkAxruIHVKU^LL_...K1LOsT3WAmmgC8xULxlGmQ0NBZhpLbR9G7Qo1H19WtacHgoD6HyKm7ENwQ227bVuFbfr5C5HlnbItLtzHtJJC9YSLhFGYi8bC')], 'method': 'POST'}
+kwargs = {}
+response = {'status': 409, 'reason': 'Conflict', 'headers': [('Date', 'Wed, 03 Sep 2025 10:33:38 GMT'), ('Expires', 'Thu, 26 Oct ...rame-Options', 'SAMEORIGIN'), ('Server', 'Splunkd')], 'body': &lt;splunklib.binding.ResponseReader object at 0x101ee16d0&gt;}
+
+    def request(self, url, message, **kwargs):
+        """Issues an HTTP request to a URL.
+    
+        :param url: The URL.
+        :type url: ``string``
+        :param message: A dictionary with the format as described in
+            :class:`HttpLib`.
+        :type message: ``dict``
+        :param kwargs: Additional keyword arguments (optional). These arguments
+            are passed unchanged to the handler.
+        :type kwargs: ``dict``
+        :returns: A dictionary describing the response (see :class:`HttpLib` for
+            its structure).
+        :rtype: ``dict``
+        """
+        while True:
+            try:
+                response = self.handler(url, message, **kwargs)
+                break
+            except Exception:
+                if self.retries &lt;= 0:
+                    raise
+                else:
+                    time.sleep(self.retryDelay)
+                    self.retries -= 1
+        response = record(response)
+        if 400 &lt;= response.status:
+&gt;           raise HTTPError(response)
+E           splunklib.binding.HTTPError: HTTP 409 Conflict -- An object with name=SDKTestMacro already exists
+
+splunklib/binding.py:1411: HTTPError</failure></testcase></testsuite></testsuites>

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,67 @@ 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")
+
+    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):
+        for macro in self.service.macros:
+            if macro.name == self.macro_name:
+                self.service.macros.delete(self.macro_name, namespace=self.namespace)
+
+    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