Python单元测试Mock InfluxDBClient失效导致超时问题排查

我需要测试一个向InfluxDB写入数据的Influx1Writer类，该类使用influxdb.InfluxDBClient进行数据库操作。在单元测试中尝试Mock influxdb.InfluxDBClient，但Mock未生效，执行测试时始终出现超时。

待测试类代码 #

''' Client for influxDB1. '''
from influxdb import InfluxDBClient

class Influx1Writer:
    ''' Client to access influxDB1. '''
        
    def __init__(self, host, port, user, password, dbname, usessl) -> None:
        ''' Initializer, will be called from constructor. '''
        self.client = InfluxDBClient(
            host=host,
            port=port,
            username=user,
            password=password,
            database=dbname,
            ssl=usessl)
        self.client.create_database(dbname)
        self.client.switch_database(dbname)
        self.client.switch_user(user, password)

原测试用例代码 #

''' Unittest writer for influxdb, version 1. '''
import unittest
from unittest.mock import patch, MagicMock

from influxdb import InfluxDBClient
from influx.influx_writer_1 import Influx1Writer

class TestInflux1Writer(unittest.TestCase):
    ''' Unittests of Influx1Writer. '''

    host = "host"
    port = 8080
    user = "user"
    password = "password"
    dbname = "dbname"
    usessl = False

    def setUp(self) -> None:
        ''' Setup test fixture. '''
        mock_influx = MagicMock('influxdb.InfluxDBClient')
        self.handler = Influx1Writer(self.host, \
                                     self.port, \
                                     self.user, \
                                     self.password, \
                                     self.dbname, \
                                     self.usessl)
        mock_influx.assert_called_with(self.host, \
                                        self.port, \
                                        self.user, \
                                        self.password, \
                                        self.dbname, \
                                        self.usessl)

        super().setUp()

    def tearDown(self) -> None:
        ''' Tear down test fixture. '''
        self.handler = None
        super().tearDown()

    @patch('influxdb.InfluxDBClient')
    @patch('influxdb.InfluxDBClient.create_database')
    @patch('influxdb.InfluxDBClient.switch_database')
    @patch('influxdb.InfluxDBClient.switch_user')
    def test_01a_init_called(self, mock_client, mock_create_database, mock_switch_database, mock_switch_user) -> None:
        ''' Check InfluxDBClient is called with parameters. '''
        mock_client.__init__.assert_called_with(self.host, self.port, self.user, self.password, self.dbname, self.usessl)
        mock_create_database.assert_called_with(self.dbname)
        mock_switch_database.assert_called_with(self.dbname)
        mock_switch_user.assert_called_with(self.dbname)
        self.assertEqual(type(self.handler), Influx1Writer)

错误原因 #

1. Mock路径错误：patch需要Mock的是被测试模块中实际引用的InfluxDBClient，即influx.influx_writer_1.InfluxDBClient，而非库的原始路径influxdb.InfluxDBClient。原代码Mock的是库本身的类，不会影响被测试类中已导入的引用。
2. 孤立Mock对象无效：setUp中创建的MagicMock只是独立对象，并未替换被测试类使用的实际InfluxDBClient，测试时仍会尝试连接真实数据库导致超时。
3. 装饰器参数顺序错误：@patch装饰器的参数从右到左传递给测试方法，原测试方法的参数顺序完全颠倒，导致断言对象对应错误。

修正后的测试代码 #

修改后的setUp与tearDown方法 #

def setUp(self) -> None:
    ''' Setup test fixture. '''
    self.patch_client = patch('influx.influx_writer_1.InfluxDBClient')
    mock_influx = self.patch_client.start()
    self.handler = Influx1Writer(self.host, \
                                self.port, \
                                self.user, \
                                self.password, \
                                self.dbname, \
                                self.usessl)
    mock_influx.assert_called_with(
        host=self.host,
        port=self.port,
        username=self.user,
        password=self.password,
        database=self.dbname,
        ssl=self.usessl
    )
    super().setUp()

def tearDown(self) -> None:
    ''' Tear down test fixture. '''
    self.patch_client.stop()
    self.handler = None
    super().tearDown()

简化后的测试方法（可选） #

@patch('influx.influx_writer_1.InfluxDBClient')
def test_01a_init_called(self, mock_client) -> None:
    ''' Check InfluxDBClient and its methods are called correctly. '''
    handler = Influx1Writer(self.host, self.port, self.user, self.password, self.dbname, self.usessl)
    
    # 断言客户端初始化参数
    mock_client.assert_called_with(
        host=self.host,
        port=self.port,
        username=self.user,
        password=self.password,
        database=self.dbname,
        ssl=self.usessl
    )
    
    # 断言实例方法调用
    mock_instance = mock_client.return_value
    mock_instance.create_database.assert_called_with(self.dbname)
    mock_instance.switch_database.assert_called_with(self.dbname)
    mock_instance.switch_user.assert_called_with(self.user, self.password)
    
    self.assertEqual(type(handler), Influx1Writer)

修正说明 #

• 锁定Mock目标：针对被测试模块内的InfluxDBClient引用Mock，确保替换生效，避免连接真实数据库。
• 管理Mock生命周期：在setUp启动Mock，tearDown停止，保证每个测试用例的Mock独立。
• 匹配关键字参数：使用与原类初始化一致的关键字参数断言，避免位置参数不匹配问题。
• 简化Mock层级：直接MockInfluxDBClient类，通过return_value获取实例Mock断言方法调用，无需单独Mock每个方法。

内容的提问来源于stack exchange，提问作者Benny