DVPL-12515: Add modular input E2E test

.github/workflows/test.yml

@@ -43,7 +43,7 @@ jobs:
         run: pip install tox
 
       - name: Test Execution
-        run: tox -e py
+        run: tox -e py -- ./tests
   fossa-scan:
     uses: splunk/oss-scanning-public/.github/workflows/oss-scan.yml@main
     secrets: inherit

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