stepsrunner_test.py 文件源码

def test_run_failure_step_group_swallows():
    """Failure step group runner swallows errors."""
    logger = pypyr.log.logger.get_logger('pypyr.stepsrunner')

    with patch('pypyr.stepsrunner.run_step_group') as mock_run_group:
        with patch.object(logger, 'error') as mock_logger_error:
            mock_run_group.side_effect = ContextError('arb error')
            pypyr.stepsrunner.run_failure_step_group(
                {'pipe': 'val'}, Context())

        mock_logger_error.assert_any_call(
            "Failure handler also failed. Swallowing.")

    mock_run_group.assert_called_once_with(pipeline_definition={'pipe': 'val'},
                                           step_group_name='on_failure',
                                           context=Context())

# ------------------------- run_failure_step_group----------------------------#

# ------------------------- run_pipeline_step---------------------------------#