Mock Kubernetes Client读取Secret时单元测试报AttributeError求助
问题:Mock读取K8s Secret的函数时触发AttributeError
运行单元测试时出现如下错误:
AttributeError: <module 'kubernetes.client' from '/usr/lib/python3.6/site-packages/kubernetes/client/init.py'> does not have the attribute 'read_namespaced_secret()'
已尝试相关Mock方案但未解决,求指出错误点。
业务脚本:read_sec.py
import base64 from kubernetes import client, config from logger import logger class kubernetesServices(): def __init__(self): pass def get_secret_vault_token(self): try: config.load_kube_config() api_instance = client.CoreV1Api() sec = api_instance.read_namespaced_secret("random-sec", "random-ns").data token = base64.b64decode(sec['token']).decode("utf-8") return token except Exception as e: logger.error("got error at get_secret_vault_token: {}".format(str(e)))
单元测试脚本:test_read_sec.py(原错误版本)
import unittest from unittest.mock import patch from read_sec import * class MockKubernetes(): def __init__(self): pass def mocker_read_namespaced_secret(*args, **kwargs): class MockReadns(): def __init__(self, json_data): self.json_data = json_data def json(self): return self.json_data return MockReadns({"data":{"token":"abc123"}}) class TestkubernetesServices(unittest.TestCase): @patch("kubernetes.client",side_effect=MockKubernetes) @patch("kubernetes.config",side_effect=MockKubernetes) @patch("kubernetes.client.read_namespaced_secret()",side_effect=mocker_read_namespaced_secret) def test_get_secret_vault_token(self,mock_client,mock_config,mock_read): k8s = kubernetesServices() token = k8s.get_secret_vault_token()
错误分析与解决方案
核心错误点
- Patch目标路径错误:
read_namespaced_secret是CoreV1Api类的实例方法,并非kubernetes.client模块的直接属性,原测试中@patch("kubernetes.client.read_namespaced_secret()")的写法完全错误(还额外加了括号)。 - Patch层级错误:应该针对业务代码中实际调用的对象进行Patch,而非直接patch顶层的
kubernetes.client模块。 - Mock实现不完整:自定义的
MockKubernetes没有模拟load_kube_config方法,也没有提供CoreV1Api类及对应的实例方法。
修正后的单元测试代码
import unittest from unittest.mock import patch, MagicMock from read_sec import kubernetesServices import base64 class TestkubernetesServices(unittest.TestCase): @patch("read_sec.config") @patch("read_sec.client.CoreV1Api") def test_get_secret_vault_token(self, mock_core_v1_api, mock_config): # 模拟config.load_kube_config方法,避免实际加载K8s配置 mock_config.load_kube_config.return_value = None # 创建CoreV1Api的Mock实例,并绑定到类的返回值 mock_api_instance = MagicMock() mock_core_v1_api.return_value = mock_api_instance # 模拟read_namespaced_secret的返回结果:构造带data属性的Secret对象 mock_secret = MagicMock() # 注意要对token做base64编码,和业务逻辑对应 mock_secret.data = {"token": base64.b64encode(b"valid-test-token").decode("utf-8")} mock_api_instance.read_namespaced_secret.return_value = mock_secret # 执行测试逻辑 k8s = kubernetesServices() token = k8s.get_secret_vault_token() # 验证结果与方法调用是否符合预期 self.assertEqual(token, "valid-test-token") mock_config.load_kube_config.assert_called_once() mock_api_instance.read_namespaced_secret.assert_called_once_with("random-sec", "random-ns")
关键修正说明
- 使用
MagicMock替代自定义Mock类,无需手动编写空类和方法,自动模拟所有属性和方法。 - Patch目标改为
read_sec.client.CoreV1Api:业务代码中通过from kubernetes import client引入,所以实际调用的是read_sec模块中的client.CoreV1Api,这是正确的Patch路径。 - 完整模拟调用链:从
config.load_kube_config到CoreV1Api实例化,再到read_namespaced_secret方法的返回值,完全匹配业务逻辑的执行流程。 - 添加断言验证:确保方法被正确调用,返回结果符合预期,提升测试的可靠性。
内容的提问来源于stack exchange,提问作者armzz_109
相关产品推荐
相关产品推荐

