Repository: apache/airflow
File: airflow-core/tests/unit/api_fastapi/core_api/routes/ui/test_grid.py (Python)

Code before:
                "task_display_name": "task_group.mapped_task",
                "max_end_date": None,
                "min_start_date": None,
                "state": None,
            },
        ]
        expected = sort_dict(expected)
        actual = sort_dict(actual)
        assert actual == expected

    def test_structure_includes_historical_removed_task_with_proper_shape(self, session, test_client):
        # Ensure the structure endpoint returns synthetic node for historical/removed task

        with assert_queries_count(7):
            response = test_client.get(f"/grid/structure/{DAG_ID_3}")
        assert response.status_code == 200
        nodes = response.json()
        # Find the historical removed task id
        t4 = next((n for n in nodes if n["id"] == TASK_ID_4), None)
        assert t4 is not None
        assert t4["label"] == TASK_ID_4
        # Optional None fields are excluded from response due to response_model_exclude_none=True
        assert "is_mapped" not in t4
        assert "children" not in t4

    def test_task_converted_to_taskgroup_doesnt_crash(self, session, test_client):
        """Test that converting a task to TaskGroup with same name doesn't crash grid view.

        Regression test for https://github.com/apache/airflow/issues/61208
        When a task is converted to a TaskGroup between DAG versions, old runs have
        children=None while new runs have children=[...]. The merge should handle this.
        """
        from airflow.models import DagRun
        from airflow.providers.standard.operators.python import PythonOperator
        from airflow.sdk import DAG

        from tests_common.test_utils.dag import sync_dag_to_db

        dag_id = "test_task_to_group_conversion"

        # Version 1: task_a is a simple task
        with DAG(
            dag_id=dag_id,
            start_date=pendulum.datetime(2024, 1, 1, tz="UTC"),
            schedule=None,
        ) as dag_v1:
            PythonOperator(task_id="task_a", python_callable=lambda: True)
            PythonOperator(task_id="task_b", python_callable=lambda: True)

        # Create DagModel and serialize old version
        sync_dag_to_db(dag_v1, bundle_name="test", session=session)

Review comment (bug):
v1 and v2 should be inverted. This test is already working on main and doesn't need your code change to work. 


On the other hand the bug appears when the task is a group, and then get transformed into a normal TI, and this is fixed by your code change. 

Basically you need to invert Version 1 and Version 2 and we should be code. And adjust the code comment

Code after:
                "task_id": "task_group.mapped_task",
                "task_display_name": "task_group.mapped_task",
                "max_end_date": None,
                "min_start_date": None,
                "state": None,
            },
        ]
        expected = sort_dict(expected)
        actual = sort_dict(actual)
        assert actual == expected

    def test_structure_includes_historical_removed_task_with_proper_shape(self, session, test_client):
        # Ensure the structure endpoint returns synthetic node for historical/removed task

        with assert_queries_count(7):
            response = test_client.get(f"/grid/structure/{DAG_ID_3}")
        assert response.status_code == 200
        nodes = response.json()
        # Find the historical removed task id
        t4 = next((n for n in nodes if n["id"] == TASK_ID_4), None)
        assert t4 is not None
        assert t4["label"] == TASK_ID_4
        # Optional None fields are excluded from response due to response_model_exclude_none=True
        assert "is_mapped" not in t4
        assert "children" not in t4

    def test_task_converted_to_task_group_doesnt_crash(self, session, dag_maker, test_client):
        """Test that converting a Task to a TaskGroup with same name doesn't crash grid view.

        Regression test for https://github.com/apache/airflow/issues/61208
        """

        dag_id = "test_task_to_group_conversion"

        # Version 1: task_a is a simple task
        with dag_maker(
            dag_id=dag_id,
            start_date=pendulum.datetime(2024, 1, 1, tz="UTC"),
            schedule=None,
        ):
            PythonOperator(task_id="task_a", python_callable=lambda: True)
            PythonOperator(task_id="task_b", python_callable=lambda: True)

        # Create another DagRun with the new version
        dag_maker.create_dagrun(
            run_id="test_run_1",
            run_type=DagRunType.MANUAL,
            logical_date=pendulum.datetime(2024, 1, 3, tz="UTC"),
        )

        response_v1 = test_client.get(f"/grid/structure/{dag_id}")