'tests xcom fetch behavior with different execution dates, using
both xcom_pull with "include_prior_dates" and without'

def test_xcom_pull_different_execution_date(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()
    exec_date += datetime.timedelta(days=1)
    ti = TI(task=task, execution_date=exec_date)
    ti.run()
    self.assertEqual(ti.xcom_pull(task_ids=u'test_xcom', key=key), None)
    self.assertEqual(ti.xcom_pull(task_ids=u'test_xcom', key=key, include_prior_dates=True), value)
