I want to mock a response from the Python Kubernetes client. Below code of my Kubernetes service:

import os

from kubernetes.client.rest import ApiException
from kubernetes import client
from kubernetes.config import load_config
from exceptions.logs_not_found_exceptions import LogsNotFound
import logging

log = logging.getLogger("services/kubernetes_service.py")


class KubernetesService:
    def __init__(self):
        super().__init__()
        if os.getenv("DISABLE_KUBERNETES_CONFIG") == "False":
            load_config()

        self.api_instance = client.CoreV1Api()

    def get_namespaces(self):
        try:
            api_response = self.api_instance.list_namespace()
            dict_response = api_response.to_dict()

            namespaces = []
            for item in dict_response['items']:
                namespaces.append(item['metadata']['name'])

            log.info(f"Retrieved the namespaces: {namespaces}")
            return namespaces
        except ApiException as e:
            raise e

When I want to mock this using mock.patch I'm getting a ModuleNotFoundError. Below code of my Test class

import os
from unittest import mock
from tests.unit_tests import utils
from services.kubernetes_service import KubernetesService

class TestKubernetesService:
    @mock.patch.dict(os.environ, {"DISABLE_KUBERNETES_CONFIG": "True"})
    def test_get_namespaces(self):
        self.service = KubernetesService()
        print(self.service.api_instance)
        with mock.patch('services.kubernetes_service.KubernetesService.api_instance.list_namespace',
                        return_value=utils.kubernetes_namespaces_response()):
            actual_result = self.service.get_namespaces()
            assert actual_result == ['default', 'kube-node-lease', 'kube-public', 'kube-system']

When I edit the path in mock.patch from services.kubernetes_service.KubernetesService.api_instance.list_namespace to services.kubernetes_service.KubernetesService.get_namespaces it successfully mocks the return value I put in. But I want to mock the response of the line self.api_instance.list_namespace() in the KubernetesService class.

Someone an idea?

CodePudding user response：

In your example, you try to patch an instance attribute of the class (api_instance). This cannot be done by just referencing it from the class, as it is not a class attribute - you need an instance instead.

There are generally two standard methods to mock an instance attribute:

• mock the whole class, in which case the return_value attribute on the mocked class will be a mock that replaces any instance of the class and can therefore be used for mocking instance attributes
• mock a concrete instance using mock.patch.object or similar - this requires that you have access to the instance in your test

Mocking the whole class is not an option in your case, as you need to use the functionality of the class, but as you have access to the instance via self.service, you can use patch.object:

    def test_get_namespaces(self):
        self.service = KubernetesService()
        with mock.patch.object(self.service.api_instance, 'list_namespace', 
                               return_value=utils.kubernetes_namespaces_response()):
            actual_result = self.service.get_namespaces()
            assert actual_result == ['default', 'kube-node-lease', 'kube-public', 'kube-system']