worker_tests.py 文件源码

def test_actually_calling_retrying_library_in_a_failing_condition(self, ):
        # Arrange
        function_side_effects = [IndentationError(), SyntaxError(), OSError()]
        func = mock.MagicMock(side_effect=function_side_effects)
        from easy_job.workers.common import call_with_retry as target
        retrying_params = {
            "stop_max_attempt_number": 3
        }
        args = (1, 2, 3)
        kwargs = {"p": "arameter"}
        # Act
        with self.assertRaises(OSError):
            target(func, args, kwargs, retry_policy=retrying_params)

        # Assert
        func.assert_has_calls([mock.call(*args, **kwargs),
                               mock.call(*args, **kwargs),
                               mock.call(*args, **kwargs)]
                              )