'test that running task which returns AirflowSkipOperator will end
up in a SKIPPED state.'

def test_run_pooling_task_with_skip(self):

    def raise_skip_exception():
        raise AirflowSkipException
    dag = models.DAG(dag_id=u'test_run_pooling_task_with_skip')
    task = PythonOperator(task_id=u'test_run_pooling_task_with_skip', dag=dag, python_callable=raise_skip_exception, owner=u'airflow', start_date=datetime.datetime(2016, 2, 1, 0, 0, 0))
    ti = TI(task=task, execution_date=datetime.datetime.now())
    ti.run()
    self.assertTrue((ti.state == models.State.SKIPPED))
