return [1, 2]
        @task
        def task_2(arg2):
            ...
        task_2.expand(arg2=task_1())
    dr = dag_maker.create_dagrun()
    ti = dr.get_task_instance(task_id="task_1")
    ti.run()
    decision = dr.task_instance_scheduling_decisions()
    assert len(decision.schedulable_tis) == 2
    
    session.add(TaskInstance(dag.get_task("task_2"), dr.execution_date, dr.run_id))
    session.flush()
    decision = dr.task_instance_scheduling_decisions()
    assert len(decision.schedulable_tis) == 2
def test_mapped_literal_length_with_no_change_at_runtime_doesnt_call_verify_integrity(dag_maker, session):
    """
    Test that when there's no change to mapped task indexes at runtime, the dagrun.verify_integrity
    is not called
    """
    from airflow.models import Variable
    Variable.set(key="arg1", value=[1, 2, 3])
    @task
    def task_1():