# pytest-kubernetes
pytest-kubernetes is a lightweight pytest plugin that makes managing (local) Kubernetes clusters a breeze. You can easily spin up a Kubernetes cluster with one [pytest fixure](https://docs.pytest.org/en/latest/explanation/fixtures.html) and remove them again.
The fixture comes with some simple functions to interact with the cluster, for example `kubectl(...)` that allows you to run typical *kubectl* commands against this cluster without worring
about the *kubeconfig* on the test machine.
**Features:**
- Set up and tear down (local) Kubernetes clusters with *minikube*, *k3d* and *kind*
- Configure the cluster to recreate for each test case (default), or keep it across multiple test cases
- Automatic management of the *kubeconfig*
- Simple functions to run kubectl commands (with *dict* output), reading logs and load custom container images
- Wait for certain conditions in the cluster
- Port forward Kubernetes-based services (using kubectl port-forward) easily during a test case
- Management utils for custom pytest-fixtures (for example pre-provisioned clusters)
## Installation
This plugin can be installed from PyPI:
- `pip install pytest-kubernetes`
- `poetry add -D pytest-kubernetes`
Note that this package provides entrypoint hooks to be automatically loaded with pytest.
## Requirements
pytest-kubernetes expects the following components to be available on the test machine:
- [`kubectl`](https://kubernetes.io/docs/reference/kubectl/)
- [`minikube`](https://minikube.sigs.k8s.io/docs/start/) (optional for minikube-based clusters)
- [`k3d`](https://k3d.io/) (optional for k3d-based clusters)
- [`kind`](https://kind.sigs.k8s.io/) (optional for kind-based clusters)
- [Docker](https://docs.docker.com/get-docker/) (optional for Docker-based Kubernetes clusters)
Please make sure they are installed to run pytest-kubernetes properly.
## Reference
### Fixture
#### k8s
The _k8s_ fixture provides access to an automatically selected Kubernetes provider (depending on the availability on the host). The priority is: k3d, kind, minikube-docker and minikube-kvm2.
The fixture passes a manager object of type *AClusterManager*.
It provides the following interface:
- `kubectl(...)`: Execute kubectl command against this cluster (defaults to `dict` as returning format)
- `apply(...)`: Apply resources to this cluster, either from YAML file, or Python dict
- `load_image(...)`: Load a container image into this cluster
- `wait(...)`: Wait for a target and a condition
- `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 (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
The interface provides proper typing and should be easy to work with.
**Example**
```python
def test_a_feature_with_k3d(k8s: AClusterManager):
k8s.create()
k8s.apply(
{
"apiVersion": "v1",
"kind": "ConfigMap",
"data": {"key": "value"},
"metadata": {"name": "myconfigmap"},
},
)
k8s.apply("./dependencies.yaml")
k8s.load_image("my-container-image:latest")
k8s.kubectl(
[
"run",
"test",
"--image",
"my-container-image:latest",
"--restart=Never",
"--image-pull-policy=Never",
]
)
```
This cluster will be deleted once the test case is over.
> Please note that you need to set *"--image-pull-policy=Never"* for images that you loaded into the cluster via the `k8s.load(name: str)` function (see example above).
### Marks
pytest-kubernetes uses [*pytest marks*](https://docs.pytest.org/en/7.1.x/how-to/mark.html) for specifying the cluster configuration for a test case
Currently the following settings are supported:
- *provider* (str): request a specific Kubernetes provider for the test case
- *cluster_name* (str): request a specific cluster name
- *keep* (bool): keep the cluster across multiple test cases
**Example**
```python
@pytest.mark.k8s(provider="minikube", cluster_name="test1", keep=True)
def test_a_feature_in_minikube(k8s: AClusterManager):
...
```
### Utils
To write custom Kubernetes-based fixtures in your project you can make use of the following util functions.
#### `select_provider_manager`
This function returns a deriving class of *AClusterManager* that is not created and wrapped in a fixture yet.
`select_provider_manager(name: Optional[str] = None) -> Type[AClusterManager]`
The returning object gets called with the init parameters of *AClusterManager*, the `cluster_name: str`.
**Example**
```python
@pytest.fixture(scope="session")
def k8s_with_workload(request):
cluster = select_provider_manager("k3d")("my-cluster")
# if minikube should be used
# cluster = select_provider_manager("minikube")("my-cluster")
cluster.create()
# init the cluster with a workload
cluster.apply("./fixtures/hello.yaml")
cluster.wait("deployments/hello-nginxdemo", "condition=Available=True")
yield cluster
cluster.delete()
```
In this example, the cluster remains active for the entire session and is only deleted once pytest is done.
> 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.
Raw data
{
"_id": null,
"home_page": "https://github.com/Blueshoe/pytest-kubernetes",
"name": "pytest-kubernetes",
"maintainer": null,
"docs_url": null,
"requires_python": "<4.0.0,>=3.11.7",
"maintainer_email": null,
"keywords": null,
"author": "Michael Schilonka",
"author_email": "michael@blueshoe.io",
"download_url": "https://files.pythonhosted.org/packages/33/60/136e21c91766f7badfb68b41e09e7c9db2a7eb55e107a2e1794c3e6c246e/pytest_kubernetes-0.5.0.tar.gz",
"platform": null,
"description": "# pytest-kubernetes\npytest-kubernetes is a lightweight pytest plugin that makes managing (local) Kubernetes clusters a breeze. You can easily spin up a Kubernetes cluster with one [pytest fixure](https://docs.pytest.org/en/latest/explanation/fixtures.html) and remove them again.\nThe fixture comes with some simple functions to interact with the cluster, for example `kubectl(...)` that allows you to run typical *kubectl* commands against this cluster without worring \nabout the *kubeconfig* on the test machine.\n\n**Features:**\n- Set up and tear down (local) Kubernetes clusters with *minikube*, *k3d* and *kind*\n- Configure the cluster to recreate for each test case (default), or keep it across multiple test cases\n- Automatic management of the *kubeconfig*\n- Simple functions to run kubectl commands (with *dict* output), reading logs and load custom container images\n- Wait for certain conditions in the cluster\n- Port forward Kubernetes-based services (using kubectl port-forward) easily during a test case\n- Management utils for custom pytest-fixtures (for example pre-provisioned clusters)\n \n## Installation\nThis plugin can be installed from PyPI:\n- `pip install pytest-kubernetes`\n- `poetry add -D pytest-kubernetes`\n\nNote that this package provides entrypoint hooks to be automatically loaded with pytest.\n\n## Requirements\npytest-kubernetes expects the following components to be available on the test machine:\n- [`kubectl`](https://kubernetes.io/docs/reference/kubectl/)\n- [`minikube`](https://minikube.sigs.k8s.io/docs/start/) (optional for minikube-based clusters)\n- [`k3d`](https://k3d.io/) (optional for k3d-based clusters)\n- [`kind`](https://kind.sigs.k8s.io/) (optional for kind-based clusters)\n- [Docker](https://docs.docker.com/get-docker/) (optional for Docker-based Kubernetes clusters)\n\nPlease make sure they are installed to run pytest-kubernetes properly.\n\n## Reference\n\n### Fixture\n\n#### k8s\nThe _k8s_ fixture provides access to an automatically selected Kubernetes provider (depending on the availability on the host). The priority is: k3d, kind, minikube-docker and minikube-kvm2.\n\nThe fixture passes a manager object of type *AClusterManager*.\n\nIt provides the following interface:\n- `kubectl(...)`: Execute kubectl command against this cluster (defaults to `dict` as returning format)\n- `apply(...)`: Apply resources to this cluster, either from YAML file, or Python dict\n- `load_image(...)`: Load a container image into this cluster\n- `wait(...)`: Wait for a target and a condition\n- `port_forwarding(...)`: Port forward a target\n- `logs(...)`: Get the logs of a pod\n- `version()`: Get the Kubernetes version of this cluster\n- `create(...)`: Create this cluster (pass special cluster arguments with `options: List[str]` to the CLI command)\n- `delete()`: Delete this cluster\n- `reset()`: Delete this cluster (if it exists) and create it again\n\nThe interface provides proper typing and should be easy to work with.\n\n**Example**\n\n```python\ndef test_a_feature_with_k3d(k8s: AClusterManager):\n k8s.create()\n k8s.apply(\n {\n \"apiVersion\": \"v1\",\n \"kind\": \"ConfigMap\",\n \"data\": {\"key\": \"value\"},\n \"metadata\": {\"name\": \"myconfigmap\"},\n },\n )\n k8s.apply(\"./dependencies.yaml\")\n k8s.load_image(\"my-container-image:latest\")\n k8s.kubectl(\n [\n \"run\",\n \"test\",\n \"--image\",\n \"my-container-image:latest\",\n \"--restart=Never\",\n \"--image-pull-policy=Never\",\n ]\n )\n```\nThis cluster will be deleted once the test case is over.\n\n> Please note that you need to set *\"--image-pull-policy=Never\"* for images that you loaded into the cluster via the `k8s.load(name: str)` function (see example above).\n\n### Marks\npytest-kubernetes uses [*pytest marks*](https://docs.pytest.org/en/7.1.x/how-to/mark.html) for specifying the cluster configuration for a test case\n\nCurrently the following settings are supported:\n\n- *provider* (str): request a specific Kubernetes provider for the test case \n- *cluster_name* (str): request a specific cluster name\n- *keep* (bool): keep the cluster across multiple test cases\n\n\n**Example**\n```python\n@pytest.mark.k8s(provider=\"minikube\", cluster_name=\"test1\", keep=True)\ndef test_a_feature_in_minikube(k8s: AClusterManager):\n ...\n```\n\n### Utils\nTo write custom Kubernetes-based fixtures in your project you can make use of the following util functions.\n\n\n#### `select_provider_manager`\nThis function returns a deriving class of *AClusterManager* that is not created and wrapped in a fixture yet.\n\n`select_provider_manager(name: Optional[str] = None) -> Type[AClusterManager]`\n\nThe returning object gets called with the init parameters of *AClusterManager*, the `cluster_name: str`.\n\n**Example**\n```python\n@pytest.fixture(scope=\"session\")\ndef k8s_with_workload(request):\n cluster = select_provider_manager(\"k3d\")(\"my-cluster\")\n # if minikube should be used\n # cluster = select_provider_manager(\"minikube\")(\"my-cluster\")\n cluster.create()\n # init the cluster with a workload\n cluster.apply(\"./fixtures/hello.yaml\")\n cluster.wait(\"deployments/hello-nginxdemo\", \"condition=Available=True\")\n yield cluster\n cluster.delete()\n```\nIn this example, the cluster remains active for the entire session and is only deleted once pytest is done.\n\n> Note that `yield` notation that is prefered by pytest to express clean up tasks for this fixture.\n\n#### Special cluster options\nYou can pass more options using `kwargs['options']: List[str]` to the `create(options=...)` function when creating the cluster like so:\n```python\n cluster = select_provider_manager(\"k3d\")(\"my-cluster\")\n # bind ports of this k3d cluster\n cluster.create(options=[\"--agents\", \"1\", \"-p\", \"8080:80@agent:0\", \"-p\", \"31820:31820/UDP@agent:0\"])\n```\n\n## Examples\nPlease 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.\n",
"bugtrack_url": null,
"license": null,
"summary": null,
"version": "0.5.0",
"project_urls": {
"Homepage": "https://github.com/Blueshoe/pytest-kubernetes",
"Repository": "https://github.com/Blueshoe/pytest-kubernetes"
},
"split_keywords": [],
"urls": [
{
"comment_text": "",
"digests": {
"blake2b_256": "47408dfa6cac2d8a81e1dcf02c8705aaafc04b684a4379591a1317f39dbddbfc",
"md5": "a8aa9c96c1b292399edce737ddccf87c",
"sha256": "972d48a2fb1f222f60c1df400efe97f7ecd47ff20e77b2f0569de8d025920520"
},
"downloads": -1,
"filename": "pytest_kubernetes-0.5.0-py3-none-any.whl",
"has_sig": false,
"md5_digest": "a8aa9c96c1b292399edce737ddccf87c",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": "<4.0.0,>=3.11.7",
"size": 16456,
"upload_time": "2024-10-28T10:03:11",
"upload_time_iso_8601": "2024-10-28T10:03:11.543247Z",
"url": "https://files.pythonhosted.org/packages/47/40/8dfa6cac2d8a81e1dcf02c8705aaafc04b684a4379591a1317f39dbddbfc/pytest_kubernetes-0.5.0-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": "",
"digests": {
"blake2b_256": "3360136e21c91766f7badfb68b41e09e7c9db2a7eb55e107a2e1794c3e6c246e",
"md5": "acdf141e9b804d74e49642fe2f3ee68a",
"sha256": "73b3f7beec3103ccd282b09007fb2d84411e11f10b2419f377ab6c6e47d416c5"
},
"downloads": -1,
"filename": "pytest_kubernetes-0.5.0.tar.gz",
"has_sig": false,
"md5_digest": "acdf141e9b804d74e49642fe2f3ee68a",
"packagetype": "sdist",
"python_version": "source",
"requires_python": "<4.0.0,>=3.11.7",
"size": 14359,
"upload_time": "2024-10-28T10:03:12",
"upload_time_iso_8601": "2024-10-28T10:03:12.780397Z",
"url": "https://files.pythonhosted.org/packages/33/60/136e21c91766f7badfb68b41e09e7c9db2a7eb55e107a2e1794c3e6c246e/pytest_kubernetes-0.5.0.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2024-10-28 10:03:12",
"github": true,
"gitlab": false,
"bitbucket": false,
"codeberg": false,
"github_user": "Blueshoe",
"github_project": "pytest-kubernetes",
"travis_ci": false,
"coveralls": false,
"github_actions": true,
"lcname": "pytest-kubernetes"
}
JSON
