Writing Tests

k8scheck is designed to interface with a Kubernetes cluster, so before you begin writing tests with k8scheck, be sure that you have access to a cluster, whether on Google Container Engine, via minikube, or through your own custom cluster. Generally, where the cluster runs shouldn’t be an issue, as long as you can access it from wherever the tests are being run.

Cluster Configuration

By default, k8scheck will look for a config file at ~/.kube/config and the current context – this is the same behavior that kubectl utilizes for the resolving cluster config. Generally, if you can reach your cluster via. kubectl, you should be able to use it with k8scheck.

If you wish to specify a different config file and/or context, you can pass it in via the --kube-config and --kube-context flags. See Command Line Usage for more details.

You can also write a kubeconfig fixture which provides the path to the config file and/or a kubecontext fixture which provides the name of the context to be used. This may be useful in case your cluster is generated as part of the tests or you wish to use specific contexts in different parts of the suite.

import pytest
import subprocess
from typing import Optional


@pytest.fixture
def kubeconfig() -> str:
    # Here, Terraform creates a cluster and outputs a kubeconfig
    # at somepath
    subprocess.check_call(['terraform', 'apply'])
    return 'somepath/kubeconfig'


@pytest.fixture
def kubecontext() -> Optional[str]:
    # Return None to use the current context as set in the kubeconfig
    # Or return the name of a specific context in the kubeconfig
    return 'k8scheck-cluster'


 def test_my_terraformed_cluster(kube):
     # Use your cluster!
     pass

Loading Manifests

It is recommended, though not required, to test against pre-defined manifest files. These files can be kept anywhere relative to your tests and can be organized however you like. Each test can have its own directory of manifests, or you can pick and choose individual manifest files for the test case.

While you can generate your own manifests within the tests themselves (e.g. by initializing a Kubernetes API object), this can become tedious and clutter up the tests. If you do choose to go this route, you can still use all of the k8scheck functionality by wrapping supported objects with their equivalent k8scheck wrapper. For example,

from kubernetes import client
from k8scheck.objects import Deployment


# Create a Kubernetes API Object
raw_deployment = client.V1Deployment(
    metadata=client.V1ObjectMeta(
        name='test-deployment'
    ),
    spec=client.V1DeploymentSpec(
        replicas=2,
        template=client.V1PodTemplateSpec(
            ...
        )
    )
)

# Wrap it in the k8scheck wrapper
wrapped_deployment = Deployment(raw_deployment)

If you use manifest files, you can load them directly into wrapped API objects easily via the k8scheck Client, which is provided to a test case via the kube fixture.

def test_something(kube):

    f = os.path.join(
        os.path.dirname(os.path.realpath(__file__)),
        'manifests',
        'deployment.yaml'
    )

   deployment = kube.load_deployment(f)

Often, tests will multiple resources that need to be loaded from manifest YAMLs. It can be tedious to construct all of the paths, load them, and create them at the start of a test. k8scheck provides the Apply Manifests marker that allows you to specify an entire directory to load, or specific files from a directory. The example below loads the same file as the previous example using the applymanifests marker.

@pytest.mark.applymanifests('manifests', files=[
    'deployment.yaml'
])
def test_something(kube):
    ...

Once a manifest is loaded, you will have (or be able to get) a reference to the created API Objects which offer more functionality.

Creating Resources

If you use the Apply Manifests, as described in the previous section, the manifest will be loaded and created for you in the test case namespace of your cluster (test case namespaces are automatically managed via the kube).

You may want to load resources manually, or load and create some at a later time in the test. This can be done via the kube client

def test_something(kube):

    # ...
    # do something first
    # ...

    deployment = kube.load_deployment('path/to/deployment.yaml')
    kube.create(deployment)

It can also be done through the resource reference itself

def test_something(kube):

    # ...
    # do something first
    # ...

    deployment = kube.load_deployment('path/to/deployment.yaml')
    deployment.create()

Deleting Resources

It is not necessary to delete resources at the end of a test case. k8scheck automatically manages the namespace for the test case. When the test completes, it will delete the namespace from the cluster which will also delete any remaining resources in that namespace.

It can still be useful to delete things while testing, e.g. to simulate a service failure and to test the subsequent disaster recovery process. Similar to resource creation, resource deletion can be done either through the object reference or through the kube client

def test_something(kube):

    # ...
    # created resource, did some testing, now need to remove
    # the resource
    # ...

    # Method #1 - delete via the kube client
    kube.delete(deployment)

    # Method #2 - delete via the object reference
    deployment.delete()

Test Namespaces

By default, k8scheck will automatically generate a new Namespace for each test case, using the test name and a timestamp for the namespace name to ensure uniqueness. This behavior may not be desired in all cases, such as when users may not have permissions to create a new namespace on the cluster, or the tests are written against an already-running deployment in an existing namespace. In such cases, the _namespace_marker may be used.

Waiting

The time it takes for a resource to start, stop, or become ready can vary across numerous factors. It is not always reliable to just time.sleep(10) and hope that the desired state is met (nor is it efficient). To help with this, there are a number of wait functions provided by k8scheck. For a full accounting of all wait functions, see the API Reference.

Below are some simple examples of select wait function usage.

Ready Nodes

If you are running on a cluster that can scale automatically, you may need to wait for the correct number of nodes to be available and ready before the test can run.

@pytest.mark.applymanifests('manifests')
def test_something(kube):
    # wait for 3 nodes to be available and ready
    kube.wait_for_ready_nodes(3, timeout=5 * 60)

Created Object

Wait until an object has been created on the cluster.

def test_something(kube):
    deployment = kube.load_deployment('path/to/deployment.yaml')
    kube.create(deployment)
    kube.wait_until_created(deployment, timeout=30)

Pod Containers Start

Wait until a Pod’s containers have all started.

@pytest.mark.applymanifests('manifests')
def test_something(kube):
    pods = kube.get_pods()
    for pod in pods.values():
        pod.wait_until_containers_start(timeout=60)