Fixes for Jsonnet global inventory and inventory dumping

kapitan/cli.py

@@ -94,7 +94,7 @@ def trigger_compile(args):
  jinja2_filters=args.jinja2_filters,
  verbose=args.verbose,
  use_go_jsonnet=args.use_go_jsonnet,
- compose_target_name=args.compose_target_name
+ compose_target_name=args.compose_target_name,
  )
 
 
@@ -119,10 +119,13 @@ def build_parser():
  )
 
  inventory_backend_parser.add_argument(
- "--compose-target-name", "--compose-target-name",
+ "--compose-target-name",
+ "--compose-target-name",
  help="Create same subfolder structure from inventory/targets inside compiled folder",
  action="store_true",
- default=from_dot_kapitan("global", "compose-target-name", from_dot_kapitan("compile", "compose-node-name", False)),
+ default=from_dot_kapitan(
+ "global", "compose-target-name", from_dot_kapitan("compile", "compose-node-name", False)
+ ),
  )
 
  eval_parser = subparser.add_parser("eval", aliases=["e"], help="evaluate jsonnet file")
@@ -395,6 +398,12 @@ def build_parser():
  default=from_dot_kapitan("inventory", "multiline-string-style", "double-quotes"),
  help="set multiline string style to STYLE, default is 'double-quotes'",
  )
+ inventory_parser.add_argument(
+ "--yaml-dump-null-as-empty",
+ default=from_dot_kapitan("inventory", "yaml-dump-null-as-empty", False),
+ action="store_true",
+ help="dumps all none-type entries as empty, default is dumping as 'null'",
+ )
 
  searchvar_parser = subparser.add_parser(
  "searchvar", aliases=["sv"], help="show all inventory files where var is declared"
@@ -601,7 +610,7 @@ def build_parser():
  "validate",
  aliases=["v"],
  help="validates the compile output against schemas as specified in inventory",
- parents=[inventory_backend_parser]
+ parents=[inventory_backend_parser],
  )
  validate_parser.set_defaults(func=schema_validate_compiled, name="validate")
 

kapitan/inventory/inv_reclass.py

@@ -16,8 +16,9 @@
 
 
 class ReclassInventory(Inventory):
-
- def render_targets(self, targets: list[InventoryTarget] = None, ignore_class_notfound: bool = False) -> None:
+ def render_targets(
+ self, targets: list[InventoryTarget] | None = None, ignore_class_notfound: bool = False
+ ) -> None:
  """
  Runs a reclass inventory in inventory_path
  (same output as running ./reclass.py -b inv_base_uri/ --inventory)

kapitan/inventory/inv_reclass_rs.py

@@ -27,7 +27,9 @@ def _make_reclass_rs(self, ignore_class_notfound: bool):
  )
  return reclass_rs.Reclass.from_config(config)
 
- def render_targets(self, targets: list = None, ignore_class_notfound: bool = False):
+ def render_targets(
+ self, targets: list[InventoryTarget] | None = None, ignore_class_notfound: bool = False
+ ):
  try:
  r = self._make_reclass_rs(ignore_class_notfound)
  start = datetime.now()

kapitan/inventory/inventory.py

@@ -10,7 +10,7 @@
 from abc import ABC, abstractmethod
 from pydantic import BaseModel, Field
 from kapitan.errors import KapitanError
-from typing import Annotated, Dict, Any, Optional
+
 logger = logging.getLogger(__name__)
 
 
@@ -24,18 +24,23 @@ class InventoryTarget(BaseModel):
 
 
 class Inventory(ABC):
- def __init__(self, inventory_path: str = "inventory", compose_target_name: bool = False, ignore_class_notfound=False):
+ def __init__(
+ self,
+ inventory_path: str = "inventory",
+ compose_target_name: bool = False,
+ ignore_class_notfound=False,
+ ):
  self.inventory_path = inventory_path
  self.compose_target_name = compose_target_name
- self.targets_path = os.path.join(self.inventory_path, 'targets')
- self.classes_path = os.path.join(self.inventory_path, 'classes')
+ self.targets_path = os.path.join(self.inventory_path, "targets")
+ self.classes_path = os.path.join(self.inventory_path, "classes")
  self.initialised: bool = False
  self.targets: dict[str, InventoryTarget] = {}
- 
+
  self.__initialise(ignore_class_notfound=ignore_class_notfound)
- 
+
  @property
- def inventory(self) -> dict:
+ def inventory(self) -> dict[str, InventoryTarget]:
  """
  get all targets from inventory
  """
@@ -57,7 +62,7 @@ def __initialise(self, ignore_class_notfound) -> bool:
  name = name.replace(os.sep, ".")
  else:
  name, ext = os.path.splitext(file)
- 
+
  if ext not in (".yml", ".yaml"):
  logger.debug(f"ignoring {file}: targets have to be .yml or .yaml files.")
  continue
@@ -69,9 +74,9 @@ def __initialise(self, ignore_class_notfound) -> bool:
  f"Conflicting targets {target.name}: {target.path} and {self.targets[target.name].path}. "
  f"Consider using '--compose-target-name'."
  )
- 
+
  self.targets[target.name] = target
- 
+
  self.render_targets(self.targets, ignore_class_notfound=ignore_class_notfound)
  self.initialised = True
  return self.initialised
@@ -82,16 +87,22 @@ def get_target(self, target_name: str, ignore_class_not_found: bool = False) ->
  """
  return self.targets.get(target_name)
 
- def get_targets(self, target_names: list[str] = [], ignore_class_not_found: bool = False) -> dict:
+ def get_targets(
+ self, target_names: list[str] = [], ignore_class_not_found: bool = False
+ ) -> dict[str, InventoryTarget]:
  """
  helper function to get rendered InventoryTarget objects for multiple targets
  """
- 
+
  if target_names:
- return {target_name: self.targets[target_name] for target_name in target_names if target_name in self.targets}
+ return {
+ target_name: self.targets[target_name]
+ for target_name in target_names
+ if target_name in self.targets
+ }
  else:
  return self.targets
- 
+
  def get_parameters(self, target_names: str | list[str], ignore_class_not_found: bool = False) -> dict:
  """
  helper function to get rendered parameters for single target or multiple targets
@@ -100,10 +111,12 @@ def get_parameters(self, target_names: str | list[str], ignore_class_not_found:
  target = self.get_target(target_names, ignore_class_not_found)
  return target.parameters
 
- return {name: target.parameters for name, target in self.get_targets(target_names)}
+ return {name: target.parameters for name, target in self.get_targets(target_names).items()}
 
  @abstractmethod
- def render_targets(self, targets: list[InventoryTarget] = None, ignore_class_notfound: bool = False) -> None:
+ def render_targets(
+ self, targets: list[InventoryTarget] | None = None, ignore_class_notfound: bool = False
+ ) -> None:
  """
  create the inventory depending on which backend gets used
  """

kapitan/resources.py

@@ -245,7 +245,9 @@ def search_imports(cwd, import_str, search_paths):
  return normalised_path, normalised_path_content.encode()
 
 
-def inventory(search_paths: list, target_name: str = None, inventory_path: str = "./inventory"):
+def inventory(
+ search_paths: list, target_name: str = None, inventory_path: str = "./inventory"
+) -> dict[str, dict]:
  """
  Reads inventory (set by inventory_path) in search_paths.
  set nodes_uri to change reclass nodes_uri the default value
@@ -278,7 +280,7 @@ def inventory(search_paths: list, target_name: str = None, inventory_path: str =
  target = inv.get_target(target_name)
  return target.model_dump()
 
- return inv.inventory
+ return {k: v.model_dump() for k, v in inv.inventory.items()}
 
 
 def generate_inventory(args):
@@ -327,9 +329,13 @@ def get_inventory(inventory_path, ignore_class_notfound: bool = False) -> Invent
  compose_target_name = hasattr(cached.args, "compose_target_name") and cached.args.compose_target_name
  backend = AVAILABLE_BACKENDS.get(backend_id, AVAILABLE_BACKENDS.get("reclass"))
  inventory_backend: Inventory = None
- 
+
  logger.debug(f"Using {backend_id} as inventory backend")
- inventory_backend = backend(inventory_path=inventory_path, compose_target_name=compose_target_name, ignore_class_notfound=ignore_class_notfound)
+ inventory_backend = backend(
+ inventory_path=inventory_path,
+ compose_target_name=compose_target_name,
+ ignore_class_notfound=ignore_class_notfound,
+ )
 
  cached.inv = inventory_backend
  # migrate inventory to selected inventory backend

tests/test_inventory.py

@@ -27,10 +27,20 @@ def test_inventory_target(self):
  inv = inventory(["examples/kubernetes"], "minikube-es")
  self.assertEqual(inv["parameters"]["cluster"]["name"], "minikube")
 
+ def test_inventory_target_type(self):
+ inv = inventory(["examples/kubernetes"], "minikube-es")
+ self.assertIsInstance(inv, dict)
+ self.assertIsInstance(inv["parameters"], dict)
+
  def test_inventory_all_targets(self):
  inv = inventory(["examples/kubernetes"], None)
  self.assertNotEqual(inv.get("minikube-es"), None)
 
+ def test_inventory_all_targets_type(self):
+ inv = inventory(["examples/kubernetes"], None)
+ self.assertIsInstance(inv, dict)
+ self.assertIsInstance(inv["minikube-es"], dict)
+
 
 class InventoryTargetTestReclassRs(InventoryTargetTest):
  def setUp(self):