DateTimeSensorAsync breaks if target_time is timezone-aware

[Gollum999]

Apache Airflow version

2.5.1

What happened

DateTimeSensorAsync fails with the following error if target_time is aware:

[2022-06-29, 05:09:11 CDT] {taskinstance.py:1889} ERROR - Task failed with exception
Traceback (most recent call last):a
  File "/opt/conda/envs/production/lib/python3.9/site-packages/airflow/sensors/time_sensor.py", line 60, in execute
    trigger=DateTimeTrigger(moment=self.target_datetime),
  File "/opt/conda/envs/production/lib/python3.9/site-packages/airflow/triggers/temporal.py", line 42, in __init__
    raise ValueError(f"The passed datetime must be using Pendulum's UTC, not {moment.tzinfo!r}")
ValueError: The passed datetime must be using Pendulum's UTC, not Timezone('America/Chicago')

What you think should happen instead

Given the fact that DateTimeSensor correctly handles timezones, this seems like a bug. DateTimeSensorAsync should be a drop-in replacement for DateTimeSensor, and therefore should have the same timezone behavior.

How to reproduce

#!/usr/bin/env python3
import datetime

from airflow.decorators import dag
from airflow.sensors.date_time import DateTimeSensor, DateTimeSensorAsync
import pendulum


@dag(
    start_date=datetime.datetime(2022, 6, 29),
    schedule='@daily',
)
def datetime_sensor_dag():
    naive_time1 = datetime.datetime(2023, 2, 16,  0,  1)
    aware_time1 = datetime.datetime(2023, 2, 16,  0,  1).replace(tzinfo=pendulum.local_timezone())
    naive_time2 = pendulum.datetime(2023, 2, 16, 23, 59)
    aware_time2 = pendulum.datetime(2023, 2, 16, 23, 59).replace(tzinfo=pendulum.local_timezone())

    DateTimeSensor(task_id='naive_time1', target_time=naive_time1, mode='reschedule')
    DateTimeSensor(task_id='naive_time2', target_time=naive_time2, mode='reschedule')
    DateTimeSensor(task_id='aware_time1', target_time=aware_time1, mode='reschedule')
    DateTimeSensor(task_id='aware_time2', target_time=aware_time2, mode='reschedule')

    DateTimeSensorAsync(task_id='async_naive_time1', target_time=naive_time1)
    DateTimeSensorAsync(task_id='async_naive_time2', target_time=naive_time2)
    DateTimeSensorAsync(task_id='async_aware_time1', target_time=aware_time1)  # fails
    DateTimeSensorAsync(task_id='async_aware_time2', target_time=aware_time2)  # fails

datetime_sensor_dag()

This can also happen if the target_time is naive and core.default_timezone = system.

Operating System

CentOS Stream 8

Versions of Apache Airflow Providers

N/A

Deployment

Other

Deployment details

Standalone

Anything else

This appears to be nearly identical to #24736. Probably worth checking other time-related sensors as well.

Are you willing to submit PR?

• Yes I am willing to submit a PR!

Code of Conduct

• I agree to follow this project's Code of Conduct