text: import time
  from contextlib import contextmanager
  from datetime import datetime, timedelta
  from pickle import dumps, loads
  from unittest.mock import Mock
  
  import pytest
  import pytz
  from case import Case
  
  from celery.schedules import (ParseException, crontab, crontab_parser,
                                schedule, solar)
  
  assertions = Case('__init__')
  
  
  @contextmanager
  def patch_crontab_nowfun(cls, retval):
      prev_nowfun = cls.nowfun
      cls.nowfun = lambda: retval
      try:
          yield
      finally:
          cls.nowfun = prev_nowfun
  
  
  class test_solar:
  
      def setup(self):
          pytest.importorskip('ephem0')
          self.s = solar('sunrise', 60, 30, app=self.app)
  
      def test_reduce(self):
          fun, args = self.s.__reduce__()
          assert fun(*args) == self.s
  
      def test_eq(self):
          assert self.s == solar('sunrise', 60, 30, app=self.app)
          assert self.s != solar('sunset', 60, 30, app=self.app)
          assert self.s != schedule(10)
  
      def test_repr(self):
          assert repr(self.s)
  
      def test_is_due(self):
          self.s.remaining_estimate = Mock(name='rem')
          self.s.remaining_estimate.return_value = timedelta(seconds=0)
          assert self.s.is_due(datetime.utcnow()).is_due
  
      def test_is_due__not_due(self):
          self.s.remaining_estimate = Mock(name='rem')
          self.s.remaining_estimate.return_value = timedelta(hours=10)
          assert not self.s.is_due(datetime.utcnow()).is_due
  
      def test_remaining_estimate(self):
          self.s.cal = Mock(name='cal')
          self.s.cal.next_rising().datetime.return_value = datetime.utcnow()
          self.s.remaining_estimate(datetime.utcnow())
  
      def test_coordinates(self):
          with pytest.raises(ValueError):
              solar('sunrise', -120, 60, app=self.app)
          with pytest.raises(ValueError):
              solar('sunrise', 120, 60, app=self.app)
          with pytest.raises(ValueError):
              solar('sunrise', 60, -200, app=self.app)
          with pytest.raises(ValueError):
              solar('sunrise', 60, 200, app=self.app)
  
      def test_invalid_event(self):
          with pytest.raises(ValueError):
              solar('asdqwewqew', 60, 60, app=self.app)
  
      def test_event_uses_center(self):
          s = solar('solar_noon', 60, 60, app=self.app)
          for ev, is_center in s._use_center_l.items():
              s.method = s._methods[ev]
              s.is_center = s._use_center_l[ev]
              try:
                  s.remaining_estimate(datetime.utcnow())
              except TypeError:
                  pytest.fail(f"{s.method} was called with 'use_center' which is not a "
                              "valid keyword for the function.")
  
  
  class test_schedule:
  
      def test_ne(self):
          s1 = schedule(10, app=self.app)
          s2 = schedule(12, app=self.app)
          s3 = schedule(10, app=self.app)
          assert s1 == s3
          assert s1 != s2
  
      def test_pickle(self):
          s1 = schedule(10, app=self.app)
          fun, args = s1.__reduce__()
          s2 = fun(*args)
          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):
          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):
              crontab_parser(24)._range_steps([1])
  
      def test_parse_star(self):
          assert crontab_parser(24).parse('*') == set(range(24))
          assert crontab_parser(60).parse('*') == set(range(60))
          assert crontab_parser(7).parse('*') == set(range(7))
          assert crontab_parser(31, 1).parse('*') == set(range(1, 31 + 1))
          assert crontab_parser(12, 1).parse('*') == set(range(1, 12 + 1))
  
      def test_parse_range(self):
          assert crontab_parser(60).parse('1-10') == set(range(1, 10 + 1))
          assert crontab_parser(24).parse('0-20') == set(range(0, 20 + 1))
          assert crontab_parser().parse('2-10') == set(range(2, 10 + 1))
          assert crontab_parser(60, 1).parse('1-10') == set(range(1, 10 + 1))
  
      def test_parse_range_wraps(self):
          assert crontab_parser(12).parse('11-1') == {11, 0, 1}
          assert crontab_parser(60, 1).parse('2-1') == set(range(1, 60 + 1))
  
      def test_parse_groups(self):
          assert crontab_parser().parse('1,2,3,4') == {1, 2, 3, 4}
          assert crontab_parser().parse('0,15,30,45') == {0, 15, 30, 45}
          assert crontab_parser(min_=1).parse('1,2,3,4') == {1, 2, 3, 4}
  
      def test_parse_steps(self):
          assert crontab_parser(8).parse('*/2') == {0, 2, 4, 6}
          assert crontab_parser().parse('*/2') == {i * 2 for i in range(30)}
          assert crontab_parser().parse('*/3') == {i * 3 for i in range(20)}
          assert crontab_parser(8, 1).parse('*/2') == {1, 3, 5, 7}
          assert crontab_parser(min_=1).parse('*/2') == {
              i * 2 + 1 for i in range(30)
          }
          assert crontab_parser(min_=1).parse('*/3') == {
              i * 3 + 1 for i in range(20)
          }
  
      def test_parse_composite(self):
          assert crontab_parser(8).parse('*/2') == {0, 2, 4, 6}
          assert crontab_parser().parse('2-9/5') == {2, 7}
          assert crontab_parser().parse('2-10/5') == {2, 7}
          assert crontab_parser(min_=1).parse('55-5/3') == {55, 58, 1, 4}
          assert crontab_parser().parse('2-11/5,3') == {2, 3, 7}
          assert crontab_parser().parse('2-4/3,*/5,0-21/4') == {
              0, 2, 4, 5, 8, 10, 12, 15, 16, 20, 25, 30, 35, 40, 45, 50, 55,
          }
          assert crontab_parser().parse('1-9/2') == {1, 3, 5, 7, 9}
          assert crontab_parser(8, 1).parse('*/2') == {1, 3, 5, 7}
          assert crontab_parser(min_=1).parse('2-9/5') == {2, 7}
          assert crontab_parser(min_=1).parse('2-10/5') == {2, 7}
          assert crontab_parser(min_=1).parse('2-11/5,3') == {2, 3, 7}
          assert crontab_parser(min_=1).parse('2-4/3,*/5,1-21/4') == {
              1, 2, 5, 6, 9, 11, 13, 16, 17, 21, 26, 31, 36, 41, 46, 51, 56,
          }
          assert crontab_parser(min_=1).parse('1-9/2') == {1, 3, 5, 7, 9}
  
      def test_parse_errors_on_empty_string(self):
          with pytest.raises(ParseException):
              crontab_parser(60).parse('')
  
      def test_parse_errors_on_empty_group(self):
          with pytest.raises(ParseException):
              crontab_parser(60).parse('1,,2')
  
      def test_parse_errors_on_empty_steps(self):
          with pytest.raises(ParseException):
              crontab_parser(60).parse('*/')
  
      def test_parse_errors_on_negative_number(self):
          with pytest.raises(ParseException):
              crontab_parser(60).parse('-20')
  
      def test_parse_errors_on_lt_min(self):
          crontab_parser(min_=1).parse('1')
          with pytest.raises(ValueError):
              crontab_parser(12, 1).parse('0')
          with pytest.raises(ValueError):
              crontab_parser(24, 1).parse('12-0')
  
      def test_parse_errors_on_gt_max(self):
          crontab_parser(1).parse('0')
          with pytest.raises(ValueError):
              crontab_parser(1).parse('1')
          with pytest.raises(ValueError):
              crontab_parser(60).parse('61-0')
  
      def test_expand_cronspec_eats_iterables(self):
          assert crontab._expand_cronspec(iter([1, 2, 3]), 100) == {1, 2, 3}
          assert crontab._expand_cronspec(iter([1, 2, 3]), 100, 1) == {1, 2, 3}
  
      def test_expand_cronspec_invalid_type(self):
          with pytest.raises(TypeError):
              crontab._expand_cronspec(object(), 100)
  
      def test_repr(self):
          assert '*' in repr(self.crontab('*'))
  
      def test_eq(self):
          assert (self.crontab(day_of_week='1, 2') ==
                  self.crontab(day_of_week='1-2'))
          assert (self.crontab(day_of_month='1, 16, 31') ==
                  self.crontab(day_of_month='*/15'))
          assert (
              self.crontab(
                  minute='1', hour='2', day_of_week='5',
                  day_of_month='10', month_of_year='5') ==
              self.crontab(
                  minute='1', hour='2', day_of_week='5',
                  day_of_month='10', month_of_year='5'))
          assert crontab(minute='1') != crontab(minute='2')
          assert (self.crontab(month_of_year='1') !=
                  self.crontab(month_of_year='2'))
          assert object() != self.crontab(minute='1')
          assert self.crontab(minute='1') != object()
          assert crontab(month_of_year='1') != schedule(10)
  
  
  class test_crontab_remaining_estimate:
  
      def crontab(self, *args, **kwargs):
          return crontab(*args, **dict(kwargs, app=self.app))
  
      def next_ocurrance(self, crontab, now):
          crontab.nowfun = lambda: now
          return now + crontab.remaining_estimate(now)
  
      def test_next_minute(self):
          next = self.next_ocurrance(
              self.crontab(), datetime(2010, 9, 11, 14, 30, 15),
          )
          assert next == datetime(2010, 9, 11, 14, 31)
  
      def test_not_next_minute(self):
          next = self.next_ocurrance(
              self.crontab(), datetime(2010, 9, 11, 14, 59, 15),
          )
          assert next == datetime(2010, 9, 11, 15, 0)
  
      def test_this_hour(self):
          next = self.next_ocurrance(
              self.crontab(minute=[5, 42]), datetime(2010, 9, 11, 14, 30, 15),
          )
          assert next == datetime(2010, 9, 11, 14, 42)
  
      def test_not_this_hour(self):
          next = self.next_ocurrance(
              self.crontab(minute=[5, 10, 15]),
              datetime(2010, 9, 11, 14, 30, 15),
          )
          assert next == datetime(2010, 9, 11, 15, 5)
  
      def test_today(self):
          next = self.next_ocurrance(
              self.crontab(minute=[5, 42], hour=[12, 17]),
              datetime(2010, 9, 11, 14, 30, 15),
          )
          assert next == datetime(2010, 9, 11, 17, 5)
  
      def test_not_today(self):
          next = self.next_ocurrance(
              self.crontab(minute=[5, 42], hour=[12]),
              datetime(2010, 9, 11, 14, 30, 15),
          )
          assert next == datetime(2010, 9, 12, 12, 5)
  
      def test_weekday(self):
          next = self.next_ocurrance(
              self.crontab(minute=30, hour=14, day_of_week='sat'),
              datetime(2010, 9, 11, 14, 30, 15),
          )
          assert next == datetime(2010, 9, 18, 14, 30)
  
      def test_not_weekday(self):
          next = self.next_ocurrance(
              self.crontab(minute=[5, 42], day_of_week='mon-fri'),
              datetime(2010, 9, 11, 14, 30, 15),
          )
          assert next == datetime(2010, 9, 13, 0, 5)
  
      def test_monthday(self):
          next = self.next_ocurrance(
              self.crontab(minute=30, hour=14, day_of_month=18),
              datetime(2010, 9, 11, 14, 30, 15),
          )
          assert next == datetime(2010, 9, 18, 14, 30)
  
      def test_not_monthday(self):
          next = self.next_ocurrance(
              self.crontab(minute=[5, 42], day_of_month=29),
              datetime(2010, 1, 22, 14, 30, 15),
          )
          assert next == datetime(2010, 1, 29, 0, 5)
  
      def test_weekday_monthday(self):
          next = self.next_ocurrance(
              self.crontab(minute=30, hour=14,
                           day_of_week='mon', day_of_month=18),
              datetime(2010, 1, 18, 14, 30, 15),
          )
          assert next == datetime(2010, 10, 18, 14, 30)
  
      def test_monthday_not_weekday(self):
          next = self.next_ocurrance(
              self.crontab(minute=[5, 42], day_of_week='sat', day_of_month=29),
              datetime(2010, 1, 29, 0, 5, 15),
          )
          assert next == datetime(2010, 5, 29, 0, 5)
  
      def test_weekday_not_monthday(self):
          next = self.next_ocurrance(
              self.crontab(minute=[5, 42], day_of_week='mon', day_of_month=18),
              datetime(2010, 1, 11, 0, 5, 15),
          )
          assert next == datetime(2010, 1, 18, 0, 5)
  
      def test_not_weekday_not_monthday(self):
          next = self.next_ocurrance(
              self.crontab(minute=[5, 42], day_of_week='mon', day_of_month=18),
              datetime(2010, 1, 10, 0, 5, 15),
          )
          assert next == datetime(2010, 1, 18, 0, 5)
  
      def test_leapday(self):
          next = self.next_ocurrance(
              self.crontab(minute=30, hour=14, day_of_month=29),
              datetime(2012, 1, 29, 14, 30, 15),
          )
          assert next == datetime(2012, 2, 29, 14, 30)
  
      def test_not_leapday(self):
          next = self.next_ocurrance(
              self.crontab(minute=30, hour=14, day_of_month=29),
              datetime(2010, 1, 29, 14, 30, 15),
          )
          assert next == datetime(2010, 3, 29, 14, 30)
  
      def test_weekmonthdayyear(self):
          next = self.next_ocurrance(
              self.crontab(minute=30, hour=14, day_of_week='fri',
                           day_of_month=29, month_of_year=1),
              datetime(2010, 1, 22, 14, 30, 15),
          )
          assert next == datetime(2010, 1, 29, 14, 30)
  
      def test_monthdayyear_not_week(self):
          next = self.next_ocurrance(
              self.crontab(minute=[5, 42], day_of_week='wed,thu',
                           day_of_month=29, month_of_year='1,4,7'),
              datetime(2010, 1, 29, 14, 30, 15),
          )
          assert next == datetime(2010, 4, 29, 0, 5)
  
      def test_weekdaymonthyear_not_monthday(self):
          next = self.next_ocurrance(
              self.crontab(minute=30, hour=14, day_of_week='fri',
                           day_of_month=29, month_of_year='1-10'),
              datetime(2010, 1, 29, 14, 30, 15),
          )
          assert next == datetime(2010, 10, 29, 14, 30)
  
      def test_weekmonthday_not_monthyear(self):
          next = self.next_ocurrance(
              self.crontab(minute=[5, 42], day_of_week='fri',
                           day_of_month=29, month_of_year='2-10'),
              datetime(2010, 1, 29, 14, 30, 15),
          )
          assert next == datetime(2010, 10, 29, 0, 5)
  
      def test_weekday_not_monthdayyear(self):
          next = self.next_ocurrance(
              self.crontab(minute=[5, 42], day_of_week='mon',
                           day_of_month=18, month_of_year='2-10'),
              datetime(2010, 1, 11, 0, 5, 15),
          )
          assert next == datetime(2010, 10, 18, 0, 5)
  
      def test_monthday_not_weekdaymonthyear(self):
          next = self.next_ocurrance(
              self.crontab(minute=[5, 42], day_of_week='mon',
                           day_of_month=29, month_of_year='2-4'),
              datetime(2010, 1, 29, 0, 5, 15),
          )
          assert next == datetime(2010, 3, 29, 0, 5)
  
      def test_monthyear_not_weekmonthday(self):
          next = self.next_ocurrance(
              self.crontab(minute=[5, 42], day_of_week='mon',
                           day_of_month=29, month_of_year='2-4'),
              datetime(2010, 2, 28, 0, 5, 15),
          )
          assert next == datetime(2010, 3, 29, 0, 5)
  
      def test_not_weekmonthdayyear(self):
          next = self.next_ocurrance(
              self.crontab(minute=[5, 42], day_of_week='fri,sat',
                           day_of_month=29, month_of_year='2-10'),
              datetime(2010, 1, 28, 14, 30, 15),
          )
          assert next == datetime(2010, 5, 29, 0, 5)
  
      def test_invalid_specification(self):
          # *** WARNING ***
          # This test triggers an infinite loop in case of a regression
          with pytest.raises(RuntimeError):
              self.next_ocurrance(
                  self.crontab(day_of_month=31, month_of_year=4),
                  datetime(2010, 1, 28, 14, 30, 15),
              )
  
      def test_leapyear(self):
          next = self.next_ocurrance(
              self.crontab(minute=30, hour=14, day_of_month=29, month_of_year=2),
              datetime(2012, 2, 29, 14, 30),
          )
          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:
  
      def setup(self):
          self.now = self.app.now()
          self.next_minute = 60 - self.now.second - 1e-6 * self.now.microsecond
          self.every_minute = self.crontab()
          self.quarterly = self.crontab(minute='*/15')
          self.hourly = self.crontab(minute=30)
          self.daily = self.crontab(hour=7, minute=30)
          self.weekly = self.crontab(hour=7, minute=30, day_of_week='thursday')
          self.monthly = self.crontab(
              hour=7, minute=30, day_of_week='thursday', day_of_month='8-14',
          )
          self.monthly_moy = self.crontab(
              hour=22, day_of_week='*', month_of_year='2',
              day_of_month='26,27,28',
          )
          self.yearly = self.crontab(
              hour=7, minute=30, day_of_week='thursday',
              day_of_month='8-14', month_of_year=3,
          )
  
      def crontab(self, *args, **kwargs):
          return crontab(*args, app=self.app, **kwargs)
  
      def test_default_crontab_spec(self):
          c = self.crontab()
          assert c.minute == set(range(60))
          assert c.hour == set(range(24))
          assert c.day_of_week == set(range(7))
          assert c.day_of_month == set(range(1, 32))
          assert c.month_of_year == set(range(1, 13))
  
      def test_simple_crontab_spec(self):
          c = self.crontab(minute=30)
          assert c.minute == {30}
          assert c.hour == set(range(24))
          assert c.day_of_week == set(range(7))
          assert c.day_of_month == set(range(1, 32))
          assert c.month_of_year == set(range(1, 13))
  
      @pytest.mark.parametrize('minute,expected', [
          (30, {30}),
          ('30', {30}),
          ((30, 40, 50), {30, 40, 50}),
          ((30, 40, 50, 51), {30, 40, 50, 51})
      ])
      def test_crontab_spec_minute_formats(self, minute, expected):
          c = self.crontab(minute=minute)
          assert c.minute == expected
  
      @pytest.mark.parametrize('minute', [60, '0-100'])
      def test_crontab_spec_invalid_minute(self, minute):
          with pytest.raises(ValueError):
              self.crontab(minute=minute)
  
      @pytest.mark.parametrize('hour,expected', [
          (6, {6}),
          ('5', {5}),
          ((4, 8, 12), {4, 8, 12}),
      ])
      def test_crontab_spec_hour_formats(self, hour, expected):
          c = self.crontab(hour=hour)
          assert c.hour == expected
  
      @pytest.mark.parametrize('hour', [24, '0-30'])
      def test_crontab_spec_invalid_hour(self, hour):
          with pytest.raises(ValueError):
              self.crontab(hour=hour)
  
      @pytest.mark.parametrize('day_of_week,expected', [
          (5, {5}),
          ('5', {5}),
          ('fri', {5}),
          ('tuesday,sunday,fri', {0, 2, 5}),
          ('mon-fri', {1, 2, 3, 4, 5}),
          ('*/2', {0, 2, 4, 6}),
      ])
      def test_crontab_spec_dow_formats(self, day_of_week, expected):
          c = self.crontab(day_of_week=day_of_week)
          assert c.day_of_week == expected
  
      @pytest.mark.parametrize('day_of_week', [
          'fooday-barday', '1,4,foo', '7', '12',
      ])
      def test_crontab_spec_invalid_dow(self, day_of_week):
          with pytest.raises(ValueError):
              self.crontab(day_of_week=day_of_week)
  
      @pytest.mark.parametrize('day_of_month,expected', [
          (5, {5}),
          ('5', {5}),
          ('2,4,6', {2, 4, 6}),
          ('*/5', {1, 6, 11, 16, 21, 26, 31}),
      ])
      def test_crontab_spec_dom_formats(self, day_of_month, expected):
          c = self.crontab(day_of_month=day_of_month)
          assert c.day_of_month == expected
  
      @pytest.mark.parametrize('day_of_month', [0, '0-10', 32, '31,32'])
      def test_crontab_spec_invalid_dom(self, day_of_month):
          with pytest.raises(ValueError):
              self.crontab(day_of_month=day_of_month)
  
      @pytest.mark.parametrize('month_of_year,expected', [
          (1, {1}),
          ('1', {1}),
          ('2,4,6', {2, 4, 6}),
          ('*/2', {1, 3, 5, 7, 9, 11}),
          ('2-12/2', {2, 4, 6, 8, 10, 12}),
      ])
      def test_crontab_spec_moy_formats(self, month_of_year, expected):
          c = self.crontab(month_of_year=month_of_year)
          assert c.month_of_year == expected
  
      @pytest.mark.parametrize('month_of_year', [0, '0-5', 13, '12,13'])
      def test_crontab_spec_invalid_moy(self, month_of_year):
          with pytest.raises(ValueError):
              self.crontab(month_of_year=month_of_year)
  
      def seconds_almost_equal(self, a, b, precision):
          for index, skew in enumerate((+1, -1, 0)):
              try:
                  assertions.assertAlmostEqual(a, b + skew, precision)
              except Exception as exc:
                  # AssertionError != builtins.AssertionError in pytest
                  if 'AssertionError' in str(exc):
                      if index + 1 >= 3:
                          raise
              else:
                  break
  
      def test_every_minute_execution_is_due(self):
          last_ran = self.now - timedelta(seconds=61)
          due, remaining = self.every_minute.is_due(last_ran)
          self.assert_relativedelta(self.every_minute, last_ran)
          assert due
          self.seconds_almost_equal(remaining, self.next_minute, 1)
  
      def assert_relativedelta(self, due, last_ran):
          try:
              from dateutil.relativedelta import relativedelta
          except ImportError:
              return
          l1, d1, n1 = due.remaining_delta(last_ran)
          l2, d2, n2 = due.remaining_delta(last_ran, ffwd=relativedelta)
          if not isinstance(d1, relativedelta):
              assert l1 == l2
              for field, value in d1._fields().items():
                  assert getattr(d1, field) == value
              assert not d2.years
              assert not d2.months
              assert not d2.days
              assert not d2.leapdays
              assert not d2.hours
              assert not d2.minutes
              assert not d2.seconds
              assert not d2.microseconds
  
      def test_every_minute_execution_is_not_due(self):
          last_ran = self.now - timedelta(seconds=self.now.second)
          due, remaining = self.every_minute.is_due(last_ran)
          assert not due
          self.seconds_almost_equal(remaining, self.next_minute, 1)
  
      def test_execution_is_due_on_saturday(self):
          # 29th of May 2010 is a saturday
          with patch_crontab_nowfun(self.hourly, datetime(2010, 5, 29, 10, 30)):
              last_ran = self.now - timedelta(seconds=61)
              due, remaining = self.every_minute.is_due(last_ran)
              assert due
              self.seconds_almost_equal(remaining, self.next_minute, 1)
  
      def test_execution_is_due_on_sunday(self):
          # 30th of May 2010 is a sunday
          with patch_crontab_nowfun(self.hourly, datetime(2010, 5, 30, 10, 30)):
              last_ran = self.now - timedelta(seconds=61)
              due, remaining = self.every_minute.is_due(last_ran)
              assert due
              self.seconds_almost_equal(remaining, self.next_minute, 1)
  
      def test_execution_is_due_on_monday(self):
          # 31st of May 2010 is a monday
          with patch_crontab_nowfun(self.hourly, datetime(2010, 5, 31, 10, 30)):
              last_ran = self.now - timedelta(seconds=61)
              due, remaining = self.every_minute.is_due(last_ran)
              assert due
              self.seconds_almost_equal(remaining, self.next_minute, 1)
  
      def test_every_hour_execution_is_due(self):
          with patch_crontab_nowfun(self.hourly, datetime(2010, 5, 10, 10, 30)):
              due, remaining = self.hourly.is_due(datetime(2010, 5, 10, 6, 30))
              assert due
              assert remaining == 60 * 60
  
      def test_every_hour_execution_is_not_due(self):
          with patch_crontab_nowfun(self.hourly, datetime(2010, 5, 10, 10, 29)):
              due, remaining = self.hourly.is_due(datetime(2010, 5, 10, 9, 30))
              assert not due
              assert remaining == 60
  
      def test_first_quarter_execution_is_due(self):
          with patch_crontab_nowfun(
                  self.quarterly, datetime(2010, 5, 10, 10, 15)):
              due, remaining = self.quarterly.is_due(
                  datetime(2010, 5, 10, 6, 30),
              )
              assert due
              assert remaining == 15 * 60
  
      def test_second_quarter_execution_is_due(self):
          with patch_crontab_nowfun(
                  self.quarterly, datetime(2010, 5, 10, 10, 30)):
              due, remaining = self.quarterly.is_due(
                  datetime(2010, 5, 10, 6, 30),
              )
              assert due
              assert remaining == 15 * 60
  
      def test_first_quarter_execution_is_not_due(self):
          with patch_crontab_nowfun(
                  self.quarterly, datetime(2010, 5, 10, 10, 14)):
              due, remaining = self.quarterly.is_due(
                  datetime(2010, 5, 10, 10, 0),
              )
              assert not due
              assert remaining == 60
  
      def test_second_quarter_execution_is_not_due(self):
          with patch_crontab_nowfun(
                  self.quarterly, datetime(2010, 5, 10, 10, 29)):
              due, remaining = self.quarterly.is_due(
                  datetime(2010, 5, 10, 10, 15),
              )
              assert not due
              assert remaining == 60
  
      def test_daily_execution_is_due(self):
          with patch_crontab_nowfun(self.daily, datetime(2010, 5, 10, 7, 30)):
              due, remaining = self.daily.is_due(datetime(2010, 5, 9, 7, 30))
              assert due
              assert remaining == 24 * 60 * 60
  
      def test_daily_execution_is_not_due(self):
          with patch_crontab_nowfun(self.daily, datetime(2010, 5, 10, 10, 30)):
              due, remaining = self.daily.is_due(datetime(2010, 5, 10, 7, 30))
              assert not due
              assert remaining == 21 * 60 * 60
  
      def test_weekly_execution_is_due(self):
          with patch_crontab_nowfun(self.weekly, datetime(2010, 5, 6, 7, 30)):
              due, remaining = self.weekly.is_due(datetime(2010, 4, 30, 7, 30))
              assert due
              assert remaining == 7 * 24 * 60 * 60
  
      def test_weekly_execution_is_not_due(self):
          with patch_crontab_nowfun(self.weekly, datetime(2010, 5, 7, 10, 30)):
              due, remaining = self.weekly.is_due(datetime(2010, 5, 6, 7, 30))
              assert not due
              assert remaining == 6 * 24 * 60 * 60 - 3 * 60 * 60
  
      def test_monthly_execution_is_due(self):
          with patch_crontab_nowfun(self.monthly, datetime(2010, 5, 13, 7, 30)):
              due, remaining = self.monthly.is_due(datetime(2010, 4, 8, 7, 30))
              assert due
              assert remaining == 28 * 24 * 60 * 60
  
      def test_monthly_execution_is_not_due(self):
          with patch_crontab_nowfun(self.monthly, datetime(2010, 5, 9, 10, 30)):
              due, remaining = self.monthly.is_due(datetime(2010, 4, 8, 7, 30))
              assert not due
              assert remaining == 4 * 24 * 60 * 60 - 3 * 60 * 60
  
      def test_monthly_moy_execution_is_due(self):
          with patch_crontab_nowfun(
                  self.monthly_moy, datetime(2014, 2, 26, 22, 0)):
              due, remaining = self.monthly_moy.is_due(
                  datetime(2013, 7, 4, 10, 0),
              )
              assert due
              assert remaining == 60.0
  
      @pytest.mark.skip('TODO: unstable test')
      def test_monthly_moy_execution_is_not_due(self):
          with patch_crontab_nowfun(
                  self.monthly_moy, datetime(2013, 6, 28, 14, 30)):
              due, remaining = self.monthly_moy.is_due(
                  datetime(2013, 6, 28, 22, 14),
              )
              assert not due
              attempt = (
                  time.mktime(datetime(2014, 2, 26, 22, 0).timetuple()) -
                  time.mktime(datetime(2013, 6, 28, 14, 30).timetuple()) -
                  60 * 60
              )
              assert remaining == attempt
  
      def test_monthly_moy_execution_is_due2(self):
          with patch_crontab_nowfun(
                  self.monthly_moy, datetime(2014, 2, 26, 22, 0)):
              due, remaining = self.monthly_moy.is_due(
                  datetime(2013, 2, 28, 10, 0),
              )
              assert due
              assert remaining == 60.0
  
      def test_monthly_moy_execution_is_not_due2(self):
          with patch_crontab_nowfun(
                  self.monthly_moy, datetime(2014, 2, 26, 21, 0)):
              due, remaining = self.monthly_moy.is_due(
                  datetime(2013, 6, 28, 22, 14),
              )
              assert not due
              attempt = 60 * 60
              assert remaining == attempt
  
      def test_yearly_execution_is_due(self):
          with patch_crontab_nowfun(self.yearly, datetime(2010, 3, 11, 7, 30)):
              due, remaining = self.yearly.is_due(datetime(2009, 3, 12, 7, 30))
              assert due
              assert remaining == 364 * 24 * 60 * 60
  
      def test_yearly_execution_is_not_due(self):
          with patch_crontab_nowfun(self.yearly, datetime(2010, 3, 7, 10, 30)):
              due, remaining = self.yearly.is_due(datetime(2009, 3, 12, 7, 30))
              assert not due
              assert remaining == 4 * 24 * 60 * 60 - 3 * 60 * 60
  