refactor: make t8s HelmRelease identifiers instance attributes

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):
     """
@@ -69,6 +45,32 @@ def __init__(self, config: dict[str, Any], basepath: str = ".", cwd: str = ".")
         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",
+        }
+        self.hr_values_fixed: dict[str, Any] = {
+            "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},
+            },
+        }
+
         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 +93,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 +107,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 +134,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 +150,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 +172,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"
@@ -178,7 +173,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):