Replace start timezone on remaining() estimation if timezone changed (#1604) (#4403)

When computing the remaining time after DST change, last_run_at time
was localized in the previous time zone (before DST)
The resulting end date was then also localized in the previous time zone
In case UTC offets are different, Replace tzinfo with
the current one

celery/utils/time.py

@@ -209,6 +209,9 @@ def remaining(start, ends_in, now=None, relative=False):
         ~datetime.timedelta: Remaining time.
     """
     now = now or datetime.utcnow()
+    if now.utcoffset() != start.utcoffset():
+        # Timezone has changed, or DST started/ended
+        start = start.replace(tzinfo=now.tzinfo)
     end_date = start + ends_in
     if relative:
         end_date = delta_resolution(end_date, ends_in)

t/unit/app/test_schedules.py

@@ -1,6 +1,7 @@
 from __future__ import absolute_import, unicode_literals
 
 import time
+import pytz
 from contextlib import contextmanager
 from datetime import datetime, timedelta
 from pickle import dumps, loads
@@ -439,6 +440,40 @@ class test_crontab_remaining_estimate:
         )
         assert next == datetime(2016, 2, 29, 14, 30)
 
+    def test_day_after_dst_end(self):
+        # Test for #1604 issue with region configuration using DST
+        tzname = "Europe/Paris"
+        self.app.timezone = tzname
+        tz = pytz.timezone(tzname)
+        crontab = self.crontab(minute=0, hour=9)
+
+        # Set last_run_at Before DST end
+        last_run_at = tz.localize(datetime(2017, 10, 28, 9, 0))
+        # Set now after DST end
+        now = tz.localize(datetime(2017, 10, 29, 7, 0))
+        crontab.nowfun = lambda: now
+        next = now + crontab.remaining_estimate(last_run_at)
+
+        assert next.utcoffset().seconds == 3600
+        assert next == tz.localize(datetime(2017, 10, 29, 9, 0))
+
+    def test_day_after_dst_start(self):
+        # Test for #1604 issue with region configuration using DST
+        tzname = "Europe/Paris"
+        self.app.timezone = tzname
+        tz = pytz.timezone(tzname)
+        crontab = self.crontab(minute=0, hour=9)
+
+        # Set last_run_at Before DST start
+        last_run_at = tz.localize(datetime(2017, 3, 25, 9, 0))
+        # Set now after DST start
+        now = tz.localize(datetime(2017, 3, 26, 7, 0))
+        crontab.nowfun = lambda: now
+        next = now + crontab.remaining_estimate(last_run_at)
+
+        assert next.utcoffset().seconds == 7200
+        assert next == tz.localize(datetime(2017, 3, 26, 9, 0))
+
 
 class test_crontab_is_due: