chore: restructure code and add test cases (#1329)

**Issue number:**
[ADDON-73368](https://splunk.atlassian.net/browse/ADDON-73368)

## Summary

### Changes

> Restructuring of code - moving helper functions in `utils.py` and
cleaning up duplicate code present elsewhere.

### User experience

> No change, the build process would work as it did before.

## Checklist

If your change doesn't seem to apply, please leave them unchecked.

* [x] I have performed a self-review of this change
* [x] Changes have been tested
* [ ] Changes are documented
* [x] PR title follows [conventional commit
semantics](https://www.conventionalcommits.org/en/v1.0.0/)

---------

Co-authored-by: renovate[bot] <29139614+renovate[bot]@users.noreply.github.com>
Co-authored-by: Artem Rys <[email protected]>
Co-authored-by: sgoral-splunk <[email protected]>

mkdocs.yml

@@ -43,6 +43,7 @@ nav:
   - Home: "index.md"
   - Quickstart: "quickstart.md"
   - ".conf files": "dot_conf_files.md"
+  - Generated files: "generated_files.md"
   - Inputs:
       - "inputs/index.md"
       - Introduction: "inputs/index.md"

splunk_add_on_ucc_framework/commands/modular_alert_builder/alert_actions_merge.py

@@ -14,8 +14,6 @@
 # limitations under the License.
 #
 import logging
-import os
-from os.path import basename as bn
 from typing import Any
 
 import addonfactory_splunk_conf_parser_lib as conf_parser
@@ -27,9 +25,6 @@
     arf_consts as ac,
 )
 
-merge_deny_list = ["default.meta", "README.txt"]
-merge_mode_config = {"app.conf": "item_overwrite"}
-
 logger = logging.getLogger("ucc_gen")
 
 
@@ -64,48 +59,3 @@ def remove_alert_from_conf_file(alert: Any, conf_file: str) -> None:
 
     with open(conf_file, "w") as cf:
         parser.write(cf)
-
-
-def merge_conf_file(
-    src_file: str, dst_file: str, merge_mode: str = "stanza_overwrite"
-) -> None:
-    if not os.path.isfile(src_file):
-        return
-    if not os.path.isfile(dst_file):
-        return
-    if bn(src_file) in merge_deny_list:
-        return
-
-    sparser = conf_parser.TABConfigParser()
-    sparser.read(src_file)
-    src_dict = sparser.item_dict()
-    parser = conf_parser.TABConfigParser()
-    parser.read(dst_file)
-    dst_dict = parser.item_dict()
-
-    if merge_mode == "stanza_overwrite":
-        for stanza, key_values in list(src_dict.items()):
-            if stanza not in dst_dict:
-                parser.add_section(stanza)
-            else:
-                parser.remove_section(stanza)
-                parser.add_section(stanza)
-
-            for k, v in list(key_values.items()):
-                parser.set(stanza, k, v)
-    elif merge_mode == "item_overwrite":
-        for stanza, key_values in list(src_dict.items()):
-            if stanza not in dst_dict:
-                parser.add_section(stanza)
-
-            for k, v in list(key_values.items()):
-                if v:
-                    parser.set(stanza, k, v)
-                else:
-                    parser.remove_option(stanza, k)
-    else:
-        # overwrite the whole file
-        parser.read(src_file)
-
-    with open(dst_file, "w") as df:
-        parser.write(df)

splunk_add_on_ucc_framework/commands/modular_alert_builder/alert_actions_py_gen.py

@@ -23,9 +23,7 @@
 from splunk_add_on_ucc_framework.commands.modular_alert_builder import (
     arf_consts as ac,
 )
-from splunk_add_on_ucc_framework.commands.modular_alert_builder.alert_actions_helper import (
-    write_file,
-)
+from splunk_add_on_ucc_framework.utils import write_file
 
 logger = logging.getLogger("ucc_gen")
 

splunk_add_on_ucc_framework/generators/file_generator.py

@@ -20,7 +20,7 @@
 
 from jinja2 import Environment, FileSystemLoader, select_autoescape
 
-from splunk_add_on_ucc_framework.commands.modular_alert_builder.alert_actions_helper import (
+from splunk_add_on_ucc_framework.utils import (
     write_file,
 )
 from splunk_add_on_ucc_framework.commands.rest_builder.global_config_builder_schema import (

splunk_add_on_ucc_framework/utils.py

@@ -14,23 +14,27 @@
 # limitations under the License.
 #
 import json
-import os
+import logging
 import shutil
+from os import listdir, makedirs, path, remove, sep
+from os.path import basename as bn
+from os.path import dirname, exists, isdir, join
 from typing import Any, Dict
 
+import addonfactory_splunk_conf_parser_lib as conf_parser
 import dunamai
 import jinja2
 import yaml
 
 from splunk_add_on_ucc_framework import exceptions
 
+logger = logging.getLogger("ucc_gen")
+
 
 def get_j2_env() -> jinja2.Environment:
     # nosemgrep: splunk.autoescape-disabled, python.jinja2.security.audit.autoescape-disabled.autoescape-disabled
     return jinja2.Environment(
-        loader=jinja2.FileSystemLoader(
-            os.path.join(os.path.dirname(__file__), "templates")
-        )
+        loader=jinja2.FileSystemLoader(join(dirname(__file__), "templates"))
     )
 
 
@@ -44,17 +48,15 @@ def recursive_overwrite(src: str, dest: str, ui_source_map: bool = False) -> Non
         ui_source_map (bool): flag that decides if source map files should be copied
     """
     # TODO: move to shutil.copytree("src", "dst", dirs_exist_ok=True) when Python 3.8+.
-    if os.path.isdir(src):
-        if not os.path.isdir(dest):
-            os.makedirs(dest)
-        files = os.listdir(src)
+    if isdir(src):
+        if not isdir(dest):
+            makedirs(dest)
+        files = listdir(src)
         for f in files:
-            recursive_overwrite(
-                os.path.join(src, f), os.path.join(dest, f), ui_source_map
-            )
+            recursive_overwrite(join(src, f), join(dest, f), ui_source_map)
     else:
-        if os.path.exists(dest):
-            os.remove(dest)
+        if exists(dest):
+            remove(dest)
 
         if (".js.map" not in dest) or ui_source_map:
             shutil.copy(src, dest)
@@ -72,11 +74,11 @@ def get_os_path(path: str) -> str:
     """
 
     if "\\\\" in path:
-        path = path.replace("\\\\", os.sep)
+        path = path.replace("\\\\", sep)
     else:
-        path = path.replace("\\", os.sep)
-    path = path.replace("/", os.sep)
-    return path.strip(os.sep)
+        path = path.replace("\\", sep)
+    path = path.replace("/", sep)
+    return path.strip(sep)
 
 
 def dump_json_config(config: Dict[Any, Any], file_path: str) -> None:
@@ -104,3 +106,84 @@ def get_version_from_git() -> str:
     except ValueError:
         raise exceptions.CouldNotVersionFromGitException()
     return f"{version.base}{stage}{version.commit}"
+
+
+def merge_conf_file(
+    src_file: str, dst_file: str, merge_mode: str = "stanza_overwrite"
+) -> None:
+    merge_deny_list = ["default.meta", "README.txt"]
+    if bn(src_file) in merge_deny_list:
+        return
+
+    sparser = conf_parser.TABConfigParser()
+    sparser.read(src_file)
+    src_dict = sparser.item_dict()
+    parser = conf_parser.TABConfigParser()
+    parser.read(dst_file)
+    dst_dict = parser.item_dict()
+
+    if merge_mode == "stanza_overwrite":
+        for stanza, key_values in src_dict.items():
+            if stanza not in dst_dict:
+                parser.add_section(stanza)
+            else:
+                parser.remove_section(stanza)
+                parser.add_section(stanza)
+
+            for k, v in key_values.items():
+                parser.set(stanza, k, v)
+    elif merge_mode == "item_overwrite":
+        for stanza, key_values in src_dict.items():
+            if stanza not in dst_dict:
+                parser.add_section(stanza)
+
+            for k, v in key_values.items():
+                if v:
+                    parser.set(stanza, k, v)
+                else:
+                    parser.remove_option(stanza, k)
+    else:
+        # overwrite the whole file
+        parser.read(src_file)
+
+    with open(dst_file, "w") as df:
+        parser.write(df)
+
+
+def write_file(file_name: str, file_path: str, content: str, **kwargs: str) -> None:
+    """
+    :param merge_mode: only supported for .conf and .conf.spec files.
+    """
+    logger.debug('operation="write", object="%s" object_type="file"', file_path)
+
+    merge_mode = kwargs.get("merge_mode", "stanza_overwrite")
+    do_merge = False
+    if file_name.endswith(".conf") or file_name.endswith(".conf.spec"):
+        do_merge = True
+    else:
+        logger.debug(
+            f"'{file_name}' is not going to be merged, only .conf and "
+            f".conf.spec files are supported."
+        )
+
+    new_file = None
+    if path.exists(file_path) and do_merge:
+        new_file = path.join(path.dirname(file_path), "new_" + file_name)
+    if new_file:
+        try:
+            with open(new_file, "w+") as fhandler:
+                fhandler.write(content)
+            merge_conf_file(new_file, file_path, merge_mode=merge_mode)
+        finally:
+            if path.exists(new_file):
+                remove(new_file)
+    else:
+        if not path.exists(path.dirname(file_path)):
+            makedirs(path.dirname(file_path))
+        with open(file_path, "w+") as fhandler:
+            fhandler.write(content)
+        if do_merge:
+            parser = conf_parser.TABConfigParser()
+            parser.read(file_path)
+            with open(file_path, "w") as df:
+                parser.write(df)