'tests xcom set/clear relative to a task in a \'success\' rerun scenario'

def test_xcom_pull_after_success(self):

    key = u'xcom_key'
    value = u'xcom_value'
    dag = models.DAG(dag_id=u'test_xcom', schedule_interval=u'@monthly')
    task = DummyOperator(task_id=u'test_xcom', dag=dag, pool=u'test_xcom', owner=u'airflow', start_date=datetime.datetime(2016, 6, 2, 0, 0, 0))
    exec_date = datetime.datetime.now()
    ti = TI(task=task, execution_date=exec_date)
    ti.run(mark_success=True)
    ti.xcom_push(key=key, value=value)
    self.assertEqual(ti.xcom_pull(task_ids=u'test_xcom', key=key), value)
    ti.run()
    self.assertEqual(ti.xcom_pull(task_ids=u'test_xcom', key=key), value)
    ti.run(ignore_all_deps=True, mark_success=True)
    self.assertEqual(ti.xcom_pull(task_ids=u'test_xcom', key=key), value)
    ti.run(ignore_all_deps=True)
    self.assertEqual(ti.xcom_pull(task_ids=u'test_xcom', key=key), None)
