使用mock.patch()时出现ModuleNotFoundError

Question

我想模拟Python-Kubernetes客户端的响应。下面是我的Kubernetes服务代码：

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

当我想使用mock.patch对此进行模拟时，我会得到一个ModuleNotFoundError。下面是我的测试类的代码

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']

当我编辑mock.patch中从services.kubernetes_service.KubernetesService.api_instance.list_namespace到services.kubernetes_service.KubernetesService.get_namespaces的路径时，它成功地模拟了我输入的返回值。但我想模拟KubernetsService类中self.api_instance.list_namespace()行的响应。

有人有主意吗？

Answer 1

在您的示例中，您尝试修补类(api_instance(的实例属性。这不能仅仅通过从类中引用它来实现，因为它不是类属性，而是需要一个实例。

通常有两种标准方法来模拟实例属性：

• 模拟整个类，在这种情况下，模拟类上的return_value属性将是一个替换类的任何实例的模拟，因此可以用于模拟实例属性
• 使用mock.patch.object或类似的方法模拟一个具体实例-这要求您在测试中有权访问该实例

在您的情况下，模拟整个类不是一个选项，因为您需要使用类的功能，但由于您可以通过self.service访问实例，因此可以使用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']