test: fix integration tests, add deprecation for create_from_token

Author: sgoral-splunk

solnlib/modular_input/event_writer.py

@@ -23,6 +23,7 @@
 import threading
 import time
 import traceback
+import warnings
 from abc import ABCMeta, abstractmethod
 from random import randint
 from typing import List, Union
@@ -39,6 +40,16 @@
 __all__ = ["ClassicEventWriter", "HECEventWriter"]
 
 
+deprecation_msg = (
+    "Function 'create_from_token' is deprecated and incompatible with 'global_settings_schema=True'. "
+    "Use 'create_from_token_with_session_key' instead."
+)
+
+
+class FunctionDeprecated(Exception):
+    pass
+
+
 class EventWriter(metaclass=ABCMeta):
     """Base class of event writer."""
 
@@ -272,12 +283,16 @@ def create_from_token(
             Created HECEventWriter.
         """
 
+        warnings.warn(deprecation_msg, DeprecationWarning, stacklevel=2)
+
+        if global_settings_schema:
+            raise FunctionDeprecated(deprecation_msg)
+
         return HECEventWriter(
             None,
             None,
             None,
             None,
-            None,
             hec_uri=hec_uri,
             hec_token=hec_token,
             global_settings_schema=global_settings_schema,

tests/integration/test_conf_manager.py

@@ -32,50 +32,6 @@
 }
 
 
-# conftest.py or test file
-
-import tempfile
-import shutil
-import re
-import importlib.util
-import sys
-import os
-import pytest
-
-
-@pytest.fixture
-def patch_splunk_cli_common_module():
-    import splunk.clilib.cli_common as comm
-
-    original_path = comm.__file__
-
-    with open(original_path, "r") as f:
-        source = f.read()
-
-    pattern = r'procArgs\s*=\s*\[\s*os\.path\.join\(\s*os\.environ\[\s*["\']SPLUNK_HOME["\']\s*\],\s*["\']bin["\'],\s*["\']splunkd["\']\s*\)\s*,\s*["\']local-rest-uri["\']\s*,\s*["\']-p["\']\s*,\s*mgmtPort\s*\]'
-    replacement = 'procArgs = [os.path.join(os.environ["SPLUNK_HOME"], "bin", "splunk"), "cmd", "splunkd", "local-rest-uri", "-p", mgmtPort]'
-
-    new_content = re.sub(pattern, replacement, source)
-
-    temp_dir = tempfile.mkdtemp()
-    patched_module_path = os.path.join(temp_dir, "module.py")
-
-    with open(patched_module_path, "w") as f:
-        f.write(new_content)
-
-    spec = importlib.util.spec_from_file_location("splunk.clilib.cli_common", patched_module_path)
-    patched_module = importlib.util.module_from_spec(spec)
-    sys.modules["splunk.clilib.cli_common"] = patched_module
-    spec.loader.exec_module(patched_module)
-
-    yield
-
-    shutil.rmtree(temp_dir)
-    importlib.invalidate_caches()
-    sys.modules.pop("splunk.clilib.cli_common", None)
-
-
-
 def _build_conf_manager(session_key: str) -> conf_manager.ConfManager:
     return conf_manager.ConfManager(
         session_key,
@@ -189,8 +145,9 @@ def test_conf_manager_update_conf_with_encrypted_keys():
     assert conf_file.get("stanza")["key2"] == "value2"
 
 
-def test_get_log_level(patch_splunk_cli_common_module, monkeypatch):
-
+@mock.patch("splunk.clilib.cli_common.getMgmtUri")
+def test_get_log_level(mock_mgm_uri, monkeypatch):
+    mock_mgm_uri.return_value = "https://127.0.0.1:8089"
     conftest.mock_splunk(monkeypatch)
 
     session_key = context.get_session_key()

tests/unit/test_modular_input_event_writer.py

@@ -23,6 +23,7 @@
 from unittest.mock import patch
 
 from solnlib.modular_input import ClassicEventWriter, HECEventWriter
+from solnlib.modular_input.event_writer import FunctionDeprecated, deprecation_msg
 
 
 def test_classic_event_writer(monkeypatch):
@@ -299,7 +300,6 @@ def mock_get_hec_config(
         (create_hec_event_writer__create_from_input, False, "https"),
         (create_hec_event_writer__create_from_token_with_session_key, True, "https"),
         (create_hec_event_writer__create_from_token_with_session_key, False, "https"),
-        (create_hec_event_writer__create_from_token, True, "https"),
         (create_hec_event_writer__create_from_token, False, "https"),
     ],
 )
@@ -318,3 +318,24 @@ def mock_get_hec_config(
 
     ev = create_hec_event_writer(hec)
     assert ev._rest_client.scheme == expected_scheme
+
+
+@pytest.mark.parametrize("hec", [False, True])
+def test_hec_event_writer_create_from_token_deprecation(hec, monkeypatch, caplog):
+    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)
+
+    if hec:
+        with pytest.raises(FunctionDeprecated) as e:
+            create_hec_event_writer__create_from_token(hec=True)
+        assert e.value.args[0] == deprecation_msg
+    else:
+        with pytest.warns(DeprecationWarning, match=deprecation_msg):
+            ev = create_hec_event_writer__create_from_token(hec=False)
+            assert ev._rest_client.scheme == "https"