return value * 2
        @dag.task
        def consumer(value):
            nonlocal result
            result = list(value)
        consumer(value=double.expand(value=make_list()))
    dr: DagRun = dag_maker.create_dagrun()
    def _task_ids(tis):
        return [ti.task_id for ti in tis]
    
    decision = dr.task_instance_scheduling_decisions(session=session)
    assert _task_ids(decision.schedulable_tis) == ["make_list"]
    
    decision.schedulable_tis[0].run(verbose=False, session=session)
    decision = dr.task_instance_scheduling_decisions(session=session)
    assert _task_ids(decision.schedulable_tis) == ["double", "double"]
    
    decision.schedulable_tis[0].run(verbose=False, session=session)
    decision = dr.task_instance_scheduling_decisions(session=session)
    assert _task_ids(decision.schedulable_tis) == ["double"]
    
    decision.schedulable_tis[0].run(verbose=False, session=session)
    decision = dr.task_instance_scheduling_decisions(session=session)
    assert _task_ids(decision.schedulable_tis) == ["consumer"]
    
    decision.schedulable_tis[0].run(verbose=False, session=session)
    decision = dr.task_instance_scheduling_decisions(session=session)
    assert decision.schedulable_tis == []
    assert result == [2, 4]
def test_schedule_tis_map_index(dag_maker, session):