tests: add unit tests

Author: sgoral-splunk

solnlib/splunkenv.py

@@ -20,8 +20,22 @@
 import os.path as op
 import socket
 from typing import List, Optional, Tuple, Union
-from splunk.clilib.cli_common import getAppConf
-from splunk.clilib.bundle_paths import make_splunkhome_path as mksplhomepath
+
+try:
+    from splunk.clilib.cli_common import getAppConf
+except ImportError:
+
+    def getAppConf(*args, **kwargs):
+        return None
+
+
+try:
+    from splunk.clilib.bundle_paths import make_splunkhome_path as mksplhomepath
+except ImportError:
+
+    def mksplhomepath(*args, **kwargs):
+        return None
+
 
 from .utils import is_true
 
@@ -34,7 +48,7 @@
     "get_splunkd_uri",
     "get_conf_key_value",
     "get_conf_stanza",
-    "get_conf_stanzas",
+    "_get_conf_stanzas",
 ]
 
 ETC_LEAF = "etc"
@@ -224,7 +238,7 @@ def get_conf_key_value(
         KeyError: If `stanza` or `key` doesn't exist.
     """
 
-    stanzas = get_conf_stanzas(conf_name, app_name)
+    stanzas = _get_conf_stanzas(conf_name, app_name)
     return stanzas[stanza][key]
 
 
@@ -245,11 +259,11 @@ def get_conf_stanza(
          KeyError: If stanza doesn't exist.
     """
 
-    stanzas = get_conf_stanzas(conf_name, app_name)
+    stanzas = _get_conf_stanzas(conf_name, app_name)
     return stanzas[stanza]
 
 
-def get_conf_stanzas(conf_name: str, app_name: str, logger=None) -> dict:
+def _get_conf_stanzas(conf_name: str, app_name: str, logger=None) -> dict:
     """Get stanzas of `conf_name`
 
     Arguments:
@@ -260,21 +274,16 @@ def get_conf_stanzas(conf_name: str, app_name: str, logger=None) -> dict:
         Config stanzas.
 
     Examples:
-       >>> stanzas = get_conf_stanzas('server')
+       >>> stanzas = _get_conf_stanzas('server')
        >>> return: {'serverName': 'testServer', 'sessionTimeout': '1h', ...}
     """
 
-    path = make_splunkhome_path(["etc", "apps", app_name])
+    if app_name == APP_SYSTEM:
+        path = make_splunkhome_path(["etc", app_name])
+    else:
+        path = make_splunkhome_path(["etc", "apps", app_name])
     app_conf = getAppConf(
         confName=conf_name, app=app_name, use_btool=False, app_path=path
     )
 
-    if logger:
-        logger.info(f"akakakakakkakakakaka 21 path:  {path}")
-
-    if logger:
-        logger.info(
-            f"akakakakakkakakakaka 22 app_conf={type(app_conf)} app_conf:  {app_conf}"
-        )
-
     return app_conf

tests/unit/common.py

@@ -14,9 +14,12 @@
 # limitations under the License.
 #
 
+import os
 import os.path as op
 import socket
-import subprocess
+
+from configparser import ConfigParser
+from io import StringIO
 
 from splunklib import binding, client
 from splunklib.data import record
@@ -32,50 +35,64 @@
 
 
 def mock_splunkhome(monkeypatch):
-    class MockPopen:
-        def __init__(
-            self,
-            args,
-            bufsize=0,
-            executable=None,
-            stdin=None,
-            stdout=None,
-            stderr=None,
-            preexec_fn=None,
-            close_fds=False,
-            shell=False,
-            cwd=None,
-            env=None,
-            universal_newlines=False,
-            startupinfo=None,
-            creationflags=0,
-        ):
-            self._conf = args[3]
-
-        def communicate(self, input=None):
-            if self._conf == "server":
-                file_path = op.sep.join(
-                    [cur_dir, "data/mock_splunk/etc/system/default/server.conf"]
-                )
-            elif self._conf == "inputs":
-                file_path = op.sep.join(
-                    [
-                        cur_dir,
-                        "data/mock_splunk/etc/apps/splunk_httpinput/local/inputs.conf",
-                    ]
-                )
-            else:
-                file_path = op.sep.join(
-                    [cur_dir, "data/mock_splunk/etc/system/default/web.conf"]
+    def get_app_conf(confName, app, use_btool, app_path):
+        if confName == "server":
+            file_path = op.sep.join(
+                [cur_dir, "data/mock_splunk/etc/system/default/server.conf"]
+            )
+        elif confName == "inputs":
+            file_path = op.sep.join(
+                [
+                    cur_dir,
+                    "data/mock_splunk/etc/apps/splunk_httpinput/local/inputs.conf",
+                ]
+            )
+        else:
+            file_path = op.sep.join(
+                [cur_dir, "data/mock_splunk/etc/system/default/web.conf"]
+            )
+
+        with open(file_path) as fp:
+            data = fp.read(), None
+
+        parser = ConfigParser(**{"strict": False})
+        parser.optionxform = str
+        parser.read_file(StringIO(data[0]))
+
+        out = {}
+        for section in parser.sections():
+            out[section] = {
+                item[0]: item[1] for item in parser.items(section, raw=True)
+            }
+        return out
+
+    def make_splunk_gome(parts):
+        relpath = os.path.normpath(os.path.join(*parts))
+        basepath = ""
+        etc_with_trailing_sep = os.path.join("etc", "")
+        if (relpath == "etc") or relpath.startswith(etc_with_trailing_sep):
+            try:
+                basepath = os.environ["SPLUNK_ETC"]
+            except KeyError:
+                basepath = op.join(os.path.normpath(os.environ["SPLUNK_HOME"]), "etc")
+            relpath = relpath[4:]
+        else:
+            basepath = os.path.normpath(os.environ["SPLUNK_HOME"])
+        fullpath = os.path.normpath(os.path.join(basepath, relpath))
+        if os.path.relpath(fullpath, basepath)[0:2] == "..":
+            raise ValueError(
+                'Illegal escape from parent directory "{}": {}'.format(
+                    basepath, fullpath
                 )
+            )
 
-            with open(file_path) as fp:
-                return fp.read(), None
+        return fullpath
 
     splunk_home = op.join(cur_dir, "data/mock_splunk/")
     monkeypatch.setenv("SPLUNK_HOME", splunk_home)
     monkeypatch.setenv("SPLUNK_ETC", op.join(splunk_home, "etc"))
-    monkeypatch.setattr(subprocess, "Popen", MockPopen)
+    monkeypatch.setattr("solnlib.splunkenv.getAppConf", get_app_conf)
+    monkeypatch.setattr("solnlib.splunkenv.mksplhomepath", make_splunk_gome)
 
 
 def mock_serverinfo(monkeypatch):