dr: DagRun = dag_maker.create_dagrun()
    tis = {(ti.task_id, ti.map_index): ti for ti in dr.task_instances}
    
    
    
    assert sorted(map_index for (task_id, map_index) in tis if task_id == "task_2") == [-1]
    assert sorted(map_index for (task_id, map_index) in tis if task_id == "task_3") == [0, 1, 2]
    assert sorted(map_index for (task_id, map_index) in tis if task_id == "task_4") == [0, 1, 2]
    tis[("task_0", -1)].run()
    tis[("task_1", -1)].run()
    
    decision = dr.task_instance_scheduling_decisions()
    assert {(ti.task_id, ti.map_index): ti.state for ti in decision.schedulable_tis} == {
        ("task_2", 0): None,
        ("task_2", 1): None,
        ("task_2", 2): None,
        ("task_3", 0): None,
        ("task_3", 1): None,
        ("task_3", 2): None,
        ("task_4", 0): None,
        ("task_4", 1): None,
        ("task_4", 2): None,
    }
def test_mapped_skip_upstream_not_deadlock(dag_maker):
    with dag_maker() as dag:
        @dag.task
        def add_one(x: int):
            return x + 1
        @dag.task
        def say_hi():