Skip to content

Instantly share code, notes, and snippets.

@chandulal

chandulal/xcoms_push_pull_test.py

Created September 8, 2019 06:35
Show Gist options
  • Star 0 You must be signed in to star a gist
  • Fork 0 You must be signed in to fork a gist
  • Save chandulal/7a2fa8e38a3bd9b39c1ba6e52c101c05 to your computer and use it in GitHub Desktop.
Save chandulal/7a2fa8e38a3bd9b39c1ba6e52c101c05 to your computer and use it in GitHub Desktop.
Download ZIP
Raw
xcoms_push_pull_test.py
def test_xcoms(self):
dag_id = 'hello_world_xcoms'
dag = self.dagbag.get_dag(dag_id)
push_to_xcoms_task = dag.get_task('push_to_xcoms')
pull_from_xcoms_task = dag.get_task('pull_from_xcoms')
execution_date = datetime.now()
push_to_xcoms_ti = TaskInstance(task=push_to_xcoms_task, execution_date=execution_date)
context = push_to_xcoms_ti.get_template_context()
push_to_xcoms_task.execute(context)
pull_from_xcoms_ti = TaskInstance(task=pull_from_xcoms_task, execution_date=execution_date)
result = pull_from_xcoms_ti.xcom_pull(key="dummyKey")
self.assertEqual(result, 'dummyValue')
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment