Merge pull request #2 from Blueshoe/feat/flexbile-cluster-options

feat: enable flexible cluster options

Author: Schille

README.md

@@ -46,7 +46,7 @@ It provides the following interface:
 - `port_forwarding(...)`: Port forward a target
 - `logs(...)`: Get the logs of a pod
 - `version()`: Get the Kubernetes version of this cluster
-- `create(...)`: Create this cluster
+- `create(...)`: Create this cluster (pass special cluster arguments with `options: List[str]` to the CLI command)
 - `delete()`: Delete this cluster
 - `reset()`: Delete this cluster (if it exists) and create it again
 
@@ -128,5 +128,13 @@ 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.
 
+#### 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
+    cluster = select_provider_manager("k3d")("my-cluster")
+    # bind ports of this k3d cluster
+    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.

pyproject.toml

@@ -1,10 +1,11 @@
 [tool.poetry]
 name = "pytest-kubernetes"
-version = "0.2.0"
+version = "0.3.0"
 description = ""
 authors = ["Michael Schilonka <michael@blueshoe.io>"]
 readme = "README.md"
 packages = [{include = "pytest_kubernetes"}]
+repository = "https://github.com/Blueshoe/pytest-kubernetes"
 classifiers = [
     "Framework :: Pytest",
     "Topic :: Software Development :: Testing",

pytest_kubernetes/options.py

@@ -8,4 +8,4 @@ class ClusterOptions:
     api_version: str = field(default="1.25.3")
     # nodes: int = None
     kubeconfig_path: Optional[Path] = None
-    cluster_timeout: int = field(default=120)
+    cluster_timeout: int = field(default=240)

pytest_kubernetes/providers/base.py

@@ -223,6 +223,7 @@ def delete(self) -> None:
         self._on_delete()
         if self.kubeconfig:
             self.kubeconfig.unlink(missing_ok=True)
+            self._cluster_options.kubeconfig_path = None
         sleep(1)
 
     def reset(self) -> None:

pytest_kubernetes/providers/k3d.py

@@ -11,6 +11,7 @@ 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",
@@ -22,6 +23,7 @@ def _on_create(self, cluster_options: ClusterOptions, **kwargs) -> None:
                 "--wait",
                 f"--timeout={cluster_options.cluster_timeout}s",
             ]
+            + opts
         )
         self._exec(
             [

pytest_kubernetes/providers/kind.py

@@ -8,6 +8,7 @@ def get_binary_name(self) -> str:
         return "kind"
 
     def _on_create(self, cluster_options: ClusterOptions, **kwargs) -> None:
+        opts = kwargs.get("options", [])
         _ = self._exec(
             [
                 "create",
@@ -18,7 +19,7 @@ def _on_create(self, cluster_options: ClusterOptions, **kwargs) -> None:
                 str(cluster_options.kubeconfig_path),
                 "--image",
                 f"kindest/node:v{cluster_options.api_version}",
-            ]
+            ] + opts
         )
 
     def _on_delete(self) -> None:

pytest_kubernetes/providers/minikube.py

@@ -16,6 +16,7 @@ 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",
@@ -26,13 +27,14 @@ def _on_create(self, cluster_options: ClusterOptions, **kwargs) -> None:
                 "--embed-certs",
                 "--kubernetes-version",
                 f"v{cluster_options.api_version}",
-            ],
+            ] + opts,
             additional_env={"KUBECONFIG": str(cluster_options.kubeconfig_path)},
         )
 
 
 class MinikubeDockerManager(MinikubeManager):
     def _on_create(self, cluster_options: ClusterOptions, **kwargs) -> None:
+        opts = kwargs.get("options", [])
         self._exec(
             [
                 "start",
@@ -43,6 +45,6 @@ def _on_create(self, cluster_options: ClusterOptions, **kwargs) -> None:
                 "--embed-certs",
                 "--kubernetes-version",
                 f"v{cluster_options.api_version}",
-            ],
+            ] + opts,
             additional_env={"KUBECONFIG": str(cluster_options.kubeconfig_path)},
         )

tests/test_providers.py

@@ -26,7 +26,13 @@ def test_a_create_simple_cluster(self):
 
     def test_b_reset_cluster(self):
         self.cluster = self.manager(self.cluster_name)
+        self.cluster.create()
+        self.cluster.kubectl(["get", "nodes"])
+        kubeconfig1 = self.cluster.kubeconfig
         self.cluster.reset()
+        kubeconfig2 = self.cluster.kubeconfig
+        self.cluster.kubectl(["get", "nodes"])
+        assert kubeconfig1 != kubeconfig2
 
     def test_c_apply_yaml_file(self):
         self.cluster.create()

tests/vendor.py

@@ -77,7 +77,7 @@ def test_e_kept_cluster_delete(k8s: AClusterManager):
     assert configmap["metadata"]["uid"] is not None
 
     k8s.reset()
-    with pytest.raises(subprocess.CalledProcessError):
+    with pytest.raises(RuntimeError):
         configmap = k8s.kubectl(["get", "configmap", "myconfigmap"])
 
     k8s.delete()
@@ -116,7 +116,7 @@ def test_g_prepopulated_cluster_kept(k8s_with_workload: AClusterManager):
     )
     assert "k3d-pytest-my-cluster" in process.stdout
     k8s.reset()
-    with pytest.raises(subprocess.CalledProcessError):
+    with pytest.raises(RuntimeError):
         k8s.kubectl(["get", "deployment", "hello-nginxdemo"])