"""Test that when the length of a mapped literal changes we remove extra TIs"""
    with dag_maker(session=session) as dag:
        @task
        def task_2(arg2):
            ...
        task_2.expand(arg2=[1, 2, 3, 4])
    dr = dag_maker.create_dagrun()
    
    dag._remove_task("task_2")
    with dag:
        mapped = task_2.expand(arg2=[1, 2]).operator
    
    
    serialized_dag = SerializedDAG.from_dict(SerializedDAG.to_dict(dag))
    dr.dag = serialized_dag
    dr.verify_integrity()
    indices = (
        session.query(TI.map_index, TI.state)
        .filter_by(task_id=mapped.task_id, dag_id=mapped.dag_id, run_id=dr.run_id)
        .order_by(TI.map_index)
        .all()
    )
    assert indices == [(0, None), (1, None), (2, TaskInstanceState.REMOVED), (3, TaskInstanceState.REMOVED)]
def test_mapped_literal_to_xcom_arg_verify_integrity(dag_maker, session):