@@ -0,0 +1,2405 @@
+from airflow import DAG%0Afrom airflow.operators.bash_operator import BashOperator%0Afrom airflow.operators.dummy_operator import DummyOperator%0Afrom airflow.operators.sensors import ExternalTaskSensor%0Afrom tests.operators.sensors import TEST_DAG_ID, DEFAULT_DATE%0Afrom datetime import datetime, timedelta%0A%0Aargs = %7B'owner': 'airflow', 'start_date': DEFAULT_DATE%7D%0A%0Abash_command_code = %22%22%22%0A%7B%25 set s=execution_date.time().second %25%7D%0Aecho %22second is %7B%7B s %7D%7D%22%0Aif %5B%5B $(( %7B%7B s %7D%7D %25 60 )) == 1 %5D%5D%0A    then%0A        exit 1%0Afi%0Aexit 0%0A%22%22%22%0A# That DAG is use to test the behavior of the ExternalTaskSensor%0A# when depending on several runs of an external task.%0A# test_without_failure should not fail, leading to%0A# test_external_task_sensor_multiple_dates_with_failure%0A# to succeed, whereas test_with_failure should fail once%0A# per minute (the DAG runs every second) leading to%0A# test_external_task_sensor_multiple_dates_with_failure%0A# to fail (because of timeout).%0Adag_external_id = TEST_DAG_ID + '_secondly_external'%0Adag_secondly_external = DAG(dag_external_id,%0A    default_args=args,%0A    schedule_interval=timedelta(seconds=1))%0Adag_secondly_external.add_task(BashOperator(%0A    task_id=%22test_with_failure%22,%0A    bash_command=bash_command_code,%0A    retries=0,%0A    depends_on_past=False,%0A    start_date=DEFAULT_DATE))%0Adag_secondly_external.add_task(DummyOperator(%0A    task_id=%22test_without_failure%22,%0A    retries=0,%0A    depends_on_past=False,%0A    start_date=DEFAULT_DATE))%0A%0Adag_id = TEST_DAG_ID + '_minutely'%0Adag_minutely = DAG(dag_id,%0A    default_args=args,%0A    schedule_interval=timedelta(minutes=1))%0Adag_minutely.add_task(ExternalTaskSensor(%0A    task_id='test_external_task_sensor_multiple_dates_without_failure',%0A    external_dag_id=dag_external_id,%0A    external_task_id='test_without_failure',%0A    execution_date_fn=lambda dt: %5Bdt + timedelta(seconds=i) for i in range(2)%5D,%0A    allowed_states=%5B'success'%5D,%0A    retries=0,%0A    timeout=1,%0A    poke_interval=1,%0A    depends_on_past=False,%0A    start_date=DEFAULT_DATE))%0Adag_minutely.add_task(ExternalTaskSensor(%0A    task_id='test_external_task_sensor_multiple_dates_with_failure',%0A    external_dag_id=dag_external_id,%0A    external_task_id='test_with_failure',%0A    execution_date_fn=lambda dt: %5Bdt + timedelta(seconds=i) for i in range(2)%5D,%0A    allowed_states=%5B'success'%5D,%0A    retries=0,%0A    depends_on_past=False,%0A    timeout=1,%0A    poke_interval=1,%0A    start_date=DEFAULT_DATE))%0A%0A

Add ability for ExternalTaskSensor to wait on multiple runs of a task