You need to enable JavaScript to run this app.
优惠活动
大模型
产品
解决方案
定价
更多

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()

错误分析与解决方案

核心错误点

  1. Patch目标路径错误:read_namespaced_secret是CoreV1Api类的实例方法,并非kubernetes.client模块的直接属性,原测试中@patch("kubernetes.client.read_namespaced_secret()")的写法完全错误(还额外加了括号)。
  2. Patch层级错误:应该针对业务代码中实际调用的对象进行Patch,而非直接patch顶层的kubernetes.client模块。
  3. 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

相关产品推荐
方舟 Agent Plan

超全模态模型 × Harness 升级,最新支持 Deepseek-V4.1-Flash、GLM-5.3 系列、Doubao-Seedream-5.0-pro、Kimi-K3 (部分), 限时 9.9 元起

最近更新时间:2026.08.14 05:15:35