ExternalTaskSensor fails with ValueError on state comparison

[BasPH]

Apache Airflow version:
2.0.0b3 (Docker)

Kubernetes version (if you are using kubernetes) (use kubectl version):
N/A

Environment:
Airflow 2.0 beta3 with Docker Compose

• Cloud provider or hardware configuration:
• OS (e.g. from /etc/os-release):
• Kernel (e.g. uname -a):
• Install tools:
• Others:

What happened:
This task fails the DAG import:

ExternalTaskSensor(
        task_id="wait_for_etl_dag3",
        external_dag_id="figure_6_19_dag_3",
        external_task_id="etl",
        dag=dag4,
        allowed_states=State.task_states,
    ),

Error:

Broken DAG: [/opt/airflow/dags/figure_6_19.py] Traceback (most recent call last):
  File "/home/airflow/.local/lib/python3.8/site-packages/airflow/utils/decorators.py", line 94, in wrapper
    result = func(*args, **kwargs)
  File "/home/airflow/.local/lib/python3.8/site-packages/airflow/sensors/external_task_sensor.py", line 118, in __init__
    raise ValueError(
ValueError: Valid values for `allowed_states` and `failed_states` when `external_task_id` is not `None`: ('success', 'running', 'failed', 'upstream_failed', 'skipped', 'up_for_retry', 'up_for_reschedule', 'queued', None, 'scheduled', 'sensing')

It appears the check on line 118 holds the following value:

{('success', 'running', 'failed', 'upstream_failed', 'skipped', 'up_for_retry', 'up_for_reschedule', 'queued', None, 'scheduled', 'sensing')}

And State.task_states holds this value:

('success', 'running', 'failed', 'upstream_failed', 'skipped', 'up_for_retry', 'up_for_reschedule', 'queued', None, 'scheduled', 'sensing')

Which fails this check:

if not total_states <= set(State.task_states):

What you expected to happen:
No failure. Probably some type conversion issue, reporting the issue now and diving into the code.

How to reproduce it:

Anything else we need to know:

[vikramkoka]

@jhtimmins I think you are working on this, correct?

[jhtimmins]

@vikramkoka I'm assigned but have not yet started. Will be starting tomorrow.

[ashb]

@BasPH I think your example has a copy-paste error -- I don't get the error you showed, unless I pass allowed_states=[State.task_states]

[ashb]

A different error we found investigating this was fixed in #12794 - but the original error seems valid - you appear to have passed a list containing a tuple, rather than the tuple directly.

[ashb]

Closing this as we think we've fixed one problem (and the original error was valid -- it was invalid.) Please re-open if there's more info/you think we're wrong.