Repository: apache/airflow
File: airflow-core/tests/unit/models/test_serialized_dag.py (Python)

Code before:
                    bundle_name="testing",
                    bundle_version=None,
                    session=session,
                )
            session.rollback()

            # Verify that no new DagVersion was committed
            # Use a fresh session to ensure we're reading from committed data
            with create_session() as fresh_session:
                final_version_count = fresh_session.scalar(
                    select(func.count()).select_from(DagVersion).where(DagVersion.dag_id == dag.dag_id)
                )
                assert final_version_count == initial_version_count, (
                    "DagVersion should not be committed when DagCode.write_code fails"
                )

                sdag = SDM.get(dag.dag_id, session=fresh_session)
                assert sdag is not None, "Original SerializedDagModel should still exist"
                assert len(sdag.dag.task_dict) == 1, (
                    "SerializedDagModel should not be updated when write fails"
                )

    def test_deadline_interval_change_triggers_new_serdag(self, testing_dag_bundle, session):
        dag_id = "test_interval_change"

        # Create a new Dag with a deadline and create a dagrun as a baseline..
        dag = DAG(
            dag_id=dag_id,
            deadline=DeadlineAlert(
                reference=DeadlineReference.DAGRUN_QUEUED_AT,
                interval=timedelta(minutes=5),
                callback=AsyncCallback(empty_callback_for_deadline),
            ),
        )
        EmptyOperator(task_id="task1", dag=dag)
        scheduler_dag = sync_dag_to_db(dag, session=session)
        scheduler_dag.create_dagrun(
            run_id="test1",
            run_after=DEFAULT_DATE,
            state=DagRunState.QUEUED,
            logical_date=DEFAULT_DATE,
            data_interval=(DEFAULT_DATE, DEFAULT_DATE),
            triggered_by=DagRunTriggeredByType.TEST,
            run_type=DagRunType.MANUAL,
        )
        session.commit()
        orig_serdag = session.scalar(select(SDM).where(SDM.dag_id == dag_id).order_by(SDM.created_at.desc()))

        # Modify the Dag's deadline interval.
        dag.deadline = DeadlineAlert(
            reference=DeadlineReference.DAGRUN_QUEUED_AT,

Review comment (suggestion):
```suggestion
        # Create a new Dag with a deadline and create a dagrun as a baseline.
```

Code after:
                    bundle_name="testing",
                    bundle_version=None,
                    session=session,
                )
            session.rollback()

            # Verify that no new DagVersion was committed
            # Use a fresh session to ensure we're reading from committed data
            with create_session() as fresh_session:
                final_version_count = fresh_session.scalar(
                    select(func.count()).select_from(DagVersion).where(DagVersion.dag_id == dag.dag_id)
                )
                assert final_version_count == initial_version_count, (
                    "DagVersion should not be committed when DagCode.write_code fails"
                )

                sdag = SDM.get(dag.dag_id, session=fresh_session)
                assert sdag is not None, "Original SerializedDagModel should still exist"
                assert len(sdag.dag.task_dict) == 1, (
                    "SerializedDagModel should not be updated when write fails"
                )

    def test_deadline_interval_change_triggers_new_serdag(self, testing_dag_bundle, session):
        dag_id = "test_interval_change"

        # Create a new Dag with a deadline and create a dagrun as a baseline.
        dag = DAG(
            dag_id=dag_id,
            deadline=DeadlineAlert(
                reference=DeadlineReference.DAGRUN_QUEUED_AT,
                interval=timedelta(minutes=5),
                callback=AsyncCallback(empty_callback_for_deadline),
            ),
        )
        EmptyOperator(task_id="task1", dag=dag)
        scheduler_dag = sync_dag_to_db(dag, session=session)
        scheduler_dag.create_dagrun(
            run_id="test1",
            run_after=DEFAULT_DATE,
            state=DagRunState.QUEUED,
            logical_date=DEFAULT_DATE,
            data_interval=(DEFAULT_DATE, DEFAULT_DATE),
            triggered_by=DagRunTriggeredByType.TEST,
            run_type=DagRunType.MANUAL,
        )
        session.commit()
        orig_serdag = session.scalar(select(SDM).where(SDM.dag_id == dag_id).order_by(SDM.created_at.desc()))

        # Modify the Dag's deadline interval.
        dag.deadline = DeadlineAlert(
            reference=DeadlineReference.DAGRUN_QUEUED_AT,