Repository: apache/airflow
File: airflow-core/tests/unit/jobs/test_scheduler_job.py (Python)

Code before:
        assert len(b_tis_in_executor) == 2

        # Global task should go to the default executor (scheduler_job.executor)
        global_tis_in_executor = [
            ti for ti in executor_to_tis.get(scheduler_job.executor, []) if ti.dag_id == "dag_global"
        ]
        assert len(global_tis_in_executor) == 1

        # Verify no cross-contamination: team pi tasks should not be in team rho executor and vice versa
        a_tis_in_wrong_executor = [
            ti for ti in executor_to_tis.get(mock_executors[1], []) if ti.dag_id == "dag_a"
        ]
        assert len(a_tis_in_wrong_executor) == 0

        b_tis_in_wrong_executor = [
            ti for ti in executor_to_tis.get(mock_executors[0], []) if ti.dag_id == "dag_b"
        ]
        assert len(b_tis_in_wrong_executor) == 0

    def task_helper(self, dag_maker, session, dag_id: str, task_num: int):
        dag_tasks = {}

        with dag_maker(dag_id=dag_id):
            for i in range(task_num):
                # Assign priority weight to certain tasks.
                if (i % 10) == 0:  # 10, 20, 30, 40, 50, ...
                    weight = int(i / 2)
                    dag_tasks[f"op{i}"] = EmptyOperator(task_id=f"dummy{i}", priority_weight=weight)
                else:
                    # No executor specified, runs on default executor
                    dag_tasks[f"op{i}"] = EmptyOperator(task_id=f"dummy{i}")

        dag_run = dag_maker.create_dagrun(run_type=DagRunType.SCHEDULED)

        task_tis = {}

        tis_list = []
        for i in range(task_num):
            task_tis[f"ti{i}"] = dag_run.get_task_instance(dag_tasks[f"op{i}"].task_id, session)
            # add
            tis_list.append(task_tis[f"ti{i}"])

        for ti in tis_list:
            ti.state = State.SCHEDULED
            ti.dag_model.max_active_tasks = 4

        session.flush()

        return tis_list

    @conf_vars(

Review comment (suggestion):
The comment suggests the condition matches tasks at indices 10, 20, 30, etc., but the condition `(i % 10) == 0` actually matches indices 0, 10, 20, 30, etc. The comment should be updated to "# 0, 10, 20, 30, 40, ..." to accurately reflect when the condition is true.
```suggestion
                if (i % 10) == 0:  # 0, 10, 20, 30, 40, 50, ...
```

Code after:
        dag_id = "SchedulerJobTest.test_find_executable_task_instances_executor"

        with dag_maker(dag_id=dag_id):
            op1 = EmptyOperator(task_id="dummy1")  # No executor specified, runs on default executor
            op2 = EmptyOperator(task_id="dummy2", executor="default_exec")
            op3 = EmptyOperator(task_id="dummy3", executor="default.exec.module.path")
            op4 = EmptyOperator(task_id="dummy4", executor="secondary_exec")
            op5 = EmptyOperator(task_id="dummy5", executor="secondary.exec.module.path")

        dag_run = dag_maker.create_dagrun(run_type=DagRunType.SCHEDULED)

        ti1 = dag_run.get_task_instance(op1.task_id, session)
        ti2 = dag_run.get_task_instance(op2.task_id, session)
        ti3 = dag_run.get_task_instance(op3.task_id, session)
        ti4 = dag_run.get_task_instance(op4.task_id, session)
        ti5 = dag_run.get_task_instance(op5.task_id, session)

        tis_tuple = (ti1, ti2, ti3, ti4, ti5)
        for ti in tis_tuple:
            ti.state = State.SCHEDULED

        session.flush()

        res = self.job_runner._executable_task_instances_to_queued(max_tis=32, session=session)

        assert len(res) == 5
        res_ti_keys = [res_ti.key for res_ti in res]
        for ti in tis_tuple:
            assert ti.key in res_ti_keys

    @conf_vars({("core", "multi_team"): "true"})
    def test_find_executable_task_instances_executor_with_teams(self, dag_maker, mock_executors, session):
        """
        Test that tasks are correctly routed to team-specific executors when multi-team is enabled
        """
        clear_db_teams()
        clear_db_dag_bundles()

        team1 = Team(name="team_a")
        team2 = Team(name="team_b")
        session.add_all([team1, team2])
        session.flush()

        bundle1 = DagBundleModel(name="bundle_a")
        bundle2 = DagBundleModel(name="bundle_b")
        bundle1.teams.append(team1)
        bundle2.teams.append(team2)
        session.add_all([bundle1, bundle2])
        session.flush()

        mock_executors[0].team_name = "team_a"