正在为Python单元测试创建AttributeError模拟kubernetes客户端

yyhrrdl8 于 2022-11-21 发布在 Kubernetes

关注(0)|答案(1)|浏览(116)

我正在模拟一个用于读取k8s secret以获取secret token的函数。但是运行unittest会产生错误-**AttributeError：〈module 'kubernetes.client' from '/usr/lib/python3.6/site-packages/kubernetes/client/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()

kubernetes

来源：https://stackoverflow.com/questions/74285773/mocking-kubernetes-client-for-python-unittest-creating-attributeerror

1条答案

按热度按时间

cidc1ykv1#

您需要模拟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

赞(0）回复(0）举报 2022-11-21

我来回答

正在为Python单元测试创建AttributeError模拟kubernetes客户端

1条答案

相关问题

热门标签

最新问答