模拟 kubernetes 客户端

Question

我正在嘲笑一个用于读取 k8s 秘密以获取秘密令牌的函数。但是运行unittest会产生错误：

AttributeError: init.py'> 确实 没有属性“read_namespaced_secret()”

我已经经历了How do you mock Python Kubernetes client CoreV1Api，但它也对我的情况没有帮助。谁能指出我在这里做错了什么？

我的脚本 - 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()

Answer 1

您需要嘲笑

kubernetes.client.CoreV1Api

而不是

kubernetes.client

。这是一个例子：

import base64
import unittest
from unittest.mock import patch, Mock

import requests
from kubernetes import client, config


class kubernetesServices():
    def get_secret_vault_token(self):
        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


class TestkubernetesServices(unittest.TestCase):
    @patch(
        'kubernetes.client.CoreV1Api',
        return_value=Mock(read_namespaced_secret=Mock(return_value=Mock(data={'token': b'YWJjMTIz'})))
    )
    @patch('kubernetes.config.load_kube_config', return_value=Mock())
    def test_get_secret_vault_token(self, mock_client, mock_config):
        k8s = kubernetesServices()
        token = k8s.get_secret_vault_token()
        self.assertEqual(token, 'abc123')

结果：

---------------------------------------------------------------------
Ran 1 tests in 0.071s

PASSED (successes=1)

JFYI：当您需要多个结果时，更适合使用

side_effect

。示例：

class TestRequest(unittest.TestCase):
    def test_side_effect(self):
        with patch('requests.get', side_effect=[1, 2, 3]):
            print(requests.get('url1'))  # 1
            print(requests.get('url2'))  # 2
            print(requests.get('url3'))  # 3