Python: why is mock not called in this instance of loop?-CodePudding

I have the following test code:

@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()

Inside wait_for_recording_result it uses a library timer.Timer.loop method (which is being mocked):

    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
        catch timer.Timeout:
            ...

All the test cases pass, except for the first test case on t.loop.assert_called() and the fourth test case seems to wait for the 60 second timeout (even though loop is mocked):

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.

How is it that the mock is called in every other circumstance, yet it still waits for 60 seconds on the 4th case, while claiming the mock was not called in the first case?

The only thing the init does of timer.Timer that I can think may cause it is self._event = threading.Event(), but this I can't see used anywhere except inside the loop method, meaning it should not be called as it is mocked.

I should also note that timer.Timeout is a custom exception, from the same timer package, inheriting from Exception so it isn't part of built-in python.

twilio_mock_testing merely creates a mock for client property of this Twilio package and not of any of the methods such as wait_for_recording_result...

with mock.patch("twilio.rest.Client"):
    return Twilio(sid="Hello", token="olleH")

I am incredibly confused by this...

CodePudding user response：

Was answered by the following change based on the comment by chepner:

    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, 'loop') as t:
                t.side_effect = [True, False] if exception is not timer.Timeout else [True, timer.Timeout]
                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.assert_called()