feat: custom cluster configs ✨

.github/workflows/pr.yaml

@@ -34,5 +34,5 @@ jobs:
         run: poetry install --no-interaction --no-root
       - name: Create k8s Kind Cluster
         uses: helm/kind-action@v1
-      - name: Run Tests
+      - name: Run Tests with kind
         run: poetry run pytest -k Testkind -v

README.md

@@ -128,6 +128,23 @@ In this example, the cluster remains active for the entire session and is only d
 
 > Note that `yield` notation that is prefered by pytest to express clean up tasks for this fixture.
 
+#### Cluster configs
+You can pass a cluster config file in the create method of a cluster:
+```python
+    cluster = select_provider_manager("k3d")("my-cluster")
+    # bind ports of this k3d cluster
+    cluster.create(
+        cluster_options=ClusterOptions(
+            cluster_config=Path("my_cluster_config.yaml")
+        )
+    )
+```
+For the different providers you have to submit different kinds of configuration files.
+- kind: https://kind.sigs.k8s.io/docs/user/configuration/#getting-started
+- k3d: https://k3d.io/v5.1.0/usage/configfile/
+- minikube: Has to be a custom yaml file that corresponds to the `minikube config` command. An example can be found in the [fixtures directory](https://github.com/Blueshoe/pytest-kubernetes/tree/main/tests/fixtures/mk_config.yaml) of this repository.
+
+
 #### Special cluster options
 You can pass more options using `kwargs['options']: List[str]` to the `create(options=...)` function when creating the cluster like so:
 ```python
@@ -136,5 +153,6 @@ You can pass more options using `kwargs['options']: List[str]` to the `create(op
     cluster.create(options=["--agents", "1", "-p", "8080:80@agent:0", "-p", "31820:31820/UDP@agent:0"])
 ```
 
+
 ## Examples
 Please find more examples in *tests/vendor.py* in this repository. These test cases are written as users of pytest-kubernetes would write test cases in their projects.

pytest_kubernetes/options.py

@@ -7,4 +7,5 @@ class ClusterOptions:
     api_version: str = field(default="1.25.3")
     # nodes: int = None
     kubeconfig_path: Path | None = None
+    cluster_config: Path | None = None  # Path to a Provider cluster config file
     cluster_timeout: int = field(default=240)

pytest_kubernetes/plugin.py

@@ -1,5 +1,6 @@
 from typing import Dict, Type
 import pytest
+from pytest import FixtureRequest
 
 from pytest_kubernetes.providers import select_provider_manager
 from pytest_kubernetes.providers.base import AClusterManager
@@ -8,7 +9,7 @@
 
 
 @pytest.fixture
-def k8s(request):
+def k8s(request: FixtureRequest):
     """Provide a Kubernetes cluster as test fixture."""
 
     provider = None
@@ -19,6 +20,7 @@ def k8s(request):
         provider = req.get("provider")
         cluster_name = req.get("cluster_name") or cluster_name
         keep = req.get("keep")
+        cluster_config = req.get("cluster_config")
     if not provider:
         provider = provider = request.config.getoption("k8s_provider")
     if not cluster_name:
@@ -31,7 +33,7 @@ def k8s(request):
         manager = cluster_cache[cache_key]
         del cluster_cache[cache_key]
     else:
-        manager: AClusterManager = manager_klass(cluster_name)
+        manager: AClusterManager = manager_klass(cluster_name, cluster_config)
 
     def delete_cluster():
         manager.delete()
@@ -52,17 +54,21 @@ def remaining_clusters_teardown():
 
 
 def pytest_addoption(parser):
-    group = parser.getgroup("k8s")
-    group.addoption(
+    k8s_group = parser.getgroup("k8s")
+    k8s_group.addoption(
         "--k8s-cluster-name",
         default="pytest",
         help="Name of the Kubernetes cluster (default 'pytest').",
     )
-    group.addoption(
+    k8s_group.addoption(
         "--k8s-provider",
         help="The default cluster provider; selects k3d, kind, minikube depending on what is available",
     )
-    group.addoption(
+    k8s_group.addoption(
         "--k8s-version",
         help="The default cluster provider; selects k3d, kind, minikube depending on what is available",
     )
+    k8s_group.addoption(
+        "--k8s-cluster-config",
+        help="Path to a Provider cluster config file",
+    )

pytest_kubernetes/providers/base.py

@@ -56,8 +56,10 @@ class AClusterManager(ABC):
     cluster_name = ""
     context = None
 
-    def __init__(self, cluster_name: str) -> None:
+    def __init__(self, cluster_name: str, cluster_config: str | None = None) -> None:
         self.cluster_name = f"pytest-{cluster_name}"
+        if cluster_config:
+            self._cluster_options.cluster_config = Path(cluster_config)
         self._ensure_executable()
 
     @classmethod
@@ -212,12 +214,18 @@ def create(
         self._on_create(self._cluster_options, **kwargs)
         _i = 0
         # check if this cluster is ready: readyz check passed and default service account is available
+        ready = "Nope"
+        sa_available = "Nope"
         while _i < timeout:
             sleep(1)
             try:
-                ready = self.kubectl(["get", "--raw='/readyz?verbose'"], as_dict=False)
-                sa_available = self.kubectl(
-                    ["get", "sa", "default", "-n", "default"], as_dict=False
+                ready = str(
+                    self.kubectl(["get", "--raw='/readyz?verbose'"], as_dict=False)
+                )
+                sa_available = str(
+                    self.kubectl(
+                        ["get", "sa", "default", "-n", "default"], as_dict=False
+                    )
                 )
             except RuntimeError:
                 _i += 1

pytest_kubernetes/providers/k3d.py

@@ -1,35 +1,67 @@
 from pytest_kubernetes.providers.base import AClusterManager
 from pytest_kubernetes.options import ClusterOptions
+import subprocess
+import yaml
+import re
 
 
 class K3dManager(AClusterManager):
     @classmethod
     def get_binary_name(self) -> str:
         return "k3d"
 
+    @classmethod
+    def get_k3d_version(self) -> str:
+        version_proc = subprocess.run(
+            "k3d --version",
+            shell=True,
+            capture_output=True,
+            check=True,
+            timeout=10,
+        )
+        version_match = re.match(
+            r"k3d version v(\d+\.\d+\.\d+)", version_proc.stdout.decode()
+        )
+        if not version_match:
+            return "0.0.0"
+        return version_match.group(1)
+
     def _translate_version(self, version: str) -> str:
         return f"rancher/k3s:v{version}-k3s1"
 
     def _on_create(self, cluster_options: ClusterOptions, **kwargs) -> None:
         opts = kwargs.get("options", [])
-        self._exec(
-            [
-                "cluster",
-                "create",
+
+        # see https://k3d.io/v5.1.0/usage/configfile/
+        if cluster_options.cluster_config and K3dManager.get_k3d_version() >= "4.0.0":
+            opts += [
+                "--config",
+                str(cluster_options.cluster_config),
+            ]
+            config_yaml = yaml.safe_load(cluster_options.cluster_config.read_text())
+        else:
+            opts += [
+                "--name",
                 self.cluster_name,
                 "--kubeconfig-update-default=0",
                 "--image",
                 self._translate_version(cluster_options.api_version),
                 "--wait",
                 f"--timeout={cluster_options.cluster_timeout}s",
             ]
+
+        self._exec(
+            [
+                "cluster",
+                "create",
+            ]
             + opts
         )
         self._exec(
             [
                 "kubeconfig",
                 "get",
-                self.cluster_name,
+                self.cluster_name if not config_yaml else config_yaml["name"],
                 ">",
                 str(cluster_options.kubeconfig_path),
             ]

pytest_kubernetes/providers/kind.py

@@ -9,17 +9,30 @@ def get_binary_name(self) -> str:
 
     def _on_create(self, cluster_options: ClusterOptions, **kwargs) -> None:
         opts = kwargs.get("options", [])
-        _ = self._exec(
-            [
-                "create",
-                "cluster",
+
+        # see https://kind.sigs.k8s.io/docs/user/configuration/#getting-started
+        if cluster_options.cluster_config:
+            opts += [
+                "--config",
+                str(cluster_options.cluster_config),
+                "--kubeconfig",
+                str(cluster_options.kubeconfig_path),
+            ]
+        else:
+            opts += [
                 "--name",
                 self.cluster_name,
                 "--kubeconfig",
                 str(cluster_options.kubeconfig_path),
                 "--image",
                 f"kindest/node:v{cluster_options.api_version}",
             ]
+
+        _ = self._exec(
+            [
+                "create",
+                "cluster",
+            ]
             + opts
         )
 

pytest_kubernetes/providers/minikube.py

@@ -1,5 +1,6 @@
 from pytest_kubernetes.providers.base import AClusterManager
 from pytest_kubernetes.options import ClusterOptions
+import yaml
 
 
 class MinikubeManager(AClusterManager):
@@ -17,17 +18,42 @@ def load_image(self, image: str) -> None:
 class MinikubeKVM2Manager(MinikubeManager):
     def _on_create(self, cluster_options: ClusterOptions, **kwargs) -> None:
         opts = kwargs.get("options", [])
-        self._exec(
-            [
-                "start",
-                "-p",
-                self.cluster_name,
+
+        if cluster_options.cluster_config:
+            config_yaml = yaml.safe_load(cluster_options.cluster_config.read_text())
+            try:
+                for config in config_yaml["cluster"]["configs"]:
+                    self._exec(
+                        [
+                            "config",
+                            config["name"],
+                            f"{config['value']}",
+                            "-p",
+                            f"{config_yaml['cluster']['profile']}",
+                        ],
+                        additional_env={
+                            "KUBECONFIG": str(cluster_options.kubeconfig_path)
+                        },
+                    )
+            except KeyError as ex:
+                raise ValueError(
+                    f"Missing key: {ex}; cluster_config for minikube setup invalid. Please refer to the docs!"
+                )
+        else:
+            opts += [
                 "--driver",
                 "kvm2",
                 "--embed-certs",
                 "--kubernetes-version",
                 f"v{cluster_options.api_version}",
             ]
+
+        self._exec(
+            [
+                "start",
+                "-p",
+                self.cluster_name,
+            ]
             + opts,
             additional_env={"KUBECONFIG": str(cluster_options.kubeconfig_path)},
         )
@@ -36,17 +62,44 @@ def _on_create(self, cluster_options: ClusterOptions, **kwargs) -> None:
 class MinikubeDockerManager(MinikubeManager):
     def _on_create(self, cluster_options: ClusterOptions, **kwargs) -> None:
         opts = kwargs.get("options", [])
-        self._exec(
-            [
-                "start",
-                "-p",
-                self.cluster_name,
+
+        if cluster_options.cluster_config:
+            config_yaml = yaml.safe_load(cluster_options.cluster_config.read_text())
+            try:
+                for config in config_yaml["cluster"]["configs"]:
+                    self._exec(
+                        [
+                            "config",
+                            config["name"],
+                            f"{config['value']}",
+                            "-p",
+                            f"{config_yaml['cluster']['profile']}",
+                        ],
+                        additional_env={
+                            "KUBECONFIG": str(cluster_options.kubeconfig_path)
+                        },
+                    )
+            except KeyError as ex:
+                raise ValueError(
+                    f"Missing key: {ex}; cluster_config for minikube setup invalid. Please refer to the docs!"
+                )
+        else:
+            opts += [
                 "--driver",
                 "docker",
                 "--embed-certs",
                 "--kubernetes-version",
                 f"v{cluster_options.api_version}",
             ]
+
+        self._exec(
+            [
+                "start",
+                "-p",
+                self.cluster_name
+                if not config_yaml
+                else config_yaml["cluster"]["profile"],
+            ]
             + opts,
             additional_env={"KUBECONFIG": str(cluster_options.kubeconfig_path)},
         )

tests/fixtures/k3d_config.yaml

@@ -0,0 +1,3 @@
+apiVersion: k3d.io/v1alpha3
+kind: Simple
+name: pytest-k3d-cluster

tests/fixtures/kind_config.yaml

@@ -0,0 +1,3 @@
+kind: Cluster
+apiVersion: kind.x-k8s.io/v1alpha4
+name: pytest-kind-cluster

tests/fixtures/mk_config.yaml

@@ -0,0 +1,10 @@
+cluster:
+  profile: pytest-mk-cluster
+  # config options have to be named the same as in the minikube config docs
+  configs:
+    - name: driver
+      value: kvm2
+    - name: EmbedCerts
+      value: true
+    - name: kubernetes-version
+      value: "v1.25.3"