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

Python测试疑问:loop实例Mock未被调用及超时问题排查

测试用例中Timer.loop Mock异常问题

测试代码

@pytest.mark.parametrize(
    "any_recordings,any_transcriptions,returns_recording,duration,exception",
    [
        (
            True,
            True,
            True,
            60,
            None
        ),
        (
            True,
            True,
            False,
            121,
            RecordingLengthException,
        ),
        (
            True,
            True,
            False,
            1,
            RecordingLengthException,
        ),
        (
            False,
            False,
            False,
            None,
            timer.Timeout,
        ),
        (
            True,
            False,
            False,
            0,
            NotTranscribedException,
        ),
    ],
)
async def test_10_mock_wait_for_recording_result(
    twilio_mock_testing: Twilio,
    any_recordings: bool,
    any_transcriptions: bool,
    returns_recording: bool,
    duration: int,
    exception: Optional[Exception],
):
    with mock.patch("twilio.rest.api.v2010.account.recording.transcription.TranscriptionInstance") as transcription_instance:
        transcription_instance.transcription_text = "Hello there."
    with mock.patch("twilio.rest.api.v2010.account.recording.RecordingInstance") as recording_instance:
        recording_instance.call_sid = "54321"
        recording_instance.date_created = datetime.datetime.now()
        recording_instance.price = "1.25"
        recording_instance.source = "whatever"
        recording_instance.transcriptions.list.return_value = [transcription_instance] if any_transcriptions else []
        recording_instance.duration = duration
        recording_instance.uri = "http://possu.xyz"
        recording_instance.status = CallStatusFinal.COMPLETED.value
        twilio_mock_testing.client.recordings.list.return_value = [recording_instance] if any_recordings else []
    on_transcribed: Callable[[Call, Recording], Any] = mock.Mock(return_value=None)
    with mock.patch.object(timer, 'Timer') as t:
        t.loop = mock.MagicMock(return_value=True)
    with pytest.raises(exception) if exception is not None else nullcontext():
        recording = twilio_mock_testing.wait_for_recording_result(
            call=Call(
                sid="54321",
                direction="abc",
                to="1234567890",
                from_="9876543210",
                status=CallStatusRunning.IN_PROGRESS.value,
                start_time=datetime.datetime.now(),
                end_time=None,
                duration=None,
                price=None,
                caller_name=None,
            ),
            raise_on_timeout=True,
            on_transcribed=on_transcribed
        )
        twilio_mock_testing.client.recordings.list.assert_called()
        if returns_recording:
            assert recording.status == CallStatusFinal.COMPLETED.value
            on_transcribed.assert_called_once()
        else:
            on_transcribed.assert_not_called()
        t.loop.assert_called()

待测试方法(wait_for_recording_result)实现

def wait_for_recording_result(
    self, *, call: Call, raise_on_timeout=False, on_transcribed: Callable[[Call, Recording], Any] = None
):
    recording_timer = timer.Timer(timeout=60, sleep=5, what="call recording")
    recording = None
    try:
        while recording_timer.loop(raise_timeout=raise_on_timeout):
            recordings = list(self.recordings(call_sid=call.sid))
            if any(recordings):
                recording = recordings[0]
                if (
                    recording.duration <= 2 or recording.duration >= 120
                ):  # See https://www.twilio.com/docs/api/errors/13617
                    raise RecordingLengthException(recording)
                if any(recording.transcriptions):
                    break
    except timer.Timeout:
        ...

测试执行结果

tests/integrations/test_twilio.py::test_10_mock_wait_for_recording_result[True-True-True-60-None] FAILED
tests/integrations/test_twilio.py::test_10_mock_wait_for_recording_result[True-True-False-121-RecordingLengthException] PASSED
tests/integrations/test_twilio.py::test_10_mock_wait_for_recording_result[True-True-False-1-RecordingLengthException] PASSED
tests/integrations/test_twilio.py::test_10_mock_wait_for_recording_result[False-False-False-None-Timeout] PASSED
tests/integrations/test_twilio.py::test_10_mock_wait_for_recording_result[True-False-False-0-NotTranscribedException] PASSED

FAILED tests/integrations/test_twilio.py::test_10_mock_wait_for_recording_result[True-True-True-60-None] - AssertionError: Expected 'loop' to have been called.

问题与背景

当前测试存在两个异常情况:

  • 第一个测试用例(True-True-True-60-None)中,断言t.loop.assert_called()失败,提示loop未被调用,但其他用例中该Mock均正常触发。
  • 第四个测试用例(False-False-False-None-Timeout)中,尽管Mock了Timer.loop,但测试仍会真实等待60秒超时,而非触发Mock逻辑。

我猜测问题可能和Timer类__init__方法中的self._event = threading.Event()有关,但该对象仅在loop方法内使用,而loop已经被Mock,无法理解为何会影响测试。

补充背景:

  • *timer.Timeout*是自定义异常,非Python内置;
  • twilio_mock_testing仅Mock了Twilio类的client属性,未Mockwait_for_recording_result等方法,Mock代码如下:
with mock.patch("twilio.rest.Client"):
    return Twilio(sid="Hello", token="olleH")

内容的提问来源于stack exchange,提问作者NotoriousPyro

相关产品推荐
方舟 Agent Plan

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

最近更新时间:2026.08.02 05:35:42