"""Test recur.py""" import os import recur import time import unittest import datetime dtd = datetime.date dtdt = datetime.datetime dtt = datetime.time class RecurTest(unittest.TestCase): def test_0_sane_date(self): # Sane dates self.assertEqual(recur.sane_date(1999, 12, 31), dtd(1999, 12, 31)) self.assertEqual(recur.sane_date(2000, 1, 1), dtd(2000, 1, 1)) self.assertEqual(recur.sane_date(2004, 5, 4), dtd(2004, 5, 4)) # Insane dates self.assertEqual(recur.sane_date(1999, 12, 0), dtd(1999, 11, 30)) self.assertEqual(recur.sane_date(2000, 1, -1), dtd(1999, 12, 30)) self.assertEqual(recur.sane_date(2004, 5, -4), dtd(2004, 4, 26)) self.assertEqual(recur.sane_date(1999, 12, 40), dtd(2000, 1, 9)) self.assertEqual(recur.sane_date(2000, 1, 81), dtd(2000, 3, 21)) self.assertEqual(recur.sane_date(2004, 5, 365), dtd(2005, 4, 30)) self.assertEqual(recur.sane_date(1999, 14, 28, True), dtd(2000, 2, 28)) self.assertEqual(recur.sane_date(1999, -10, 28, True), dtd(1998, 2, 28)) def test_1_sane_time(self): # Sane times self.assertEqual(recur.sane_time(0, 6, 0, 0), (0, dtt(6, 0, 0))) self.assertEqual(recur.sane_time(1, 3, 59, 5), (1, dtt(3, 59, 5))) self.assertEqual(recur.sane_time(5, 16, 0, 10), (5, dtt(16, 0, 10))) # Insane times self.assertEqual(recur.sane_time(0, 26, 0, 0), (1, dtt(2, 0, 0))) self.assertEqual(recur.sane_time(0, 6, 80, 10), (0, dtt(7, 20, 10))) self.assertEqual(recur.sane_time(0, 0, 0, -1), (-1, dtt(23, 59, 59))) self.assertEqual(recur.sane_time(0, 23, 59, 64), (1, dtt(0, 0, 4))) self.assertEqual(recur.sane_time(-4, -2, 0, 0), (-5, dtt(22, 0, 0))) def test_seconds(self): dates = recur.seconds(dtdt(1999, 12, 28, 23, 59), 10, dtdt(1999, 12, 29, 0)) self.assertEqual([x for x in dates], [dtdt(1999, 12, 28, 23, 59, 0), dtdt(1999, 12, 28, 23, 59, 10), dtdt(1999, 12, 28, 23, 59, 20), dtdt(1999, 12, 28, 23, 59, 30), dtdt(1999, 12, 28, 23, 59, 40), dtdt(1999, 12, 28, 23, 59, 50), dtdt(1999, 12, 29, 0, 0, 0), ] ) def test_eachminute(self): dates = recur.eachminute(dtdt(1999, 1, 1, 1, 55), 15, dtdt(1999, 1, 1, 2, 2)) self.assertEqual([x for x in dates], [dtdt(1999, 1, 1, 1, 55, 15), dtdt(1999, 1, 1, 1, 56, 15), dtdt(1999, 1, 1, 1, 57, 15), dtdt(1999, 1, 1, 1, 58, 15), dtdt(1999, 1, 1, 1, 59, 15), dtdt(1999, 1, 1, 2, 0, 15), dtdt(1999, 1, 1, 2, 1, 15), ] ) def test_minutes(self): dates = recur.minutes(dtdt(1999, 12, 28, 23), 15, dtdt(1999, 12, 29, 1)) self.assertEqual([x for x in dates], [dtdt(1999, 12, 28, 23), dtdt(1999, 12, 28, 23, 15), dtdt(1999, 12, 28, 23, 30), dtdt(1999, 12, 28, 23, 45), dtdt(1999, 12, 29, 0, 0), dtdt(1999, 12, 29, 0, 15), dtdt(1999, 12, 29, 0, 30), dtdt(1999, 12, 29, 0, 45), dtdt(1999, 12, 29, 1, 0), ] ) def test_eachhour(self): dates = recur.eachhour(dtdt(1999, 12, 29, 22), 59, 59, dtdt(1999, 12, 30, 5)) self.assertEqual([x for x in dates], [dtdt(1999, 12, 29, 22, 59, 59), dtdt(1999, 12, 29, 23, 59, 59), dtdt(1999, 12, 30, 0, 59, 59), dtdt(1999, 12, 30, 1, 59, 59), dtdt(1999, 12, 30, 2, 59, 59), dtdt(1999, 12, 30, 3, 59, 59), dtdt(1999, 12, 30, 4, 59, 59), ] ) def test_hours(self): dates = recur.hours(dtd(1999, 12, 28), 4, dtd(1999, 12, 29)) self.assertEqual([x for x in dates], [dtdt(1999, 12, 28, 0), dtdt(1999, 12, 28, 4), dtdt(1999, 12, 28, 8), dtdt(1999, 12, 28, 12), dtdt(1999, 12, 28, 16), dtdt(1999, 12, 28, 20), dtdt(1999, 12, 29, 0), dtdt(1999, 12, 29, 4), dtdt(1999, 12, 29, 8), dtdt(1999, 12, 29, 12), dtdt(1999, 12, 29, 16), dtdt(1999, 12, 29, 20), ] ) # Test a startDate and endDate with a time component. dates = recur.hours(dtdt(1999, 12, 28, 9), 4, dtdt(1999, 12, 29, 7)) self.assertEqual([x for x in dates], [dtdt(1999, 12, 28, 9), dtdt(1999, 12, 28, 13), dtdt(1999, 12, 28, 17), dtdt(1999, 12, 28, 21), dtdt(1999, 12, 29, 1), dtdt(1999, 12, 29, 5), ] ) def test_time_from_str(self): # Valid values self.assertEqual(recur.time_from_str("4:03:18"), dtt(4, 3, 18)) self.assertEqual(recur.time_from_str("14:5:48"), dtt(14, 5, 48)) self.assertEqual(recur.time_from_str("0"), dtt(0)) self.assertEqual(recur.time_from_str("0:0:0"), dtt(0)) self.assertEqual(recur.time_from_str("23:59:59"), dtt(23, 59, 59)) # Invalid values self.assertRaises(ValueError, recur.time_from_str, "28:59:59") self.assertRaises(ValueError, recur.time_from_str, "no:time:to:lose") def test_eachday(self): dates = recur.eachday(dtd(1999, 12, 28), dtt(6, 43, 21), dtd(2000, 1, 3)) self.assertEqual([x for x in dates], [dtdt(1999, 12, 28, 6, 43, 21), dtdt(1999, 12, 29, 6, 43, 21), dtdt(1999, 12, 30, 6, 43, 21), dtdt(1999, 12, 31, 6, 43, 21), dtdt(2000, 1, 1, 6, 43, 21), dtdt(2000, 1, 2, 6, 43, 21), dtdt(2000, 1, 3, 6, 43, 21), ] ) # Test start and end dates that have time components. # If 'inside' the given time, those extrema should # not be included. dates = recur.eachday(dtdt(1999, 12, 28, 8), dtt(6, 43, 21), dtdt(2000, 1, 3, 2)) self.assertEqual([x for x in dates], [dtdt(1999, 12, 29, 6, 43, 21), dtdt(1999, 12, 30, 6, 43, 21), dtdt(1999, 12, 31, 6, 43, 21), dtdt(2000, 1, 1, 6, 43, 21), dtdt(2000, 1, 2, 6, 43, 21), ] ) def test_days(self): dates = recur.days(dtd(1999, 12, 28), 1, dtd(2000, 1, 3)) self.assertEqual([x for x in dates], [dtd(1999, 12, 28), dtd(1999, 12, 29), dtd(1999, 12, 30), dtd(1999, 12, 31), dtd(2000, 1, 1), dtd(2000, 1, 2), dtd(2000, 1, 3), ] ) dates = recur.days(dtd(1999, 12, 28), 3, dtd(2000, 1, 3)) self.assertEqual([x for x in dates], [dtd(1999, 12, 28), dtd(1999, 12, 31), dtd(2000, 1, 3), ] ) def test_eachweekday(self): dates = recur.eachweekday(dtd(1999, 12, 28), 0, dtt(6, 43, 21), dtd(2000, 1, 30)) self.assertEqual([x for x in dates], [dtdt(2000, 1, 3, 6, 43, 21), dtdt(2000, 1, 10, 6, 43, 21), dtdt(2000, 1, 17, 6, 43, 21), dtdt(2000, 1, 24, 6, 43, 21), ] ) # Test start and end dates that have time components. # If 'inside' the given time, those extrema should # not be included. dates = recur.eachweekday(dtdt(2000, 1, 3, 8), 0, dtt(6, 43, 21), dtdt(2000, 1, 31, 2)) self.assertEqual([x for x in dates], [dtdt(2000, 1, 10, 6, 43, 21), dtdt(2000, 1, 17, 6, 43, 21), dtdt(2000, 1, 24, 6, 43, 21), ] ) def test_eachweek(self): knowndates = [dtd(1999, 12, 31), dtd(2000, 1, 7), dtd(2000, 1, 14), dtd(2000, 1, 21), dtd(2000, 1, 28), dtd(2000, 2, 4), dtd(2000, 2, 11), ] dates = recur.eachweek(dtd(1999, 12, 28), 4, dtd(2000, 2, 13)) self.assertEqual([x for x in dates], knowndates) # Try a weekday that is out of bounds. dates = recur.eachweek(dtd(1999, 12, 28), 11, dtd(2000, 2, 13)) self.assertEqual([x for x in dates], knowndates) def test_weeks(self): dates = recur.weeks(dtd(1999, 12, 28), 1, dtd(2000, 2, 13)) self.assertEqual([x for x in dates], [dtd(1999, 12, 28), dtd(2000, 1, 4), dtd(2000, 1, 11), dtd(2000, 1, 18), dtd(2000, 1, 25), dtd(2000, 2, 1), dtd(2000, 2, 8), ] ) dates = recur.weeks(dtd(1999, 12, 28), 3, dtd(2000, 2, 13)) self.assertEqual([x for x in dates], [dtd(1999, 12, 28), dtd(2000, 1, 18), dtd(2000, 2, 8), ] ) def test_eachmonth(self): knowndates = [dtd(1999, 12, 20), dtd(2000, 1, 20), dtd(2000, 2, 20), dtd(2000, 3, 20), dtd(2000, 4, 20), dtd(2000, 5, 20), dtd(2000, 6, 20), ] dates = recur.eachmonth(dtd(1999, 11, 28), 20, dtd(2000, 7, 13)) self.assertEqual([x for x in dates], knowndates) # Try a day that is out of bounds. knowndates = [datetime.date(1999, 12, 10), datetime.date(2000, 2, 9), datetime.date(2000, 4, 9), datetime.date(2000, 6, 9)] dates = recur.eachmonth(dtd(1999, 11, 28), 40, dtd(2000, 7, 13)) self.assertEqual([x for x in dates], knowndates) def test_months(self): dates = recur.months(dtd(1999, 12, 31), 1, dtd(2000, 7, 13)) self.assertEqual([x for x in dates], [dtd(1999, 12, 31), dtd(2000, 1, 31), dtd(2000, 3, 2), dtd(2000, 3, 31), dtd(2000, 5, 1), dtd(2000, 5, 31), dtd(2000, 7, 1), ] ) dates = recur.months(dtd(1999, 12, 31), 3, dtd(2000, 7, 13)) self.assertEqual([x for x in dates], [dtd(1999, 12, 31), dtd(2000, 3, 31), dtd(2000, 7, 1), ] ) def test_eachyear(self): knowndates = [dtd(1999, 3, 1), dtd(2000, 2, 29), dtd(2001, 3, 1), dtd(2002, 3, 1), dtd(2003, 3, 1), dtd(2004, 2, 29), dtd(2005, 3, 1), ] dates = recur.eachyear(dtd(1998, 11, 28), 2, 29, dtd(2005, 7, 13)) self.assertEqual([x for x in dates], knowndates) # Try a day that is out of bounds. dates = recur.eachyear(dtd(1998, 11, 28), 1, 60, dtd(2005, 7, 13)) self.assertEqual([x for x in dates], knowndates) # Try a negative month. dates = recur.eachyear(dtd(1998, 11, 28), -10, 29, dtd(2005, 7, 13)) for x, y in enumerate(dates): self.assertEqual(y, knowndates[x]) def test_years(self): dates = recur.years(dtd(1999, 12, 31), 1, dtd(2006, 7, 13)) self.assertEqual([x for x in dates], [dtd(1999, 12, 31), dtd(2000, 12, 31), dtd(2001, 12, 31), dtd(2002, 12, 31), dtd(2003, 12, 31), dtd(2004, 12, 31), dtd(2005, 12, 31), ] ) dates = recur.years(dtd(2000, 2, 29), 3, dtd(2012, 7, 13)) self.assertEqual([x for x in dates], [dtd(2000, 2, 29), dtd(2003, 3, 1), dtd(2006, 3, 1), dtd(2009, 3, 1), dtd(2012, 2, 29), ] ) def test_Recurrence(self): # Test seconds n = datetime.datetime.now() dates = list(recur.Recurrence(n, "1 second", n + datetime.timedelta(seconds=5))) knowndates = [n + datetime.timedelta(seconds=x) for x in xrange(6)] self.assertEqual(dates, knowndates) # Test eachday knowndates = [dtdt(600, 1, 1, 6, 35), dtdt(600, 1, 2, 6, 35), dtdt(600, 1, 3, 6, 35), dtdt(600, 1, 4, 6, 35), dtdt(600, 1, 5, 6, 35), ] dates = recur.Recurrence(dtd(600, 1, 1), "6:35 every day", dtd(600, 1, 5)) self.assertEqual([x for x in dates], knowndates) # Test eachweekday knowndates = [dtdt(1999, 12, 28, 6, 35), dtdt(2000, 1, 4, 6, 35), dtdt(2000, 1, 11, 6, 35), dtdt(2000, 1, 18, 6, 35), dtdt(2000, 1, 25, 6, 35), ] dates = recur.Recurrence(dtd(1999, 12, 28), "6:35 every tuesday", dtd(2000, 1, 30)) self.assertEqual([x for x in dates], knowndates) # Test eachweek knowndates = [dtd(1999, 12, 31), dtd(2000, 1, 7), dtd(2000, 1, 14), dtd(2000, 1, 21), dtd(2000, 1, 28), dtd(2000, 2, 4), dtd(2000, 2, 11), ] dates = recur.Recurrence(dtd(1999, 12, 28), "fri", dtd(2000, 2, 13)) self.assertEqual([x for x in dates], knowndates) dates = recur.Recurrence(dtd(1999, 12, 28), "friday", dtd(2000, 2, 13)) self.assertEqual([x for x in dates], knowndates) # Test eachmonth knowndates = [dtd(1999, 12, 20), dtd(2000, 1, 20), dtd(2000, 2, 20), dtd(2000, 3, 20), dtd(2000, 4, 20), dtd(2000, 5, 20), dtd(2000, 6, 20), ] dates = recur.Recurrence(dtd(1999, 11, 28), "20 each month", dtd(2000, 7, 13)) self.assertEqual([x for x in dates], knowndates) # Test negative eachmonth knowndates = [dtd(1999, 12, 26), dtd(2000, 1, 26), dtd(2000, 2, 24), dtd(2000, 3, 26), dtd(2000, 4, 25), dtd(2000, 5, 26), dtd(2000, 6, 25), ] dates = recur.Recurrence(dtd(1999, 11, 28), "-5 each month", dtd(2000, 7, 13)) self.assertEqual([x for x in dates], knowndates) # Test eachyear knowndates = [dtd(1999, 3, 1), dtd(2000, 2, 29), dtd(2001, 3, 1), dtd(2002, 3, 1), dtd(2003, 3, 1), dtd(2004, 2, 29), dtd(2005, 3, 1), ] dates = recur.Recurrence(dtd(1998, 11, 28), "Feb 29", dtd(2005, 7, 13)) self.assertEqual([x for x in dates], knowndates) # Make sure Recurrence objects are unique. dates2 = recur.Recurrence(dtd(1998, 11, 28), "Feb 29", dtd(2005, 7, 13)) self.assertNotEqual(dates, dates2) self.assertNotEqual(dates.generator, dates2.generator) # Try a "years" test dates = recur.Recurrence(dtd(1999, 12, 31), "1 year", dtd(2006, 7, 13)) self.assertEqual([x for x in dates], [dtd(1999, 12, 31), dtd(2000, 12, 31), dtd(2001, 12, 31), dtd(2002, 12, 31), dtd(2003, 12, 31), dtd(2004, 12, 31), dtd(2005, 12, 31), ] ) # Try a non-matching recurrence. self.assertRaises(ValueError, recur.Recurrence, dtd(1998, 11, 28), "Umptember 29", dtd(2005, 7, 13)) def test_Workers(self): count = [0] class Counter(recur.Worker): def work(self): count[0] += 1 import time another = [0] class AnotherCounter(recur.Worker): def work(self): another[0] += 1 paused = [0] class PausedWorker(recur.Worker): """A worker that starts paused (active==False).""" def __init__(self, *args, **params): recur.Worker.__init__(self, *args, **params) self.active = False def work(self): # This worker starts paused (inactive) and shouldn't do # any work paused[0] += 1 # the aptly named... class WorkerNotAppearingInThisFilm(recur.Worker): def work(self): pass s = recur.Scheduler({'mycounter': Counter("1 second")}) s.start() time.sleep(2.5) s.stop() # count == 3 because the first recurrence value == now. self.assertEqual(count[0], 3) s = recur.Scheduler({'c1': Counter("1 second"), 'c2': AnotherCounter("2 seconds"), # A worker that starts paused (inactive). 'c3': PausedWorker("1 second"), # A worker with an empty recurrence string. # This is allowed in __init__, and sets # self.recurrence to None. 'c4': WorkerNotAppearingInThisFilm("")}) s.start() try: # Here's how this should work: # elapsed time count another # 0.0 + + # 0.5 =4 =1 # 1.0 + # 1.5 =5 =1 pause # 2.0 X X # 2.5 =5 =1 resume # 3.0 + # 3.5 =6 =1 # 4.0 + + # 4.5 =7 =2 time.sleep(0.5) self.assertEqual(count[0], 4) self.assertEqual(another[0], 1) time.sleep(1) self.assertEqual(count[0], 5) self.assertEqual(another[0], 1) s.paused = True time.sleep(1) self.assertEqual(count[0], 5) self.assertEqual(another[0], 1) s.paused = False time.sleep(1) self.assertEqual(count[0], 6) # We skipped "another" when we paused, above. self.assertEqual(another[0], 1) time.sleep(1) self.assertEqual(count[0], 7) self.assertEqual(another[0], 2) # Make sure the PausedWorker hasn't done anything self.assertEqual(paused[0], 0) finally: s.stop() # Test a Scheduler with a single Worker with an empty recurrence string s = recur.Scheduler({'w': WorkerNotAppearingInThisFilm("")}) s.start() time.sleep(0.5) s.stop() if __name__ == "__main__": unittest.main()