Merge branch 'main' into develop

Author: kkedziak-splunk

poetry.lock

@@ -16,7 +16,7 @@
 
 [tool.poetry]
 name = "solnlib"
-version = "5.1.0-beta.4"
+version = "5.1.2"
 description = "The Splunk Software Development Kit for Splunk Solutions"
 authors = ["Splunk <addonfactory@splunk.com>"]
 license = "Apache-2.0"

pyproject.toml

@@ -56,4 +56,4 @@
     "utils",
 ]
 
-__version__ = "5.1.0-beta.4"
+__version__ = "5.1.2"

solnlib/__init__.py

@@ -229,12 +229,11 @@ def __init__(
         else:
             self.logger = logging
 
-        if not all([scheme, host, port]):
-            scheme, host, port = get_splunkd_access_info()
-
         if hec_uri and hec_token:
             scheme, host, hec_port = utils.extract_http_scheme_host_port(hec_uri)
         else:
+            if not all([scheme, host, port]):
+                scheme, host, port = get_splunkd_access_info()
             hec_port, hec_token = self._get_hec_config(
                 hec_input_name, session_key, scheme, host, port, **context
             )

solnlib/modular_input/event_writer.py

@@ -20,7 +20,17 @@
 import json
 from typing import Any, Dict, Optional
 
-from splunk.rest import getWebCertFile, getWebKeyFile
+try:
+    from splunk.rest import getWebCertFile, getWebKeyFile
+except (ModuleNotFoundError, ImportError):
+
+    def getWebCertFile():
+        return None
+
+    def getWebKeyFile():
+        return None
+
+
 from splunklib import binding
 from solnlib import splunk_rest_client as rest_client
 from solnlib import utils

solnlib/server_info.py

@@ -18,6 +18,7 @@
 import sys
 
 import common
+import pytest
 from splunklib import binding
 
 from solnlib.modular_input import ClassicEventWriter, HECEventWriter
@@ -85,7 +86,54 @@ def flush(self):
     assert mock_stdout.write_count == 1
 
 
-def test_hec_event_writer(monkeypatch):
+def create_hec_event_writer__create_from_input(hec=False):
+    return HECEventWriter.create_from_input(
+        "HECTestInput",
+        "https://localhost:8089",
+        common.SESSION_KEY,
+        global_settings_schema=hec,
+    )
+
+
+def create_hec_event_writer__create_from_token_with_session_key(hec=False):
+    return HECEventWriter.create_from_token_with_session_key(
+        "https://localhost:8089",
+        common.SESSION_KEY,
+        "https://localhost:8090",
+        "test_token",
+        global_settings_schema=hec,
+    )
+
+
+def create_hec_event_writer__create_from_token(hec=False):
+    return HECEventWriter.create_from_token(
+        "https://localhost:8090", "test_token", global_settings_schema=hec
+    )
+
+
+def create_hec_event_writer__create_from_token__external_host(hec=False):
+    return HECEventWriter.create_from_token(
+        "https://external:8090", "test_token", global_settings_schema=hec
+    )
+
+
+def create_hec_event_writer__constructor(hec=False):
+    return HECEventWriter(
+        "HECTestInput", common.SESSION_KEY, global_settings_schema=hec
+    )
+
+
+@pytest.mark.parametrize(
+    "create_hec_event_writer, has_splunk_home",
+    [
+        (create_hec_event_writer__constructor, True),
+        (create_hec_event_writer__create_from_input, True),
+        (create_hec_event_writer__create_from_token_with_session_key, True),
+        (create_hec_event_writer__create_from_token, True),
+        (create_hec_event_writer__create_from_token__external_host, False),
+    ],
+)
+def test_hec_event_writer(monkeypatch, create_hec_event_writer, has_splunk_home):
     def mock_get(self, path_segment, owner=None, app=None, sharing=None, **query):
         if path_segment.endswith("/http"):
             return common.make_response_record(
@@ -120,42 +168,46 @@ def mock_get_hec_config(
     ):
         return "8088", "87de04d1-0823-11e6-9c94-a45e60e"
 
-    common.mock_splunkhome(monkeypatch)
+    if has_splunk_home:
+        common.mock_splunkhome(monkeypatch)
+    else:
+        # simulate an environment that has no splunk installation
+        monkeypatch.delenv("SPLUNK_HOME", raising=False)
+        monkeypatch.delenv("SPLUNK_ETC", raising=False)
     monkeypatch.setattr(binding.Context, "get", mock_get)
     monkeypatch.setattr(binding.Context, "post", mock_post)
     monkeypatch.setattr(HECEventWriter, "_get_hec_config", mock_get_hec_config)
 
-    for i in range(4):
-        ew = create_hec_event_writer(i)
-
-        events = []
-        events.append(
-            ew.create_event(
-                data="This is a test data1.",
-                time=1372274622.493,
-                index="main",
-                host="localhost",
-                source="Splunk",
-                sourcetype="misc",
-                stanza="test_scheme://test",
-                unbroken=True,
-                done=False,
-            )
+    ew = create_hec_event_writer(hec=False)
+
+    events = []
+    events.append(
+        ew.create_event(
+            data="This is a test data1.",
+            time=1372274622.493,
+            index="main",
+            host="localhost",
+            source="Splunk",
+            sourcetype="misc",
+            stanza="test_scheme://test",
+            unbroken=True,
+            done=False,
         )
-        events.append(
-            ew.create_event(
-                data="This is a test data2.",
-                time=1372274622.493,
-                index="main",
-                host="localhost",
-                source="Splunk",
-                sourcetype="misc",
-                stanza="test_scheme://test",
-                unbroken=True,
-                done=True,
-            )
+    )
+    events.append(
+        ew.create_event(
+            data="This is a test data2.",
+            time=1372274622.493,
+            index="main",
+            host="localhost",
+            source="Splunk",
+            sourcetype="misc",
+            stanza="test_scheme://test",
+            unbroken=True,
+            done=True,
         )
-        ew.write_events(events)
+    )
+    ew.write_events(events)
 
     # length of this list will indicate how many times post was called
     times_post_called = []
@@ -170,7 +222,7 @@ def mock_post_2(
     # test that there are 2 event batches created for write_event and post is called 2 times
     # max batch size is 1,000,000. If the max size is exceeded then a new batch is created.
     assert len(times_post_called) == 0
-    ew = create_hec_event_writer(1)
+
     events = []
 
     # each event length will be ~500 characters, 3000 events length will equal ~1,500,000 characters
@@ -207,35 +259,31 @@ def mock_post_2(
     # test that post is called 2 times
     assert len(times_post_called) == 2
 
-    for i in range(4):
-        ev = create_hec_event_writer(i)
-        assert ev._rest_client.scheme == "https"
-    for i in range(4):
-        ev = create_hec_event_writer(i, hec=True)
-        assert ev._rest_client.scheme == "http"
-
-
-def create_hec_event_writer(i, hec=False):
-    if i == 1:
-        return HECEventWriter.create_from_input(
-            "HECTestInput",
-            "https://localhost:8089",
-            common.SESSION_KEY,
-            global_settings_schema=hec,
-        )
-    elif i == 2:
-        return HECEventWriter.create_from_token_with_session_key(
-            "https://localhost:8089",
-            common.SESSION_KEY,
-            "https://localhost:8090",
-            "test_token",
-            global_settings_schema=hec,
-        )
-    elif i == 3:
-        return HECEventWriter.create_from_token(
-            "https://localhost:8090", "test_token", global_settings_schema=hec
-        )
-    else:
-        return HECEventWriter(
-            "HECTestInput", common.SESSION_KEY, global_settings_schema=hec
-        )
+
+@pytest.mark.parametrize(
+    "create_hec_event_writer, hec, expected_scheme",
+    [
+        (create_hec_event_writer__constructor, True, "http"),
+        (create_hec_event_writer__constructor, False, "https"),
+        (create_hec_event_writer__create_from_input, True, "http"),
+        (create_hec_event_writer__create_from_input, False, "https"),
+        (create_hec_event_writer__create_from_token_with_session_key, True, "http"),
+        (create_hec_event_writer__create_from_token_with_session_key, False, "https"),
+        (create_hec_event_writer__create_from_token, True, "http"),
+        (create_hec_event_writer__create_from_token, False, "https"),
+    ],
+)
+def test_hec_event_writer_gets_scheme_from_global_settings_if_requested(
+    monkeypatch, create_hec_event_writer, hec, expected_scheme
+):
+    common.mock_splunkhome(monkeypatch)
+
+    def mock_get_hec_config(
+        self, hec_input_name, session_key, scheme, host, port, **context
+    ):
+        return "8088", "87de04d1-0823-11e6-9c94-a45e60e"
+
+    monkeypatch.setattr(HECEventWriter, "_get_hec_config", mock_get_hec_config)
+
+    ev = create_hec_event_writer(hec)
+    assert ev._rest_client.scheme == expected_scheme

tests/unit/conftest.py

@@ -161,3 +161,32 @@ def test_server_info_object_with_key_file(
             assert kwargs.get("cert_file") is None
             assert kwargs.get("key_file") is None
             assert kwargs.get("verify") is None
+
+    @patch("solnlib.server_info.os.environ", autospec=True, return_value="$SPLUNK_HOME")
+    @patch(
+        "solnlib.server_info.get_splunkd_access_info",
+        autospec=True,
+        return_value=("https", "127.0.0.1", "8089"),
+    )
+    @patch("solnlib.server_info.rest_client", autospec=True)
+    def test_server_info_object_with_no_splunk_import(
+        self,
+        mock_rest_client,
+        mock_splunkd,
+        mock_os_env,
+    ):
+        mock_rest_client.SplunkRestClient = MagicMock()
+
+        # we want to raise 'ModuleNotFoundError' when importing the required functions
+        with patch.dict("sys.modules", {"splunk": ModuleNotFoundError}):
+            server_info.ServerInfo(common.SESSION_KEY)
+
+            for call_arg in mock_rest_client.SplunkRestClient.call_args_list:
+                _, kwargs = call_arg
+                assert (
+                    kwargs.get("cert_file") is None
+                )  # comes from the empty function in except
+                assert (
+                    kwargs.get("key_file") is None
+                )  # comes from the empty function in except
+                assert kwargs.get("verify") is None