如何在不修改生产代码的前提下,使用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

