В моем коде (
Код: Выделить всё
main.py
Код: Выделить всё
import time
from utils import *
def check_condition():
if something_wrong:
return False # Only get False by an external condition, not by the function's internal logic
return True
queue = get_MQ()
while True:
# some logic
if not check_condition():
break
# some logic
queue.close_connect()
time.sleep(30)
Я пытался исправить проверку_контроли с помощью side_effect = [true, false] , чтобы моделировать это в тестировании, но во время тестирования всегда возвращает true , и Loop никогда не исследует. Напечатано несколько журналов внутри main.py и заметил, что код всегда выполняется полностью во время импорта, прежде чем патч даже будет применен. Он достигает времени. Sleep (30) в конце, а затем продолжает перезагружать в бесконечном цикле - вот почему патч не имеет эффекта.
Вот упрощенная версия моего тестового примера:
Код: Выделить всё
@pytest.fixture()
def mock_message_queue():
with patch('mq.pika.BlockingConnection') as mock_conn, \
patch('mq.pika.PlainCredentials'), \
patch('mq.pika.ConnectionParameters'):
mock_conn_instance = Mock()
mock_channel = Mock()
# Setup connection mock
mock_conn_instance.is_open = True
mock_conn_instance.channel.return_value = mock_channel
mock_conn.return_value = mock_conn_instance
# Setup channel mock
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 = { } # Input data when run file main.py
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]):
import main #
Подробнее здесь: [url]https://stackoverflow.com/questions/79718529/using-pytest-to-test-a-while-loop-in-module-level[/url]