Add modular input e2e test

Currently, we only have unit tests that verify the generated XML output
of modular inputs. This change adds E2E tests to confirm that our
modular input machinery works correctly with a Splunk instance.

Author: mateusz834

docker-compose.yml

@@ -22,7 +22,9 @@ services:
             - "./tests/system/test_apps/generating_app:/opt/splunk/etc/apps/generating_app"
             - "./tests/system/test_apps/reporting_app:/opt/splunk/etc/apps/reporting_app"
             - "./tests/system/test_apps/streaming_app:/opt/splunk/etc/apps/streaming_app"
+            - "./tests/system/test_apps/modularinput_app:/opt/splunk/etc/apps/modularinput_app"
             - "./splunklib:/opt/splunk/etc/apps/eventing_app/lib/splunklib"
             - "./splunklib:/opt/splunk/etc/apps/generating_app/lib/splunklib"
             - "./splunklib:/opt/splunk/etc/apps/reporting_app/lib/splunklib"
             - "./splunklib:/opt/splunk/etc/apps/streaming_app/lib/splunklib"
+            - "./splunklib:/opt/splunk/etc/apps/modularinput_app/lib/splunklib"

tests/system/test_apps/modularinput_app/README/inputs.conf.spec

@@ -0,0 +1,3 @@
+[modularinput://<name>]
+
+endpoint = <value>

tests/system/test_apps/modularinput_app/bin/modularinput.py

@@ -0,0 +1,58 @@
+#!/usr/bin/env python
+
+# Copyright © 2011-2025 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.
+
+import sys
+import os
+from urllib import parse
+
+sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "lib"))
+from splunklib.modularinput import Scheme, Argument, Script, Event
+
+
+class ModularInput(Script):
+    endpoint_arg = "endpoint"
+
+    def get_scheme(self):
+        scheme = Scheme("modularinput")
+
+        scheme.use_external_validation = True
+        scheme.use_single_instance = True
+
+        endpoint = Argument(self.endpoint_arg)
+        endpoint.title = "URL"
+        endpoint.data_type = Argument.data_type_string
+        endpoint.description = "URL"
+        endpoint.required_on_create = True
+        scheme.add_argument(endpoint)
+
+        return scheme
+
+    def validate_input(self, definition):
+        url = definition.parameters[self.endpoint_arg]
+        parsed = parse.urlparse(url)
+        if parsed.scheme != "https":
+            raise ValueError(f"non-supported scheme {parsed.scheme}")
+
+    def stream_events(self, inputs, ew):
+        for input_name, input_item in list(inputs.inputs.items()):
+            event = Event()
+            event.stanza = input_name
+            event.data = "example message"
+            ew.write_event(event)
+
+
+if __name__ == "__main__":
+    sys.exit(ModularInput().run(sys.argv))

tests/system/test_apps/modularinput_app/default/app.conf

@@ -0,0 +1,14 @@
+[install]
+is_configured = 0
+
+[ui]
+is_visible = 1
+label = Modular Input test app
+
+[launcher]
+author=Splunk
+description=Modular input test app
+version = 1.0
+
+[package]
+check_for_updates = false

tests/system/test_apps/modularinput_app/default/inputs.conf

@@ -0,0 +1,2 @@
+[modularinput]
+python.version = python3

tests/system/test_modularinput_app.py

@@ -0,0 +1,90 @@
+#!/usr/bin/env python
+#
+# Copyright © 2011-2025 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 splunklib import results
+from tests import testlib
+from splunklib.binding import HTTPError
+
+
+class ModularInput(testlib.SDKTestCase):
+    index_name = "test_modular_input"
+    input_name = "test_modular_input"
+    input_kind = "modularinput"
+
+    def setUp(self):
+        super().setUp()
+
+        app_found = False
+        for kind in self.service.modular_input_kinds:
+            if kind.name == self.input_kind:
+                app_found = True
+
+        self.assertTrue(app_found, f"{self.input_kind} modular input not installed")
+        self.clean()
+
+    def tearDown(self):
+        super().tearDown()
+        self.clean()
+
+    def clean(self):
+        for input in self.service.inputs:
+            if input.name == self.input_name and input.kind == self.input_kind:
+                self.service.inputs.delete(self.input_name, self.input_kind)
+
+        for index in self.service.indexes:
+            if index.name == self.input_name:
+                self.service.indexes.delete(self.input_name)
+
+    def test_modular_input(self):
+        self.service.indexes.create(self.index_name)
+
+        self.service.inputs.create(
+            self.input_name,
+            self.input_kind,
+            endpoint="https://example.com/api/endpoint",
+            index=self.index_name,
+        )
+
+        def query():
+            stream = self.service.jobs.oneshot(
+                f'search index="{self.index_name}"', output_mode="json"
+            )
+            reader = results.JSONResultsReader(stream)
+            return list(reader)
+
+        # Wait until the modular input is executed by splunk.
+        self.assertEventuallyTrue(lambda: len(query()) != 0, timeout=10)
+
+        items = query()
+        self.assertTrue(len(items) == 1)
+        self.assertEqual(items[0]["_raw"], "example message")
+
+    def test_external_validator(self):
+        def create():
+            self.service.inputs.create(
+                self.input_name,
+                self.input_kind,
+                endpoint="http://example.com/api/endpoint",
+                index=self.index_name,
+            )
+
+        self.assertRaisesRegex(HTTPError, "non-supported scheme http", create)
+
+
+if __name__ == "__main__":
+    import unittest
+
+    unittest.main()