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.