Simplify value decoding

flux_local/helm.py

@@ -250,7 +250,7 @@ def _find_object(name: str, namespace: str, objects: Sequence[_T]) -> _T | None:
     return None
 
 
-def _decode_config_or_secret_value(name: str, string_data: str | None, binary_data: str | None) -> str:
+def _decode_config_or_secret_value(name: str, string_data: str | None, binary_data: str | None) -> dict[str, str]:
     """Return the config or secret data."""
     if binary_data:
         try:
@@ -266,19 +266,15 @@ def _decode_config_or_secret_value(name: str, string_data: str | None, binary_da
 
 
 
-def _get_secret_data(
-    name: str, namespace: str, ks: Kustomization
-) -> dict[str, Any] | None:
+def _get_secret_data(name: str, namespace: str, ks: Kustomization) -> dict[str, str] | None:
     """Find the secret value in the kustomization."""
     found: Secret | None = _find_object(name, namespace, ks.secrets)
     if not found:
         return None
     return _decode_config_or_secret_value(f"{namespace}/{name}", found.string_data, found.data)
 
 
-def _get_configmap_data(
-    name: str, namespace: str, ks: Kustomization
-) -> dict[str, Any] | None:
+def _get_configmap_data(name: str, namespace: str, ks: Kustomization) -> dict[str, str] | None:
     """Find the configmap value in the kustomization."""
     found: ConfigMap | None = _find_object(name, namespace, ks.config_maps)
     if not found:
@@ -296,41 +292,35 @@ def expand_value_references(
     values = helm_release.values or {}
     for ref in helm_release.values_from:
         _LOGGER.debug("Expanding value reference %s", ref)
-        found_data: dict[str, Any] | None = None
+        found_data: str | None = None
         if ref.kind == SECRET_KIND:
             found_data = _get_secret_data(
                 ref.name, helm_release.namespace, kustomization
             )
-            if not found_data:
-                if not ref.optional:
-                    _LOGGER.warning(
-                        "Unable to find secret %s/%s referenced in HelmRelease %s",
-                        helm_release.namespace,
-                        ref.name,
-                        helm_release.namespaced_name,
-                    )
-                continue
         elif ref.kind == CONFIG_MAP_KIND:
             found_data = _get_configmap_data(
                 ref.name, helm_release.namespace, kustomization
             )
-            if not found_data:
-                if not ref.optional:
-                    _LOGGER.warning(
-                        "Unable to find configmap %s/%s referenced in HelmRelease %s",
-                        helm_release.namespace,
-                        ref.name,
-                        helm_release.namespaced_name,
-                    )
-                continue
         else:
             _LOGGER.warning(
                 "Unsupported valueFrom kind %s in HelmRelease %s",
                 ref.kind,
                 helm_release.namespaced_name,
             )
             continue
-        if (found_data := found_data.get(ref.values_key)) is None:
+
+        if not found_data:
+            if not ref.optional:
+                _LOGGER.warning(
+                    "Unable to find %s %s/%s referenced in HelmRelease %s",
+                    ref.kind.
+                    helm_release.namespace,
+                    ref.name,
+                    helm_release.namespaced_name,
+                )
+            continue
+
+        if (found_value := found_data.get(ref.values_key)) is None:
             _LOGGER.warning(
                 "Unable to find key %s in %s/%s referenced in HelmRelease %s",
                 ref.values_key,
@@ -341,7 +331,6 @@ def expand_value_references(
             continue
 
         if ref.target_path:
-            _LOGGER.debug("Updating %s with %s", ref.target_path, found_data)
             parts = ref.target_path.split(".")
             inner_values = values
             for part in parts[:-1]:
@@ -353,9 +342,9 @@ def expand_value_references(
                     )
                 inner_values = inner_values[part]
 
-            inner_values[parts[-1]] = found_data
+            inner_values[parts[-1]] = found_value
         else:
-            obj = yaml.load(found_data, Loader=yaml.SafeLoader)
+            obj = yaml.load(found_value, Loader=yaml.SafeLoader)
             if not obj or not isinstance(obj, dict):
                 raise HelmException(
                     f"While building HelmRelease '{helm_release.namespaced_name}': Expected '{ref.name}' field '{ref.target_path}' values to be valid yaml, found {type(values)}"

flux_local/manifest.py

@@ -39,6 +39,8 @@
 SECRET_KIND = "Secret"
 CONFIG_MAP_KIND = "ConfigMap"
 DEFAULT_NAMESPACE = "flux-system"
+VALUE_PLACEHOLDER = "**PLACEHOLDER**"
+VALUE_B64_PLACEHOLDER = base64.b64encode(VALUE_PLACEHOLDER.encode())
 
 REPO_TYPE_DEFAULT = "default"
 REPO_TYPE_OCI = "oci"
@@ -372,10 +374,10 @@ def parse_doc(cls, doc: dict[str, Any]) -> "Secret":
         # placeholder values anyway.
         if data := doc.get("data"):
             for key, value in data.items():
-                data[key] = base64.b64encode("**PLACEHOLDER**".encode())
+                data[key] = VALUE_B64_PLACEHOLDER
         if string_data := doc.get("stringData"):
             for key, value in string_data.items():
-                string_data[key] = "**PLACEHOLDER**"
+                string_data[key] = VALUE_PLACEHOLDER
         return Secret(
             name=name, namespace=namespace, data=data, string_data=string_data
         )