'Test to check that a DAG with a system.exit() doesn\'t break the scheduler.'

def test_dag_with_system_exit(self):

    dag_id = u'exit_test_dag'
    dag_ids = [dag_id]
    dag_directory = os.path.join(settings.DAGS_FOLDER, u'..', u'dags_with_system_exit')
    dag_file = os.path.join(dag_directory, u'b_test_scheduler_dags.py')
    dagbag = DagBag(dag_folder=dag_file)
    for dag_id in dag_ids:
        dag = dagbag.get_dag(dag_id)
        dag.clear()
    scheduler = SchedulerJob(dag_ids=dag_ids, subdir=dag_directory, num_runs=1, **self.default_scheduler_args)
    scheduler.run()
    session = settings.Session()
    self.assertEqual(len(session.query(TI).filter((TI.dag_id == dag_id)).all()), 1)
