with dag_maker(session=session):
        @task(depends_on_past=True)
        def print_value(value):
            print(value)
        print_value.expand_kwargs([{"value": i} for i in range(2)])
    dr1: DagRun = dag_maker.create_dagrun(run_type=DagRunType.SCHEDULED)
    dr2: DagRun = dag_maker.create_dagrun_after(dr1, run_type=DagRunType.SCHEDULED)
    
    decision = dr2.task_instance_scheduling_decisions(session=session)
    assert len(decision.schedulable_tis) == 0
    
    decision = dr1.task_instance_scheduling_decisions(session=session)
    assert len(decision.schedulable_tis) == 2
    for ti in decision.schedulable_tis:
        ti.run(session=session)
    
    decision = dr2.task_instance_scheduling_decisions(session=session)
    assert len(decision.schedulable_tis) == 2
    for ti in decision.schedulable_tis:
        ti.run(session=session)
    
    decision = dr1.task_instance_scheduling_decisions(session=session)
    assert len(decision.unfinished_tis) == 0
    decision = dr2.task_instance_scheduling_decisions(session=session)
    assert len(decision.unfinished_tis) == 0
def test_clearing_task_and_moving_from_non_mapped_to_mapped(dag_maker, session):