added discriminator

kapitan/inputs/helm.py

@@ -33,9 +33,9 @@ class Helm(InputType):
     def __init__(self, compile_path, search_paths, ref_controller, args):
         super().__init__("helm", compile_path, search_paths, ref_controller)
 
-        self.helm_values_files = args.get("helm_values_files")
-        self.helm_params = args.get("helm_params") or {}
-        self.helm_path = args.get("helm_path")
+        self.helm_values_files = args.helm_values_files
+        self.helm_params = args.helm_params
+        self.helm_path = args.helm_path
         self.file_path = None
 
         self.helm_values_file = None

kapitan/inventory/model/__init__.py

@@ -1,5 +1,5 @@
 from enum import StrEnum, auto
-from typing import List, Optional
+from typing import Annotated, List, Literal, Optional, Union
 
 from pydantic import BaseModel, ConfigDict, Field
 
@@ -52,7 +52,7 @@ class KapitanSecretsConfig(BaseModel):
     azkms: Optional[KapitanSecretsAZKMSConfig] = None
 
 
-class InputType(StrEnum):
+class InputTypes(StrEnum):
     JSONNET = auto()
     JINJA2 = auto()
     HELM = auto()
@@ -71,60 +71,73 @@ class OutputType(StrEnum):
 
 
 class KapitanDependencyTypes(StrEnum):
-    HELM = auto()
-    HTTP = auto()
-    HTTPS = auto()
-    GIT = auto()
+    HELM = "helm"
+    HTTP = "http"
+    HTTPS = "https"
+    GIT = "git"
 
 
 class KapitanCompileBaseConfig(BaseModel):
+    model_config = ConfigDict(extra="forbid")
+    name: Optional[str] = None
+    input_type: InputTypes
     output_path: str
-    input_paths: List[str]
-    input_type: InputType
+    input_params: dict = {}
+    continue_on_compile_error: bool = False
+
     output_type: OutputType = OutputType.YAML
     ignore_missing: bool = False
     prune: bool = True
-    continue_on_compile_error: bool = False
 
 
 class KapitanCompileExternalConfig(KapitanCompileBaseConfig):
+    input_type: Literal[InputTypes.EXTERNAL] = InputTypes.EXTERNAL
     env_vars: dict[str, str] = {}
-    input_type: InputType = InputType.EXTERNAL
-    input_params: dict = {}
+    args: List[str] = []
+    input_paths: List[str]
 
 
 class KapitanCompileCopy(KapitanCompileBaseConfig):
-    input_type: InputType = InputType.COPY
+    input_type: Literal[InputTypes.COPY] = InputTypes.COPY
+    input_paths: List[str]
     ignore_missing: bool = False
 
 
 class KapitanCompileJinja2Config(KapitanCompileBaseConfig):
-    input_type: InputType = InputType.JINJA2
-    output_type: OutputType = OutputType.PLAIN
-    input_params: dict = {}
-    ignore_missing: bool = True
-    suffix_remove: bool = False
-    suffix_stripped: str = ".j2"
+    input_type: Literal[InputTypes.JINJA2] = InputTypes.JINJA2
+    input_paths: List[str]
+    output_type: Optional[OutputType] = OutputType.PLAIN
+    ignore_missing: Optional[bool] = True
+    suffix_remove: Optional[bool] = False
+    suffix_stripped: Optional[str] = ".j2"
 
 
 class KapitanCompileHelmConfig(KapitanCompileBaseConfig):
-    input_type: InputType = InputType.HELM
-    output_type: OutputType = OutputType.YAML
-    input_params: dict = {}
-    helm_values: dict = {}
+    input_type: Literal[InputTypes.HELM] = InputTypes.HELM
     helm_params: dict = {}
+    helm_values: Optional[dict] = {}
+    helm_values_files: Optional[List[str]] = []
+    helm_path: Optional[str] = None
+    input_paths: List[str]
 
 
 class KapitanCompileJsonnetConfig(KapitanCompileBaseConfig):
-    input_type: InputType = InputType.JSONNET
+    input_type: Literal[InputTypes.JSONNET] = InputTypes.JSONNET
     output_type: OutputType = OutputType.JSON
-    input_params: dict = {}
+    input_paths: List[str]
+    input_value: Optional[dict] = None
 
 
 class KapitanCompileKadetConfig(KapitanCompileBaseConfig):
-    input_type: InputType = InputType.KADET
+    input_type: Literal[InputTypes.KADET] = InputTypes.KADET
     output_type: OutputType = OutputType.YAML
-    input_params: dict = {}
+    input_paths: List[str]
+    input_value: Optional[dict] = None
+
+
+class KapitanCompileRemoveConfig(KapitanCompileBaseConfig):
+    input_type: Literal[InputTypes.REMOVE]
+    input_paths: List[str]
 
 
 class KapitanEssentialVars(BaseModel):
@@ -138,46 +151,48 @@ class KapitanDependencyBaseConfig(BaseModel):
     output_path: str
 
 
-class KapitanDependendencyHelmConfig(KapitanDependencyBaseConfig):
-    type: KapitanDependencyTypes = KapitanDependencyTypes.HELM
-    version: str
+class KapitanDependencyHelmConfig(KapitanDependencyBaseConfig):
+    type: Literal[KapitanDependencyTypes.HELM] = KapitanDependencyTypes.HELM
     chart_name: str
+    version: Optional[str] = None
     helm_path: Optional[str] = None
 
 
-class KapitanDependendencyGitConfig(KapitanDependencyBaseConfig):
-    type: KapitanDependencyTypes = KapitanDependencyTypes.GIT
+class KapitanDependencyGitConfig(KapitanDependencyBaseConfig):
+    type: Literal[KapitanDependencyTypes.GIT] = KapitanDependencyTypes.GIT
     ref: Optional[str] = "master"
     subdir: Optional[str] = None
     submodules: Optional[bool] = False
 
 
-class KapitanDependendencyHttpConfig(KapitanDependencyBaseConfig):
-    type: KapitanDependencyTypes = KapitanDependencyTypes.HTTP
+class KapitanDependencyHttpsConfig(KapitanDependencyBaseConfig):
+    type: Literal[KapitanDependencyTypes.HTTPS, KapitanDependencyTypes.HTTP]
     unpack: bool = False
 
 
-class KapitanDependendencyHttpsConfig(KapitanDependendencyHttpConfig):
-    type: KapitanDependencyTypes = KapitanDependencyTypes.HTTPS
+CompileInputTypeConfig = Annotated[
+    Union[
+        KapitanCompileJinja2Config,
+        KapitanCompileExternalConfig,
+        KapitanCompileCopy,
+        KapitanCompileKadetConfig,
+        KapitanCompileJsonnetConfig,
+        KapitanCompileHelmConfig,
+        KapitanCompileRemoveConfig,
+    ],
+    Field(discriminator="input_type"),
+]
 
+DependencyTypeConfig = Union[
+    KapitanDependencyHelmConfig, KapitanDependencyHttpsConfig, KapitanDependencyGitConfig
+]
 
-class KapitanInventorySettings(BaseModel):
 
-    compile: List[
-        KapitanCompileJinja2Config
-        | KapitanCompileExternalConfig
-        | KapitanCompileKadetConfig
-        | KapitanCompileJsonnetConfig
-        | KapitanCompileHelmConfig
-    ] = []
+class KapitanInventorySettings(BaseModel):
+    compile: List[CompileInputTypeConfig] = []
     vars: KapitanEssentialVars = KapitanEssentialVars()
     labels: dict[str, str] = {}
-    dependencies: List[
-        KapitanDependendencyHelmConfig
-        | KapitanDependendencyHttpConfig
-        | KapitanDependendencyHttpsConfig
-        | KapitanDependendencyGitConfig
-    ] = []
+    dependencies: List[DependencyTypeConfig] = []
     target_full_path: str = ""
     secrets: Optional[KapitanSecretsConfig] = None
     validate_: list[dict] = Field(alias="validate", default=[])

kapitan/targets.py

@@ -27,6 +27,7 @@
 from kapitan.inputs.jsonnet import Jsonnet
 from kapitan.inputs.kadet import Kadet
 from kapitan.inputs.remove import Remove
+from kapitan.inventory.model import InputTypes
 from kapitan.resources import get_inventory
 
 logger = logging.getLogger(__name__)
@@ -283,25 +284,23 @@ def compile_target(target_obj, search_paths, compile_path, ref_controller, globa
         input_params = comp_obj.input_params
         continue_on_compile_error = comp_obj.continue_on_compile_error
 
-        if input_type == "jinja2":
+        if input_type == input_type.JINJA2:
             input_compiler = Jinja2(compile_path, search_paths, ref_controller, comp_obj)
-        elif input_type == "jsonnet":
+        elif input_type == input_type.JSONNET:
             input_compiler = Jsonnet(compile_path, search_paths, ref_controller, use_go=use_go_jsonnet)
-        elif input_type == "kadet":
+        elif input_type == input_type.KADET:
             input_compiler = Kadet(compile_path, search_paths, ref_controller, input_params=input_params)
-        elif input_type == "helm":
+        elif input_type == input_type.HELM:
             input_compiler = Helm(compile_path, search_paths, ref_controller, comp_obj)
-        elif input_type == "copy":
+        elif input_type == input_type.COPY:
             ignore_missing = comp_obj.ignore_missing
             input_compiler = Copy(compile_path, search_paths, ref_controller, ignore_missing)
-        elif input_type == "remove":
+        elif input_type == input_type.REMOVE:
             input_compiler = Remove(compile_path, search_paths, ref_controller)
-        elif input_type == "external":
+        elif input_type == input_type.EXTERNAL:
             input_compiler = External(compile_path, search_paths, ref_controller)
-            if "args" in comp_obj:
-                input_compiler.set_args(comp_obj.args)
-            if "env_vars" in comp_obj:
-                input_compiler.set_env_vars(comp_obj.env_vars)
+            input_compiler.set_args(comp_obj.args)
+            input_compiler.set_env_vars(comp_obj.env_vars)
         else:
             err_msg = 'Invalid input_type: "{}". Supported input_types: jsonnet, jinja2, kadet, helm, copy, remove, external'
             raise CompileError(err_msg.format(input_type))

tests/test_dependency_manager.py

@@ -23,6 +23,12 @@
     fetch_http_source,
 )
 from kapitan.errors import HelmFetchingError
+from kapitan.inventory.model import (
+    KapitanDependencyGitConfig,
+    KapitanDependencyHelmConfig,
+    KapitanDependencyHttpsConfig,
+    KapitanInventorySettings,
+)
 
 
 class DependencyManagerTest(unittest.TestCase):
@@ -68,11 +74,15 @@ def test_clone_repo_subdir(self):
         output_dir = tempfile.mkdtemp()
         source = "https://github.com/kapicorp/kapitan.git"
         dep = [
-            {
-                "output_path": os.path.join(output_dir, "subdir"),
-                "ref": "master",
-                "subdir": "tests",
-            }
+            KapitanDependencyGitConfig(
+                **{
+                    "type": "git",
+                    "source": source,
+                    "output_path": os.path.join(output_dir, "subdir"),
+                    "ref": "master",
+                    "subdir": "tests",
+                }
+            )
         ]
         fetch_git_dependency((source, dep), temp_dir, force=False)
         self.assertTrue(os.path.isdir(os.path.join(output_dir, "subdir")))
@@ -90,12 +100,14 @@ def test_fetch_helm_chart(self):
         version = "11.3.0"
         repo = "https://github.com/BurdenBear/kube-charts-mirror/raw/master/docs/"
         dep = [
-            {
-                "output_path": output_chart_dir,
-                "version": version,
-                "chart_name": chart_name,
-                "source": repo,
-            }
+            KapitanDependencyHelmConfig(
+                **{
+                    "output_path": output_chart_dir,
+                    "version": version,
+                    "chart_name": chart_name,
+                    "source": repo,
+                }
+            )
         ]
         fetch_helm_chart((HelmSource(repo, chart_name, version, None), dep), temp_dir, force=False)
         self.assertTrue(os.path.isdir(output_chart_dir))
@@ -115,12 +127,14 @@ def test_fetch_helm_chart_version_that_does_not_exist(self):
         version = "10.7.0"
         repo = "https://github.com/BurdenBear/kube-charts-mirror/raw/master/docs/"
         dep = [
-            {
-                "output_path": output_chart_dir,
-                "version": version,
-                "chart_name": chart_name,
-                "source": repo,
-            }
+            KapitanDependencyHelmConfig(
+                **{
+                    "output_path": output_chart_dir,
+                    "version": version,
+                    "chart_name": chart_name,
+                    "source": repo,
+                }
+            )
         ]
         with self.assertRaises(HelmFetchingError):
             fetch_helm_chart((HelmSource(repo, chart_name, version, None), dep), temp_dir, force=False)
@@ -134,34 +148,33 @@ def test_fetch_dependencies_unpack_parallel(self):
         save_dir = tempfile.mkdtemp()
         # use default parallelism of 4 for test
         with multiprocessing.Pool(4) as pool:
-            target_objs = [
+            dependencies = [
                 {
-                    "dependencies": [
-                        {
-                            "type": "https",
-                            "source": "https://github.com/BurdenBear/kube-charts-mirror/raw/master/docs/nfs-client-provisioner-1.2.8.tgz",
-                            "output_path": "nfs-client-provisioner",
-                            "unpack": True,
-                        },
-                        {
-                            "type": "https",
-                            "source": "https://github.com/BurdenBear/kube-charts-mirror/raw/master/docs/prometheus-pushgateway-1.2.13.tgz",
-                            "output_path": "prometheus-pushgateway",
-                            "unpack": True,
-                        },
-                    ]
-                }
+                    "type": "https",
+                    "source": "https://github.com/BurdenBear/kube-charts-mirror/raw/master/docs/nfs-client-provisioner-1.2.8.tgz",
+                    "output_path": "nfs-client-provisioner",
+                    "unpack": True,
+                },
+                {
+                    "type": "https",
+                    "source": "https://github.com/BurdenBear/kube-charts-mirror/raw/master/docs/prometheus-pushgateway-1.2.13.tgz",
+                    "output_path": "prometheus-pushgateway",
+                    "unpack": True,
+                },
             ]
+
+            inventory = KapitanInventorySettings(dependencies=dependencies)
+            target_objs = [inventory]
             try:
                 fetch_dependencies(output_path, target_objs, save_dir, False, pool)
                 pool.close()
             except Exception as e:
                 pool.terminate()
                 raise e
 
-        for obj in target_objs[0]["dependencies"]:
-            self.assertTrue(os.path.isdir(os.path.join(output_path, obj["output_path"])))
-            self.assertTrue(os.path.isdir(os.path.join(save_dir, obj["output_path"])))
+        for obj in target_objs[0].dependencies:
+            self.assertTrue(os.path.isdir(os.path.join(output_path, obj.output_path)))
+            self.assertTrue(os.path.isdir(os.path.join(save_dir, obj.output_path)))
         rmtree(output_path)
         rmtree(save_dir)
 

tests/test_omegaconf.py

@@ -61,10 +61,10 @@ def test_load_and_resolve_single_target(self):
         # Loads the target using the inventory
         inventory.load_target(target)
 
-        self.assertDictEqual(target_kapitan_metadata["_kapitan_"], target.parameters["_kapitan_"])
-        self.assertEqual(target.parameters["_kapitan_"]["name"]["short"], "minikube")
-        self.assertEqual(target.parameters["target_name"], "minikube-es")
-        self.assertEqual(target.parameters["kubectl"]["insecure_skip_tls_verify"], False)
+        self.assertDictEqual(target_kapitan_metadata["_kapitan_"], target.parameters._kapitan_)
+        self.assertEqual(target.parameters._kapitan_.name.short, "minikube")
+        self.assertEqual(target.parameters.target_name, "minikube-es")
+        self.assertEqual(target.parameters.kubectl.insecure_skip_tls_verify, False)
 
     def tearDown(self) -> None:
         shutil.rmtree(self.temp_dir)