"""Test that when we change from literal to a XComArg the TIs are removed"""
    with dag_maker(session=session) as dag:
        t1 = BaseOperator(task_id="task_1")
        @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=t1.output).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, TaskInstanceState.REMOVED),
        (1, TaskInstanceState.REMOVED),
        (2, TaskInstanceState.REMOVED),
        (3, TaskInstanceState.REMOVED),
    ]
def test_mapped_literal_length_increase_adds_additional_ti(dag_maker, session):