day of week fix

python_modules/dagster/dagster/_seven/compat/pendulum.py

@@ -1,5 +1,6 @@
 import datetime
 from contextlib import contextmanager
+from unittest import mock
 
 import packaging.version
 import pendulum
@@ -98,7 +99,8 @@ def create_pendulum_time(year, month, day, *args, **kwargs):
 @contextmanager
 def pendulum_freeze_time(t):
     if _IS_PENDULUM_3:
-        yield from pendulum.travel_to(t, freeze=True)
+        with mock.patch("pendulum.now", return_value=t):
+            yield
     else:
         with pendulum.test(t) as frozen_time:
             yield frozen_time
@@ -112,3 +114,11 @@ def to_timezone(dt: PendulumDateTime, tz: str):
     if timestamp < 0:
         return pendulum.from_timestamp(0, tz=tz) + datetime.timedelta(seconds=timestamp)
     return pendulum.from_timestamp(dt.timestamp(), tz=tz)
+
+
+def get_crontab_day_of_week(dt: PendulumDateTime) -> int:
+    if _IS_PENDULUM_3:
+        # In pendulum 3, 0-6 is Monday-Sunday (unlike crontab, where 0-6 is Sunday-Saturday)
+        return (dt.day_of_week + 1) % 7
+    else:
+        return dt.day_of_week

python_modules/dagster/dagster/_utils/schedules.py

@@ -15,6 +15,7 @@
     TRANSITION_ERROR,
     PendulumDateTime,
     create_pendulum_time,
+    get_crontab_day_of_week,
 )
 
 # Monthly schedules with 29-31 won't reliably run every month
@@ -292,9 +293,9 @@ def _find_weekly_schedule_time(
             minute,
             pendulum_date.day,
         )
-
         # Move to the correct day of the week
-        current_day_of_week = new_time.day_of_week
+        current_day_of_week = get_crontab_day_of_week(new_time)
+
         if day_of_week != current_day_of_week:
             if ascending:
                 new_time = new_time.add(days=(day_of_week - current_day_of_week) % 7)