tests: add tests for omegaconf inv loading and rendering

kapitan/inventory.py

@@ -6,6 +6,7 @@
 
 import logging
 import os
+import tempfile
 from abc import ABC, abstractmethod
 from typing import overload
 
@@ -30,14 +31,10 @@
 class InventoryBackend(ABC):
     inventory_path: str
     ignore_class_notfound: bool
-    compose_node_name: bool
-    targets: list
 
     def __init__(self, inventory_path, ignore_class_notfound=False, compose_node_name=False, targets=[]):
         self.inventory_path = inventory_path
         self.ignore_class_notfound = ignore_class_notfound
-        self.compose_node_name = compose_node_name
-        self.targets = targets
 
     @abstractmethod
     def inventory(self) -> dict:
@@ -61,18 +58,43 @@ def migrate(self):
 
 
 class OmegaConfBackend(InventoryBackend):
-    def __init__(self, *args):
+    targets: list
+    compose_node_name: bool
+    logfile: str
+
+    def __init__(
+        self,
+        inventory_path: str,
+        ignore_class_notfound: bool = False,
+        targets: list = [],
+        compose_node_name: bool = False,
+        logfile: str = "",
+    ):
         logger.debug("Using omegaconf as inventory backend")
         logger.warning("NOTE: OmegaConf inventory is currently in experimental mode.")
-        super().__init__(*args)
+
+        self.targets = targets
+        self.compose_node_name = compose_node_name
+        self.logfile = logfile
+
+        super().__init__(inventory_path, ignore_class_notfound)
 
     def inventory(self):
         return inventory_omegaconf(
-            self.inventory_path, self.ignore_class_notfound, self.targets, self.compose_node_name
+            self.inventory_path,
+            self.ignore_class_notfound,
+            self.targets,
+            self.compose_node_name,
+            self.logfile,
         )
 
     def lint(self):
-        raise NotImplementedError()
+        temp = tempfile.mktemp()
+
+        self.inventory()
+        with open(temp, "r") as f:
+            for line in f.readlines():
+                logger.info(line)
 
     def searchvar(self):
         raise NotImplementedError()

kapitan/omegaconf_inv.py

@@ -19,6 +19,7 @@ def inventory_omegaconf(
     ignore_class_notfound: bool = False,
     targets: list = [],
     compose_node_name: bool = False,
+    logfile: str = "",
 ) -> dict:
     """
     generates inventory from yaml files using OmegaConf
@@ -64,15 +65,17 @@ def inventory_omegaconf(
     # load targets
     for target in selected_targets:
         try:
-            name, config = load_target(target, classes_searchpath, ignore_class_notfound)
+            name, config = load_target(target, classes_searchpath, ignore_class_notfound, logfile)
             inv["nodes"][name] = config
         except Exception as e:
             raise InventoryError(f"{target['name']}: {e}")
 
     return inv
 
 
-def load_target(target: dict, classes_searchpath: str, ignore_class_notfound: bool = False):
+def load_target(
+    target: dict, classes_searchpath: str, ignore_class_notfound: bool = False, logfile: str = ""
+):
     """
     load only one target with all its classes
     """
@@ -126,7 +129,10 @@ def load_target(target: dict, classes_searchpath: str, ignore_class_notfound: bo
         # merge target with loaded classes
         if target_config_parameters:
             target_config_parameters = OmegaConf.unsafe_merge(
-                class_config_parameters, target_config_parameters, list_merge_mode=ListMergeMode.EXTEND
+                class_config_parameters,
+                target_config_parameters,
+                list_merge_mode=ListMergeMode.EXTEND,
+                log_filename=logfile,
             )
         else:
             target_config_parameters = class_config_parameters

pyproject.toml

@@ -52,7 +52,7 @@ jsonschema = "^4.17.3"
 kadet = "^0.2.2"
 python-gnupg = "^0.4.7"
 pyyaml = "^6.0"
-omegaconf = { git = "https://github.com/neXenio/omegaconf.git", branch = "dev" }
+omegaconf = { git = "https://github.com/neXenio/omegaconf.git", branch = "fix-logging-redundant-keys" }
 requests = "^2.28.2"
 six = "^1.16.0"
 toml = "^0.10.2"

tests/test_omegaconf.py

@@ -92,3 +92,131 @@ def test_migration_meta_interpolation(self):
 
         migrated = OmegaConfBackend.migrate(content)
         self.assertEqual(migrated, expected)
+
+
+class OmegaConfInventoryTest(unittest.TestCase):
+    params: dict
+    logfile: str
+
+    @classmethod
+    def setUpClass(cls) -> None:
+        cls.inventory_path = tempfile.mkdtemp()
+
+        target = """
+        classes:
+            # test absolute class
+            - test.class
+        
+        parameters:
+            target_name: target
+            
+            # test value
+            value: test
+            
+            # test absolute interpolation
+            absolute_interpolation: ${value}
+            
+            # test relative interpolation
+            relative:
+                value: test
+                interpolation: ${.value}
+            
+            # test custom resolvers
+            tag: ${tag:TAG}
+            merge: ${merge:${merge1},${merge2}}
+            merge1: 
+                - value1
+            merge2:
+                - value2
+            key: ${key:}
+            full:
+                key: ${fullkey:}
+            
+            # test overwrite prefix    
+            overwrite: false
+            
+            # test redundant keys
+            redundant: redundant
+        """
+
+        testclass = """
+        classes:
+            # test absolute class
+            - .relative
+            
+        parameters:
+            # indicator for success
+            absolute_class: success
+            
+            # test redundant keys
+            redundant: redundant
+            
+            # test overwrite prefix
+            ~overwrite: true
+        """
+
+        relativeclass = """
+        parameters:
+            # indicator for success
+            relative_class: success
+        """
+
+        targets_path = os.path.join(cls.inventory_path, "targets")
+        os.makedirs(targets_path, exist_ok=True)
+        with open(os.path.join(targets_path, "target.yml"), "w") as f:
+            f.write(target)
+
+        classes_path = os.path.join(cls.inventory_path, "classes", "test")
+        os.makedirs(classes_path, exist_ok=True)
+        with open(os.path.join(classes_path, "class.yml"), "w") as f:
+            f.write(testclass)
+        with open(os.path.join(classes_path, "relative.yml"), "w") as f:
+            f.write(relativeclass)
+
+        _, cls.logfile = tempfile.mkstemp(suffix=".log")
+
+        # get inventory
+        backend = OmegaConfBackend(cls.inventory_path, False, [], cls.logfile)
+        inventory = backend.inventory()
+        cls.params = inventory["nodes"]["target"]["parameters"]
+
+    def test_absolute_class(self):
+        self.assertTrue(self.params.get("absolute_class"))
+
+    def test_relative_class(self):
+        self.assertTrue(self.params.get("relative_class"))
+
+    def test_value(self):
+        self.assertEqual(self.params.get("value"), "test")
+
+    def test_absolute_interpolation(self):
+        self.assertEqual(self.params.get("absolute_interpolation"), "test")
+
+    def test_relative_interpolation(self):
+        self.assertEqual(self.params.get("relative").get("interpolation"), "test")
+
+    def test_absolute_class(self):
+        self.assertEqual(self.params.get("value"), "test")
+
+    def test_absolute_class(self):
+        self.assertEqual(self.params.get("tag"), "${TAG}")
+        self.assertEqual(self.params.get("merge"), ["value1", "value2"])
+        self.assertEqual(self.params.get("key"), "key")
+        self.assertEqual(self.params.get("full").get("key"), "full.key")
+
+    def test_overwrite_prefix(self):
+        self.assertTrue(self.params.get("overwrite"))
+
+    def test_meta_data(self):
+        meta = self.params["_meta_"]
+        self.assertTrue(meta)
+
+    @unittest.skip
+    def test_redundant_key_check(self):
+        content = ""
+        print(self.logfile)
+        with open(self.logfile, "r") as f:
+            print(f.read())
+
+        expected = "value 'redundant' is defined redundantly in 'redundant'"
+        self.assertEqual(content, expected)