celery.schedule.crontab: fix reduce (#3826) (#3827)

celery.schedule.crontab wasn't storing its kwargs with __reduce__ and therefore
wasn't unpickling correctly.  Noteably, nowfun wasn't being wrapped into it.

celery/schedules.py

@@ -101,6 +101,11 @@ class BaseSchedule(object):
             return timezone.to_local_fallback(dt)
         return dt
 
+    def __eq__(self, other):
+        if isinstance(other, BaseSchedule):
+            return other.nowfun == self.nowfun
+        return NotImplemented
+
 
 @python_2_unicode_compatible
 class schedule(BaseSchedule):
@@ -398,6 +403,7 @@ class crontab(BaseSchedule):
         self._orig_day_of_week = cronfield(day_of_week)
         self._orig_day_of_month = cronfield(day_of_month)
         self._orig_month_of_year = cronfield(month_of_year)
+        self._orig_kwargs = kwargs
         self.hour = self._expand_cronspec(hour, 24)
         self.minute = self._expand_cronspec(minute, 60)
         self.day_of_week = self._expand_cronspec(day_of_week, 7)
@@ -529,7 +535,12 @@ class crontab(BaseSchedule):
                                  self._orig_hour,
                                  self._orig_day_of_week,
                                  self._orig_day_of_month,
-                                 self._orig_month_of_year), None)
+                                 self._orig_month_of_year), self._orig_kwargs)
+
+    def __setstate__(self, state):
+        # Calling super's init because the kwargs aren't necessarily passed in
+        # the same form as they are stored by the superclass
+        super(crontab, self).__init__(**state)
 
     def remaining_delta(self, last_run_at, tz=None, ffwd=ffwd):
         # pylint: disable=redefined-outer-name
@@ -624,7 +635,8 @@ class crontab(BaseSchedule):
                 other.day_of_month == self.day_of_month and
                 other.day_of_week == self.day_of_week and
                 other.hour == self.hour and
-                other.minute == self.minute
+                other.minute == self.minute and
+                super(crontab, self).__eq__(other)
             )
         return NotImplemented
 

t/unit/app/test_schedules.py

@@ -91,13 +91,28 @@ class test_schedule:
         assert s1 == s2
 
 
+# This is needed for test_crontab_parser because datetime.utcnow doesn't pickle
+# in python 2
+def utcnow():
+    return datetime.utcnow()
+
+
 class test_crontab_parser:
 
     def crontab(self, *args, **kwargs):
         return crontab(*args, **dict(kwargs, app=self.app))
 
     def test_crontab_reduce(self):
-        assert loads(dumps(self.crontab('*')))
+        c = self.crontab('*')
+        assert c == loads(dumps(c))
+        c = self.crontab(
+            minute='1',
+            hour='2',
+            day_of_week='3',
+            day_of_month='4',
+            month_of_year='5',
+            nowfun=utcnow)
+        assert c == loads(dumps(c))
 
     def test_range_steps_not_enough(self):
         with pytest.raises(crontab_parser.ParseException):