test_tasks.py 文件源码

def test_transcode_video_twice(self):
        factories.VideoFactory(public_id='videoid')
        mock_backend = Mock(return_value=Mock(
            start_transcoding=Mock(return_value=['job1']),
            iter_formats=Mock(return_value=[]),
        ))

        # First attempt: failure
        mock_backend.return_value.check_progress = Mock(side_effect=exceptions.TranscodingFailed)
        with override_settings(PLUGIN_BACKEND=mock_backend):
            tasks.transcode_video('videoid')

        # Second attempt: success
        mock_backend.return_value.check_progress = Mock(return_value=(100, True))
        with override_settings(PLUGIN_BACKEND=mock_backend):
            tasks.transcode_video('videoid')

        video_processing_state = models.ProcessingState.objects.get()
        self.assertEqual(models.ProcessingState.STATUS_SUCCESS, video_processing_state.status)
        self.assertEqual("", video_processing_state.message)
        self.assertEqual(100, video_processing_state.progress)