refactor: make t8s HelmRelease identifiers instance attributes

Signed-off-by: Chris Werner Rau <cwrau@cwrau.info>

Author: cwrau

Tests/kaas/plugin/plugin_t8s.py

@@ -22,30 +22,6 @@
 HR_VERSION = "v2"
 HR_PLURAL = "helmreleases"
 
-# All of these are mandated by the ValidatingAdmissionPolicy on the management cluster
-# and cannot be changed without updating both the VAP and the RBAC.
-HR_NAME = "scs-kaas-certification"
-HR_NAMESPACE = "scs-kaas-certification"
-HR_KUBECONFIG_SECRET = "scs-kaas-certification-kubeconfig"
-HR_CHART_REF = {
-    "kind": "HelmChart",
-    "name": "scs-kaas-certification",
-    "namespace": "flux-system",
-}
-HR_VALUES_FIXED = {
-    "cloud": "bfe2-prod",
-    "controlPlane": {"hosted": True},
-    "metadata": {
-        "customerID": 1111,
-        "customerName": "teuto.net Netzdienste GmbH",
-        "friendlyName": "scs-kaas-certification",
-        "serviceLevelAgreement": "None",
-    },
-    "nodePools": {
-        "pool-0": {"flavor": "standard.2.1905", "replicas": 2},
-    },
-}
-
 
 class PluginT8s(KubernetesClusterPlugin):
     """
@@ -62,13 +38,51 @@ class PluginT8s(KubernetesClusterPlugin):
         kubeconfig: t8s-kubeconfig.yaml  (management cluster kubeconfig template)
       secrets:
         token: '{{ clouds_conf.teuto_mgmt_token }}'
+      cloud:             'bfe2-prod'    (optional, defaults to 'bfe2-prod')
+      controlPlaneHosted: true          (optional, defaults to true)
+      nodePools:                        (optional, defaults to a single 'pool-0')
+        pool-0:
+          flavor: 'standard.2.1905'
+          replicas: 2
+      metadata:                         (optional; individual keys default as shown)
+        customerID:             1111
+        customerName:           'teuto.net Netzdienste GmbH'
+        friendlyName:           'scs-kaas-certification'
+        serviceLevelAgreement:  'None'
     """
 
     def __init__(self, config: dict[str, Any], basepath: str = ".", cwd: str = ".") -> None:
         self.basepath = basepath
         self.cwd = cwd
         self.env = Environment()
 
+        # All of these are mandated by the ValidatingAdmissionPolicy on the management cluster
+        # and cannot be changed without updating both the VAP and the RBAC. They are instance
+        # (rather than module-level) attributes because, in principle, different certification
+        # targets could be configured with different values here.
+        self.hr_name = "scs-kaas-certification"
+        self.hr_namespace = "scs-kaas-certification"
+        self.hr_kubeconfig_secret = "scs-kaas-certification-kubeconfig"
+        self.hr_chart_ref: dict[str, str] = {
+            "kind": "HelmChart",
+            "name": "scs-kaas-certification",
+            "namespace": "flux-system",
+        }
+        metadata_config: dict[str, Any] = config.get("metadata", {})
+        self.hr_values_fixed: dict[str, Any] = {
+            "cloud": config.get("cloud", "bfe2-prod"),
+            "controlPlane": {"hosted": config.get("controlPlaneHosted", True)},
+            "metadata": {
+                "customerID": metadata_config.get("customerID", 1111),
+                "customerName": metadata_config.get("customerName", "teuto.net Netzdienste GmbH"),
+                "friendlyName": metadata_config.get("friendlyName", "scs-kaas-certification"),
+                "serviceLevelAgreement": metadata_config.get("serviceLevelAgreement", "None"),
+            },
+            "nodePools": config.get("nodePools", {
+                "pool-0": {"flavor": "standard.2.1905", "replicas": 2},
+            }),
+        }
+
         fn: str = config["templates"]["kubeconfig"]
         with open(os.path.join(basepath, fn), "r") as f:
             self.kubeconfig_template: Template = self.env.from_string(f.read())
@@ -91,12 +105,12 @@ def _build_helmrelease(self) -> dict[str, Any]:
         return {
             "apiVersion": f"{HR_GROUP}/{HR_VERSION}",
             "kind": "HelmRelease",
-            "metadata": {"name": HR_NAME, "namespace": HR_NAMESPACE},
+            "metadata": {"name": self.hr_name, "namespace": self.hr_namespace},
             "spec": {
-                "chartRef": HR_CHART_REF,
+                "chartRef": self.hr_chart_ref,
                 "driftDetection": {"mode": "enabled"},
                 "interval": "1m",
-                "values": {**HR_VALUES_FIXED, "version": self.k8s_version},
+                "values": {**self.hr_values_fixed, "version": self.k8s_version},
             },
         }
 
@@ -105,17 +119,17 @@ def _delete_helmrelease(self, co_api: CustomObjectsApi) -> None:
             co_api.delete_namespaced_custom_object(
                 HR_GROUP,
                 HR_VERSION,
-                HR_NAMESPACE,
+                self.hr_namespace,
                 HR_PLURAL,
-                HR_NAME,
+                self.hr_name,
             )
         except ApiException as e:
             if e.status == 404:
-                logger.debug(f"HelmRelease {HR_NAME} not present, nothing to delete")
+                logger.debug(f"HelmRelease {self.hr_name} not present, nothing to delete")
                 return
             raise
         # Give Flux time to begin deletion before we try to recreate
-        logger.debug(f"HelmRelease {HR_NAME} deletion requested; waiting 30s")
+        logger.debug(f"HelmRelease {self.hr_name} deletion requested; waiting 30s")
         time.sleep(30)
 
     def _apply_helmrelease(self, api_client: ApiClient) -> None:
@@ -132,9 +146,9 @@ def _apply_helmrelease(self, api_client: ApiClient) -> None:
             path_params={
                 'group': HR_GROUP,
                 'version': HR_VERSION,
-                'namespace': HR_NAMESPACE,
+                'namespace': self.hr_namespace,
                 'plural': HR_PLURAL,
-                'name': HR_NAME,
+                'name': self.hr_name,
             },
             query_params=[('fieldManager', 'plugin_t8s'), ('force', 'true')],
             header_params={
@@ -148,15 +162,15 @@ def _apply_helmrelease(self, api_client: ApiClient) -> None:
             auth_settings=['BearerToken'],
             _return_http_data_only=True,
         )
-        logger.debug(f"HelmRelease {HR_NAME} applied")
+        logger.debug(f"HelmRelease {self.hr_name} applied")
 
     def _get_kubeconfig_from_secret(self, core_api: CoreV1Api) -> bytes:
-        secret = cast(V1Secret, core_api.read_namespaced_secret(HR_KUBECONFIG_SECRET, HR_NAMESPACE))
+        secret = cast(V1Secret, core_api.read_namespaced_secret(self.hr_kubeconfig_secret, self.hr_namespace))
         data: dict[str, str] = secret.data or {}
         if "value" in data:
             return base64.standard_b64decode(data["value"].encode())
         raise RuntimeError(
-            f"kubeconfig secret {HR_KUBECONFIG_SECRET} is missing key 'value'; has keys: {list(data)}"
+            f"kubeconfig secret {self.hr_kubeconfig_secret} is missing key 'value'; has keys: {list(data)}"
         )
 
     def _wait_for_kubeconfig_secret(self, core_api: CoreV1Api) -> bytes:
@@ -170,10 +184,10 @@ def _wait_for_kubeconfig_secret(self, core_api: CoreV1Api) -> bytes:
             timeout = next(timeouts)
             if not timeout:
                 raise RuntimeError(
-                    f"Timeout waiting for kubeconfig secret {HR_KUBECONFIG_SECRET}"
+                    f"Timeout waiting for kubeconfig secret {self.hr_kubeconfig_secret}"
                 )
             logger.debug(
-                f"waiting {timeout}s for kubeconfig secret {HR_KUBECONFIG_SECRET}"
+                f"waiting {timeout}s for kubeconfig secret {self.hr_kubeconfig_secret}"
             )
             time.sleep(timeout)
 

Tests/kaas/plugin/test_plugin_t8s.py

@@ -8,12 +8,7 @@
 from kubernetes.client import ApiException, V1Secret
 
 from plugin_t8s import (
-    HR_CHART_REF,
     HR_GROUP,
-    HR_NAME,
-    HR_NAMESPACE,
-    HR_KUBECONFIG_SECRET,
-    HR_VALUES_FIXED,
     HR_VERSION,
     PluginT8s,
 )
@@ -62,13 +57,13 @@ def test_helmrelease_kind(plugin):
 
 def test_helmrelease_name_namespace(plugin):
     hr = plugin._build_helmrelease()
-    assert hr["metadata"]["name"] == HR_NAME
-    assert hr["metadata"]["namespace"] == HR_NAMESPACE
+    assert hr["metadata"]["name"] == plugin.hr_name
+    assert hr["metadata"]["namespace"] == plugin.hr_namespace
 
 
 def test_helmrelease_chartref(plugin):
     hr = plugin._build_helmrelease()
-    assert hr["spec"]["chartRef"] == HR_CHART_REF
+    assert hr["spec"]["chartRef"] == plugin.hr_chart_ref
     assert hr["spec"]["chartRef"]["kind"] == "HelmChart"
     assert hr["spec"]["chartRef"]["name"] == "scs-kaas-certification"
     assert hr["spec"]["chartRef"]["namespace"] == "flux-system"
@@ -135,6 +130,45 @@ def test_helmrelease_single_nodepool(plugin):
     assert "pool-0" in nodepools
 
 
+def _make_plugin(tmp_path, **config_overrides):
+    tpl = tmp_path / "t8s-kubeconfig.yaml"
+    tpl.write_text(KUBECONFIG_TEMPLATE)
+    config = {
+        "kubernetesVersion": "1.35",
+        "version_patch": 2,
+        "templates": {"kubeconfig": tpl.name},
+        "secrets": {},
+        **config_overrides,
+    }
+    with patch("k8s_helper.setup_client_config"):
+        return PluginT8s(config, basepath=str(tmp_path), cwd=str(tmp_path))
+
+
+def test_helmrelease_values_from_config(tmp_path):
+    plugin = _make_plugin(
+        tmp_path,
+        cloud="other-cloud",
+        controlPlaneHosted=False,
+        nodePools={"pool-1": {"flavor": "custom.flavor", "replicas": 5}},
+        metadata={
+            "customerID": 2222,
+            "customerName": "Other Customer GmbH",
+            "friendlyName": "other-friendly-name",
+            "serviceLevelAgreement": "Gold",
+        },
+    )
+    values = plugin._build_helmrelease()["spec"]["values"]
+    assert values["cloud"] == "other-cloud"
+    assert values["controlPlane"]["hosted"] is False
+    assert values["nodePools"] == {"pool-1": {"flavor": "custom.flavor", "replicas": 5}}
+    assert values["metadata"] == {
+        "customerID": 2222,
+        "customerName": "Other Customer GmbH",
+        "friendlyName": "other-friendly-name",
+        "serviceLevelAgreement": "Gold",
+    }
+
+
 # --- version parsing ---
 
 @pytest.mark.parametrize("version,version_patch,expected", [
@@ -178,7 +212,7 @@ def test_get_kubeconfig_from_secret(plugin):
     core_api = MagicMock()
     core_api.read_namespaced_secret.return_value = secret
     assert plugin._get_kubeconfig_from_secret(core_api) == raw
-    core_api.read_namespaced_secret.assert_called_once_with(HR_KUBECONFIG_SECRET, HR_NAMESPACE)
+    core_api.read_namespaced_secret.assert_called_once_with(plugin.hr_kubeconfig_secret, plugin.hr_namespace)
 
 
 def test_get_kubeconfig_from_secret_null_data(plugin):