cki_lib.inttests

Integration tests that require other services can be run via kind (Kubernetes in Docker).

Currently, a RabbitMQ service including certificates is implemented.

Creating an integration test

For an integration test, create a unit test in /inttests, and derive the TestCase from cki_lib.inttests.cluster.KubernetesCluster or one of the derived classes. As an example, for an integration test that needs a RabbitMQ server:

from cki_lib.inttests import cluster
from cki_lib.inttests.rabbitmq import RabbitMQServer

@cluster.skip_without_requirements()
class TestStompClient(RabbitMQServer):
    pass

During setupClass(), a Kubernetes cluster and any required services will be created and configured. The services will be removed in tearDownClass(), but the cluster will be kept running to speed up successive tests.

API

KubernetesCluster - Kubernetes cluster running in Docker

• test case class: inttests.cluster.KubernetesCluster

The cluster is able to run locally and in GitLab CI as long as the kind and docker executables are available. This can be checked with the @skip_without_requirements decorator. For GitLab CI, jobs need to have the Docker-in-Docker service configured in .gitlab-ci.yml:

integration-tests:
  services:
    - docker:dind
  variables:
    DOCKER_HOST: 'tcp://docker:2375'

Fields provided by KubernetesCluster

• cls.cluster_name: str: kind cluster name
• cls.hostname: str: host name for all services usable locally and in GitLab CI
• cls.api_client: client.ApiClient: Python kubernetes API client
• cls.dynamic_client: dynamic.client.DynamicClient: Python kubernetes Dynamic client
• cls.core_v1: client.CoreV1Api: Python kubernetes Core v1 API

Methods provided by KubernetesCluster

• cls.k8s_namespace(namespace: str): context manager that creates and cleans up a Kubernetes namespace.
• cls.k8s_apply(namespace: str, body: typing.Any): apply a Kubernetes resource update via server-side apply.
• cls.k8s_delete_all(namespace: str, resources: Iterable[str]): delete all resources of the given types from a namespace.

RabbitMQServer - RabbitMQ server

• test case class: inttests.rabbitmq.RabbitMQServer

The server is configured with AMQP and STOMP access with and without TLS. For AMQP and STOMP, authentication can be done via password and client certificates (from cert.common_name) methods.

Ports provided by RabbitMQServer

Test environment variables provided by RabbitMQServer

HashiCorpVaultServer - HashiCorp Vault server

• test case class: inttests.vault.HashiCorpVaultServer

The server is started in dev mode.

Ports provided by HashiCorpVaultServer

Test environment variables provided by HashiCorpVaultServer

MariaDBServer - MariaDB server

• test case class: inttests.mariadb.MariaDBServer

The database db is created on startup.

Ports provided by MariaDBServer

Test environment variables provided by MariaDBServer

MinioServer - Minio server

• test case class: inttests.minio.MinioServer

Ports provided by MinioServer

Test environment variables provided by MinioServer

RemoteResponsesServer - Remote Responses server

• test case class: inttests.remote_responses.RemoteResponsesServer

Provides a remote URL request mocking endpoint with an API similar to responses.

Fields provided by RemoteResponsesServer

The cls.responses field contains a wrapper class with a subset of the responses API:

• add(method, url, body=None, json=None, status=200, headers=None, match=[...])
• replace(method, url, body=None, json=None, status=200, headers=None, match=[...])
• upsert(method, url, body=None, json=None, status=200, headers=None, match=[...])
• reset()
• json_params_matcher(params, strict_match=True)
• calls

The return values of add/replace/upsert are proxied BaseResponse instances and provide all expected fields:

• call_count
• calls
• …

This allows to set up test beds with

# GET request
data = {'a': 'b'}
self.responses.add('GET', f'http://ignored/pooh', json=data)

# POST request with match filter
data = {'data': {'namespace': {'fullPath': 'foo', 'rootStorageStatistics': {
    'repositorySize': 1}, 'projects': {'nodes': []}, }}}
self.responses.add('POST', f'http://ignored/api/graphql', json=data, match=[
    self.responses.json_params_matcher({"operationName": "pooh"}, strict_match=False)])

The mocked responses are available from REMOTE_RESPONSES_URL_EXTERNAL when accessed from the test bed, or from REMOTE_RESPONSES_URL when accessed from other services inside the cluster.

The mocked responses are reset before each test. For sub tests, responses can be manually reset via self.responses.reset() as well.

Ports provided by RemoteResponsesServer

Test environment variables provided by RemoteResponsesServer