### Create Custom Production-Like Kubernetes Cluster with pytest-kubernetes

Source: https://context7.com/blueshoe/pytest-kubernetes/llms.txt

Demonstrates creating a k3d cluster with custom configurations like Kubernetes version, port mappings, and agent memory. It includes setup for a specific namespace and setting KUBECONFIG, with cleanup logic.

```python
import pytest
import os
from pytest_kubernetes.providers.base import AClusterManager
from pytest_kubernetes.options import ClusterOptions

@pytest.fixture(scope="module")
def production_like_cluster(k8s_manager):
    # Create a k3d cluster with custom configuration
    k8s: AClusterManager = k8s_manager("k3d")("prod-test")

    # Check if cluster already exists (for local development)
    cluster_exists = k8s.ready(timeout=2)

    if not cluster_exists:
        # Create with specific Kubernetes version and port mappings
        k8s.create(
            ClusterOptions(api_version="1.29.5"),
            options=[
                "--agents", "2",
                "-p", "8080:80@agent:0",
                "-p", "8443:443@agent:0",
                "--agents-memory", "4G"
            ]
        )

    # Create namespace
    if "prod-test" not in k8s.kubectl(["get", "ns"], as_dict=False):
        k8s.kubectl(["create", "ns", "prod-test"])
        k8s.wait("ns/prod-test", "jsonpath='{.status.phase}'=Active")

    # Set KUBECONFIG for other tools
    os.environ["KUBECONFIG"] = str(k8s.kubeconfig)

    yield k8s

    # Cleanup: only delete if we created it
    if not cluster_exists:
        k8s.delete()

def test_with_custom_cluster(production_like_cluster):
    k8s = production_like_cluster

    # Deploy to custom namespace
    k8s.kubectl([
        "run", "nginx",
        "--image", "nginx:latest",
        "-n", "prod-test"
    ])

    k8s.wait("pod/nginx", "condition=Ready", timeout=60, namespace="prod-test")

    # Get Kubernetes version
    major, minor = k8s.version()
    assert major == 1
    assert minor >= 29

```

--------------------------------

### Command Line Options with Bash

Source: https://context7.com/blueshoe/pytest-kubernetes/llms.txt

This bash script provides examples of command line options for overriding pytest-kubernetes cluster configurations at runtime. It allows specifying provider, cluster name, Kubernetes version, and config files via pytest flags. The commands require pytest and pytest-kubernetes installed, taking CLI arguments as input to configure clusters without code changes. Limitations include dependency on available providers and config file correctness.

```bash
# Run tests with specific provider
pytest tests/ --k8s-provider k3d

# Run tests with specific cluster name
pytest tests/ --k8s-cluster-name my-test-cluster

# Run tests with specific Kubernetes version
pytest tests/ --k8s-version 1.29.0

# Use provider-specific config file
pytest tests/ --k8s-provider-config ./k3d-config.yaml

# Use existing cluster (skip creation/deletion)
pytest tests/ --k8s-kubeconfig-override ~/.kube/config --k8s-cluster-name existing-cluster

# Combine options for development workflow
pytest tests/ --k8s-provider k3d --k8s-cluster-name dev-cluster --k8s-version 1.28.0 -s -x
```

--------------------------------

### Manage Kubernetes Cluster with k8s Fixture (Python)

Source: https://context7.com/blueshoe/pytest-kubernetes/llms.txt

Demonstrates using the `k8s` fixture to create, apply manifests, load images, run commands, wait for resources, and get logs from a Kubernetes cluster. It automatically manages the cluster lifecycle, creating it before tests and deleting it afterward unless configured otherwise.

```python
import pytest
from pytest_kubernetes.providers.base import AClusterManager

def test_deploy_application(k8s: AClusterManager):
    # Create the cluster
    k8s.create()

    # Apply a ConfigMap from Python dict
    k8s.apply({
        "apiVersion": "v1",
        "kind": "ConfigMap",
        "metadata": {"name": "app-config"},
        "data": {"database_url": "postgres://db:5432"}
    })

    # Apply deployment from YAML file
    k8s.apply("./deployment.yaml")

    # Load local container image into cluster
    k8s.load_image("myapp:latest")

    # Create a pod using the loaded image
    k8s.kubectl([
        "run", "myapp",
        "--image", "myapp:latest",
        "--restart=Never",
        "--image-pull-policy=Never"
    ])

    # Wait for pod to be ready
    k8s.wait("pod/myapp", "condition=Ready", timeout=60)

    # Get pod details as dict
    pod_info = k8s.kubectl(["get", "pod", "myapp"])
    assert pod_info["status"]["phase"] == "Running"

    # Get logs from the pod
    logs = k8s.logs("myapp")
    assert "Application started" in logs

```

--------------------------------

### Manual Provider Selection with Python

Source: https://context7.com/blueshoe/pytest-kubernetes/llms.txt

This code demonstrates manual selection of Kubernetes cluster providers (k3d or kind) using the select_provider_manager function in pytest-kubernetes. It creates session-scoped fixtures for clusters with predefined configurations and workloads, applies test data, and includes cleanup. The setup requires pytest-kubernetes dependencies and accepts cluster options and YAML paths as inputs, outputting managed cluster instances. Limitations include provider availability and manual configuration without pytest auto-awareness.

```python
import pytest
from pathlib import Path
from pytest_kubernetes.providers import select_provider_manager
from pytest_kubernetes.options import ClusterOptions

@pytest.fixture(scope="session")
def k3d_cluster():
    # Manually select k3d provider
    ClusterClass = select_provider_manager("k3d")
    cluster = ClusterClass("integration-tests")

    cluster.create(
        ClusterOptions(api_version="1.29.0"),
        options=["--agents", "1", "-p", "8080:80@agent:0"]
    )

    # Pre-populate cluster with test data
    cluster.apply({
        "apiVersion": "v1",
        "kind": "Namespace",
        "metadata": {"name": "test-data"}
    })

    cluster.apply(Path("./fixtures/test-workload.yaml"))
    cluster.wait("deployment/test-app", "condition=Available", timeout=120, namespace="test-data")

    yield cluster

    # Cleanup
    cluster.delete()

@pytest.fixture(scope="session")
def kind_cluster():
    # Manually select kind provider
    ClusterClass = select_provider_manager("kind")
    cluster = ClusterClass("kind-integration")

    cluster.create(
        ClusterOptions(
            api_version="1.28.0",
            provider_config=Path("./kind-config.yaml")
        )
    )

    yield cluster
    cluster.delete()

def test_with_k3d(k3d_cluster):
    # Use pre-configured k3d cluster
    deployment = k3d_cluster.kubectl(["get", "deployment", "test-app", "-n", "test-data"])
    assert deployment["status"]["readyReplicas"] == deployment["spec"]["replicas"]

def test_with_kind(kind_cluster):
    # Use kind cluster
    namespaces = kind_cluster.kubectl(["get", "namespaces"])
    assert "items" in namespaces
```

--------------------------------

### Load Container Images into Kubernetes Cluster in Python

Source: https://context7.com/blueshoe/pytest-kubernetes/llms.txt

Loads local container images into the Kubernetes cluster for pod usage. Requires Docker to be installed and accessible. The method handles image loading and verification of deployment using custom images with appropriate pull policies.

```python
import subprocess
from pytest_kubernetes.providers.base import AClusterManager

def test_load_custom_image(k8s: AClusterManager):
    k8s.create()

    # Build a local Docker image
    subprocess.run([
        "docker", "build",
        "-t", "myapp:test",
        "-f", "./Dockerfile",
        "."
    ], check=True)

    # Load the image into the cluster
    k8s.load_image("myapp:test")

    # Deploy using the loaded image
    k8s.kubectl([
        "run", "myapp-pod",
        "--image", "myapp:test",
        "--image-pull-policy=Never",
        "--restart=Never"
    ])

    # Wait for pod to be ready
    k8s.wait("pod/myapp-pod", "condition=Ready", timeout=60)

    # Verify the pod is using the correct image
    pod = k8s.kubectl(["get", "pod", "myapp-pod"])
    assert pod["spec"]["containers"][0]["image"] == "myapp:test"
    assert pod["spec"]["containers"][0]["imagePullPolicy"] == "Never"

    # Check application logs
    logs = k8s.logs("myapp-pod")
    assert "Application started successfully" in logs
```

--------------------------------

### Configuring Clusters with ClusterOptions in pytest-kubernetes

Source: https://context7.com/blueshoe/pytest-kubernetes/llms.txt

ClusterOptions dataclass sets up Kubernetes cluster creation parameters like API version, kubeconfig path, provider config, and timeouts for use with AClusterManager's create method. It depends on pathlib and pytest-kubernetes options module. Inputs include cluster_name, api_version, cluster_timeout, provider_config, and kubeconfig_path; outputs configure and verify the cluster setup. Limitations involve provider-specific configs and timeout handling for creation.

```python
from pathlib import Path
from pytest_kubernetes.providers.base import AClusterManager
from pytest_kubernetes.options import ClusterOptions

def test_cluster_with_options(k8s: AClusterManager):
    # Create cluster with specific Kubernetes version
    k8s.create(
        ClusterOptions(
            cluster_name="custom-cluster",
            api_version="1.28.0",
            cluster_timeout=300
        )
    )

    # Verify cluster version
    major, minor = k8s.version()
    assert major == 1
    assert minor == 28

    k8s.delete()

def test_cluster_with_config_file(k8s: AClusterManager):
    # Create cluster using provider-specific config file
    k8s.create(
        ClusterOptions(
            provider_config=Path("./k3d-config.yaml"),
            cluster_timeout=240
        )
    )

    # Cluster will use settings from config file
    result = k8s.kubectl(["cluster-info"])
    assert "running" in str(result).lower()

    k8s.delete()

def test_cluster_with_custom_kubeconfig(k8s: AClusterManager):
    # Specify custom kubeconfig location
    kubeconfig_path = Path("/tmp/my-kubeconfig")

    k8s.create(
        ClusterOptions(
            cluster_name="test-cluster",
            kubeconfig_path=kubeconfig_path,
            api_version="1.29.0"
        )
    )

    # Verify kubeconfig exists at specified location
    assert kubeconfig_path.exists()
    assert k8s.kubeconfig == kubeconfig_path

    k8s.delete()
```

--------------------------------

### Retrieving Pod Logs using pytest-kubernetes

Source: https://context7.com/blueshoe/pytest-kubernetes/llms.txt

The logs method retrieves logs from Kubernetes pods, supporting single or multi-container setups and specific namespaces. It requires an AClusterManager instance, pod creation via kubectl or apply, and waiting for readiness. Inputs include pod name, optional container and namespace parameters; outputs are log strings for assertion. Limitations include dependency on cluster timeouts and pod readiness conditions.

```python
from pytest_kubernetes.providers.base import AClusterManager

def test_pod_logs(k8s: AClusterManager):
    k8s.create()

    # Create a pod that outputs logs
    k8s.kubectl([
        "run", "logger",
        "--image", "busybox",
        "--restart=Never",
        "--", "sh", "-c",
        "echo 'Starting application' && echo 'Processing data' && sleep 3600"
    ])

    k8s.wait("pod/logger", "condition=Ready", timeout=60)

    # Get logs from single-container pod
    logs = k8s.logs("logger")
    assert "Starting application" in logs
    assert "Processing data" in logs

    # Create multi-container pod
    k8s.apply({
        "apiVersion": "v1",
        "kind": "Pod",
        "metadata": {"name": "multi-container"},
        "spec": {
            "containers": [
                {
                    "name": "app",
                    "image": "busybox",
                    "command": ["sh", "-c", "echo 'App container' && sleep 3600"]
                },
                {
                    "name": "sidecar",
                    "image": "busybox",
                    "command": ["sh", "-c", "echo 'Sidecar container' && sleep 3600"]
                }
            ]
        }
    })

    k8s.wait("pod/multi-container", "condition=Ready", timeout=60)

    # Get logs from specific container
    app_logs = k8s.logs("multi-container", container="app")
    assert "App container" in app_logs

    sidecar_logs = k8s.logs("multi-container", container="sidecar")
    assert "Sidecar container" in sidecar_logs

    # Get logs from pod in specific namespace
    k8s.kubectl(["create", "ns", "logging"])
    k8s.kubectl([
        "run", "app", "--image=busybox",
        "-n", "logging",
        "--", "sh", "-c", "echo 'Namespace logs' && sleep 3600"
    ])
    k8s.wait("pod/app", "condition=Ready", timeout=60, namespace="logging")

    ns_logs = k8s.logs("app", namespace="logging")
    assert "Namespace logs" in ns_logs
```

--------------------------------

### Apply Kubernetes Resources with pytest-kubernetes

Source: https://context7.com/blueshoe/pytest-kubernetes/llms.txt

Illustrates using the `apply` method to deploy Kubernetes resources from Python dictionaries, YAML files, Path objects, and entire directories. Includes verification of created resources.

```python
from pathlib import Path
from pytest_kubernetes.providers.base import AClusterManager

def test_apply_resources(k8s: AClusterManager):
    k8s.create()

    # Apply from Python dict
    k8s.apply({
        "apiVersion": "v1",
        "kind": "Service",
        "metadata": {"name": "web-service"},
        "spec": {
            "selector": {"app": "web"},
            "ports": [{
                "protocol": "TCP",
                "port": 80,
                "targetPort": 8080
            }],
            "type": "ClusterIP"
        }
    })

    # Apply from YAML file
    k8s.apply("./manifests/deployment.yaml")

    # Apply from Path object
    k8s.apply(Path("./manifests/configmap.yaml"))

    # Verify resources were created
    service = k8s.kubectl(["get", "service", "web-service"])
    assert service["spec"]["ports"][0]["port"] == 80

    # Apply multiple resources from directory
    k8s.apply("./manifests/")

```

--------------------------------

### Create cluster with configuration file

Source: https://github.com/blueshoe/pytest-kubernetes/blob/main/README.md

Demonstrates how to create a Kubernetes cluster by providing a configuration file through the cluster_options parameter. Supports various providers including kind, k3d, and minikube which require provider-specific config formats.

```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")
    )
)
```

--------------------------------

### Execute Kubectl Commands with pytest-kubernetes

Source: https://context7.com/blueshoe/pytest-kubernetes/llms.txt

Shows how to execute kubectl commands using the `kubectl` method, supporting both dictionary and raw string outputs. It covers creating namespaces, deployments, scaling, and using custom timeouts.

```python
from pytest_kubernetes.providers.base import AClusterManager

def test_kubectl_operations(k8s: AClusterManager):
    k8s.create()

    # Create a namespace (dict output)
    result = k8s.kubectl(["create", "ns", "testing"])
    assert result["metadata"]["name"] == "testing"

    # Create a deployment
    k8s.kubectl([
        "create", "deployment", "nginx",
        "--image=nginx:1.21",
        "-n", "testing"
    ])

    # Get deployment with dict output (default)
    deployment = k8s.kubectl(["get", "deployment", "nginx", "-n", "testing"])
    assert deployment["spec"]["replicas"] == 1
    assert deployment["spec"]["template"]["spec"]["containers"][0]["image"] == "nginx:1.21"

    # Scale the deployment
    k8s.kubectl(["scale", "deployment/nginx", "--replicas=3", "-n", "testing"], as_dict=False)

    # Get raw string output
    pods_output = k8s.kubectl(["get", "pods", "-n", "testing"], as_dict=False)
    assert "nginx" in pods_output

    # Use custom timeout for long operations
    k8s.kubectl(["rollout", "status", "deployment/nginx", "-n", "testing"],
                as_dict=False, timeout=120)

```

--------------------------------

### Create custom Kubernetes fixture with workload

Source: https://github.com/blueshoe/pytest-kubernetes/blob/main/README.md

Demonstrates creating a custom Kubernetes fixture with a pre-deployed workload using select_provider_manager. The cluster persists for the test session and includes workload initialization.

```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()
```

--------------------------------

### Create cluster with special options

Source: https://github.com/blueshoe/pytest-kubernetes/blob/main/README.md

Shows how to pass additional command-line options during cluster creation using the options parameter. This allows for customization such as port binding and agent configuration specific to the cluster provider.

```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"])
```

--------------------------------

### Create and manage k3d Kubernetes cluster fixture

Source: https://github.com/blueshoe/pytest-kubernetes/blob/main/README.md

This Python fixture demonstrates how to create, manage, and clean up a k3d Kubernetes cluster for testing. It checks for existing clusters, creates new ones if needed, manages namespaces, and handles cleanup based on cluster origin.

```python
@pytest.fixture(scope="module")
def k3d(k8s_manager):
    k8s: AClusterManager = k8s_manager("k3d")("gefyra")
    # ClusterOptions() forces pytest-kubernetes to always write a new kubeconfig file to disk
    cluster_exists = k8s.ready(timeout=1)
    if not cluster_exists:
        k8s.create(
            ClusterOptions(api_version="1.29.5"),
            options=[
                "--agents",
                "1",
                "-p",
                "8080:80@agent:0",
                "-p",
                "31820:31820/UDP@agent:0",
                "--agents-memory",
                "8G",
            ],
        )
    if "gefyra" not in k8s.kubectl(["get", "ns"], as_dict=False):
        k8s.kubectl(["create", "ns", "gefyra"])
        k8s.wait("ns/gefyra", "jsonpath='{.status.phase}'=Active")
    else:
        purge_gefyra_objects(k8s)
    os.environ["KUBECONFIG"] = str(k8s.kubeconfig)
    yield k8s
    if cluster_exists:
        # delete existing bridges
        purge_gefyra_objects(k8s)
        k8s.kubectl(["delete", "ns", "gefyra"], as_dict=False)
    else:
        # we delete this cluster only when created during this run
        k8s.delete()
```

--------------------------------

### Runtime Configuration Access with Python

Source: https://context7.com/blueshoe/pytest-kubernetes/llms.txt

This Python test code illustrates how to access and verify runtime configuration options in pytest-kubernetes, including provider selection and cluster naming via CLI. It depends on pytest-kubernetes fixtures and CLI overrides, taking cluster manager as input to assert configurations. The output includes cluster properties and kubectl results. Limitations involve CLI option precedence and provider support.

```python
# Test can access runtime configuration
from pytest_kubernetes.providers.base import AClusterManager

def test_respects_cli_options(k8s: AClusterManager):
    """
    This test respects CLI options:
    pytest test_file.py --k8s-provider kind --k8s-cluster-name cli-cluster
    """
    k8s.create()

    # Cluster name comes from --k8s-cluster-name option
    assert k8s.cluster_name == "cli-cluster"

    # Provider comes from --k8s-provider option
    assert k8s.get_binary_name() == "kind"

    # Can override in mark decorator
    k8s.kubectl(["get", "nodes"])
```

--------------------------------

### Configure Kubernetes Cluster with Pytest Mark (Python)

Source: https://context7.com/blueshoe/pytest-kubernetes/llms.txt

Shows how to use the `@pytest.mark.k8s` decorator to specify cluster provider, name, and persistence settings for individual test functions. This allows for fine-grained control over cluster configuration on a per-test basis, including reusing persistent clusters.

```python
import pytest
from pytest_kubernetes.providers.base import AClusterManager

@pytest.mark.k8s(provider="k3d", cluster_name="test-cluster", keep=True)
def test_stateful_workload(k8s: AClusterManager):
    k8s.create()

    # Apply a StatefulSet
    k8s.apply({
        "apiVersion": "apps/v1",
        "kind": "StatefulSet",
        "metadata": {"name": "postgres"},
        "spec": {
            "serviceName": "postgres",
            "replicas": 1,
            "selector": {"matchLabels": {"app": "postgres"}},
            "template": {
                "metadata": {"labels": {"app": "postgres"}},
                "spec": {
                    "containers": [{
                        "name": "postgres",
                        "image": "postgres:14",
                        "env": [{"name": "POSTGRES_PASSWORD", "value": "secret"}]
                    }]
                }
            }
        }
    })

    # Wait for StatefulSet to be ready
    k8s.wait("statefulset/postgres", "jsonpath='{.status.readyReplicas}'=1", timeout=120)

    # Verify pod exists
    result = k8s.kubectl(["get", "pod", "postgres-0"])
    assert result["status"]["phase"] == "Running"

@pytest.mark.k8s(provider="k3d", cluster_name="test-cluster", keep=True)
def test_reuse_cluster(k8s: AClusterManager):
    # This test reuses the cluster from the previous test
    result = k8s.kubectl(["get", "statefulset", "postgres"])
    assert result["metadata"]["name"] == "postgres"

```

--------------------------------

### Configure Kubernetes test with pytest marks

Source: https://github.com/blueshoe/pytest-kubernetes/blob/main/README.md

Shows how to use pytest marks to specify Kubernetes cluster configuration for test cases, including provider selection, cluster naming, and persistence settings.

```python
@pytest.mark.k8s(provider="minikube", cluster_name="test1", keep=True)
def test_a_feature_in_minikube(k8s: AClusterManager):
    ...
```

--------------------------------

### Forward Ports to Local Machine in Python

Source: https://context7.com/blueshoe/pytest-kubernetes/llms.txt

Creates port forwards from localhost to Kubernetes resources using context managers. Supports pods, deployments, and services. Requires the requests library for testing connectivity and a configured AClusterManager instance.

```python
import requests
from pytest_kubernetes.providers.base import AClusterManager

def test_port_forwarding(k8s: AClusterManager):
    k8s.create()

    # Deploy a web service
    k8s.apply({
        "apiVersion": "v1",
        "kind": "Pod",
        "metadata": {"name": "web", "labels": {"app": "web"}},
        "spec": {
            "containers": [{
                "name": "nginx",
                "image": "nginx:alpine",
                "ports": [{"containerPort": 80}]
            }]
        }
    })

    k8s.wait("pod/web", "condition=Ready", timeout=60)

    # Forward port using context manager
    with k8s.port_forwarding("pod/web", 8080, 80) as pf:
        # Make HTTP request to forwarded port
        response = requests.get("http://localhost:8080")
        assert response.status_code == 200
        assert "nginx" in response.text.lower()

    # Port forward to a service
    k8s.apply({
        "apiVersion": "v1",
        "kind": "Service",
        "metadata": {"name": "web-svc"},
        "spec": {
            "selector": {"app": "web"},
            "ports": [{"protocol": "TCP", "port": 80, "targetPort": 80}]
        }
    })

    with k8s.port_forwarding("service/web-svc", 9090, 80, namespace="default") as pf:
        response = requests.get("http://localhost:9090")
        assert response.status_code == 200
```

--------------------------------

### Wait for Kubernetes Resource Conditions in Python

Source: https://context7.com/blueshoe/pytest-kubernetes/llms.txt

Blocks execution until a Kubernetes resource meets specified conditions or times out. Supports deployments, namespaces, and pods with various condition types. Requires a configured AClusterManager instance and properly defined Kubernetes resources.

```python
from pytest_kubernetes.providers.base import AClusterManager

def test_wait_for_resources(k8s: AClusterManager):
    k8s.create()

    # Create a deployment
    k8s.apply({
        "apiVersion": "apps/v1",
        "kind": "Deployment",
        "metadata": {"name": "api-server"},
        "spec": {
            "replicas": 2,
            "selector": {"matchLabels": {"app": "api"}},
            "template": {
                "metadata": {"labels": {"app": "api"}},
                "spec": {
                    "containers": [{
                        "name": "api",
                        "image": "nginx:alpine",
                        "ports": [{"containerPort": 80}]
                    }]
                }
            }
        }
    })

    # Wait for deployment to be available
    k8s.wait("deployment/api-server", "condition=Available=True", timeout=120)

    # Create a namespace and wait for it to be active
    k8s.kubectl(["create", "ns", "production"])
    k8s.wait("ns/production", "jsonpath='{.status.phase}'=Active", timeout=30)

    # Wait for pod to be running in specific namespace
    k8s.kubectl(["run", "worker", "--image=busybox", "-n", "production",
                 "--", "sleep", "3600"])
    k8s.wait("pod/worker", "condition=Ready", timeout=60, namespace="production")

    # Verify the pod is ready
    pod = k8s.kubectl(["get", "pod", "worker", "-n", "production"])
    assert pod["status"]["phase"] == "Running"
```

--------------------------------

### Using k8s Fixture for Kubernetes Cluster Management in pytest (Python)

Source: https://github.com/blueshoe/pytest-kubernetes/blob/main/README.md

This snippet demonstrates using the k8s fixture to create a Kubernetes cluster, apply configurations from dicts or YAML files, load custom images, and run kubectl commands in a pytest test function. It depends on the pytest-kubernetes plugin, kubectl, and optional tools like k3d. The fixture automatically manages kubeconfig and deletes the cluster after the test; limitations include requiring --image-pull-policy=Never for loaded images and availability of cluster providers on the host.

```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",
        ]
    )
```

=== COMPLETE CONTENT === This response contains all available snippets from this library. No additional content exists. Do not make further requests.