Convert TimeSensorAsync target_time to utc on call time

airflow/sensors/time_sensor.py

@@ -51,10 +51,12 @@ def __init__(self, *, target_time, **kwargs):
         super().__init__(**kwargs)
         self.target_time = target_time
 
-        self.target_datetime = timezone.coerce_datetime(
+        aware_time = timezone.coerce_datetime(
             datetime.datetime.combine(datetime.datetime.today(), self.target_time)
         )
 
+        self.target_datetime = timezone.convert_to_utc(aware_time)
+
     def execute(self, context: Context):
         self.defer(
             trigger=DateTimeTrigger(moment=self.target_datetime),

tests/sensors/test_time_sensor.py

@@ -68,3 +68,10 @@ def test_task_is_deferred(self):
         assert exc_info.value.trigger.moment == timezone.datetime(2020, 7, 7, 10)
         assert exc_info.value.method_name == "execute_complete"
         assert exc_info.value.kwargs is None
+
+    def test_target_time_aware(self):
+        with DAG("test_target_time_aware", start_date=timezone.datetime(2020, 1, 1, 23, 0)):
+            aware_time = time(0, 1).replace(tzinfo=pendulum.local_timezone())
+            op = TimeSensorAsync(task_id="test", target_time=aware_time)
+            assert hasattr(op.target_datetime.tzinfo, "offset")
+            assert op.target_datetime.tzinfo.offset == 0