ti.run()
    
    
    with mock.patch("airflow.models.dagrun.DagRun.verify_integrity") as mock_verify_integrity:
        dr.task_instance_scheduling_decisions()
        mock_verify_integrity.assert_not_called()
def test_calls_to_verify_integrity_with_mapped_task_increase_at_runtime(dag_maker, session):
    """
    Test increase in mapped task at runtime with calls to dagrun.verify_integrity
    """
    from airflow.models import Variable
    Variable.set(key="arg1", value=[1, 2, 3])
    @task
    def task_1():
        return Variable.get("arg1", deserialize_json=True)
    with dag_maker(session=session) as dag:
        @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()
    dr.task_instance_scheduling_decisions()
    tis = dr.get_task_instances()
    indices = [(ti.map_index, ti.state) for ti in tis if ti.map_index >= 0]
    assert sorted(indices) == [
        (0, State.NONE),
        (1, State.NONE),
        (2, State.NONE),