Mock python k8s client

Viewed 119

Here is my snippets:

folder structure

utils/
├── __init__.py
├── k8s_client.py
├── simple.py
└── tests
    ├── k8s_client_test.py

k8s_client.py

import os
from kubernetes import client, config

config.load_kube_config(os.getenv('KUBECONFIG')) if os.getenv(
    'KUBECONFIG'
) else config.load_incluster_config()


class K8sClient:

    def __init__(self) -> None:
        self.k8s_api_client = client.CustomObjectsApi()

    def get_crs(self) -> list:
  
        custom_resources = self.k8s_api_client.list_cluster_custom_object(
            group=os.environ['GROUP'],
            version=os.environ['VERSION'],
            plural=os.environ['PLURAL'],
        )
        # list with all custom resources of certain type
        return custom_resources["items"]

sample.py

from utils.k8s_client import K8sClient


def initialize_config():

    return K8sClient().get_crs()

and finally I am trying to unittest initialize_config function. Here is my test code:

from utils.simple import initialize_config
from mock import patch


@patch('utils.k8s_client.K8sClient.get_crs')
@patch('utils.k8s_client.config.load_incluster_config')
def test_sample(kube_config_mock, k8s_client_mock):
    custom_resources = initialize_config()

    assert k8s_client_mock.assert_called_once()

This is what I get for error:

kubernetes.config.config_exception.ConfigException: Service host/port is not set.

This is because don't export kubeconfig at all, but I don't want to do it and I don't wish to make a real calls to the k8s cluster to get the real custom resources from there. I just want to fake these calls and kubeconfig exporting and in my case to see if k8s_client_mock is called and if I handle with that I can give return value to k8s_client_mock to assert its length and so on. But the main idea is How to fake this k8s client and to manage it as I want.

0 Answers
Related