'Test if a task instance will be added if the dag is updated'

def test_scheduler_add_new_task(self):

    dag = DAG(dag_id=u'test_scheduler_add_new_task', start_date=DEFAULT_DATE)
    dag_task1 = DummyOperator(task_id=u'dummy', dag=dag, owner=u'airflow')
    session = settings.Session()
    orm_dag = DagModel(dag_id=dag.dag_id)
    session.merge(orm_dag)
    session.commit()
    session.close()
    scheduler = SchedulerJob()
    dag.clear()
    dr = scheduler.create_dag_run(dag)
    self.assertIsNotNone(dr)
    tis = dr.get_task_instances()
    self.assertEquals(len(tis), 1)
    dag_task2 = DummyOperator(task_id=u'dummy2', dag=dag, owner=u'airflow')
    queue = mock.Mock()
    scheduler._process_task_instances(dag, queue=queue)
    tis = dr.get_task_instances()
    self.assertEquals(len(tis), 2)
