ti2.state = State.RUNNING
        ti3 = TI(task=test_task, run_id=dr3.run_id)
        ti3.state = State.QUEUED
        ti4 = TI(task=test_task, run_id=dr4.run_id)
        ti4.state = State.RUNNING
        session = settings.Session()
        session.merge(ti1)
        session.merge(ti2)
        session.merge(ti3)
        session.merge(ti4)
        session.commit()
        assert 0 == DAG.get_num_task_instances(test_dag_id, task_ids=["fakename"], session=session)
        assert 4 == DAG.get_num_task_instances(test_dag_id, task_ids=[test_task_id], session=session)
        assert 4 == DAG.get_num_task_instances(
            test_dag_id, task_ids=["fakename", test_task_id], session=session
        )
        assert 1 == DAG.get_num_task_instances(
            test_dag_id, task_ids=[test_task_id], states=[None], session=session
        )
        assert 2 == DAG.get_num_task_instances(
            test_dag_id, task_ids=[test_task_id], states=[State.RUNNING], session=session
        )
        assert 3 == DAG.get_num_task_instances(
            test_dag_id, task_ids=[test_task_id], states=[None, State.RUNNING], session=session
        )
        assert 4 == DAG.get_num_task_instances(
            test_dag_id, task_ids=[test_task_id], states=[None, State.QUEUED, State.RUNNING], session=session
        )
        session.close()
    def test_get_task_instances_before(self):