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

如何在不修改生产代码的前提下,使用pytest测试模块级while循环的退出逻辑?

如何在不修改生产代码的前提下,使用pytest测试模块级while循环的退出逻辑?

我完全理解你的困扰——生产代码没加if __name__ == '__main__'防护,导致一导入就跑循环,patch根本赶不上生效。这种情况我之前踩过坑,给你几个可行的解决方案,都不用改原代码:

方案一:先应用所有Patch,再导入/重载main模块

问题的核心是模块级代码在import时立即执行,而你的patch是在import之后才生效的。所以要把顺序反过来:先把所有需要的patch都设置好,再导入main模块(如果之前已经导入过,就用importlib.reload重新加载)。

调整你的测试代码如下:

import importlib
import pytest
from unittest.mock import patch, Mock

@pytest.fixture()
def mock_message_queue():
    # 保持你的fixture逻辑不变
    with patch('mq.pika.BlockingConnection') as mock_conn, \
         patch('mq.pika.PlainCredentials'), \
         patch('mq.pika.ConnectionParameters'):
        mock_conn_instance = Mock()
        mock_channel = Mock()
        mock_conn_instance.is_open = True
        mock_conn_instance.channel.return_value = mock_channel
        mock_conn.return_value = mock_conn_instance
        mock_channel.is_open = True
        mock_channel.queue_declare.return_value = Mock()
        mock_channel.basic_publish.return_value = None
        mock_channel.basic_qos.return_value = None
        mock_channel.basic_consume.return_value = None
        mock_channel.start_consuming.return_value = None
        mq = MessageQueue(
            host=HOST, port=PORT, virtual_host=VIRTUAL_HOST,
            user=USER, password=PASSWORD
        )
        mq.get_channel = Mock(return_value=mock_channel)
        mq.send_message = Mock()
        mq._test_mocks = {
            'connection': mock_conn,
            'connection_instance': mock_conn_instance,
            'channel': mock_channel
        }
        yield mq
        try:
            mq.close()
        except:
            pass

def test_main(mock_message_queue):
    sample_data = {}
    # 先定义所有需要的patch,包括check_condition的side_effect
    with patch('utils.list_doc', return_value=[sample_data]), \
         patch('utils.get_MQ', return_value=mock_message_queue), \
         patch('main.check_condition', side_effect=[True, False]) as mock_check:
        # 先检查main是否已经被导入,如果是就重载,否则直接导入
        import sys
        if 'main' in sys.modules:
            importlib.reload(sys.modules['main'])
        else:
            import main
        
        # 现在断言逻辑
        assert mock_check.call_count >= 2  # 至少被调用两次(True然后False)
        assert mock_message_queue.send_message.call_count == 3

这样做的原理是:所有patch在导入/重载main模块前就已经生效,当模块级的while循环执行时,check_condition()会先返回True,然后返回False,触发break退出循环,不会卡在time.sleep(30)里。

方案二:用pytest的autouse fixture提前Patch(适合多个测试用例)

如果有多个测试用例需要测试main模块,你可以用一个autouse fixture,在所有测试开始前就patch main.check_condition,或者在导入main模块前就做好准备:

import importlib
import sys
import pytest
from unittest.mock import patch

@pytest.fixture(autouse=True)
def prevent_main_loop_before_patch():
    # 先patch check_condition为立即返回False,避免导入时循环卡住
    with patch('main.check_condition', return_value=False):
        # 如果main已经被导入,先卸载
        if 'main' in sys.modules:
            del sys.modules['main']
        yield
    # 测试结束后清理模块缓存
    if 'main' in sys.modules:
        del sys.modules['main']

def test_main_with_exit(mock_message_queue):
    sample_data = {}
    with patch('utils.list_doc', return_value=[sample_data]), \
         patch('utils.get_MQ', return_value=mock_message_queue), \
         patch('main.check_condition', side_effect=[True, False]) as mock_check:
        import main
        importlib.reload(main)
        
        assert mock_check.call_count == 2
        assert mock_message_queue.send_message.call_count == 3

这个autouse fixture会在每个测试前,先把check_conditionpatch成返回False,然后卸载main模块的缓存,这样当测试里导入main时,模块级代码执行时会因为check_condition()返回False而直接退出循环,不会卡住,之后再用测试需要的side_effect重载。

为什么之前的方法无效?

你之前的问题在于:当你在测试中import main时,模块级的while循环已经在patch生效前就开始执行了——甚至在你的测试函数还没完全设置好patch的时候,main.py的代码已经跑到time.sleep(30)了,这时候patch根本没机会作用到check_condition()上。调整导入顺序,让patch先生效,再执行模块级代码,就能解决这个问题。

额外注意事项

  • 每次测试后记得清理sys.modules里的main模块,避免下一次测试使用缓存的模块,导致patch失效。
  • 如果你的main模块里有其他模块级的全局变量,重载后会被重置,这在测试中是正常的,因为每次测试都应该是独立的。
  • 确保check_condition的patch路径正确:你需要patch的是main.check_condition,也就是main模块里的那个函数引用,而不是utils.check_condition(从你的代码看,check_condition是定义在main.py里的,所以当前路径是对的)。

内容来源于stack exchange

相关产品推荐
方舟 Agent Plan

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

最近更新时间:2026.04.07 12:44:32