[dagster-airflow] pass airflow dag timezone to dagster schedule defin…

…ition (#11663)

python_modules/libraries/dagster-airflow/dagster_airflow/dagster_pipeline_factory.py

@@ -287,7 +287,10 @@ def make_dagster_schedule_from_airflow_dag(dag, job_def):
 
     if isinstance(dag.normalized_schedule_interval, str) and is_valid_cron_schedule(cron_schedule):
         return ScheduleDefinition(
-            job=job_def, cron_schedule=cron_schedule, description=schedule_description
+            job=job_def,
+            cron_schedule=cron_schedule,
+            description=schedule_description,
+            execution_timezone=dag.timezone.name,
         )
 
 

python_modules/libraries/dagster-airflow/dagster_airflow_tests/test_dagster_job_factory/test_schedule_timezone.py

@@ -0,0 +1,27 @@
+import pendulum
+import pytest
+from airflow import __version__ as airflow_version
+from airflow.models.dag import DAG
+from dagster import job
+from dagster_airflow.dagster_pipeline_factory import make_dagster_schedule_from_airflow_dag
+
+
+@pytest.mark.skipif(airflow_version < "2.0.0", reason="requires airflow 2")
+def test_schedule_timezone():
+    args = {
+        "owner": "airflow",
+        "start_date": pendulum.today("Europe/London").add(days=-2),
+    }
+    dag = DAG(
+        dag_id="test_schedules",
+        default_args=args,
+        schedule="0 0 * * *",
+    )
+
+    @job
+    def job_def():
+        return
+
+    schedule = make_dagster_schedule_from_airflow_dag(dag=dag, job_def=job_def)
+    assert schedule.cron_schedule == "0 0 * * *"
+    assert schedule.execution_timezone == "Europe/London"