pipelinerunner_test.py 文件源码

def test_run_pipeline_pass_skip_parse_context(mocked_work_dir,
                                              mocked_get_pipe_def,
                                              mocked_get_parsed_context,
                                              mocked_run_step_group):
    """run_pipeline passes correct params to all methods."""

    pypyr.pipelinerunner.run_pipeline(
        pipeline_name='arb pipe',
        working_dir='arb/dir',
        parse_input=False)

    mocked_work_dir.assert_not_called()
    mocked_get_pipe_def.assert_called_once_with(pipeline_name='arb pipe',
                                                working_dir='arb/dir')
    mocked_get_parsed_context.assert_not_called()

    # 1st called steps, then on_success
    expected_run_step_groups = [call(context={},
                                     pipeline_definition='pipe def',
                                     step_group_name='steps'),
                                call(context={},
                                     pipeline_definition='pipe def',
                                     step_group_name='on_success')]

    mocked_run_step_group.assert_has_calls(expected_run_step_groups)