test_runners.py 文件源码

def describe_threaded_runner():

    @pytest.fixture(scope='module')
    def runner_fixture():
        return ThreadedRunner(thread_pool_executor=ThreadPoolExecutor(max_workers=2))

    def describe_run_all_threaded_behavior():
        def given_many_tasks():
            def when_first_task_fails():
                def expect_later_tasks_still_run(runner_fixture):
                    task1_mock, task2_mock = create_task_mock(count=2)
                    task1_mock.successful = False

                    with pytest.raises(TaskFailedError):
                        runner_fixture.run_all([task1_mock, task2_mock])

                    task1_mock.execute.assert_called_once()
                    task2_mock.execute.assert_called_once()