Test kubernetes connection

src/servala/core/models/service.py

@@ -1,7 +1,10 @@
+import kubernetes
 from django.core.exceptions import ValidationError
 from django.db import models
 from django.utils.translation import gettext_lazy as _
 from encrypted_fields.fields import EncryptedJSONField
+from kubernetes import config
+from kubernetes.client.rest import ApiException
 
 
 class ServiceCategory(models.Model):
@@ -108,6 +111,68 @@ class ControlPlane(models.Model):
     def __str__(self):
         return self.name
 
+    @property
+    def kubernetes_config(self):
+        conf = kubernetes.client.Configuration()
+
+        config_dict = {
+            "apiVersion": "v1",
+            "clusters": [
+                {
+                    "cluster": {
+                        "certificate-authority-data": self.api_credentials[
+                            "certificate-authority-data"
+                        ],
+                        "server": self.api_credentials["server"],
+                    },
+                    "name": self.name,
+                },
+            ],
+            "contexts": [
+                {
+                    "context": {
+                        "cluster": self.name,
+                        "namespace": "syn-appcat",  # TODO
+                        "user": "appcat-service-cluster",  # TODO
+                    },
+                    "name": self.name,
+                }
+            ],
+            "current-context": self.name,
+            "kind": "Config",
+            "preferences": {},
+            "users": [
+                {
+                    "name": "appcat-service-cluster",  # TODO
+                    "user": {"token": self.api_credentials["token"]},
+                }
+            ],
+        }
+        config.load_kube_config_from_dict(
+            config_dict=config_dict,
+            client_configuration=conf,
+        )
+        return conf
+
+    def get_kubernetes_client(self):
+        return kubernetes.client.ApiClient(self.kubernetes_config)
+
+    def test_connection(self):
+        if not self.api_credentials:
+            return False, _("No API credentials provided")
+
+        try:
+            v1 = kubernetes.client.CoreV1Api(self.get_kubernetes_client())
+            namespace_count = len(v1.list_namespace().items)
+
+            return True, _(
+                "Successfully connected to Kubernetes API. Found {} namespaces."
+            ).format(namespace_count)
+        except ApiException as e:
+            return False, _("API error: {}").format(str(e))
+        except Exception as e:
+            return False, _("Connection error: {}").format(str(e))
+
 
 class CloudProvider(models.Model):
     """