dinislage/web2py
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge pull request #1365 from niphlod/feature/scheduler_crontab

Browse Source 
repeats via cronline expression
This commit is contained in:
mdipierro
2016-06-18 07:49:08 -05:00
committed by GitHub
parent 338ca6ba5c 6bb255286a
commit ce3f5fbff2
2 changed files with 603 additions and 41 deletions
Download Patch File Download Diff File Expand all files Collapse all files
gluon/tests/test_scheduler.py
+350 -1
View File
@@ -15,7 +15,7 @@ fix_sys_path(__file__)
from gluon.storage import Storage
from gluon.languages import translator
from gluon.scheduler import JobGraph, Scheduler
from gluon.scheduler import JobGraph, Scheduler, CronParser
from gluon.dal import DAL
@@ -51,6 +51,355 @@ class BaseTestScheduler(unittest.TestCase):
pass
class CronParserTest(unittest.TestCase):
def testMinute(self):
# minute asterisk
base = datetime.datetime(2010, 1, 23, 12, 18)
itr = CronParser('*/1 * * * *', base)
n1 = itr.get_next() # 19
self.assertEqual(base.year, n1.year)
self.assertEqual(base.month, n1.month)
self.assertEqual(base.day, n1.day)
self.assertEqual(base.hour, n1.hour)
self.assertEqual(base.minute, n1.minute - 1)
for i in range(39): # ~ 58
itr.get_next()
n2 = itr.get_next()
self.assertEqual(n2.minute, 59)
n3 = itr.get_next()
self.assertEqual(n3.minute, 0)
self.assertEqual(n3.hour, 13)
itr = CronParser('*/5 * * * *', base)
n4 = itr.get_next()
self.assertEqual(n4.minute, 20)
for i in range(6):
itr.get_next()
n5 = itr.get_next()
self.assertEqual(n5.minute, 55)
n6 = itr.get_next()
self.assertEqual(n6.minute, 0)
self.assertEqual(n6.hour, 13)
base = datetime.datetime(2010, 1, 23, 12, 18)
itr = CronParser('4/34 * * * *', base)
n7 = itr.get_next()
self.assertEqual(n7.minute, 38)
self.assertEqual(n7.hour, 12)
n8 = itr.get_next()
self.assertEqual(n8.minute, 4)
self.assertEqual(n8.hour, 13)
def testHour(self):
base = datetime.datetime(2010, 1, 24, 12, 2)
itr = CronParser('0 */3 * * *', base)
n1 = itr.get_next()
self.assertEqual(n1.hour, 15)
self.assertEqual(n1.minute, 0)
for i in range(2):
itr.get_next()
n2 = itr.get_next()
self.assertEqual(n2.hour, 0)
self.assertEqual(n2.day, 25)
def testDay(self):
base = datetime.datetime(2010, 2, 24, 12, 9)
itr = CronParser('0 0 */3 * *', base)
n1 = itr.get_next()
# 1 4 7 10 13 16 19 22 25 28
self.assertEqual(n1.day, 25)
n2 = itr.get_next()
self.assertEqual(n2.day, 28)
n3 = itr.get_next()
self.assertEqual(n3.day, 1)
self.assertEqual(n3.month, 3)
# test leap year
base = datetime.datetime(1996, 2, 27)
itr = CronParser('0 0 * * *', base)
n1 = itr.get_next()
self.assertEqual(n1.day, 28)
self.assertEqual(n1.month, 2)
n2 = itr.get_next()
self.assertEqual(n2.day, 29)
self.assertEqual(n2.month, 2)
base2 = datetime.datetime(2000, 2, 27)
itr2 = CronParser('0 0 * * *', base2)
n3 = itr2.get_next()
self.assertEqual(n3.day, 28)
self.assertEqual(n3.month, 2)
n4 = itr2.get_next()
self.assertEqual(n4.day, 29)
self.assertEqual(n4.month, 2)
def testWeekDay(self):
base = datetime.datetime(2010, 2, 25)
itr = CronParser('0 0 * * sat', base)
n1 = itr.get_next()
self.assertEqual(n1.isoweekday(), 6)
self.assertEqual(n1.day, 27)
n2 = itr.get_next()
self.assertEqual(n2.isoweekday(), 6)
self.assertEqual(n2.day, 6)
self.assertEqual(n2.month, 3)
base = datetime.datetime(2010, 1, 25)
itr = CronParser('0 0 1 * wed', base)
n1 = itr.get_next()
self.assertEqual(n1.month, 1)
self.assertEqual(n1.day, 27)
self.assertEqual(n1.year, 2010)
n2 = itr.get_next()
self.assertEqual(n2.month, 2)
self.assertEqual(n2.day, 1)
self.assertEqual(n2.year, 2010)
n3 = itr.get_next()
self.assertEqual(n3.month, 2)
self.assertEqual(n3.day, 3)
self.assertEqual(n3.year, 2010)
def testMonth(self):
base = datetime.datetime(2010, 1, 25)
itr = CronParser('0 0 1 * *', base)
n1 = itr.get_next()
self.assertEqual(n1.month, 2)
self.assertEqual(n1.day, 1)
n2 = itr.get_next()
self.assertEqual(n2.month, 3)
self.assertEqual(n2.day, 1)
for i in range(8):
itr.get_next()
n3 = itr.get_next()
self.assertEqual(n3.month, 12)
self.assertEqual(n3.year, 2010)
n4 = itr.get_next()
self.assertEqual(n4.month, 1)
self.assertEqual(n4.year, 2011)
base = datetime.datetime(2010, 1, 25)
itr = CronParser('0 0 1 */4 *', base)
n1 = itr.get_next()
self.assertEqual(n1.month, 5)
self.assertEqual(n1.day, 1)
base = datetime.datetime(2010, 1, 25)
itr = CronParser('0 0 1 1-3 *', base)
n1 = itr.get_next()
self.assertEqual(n1.month, 2)
self.assertEqual(n1.day, 1)
n2 = itr.get_next()
self.assertEqual(n2.month, 3)
self.assertEqual(n2.day, 1)
n3 = itr.get_next()
self.assertEqual(n3.month, 1)
self.assertEqual(n3.day, 1)
def testSundayToThursdayWithAlphaConversion(self):
base = datetime.datetime(2010, 8, 25, 15, 56)
itr = CronParser("30 22 * * sun-thu", base)
n1 = itr.get_next()
self.assertEqual(base.year, n1.year)
self.assertEqual(base.month, n1.month)
self.assertEqual(base.day, n1.day)
self.assertEqual(22, n1.hour)
self.assertEqual(30, n1.minute)
def testISOWeekday(self):
base = datetime.datetime(2010, 2, 25)
itr = CronParser('0 0 * * 7', base)
n1 = itr.get_next()
self.assertEqual(n1.isoweekday(), 7)
self.assertEqual(n1.day, 28)
n2 = itr.get_next()
self.assertEqual(n2.isoweekday(), 7)
self.assertEqual(n2.day, 7)
self.assertEqual(n2.month, 3)
base = datetime.datetime(2010, 2, 22)
itr = CronParser('0 0 * * */2', base)
n1 = itr.get_next()
self.assertEqual(n1.isoweekday(), 2)
self.assertEqual(n1.day, 23)
n2 = itr.get_next()
self.assertEqual(n2.isoweekday(), 4)
self.assertEqual(n2.day, 25)
def testBug2(self):
base = datetime.datetime(2012, 1, 1, 0, 0)
itr = CronParser('0 * * 3 *', base)
n1 = itr.get_next()
self.assertEqual(n1.year, base.year)
self.assertEqual(n1.month, 3)
self.assertEqual(n1.day, base.day)
self.assertEqual(n1.hour, base.hour)
self.assertEqual(n1.minute, base.minute)
n2 = itr.get_next()
self.assertEqual(n2.year, base.year)
self.assertEqual(n2.month, 3)
self.assertEqual(n2.day, base.day)
self.assertEqual(n2.hour, base.hour + 1)
self.assertEqual(n2.minute, base.minute)
n3 = itr.get_next()
self.assertEqual(n3.year, base.year)
self.assertEqual(n3.month, 3)
self.assertEqual(n3.day, base.day)
self.assertEqual(n3.hour, base.hour + 2)
self.assertEqual(n3.minute, base.minute)
def testBug3(self):
base = datetime.datetime(2013, 3, 1, 12, 17, 34, 257877)
c = CronParser('00 03 16,30 * *', base)
n1 = c.get_next()
self.assertEqual(n1.month, 3)
self.assertEqual(n1.day, 16)
n2 = c.get_next()
self.assertEqual(n2.month, 3)
self.assertEqual(n2.day, 30)
n3 = c.get_next()
self.assertEqual(n3.month, 4)
self.assertEqual(n3.day, 16)
def test_rangeGenerator(self):
base = datetime.datetime(2013, 3, 4, 0, 0)
itr = CronParser('1-9/2 0 1 * *', base)
n1 = itr.get_next()
n2 = itr.get_next()
n3 = itr.get_next()
n4 = itr.get_next()
n5 = itr.get_next()
self.assertEqual(n1.minute, 1)
self.assertEqual(n2.minute, 3)
self.assertEqual(n3.minute, 5)
self.assertEqual(n4.minute, 7)
self.assertEqual(n5.minute, 9)
def test_iterGenerator(self):
base = datetime.datetime(2013, 3, 4, 0, 0)
itr = CronParser('1-9/2 0 1 * *', base)
x = 0
for n in itr:
x += 1
if x > 4:
break
self.assertEqual(n.minute, 9)
def test_invalidcron(self):
base = datetime.datetime(2013, 3, 4, 0, 0)
itr = CronParser('5 4 31 2 *', base)
self.assertRaises(ValueError, itr.get_next)
itr = CronParser('* * 5-1 * *', base)
self.assertRaises(ValueError, itr.get_next)
itr = CronParser('* * * janu-jun *', base)
self.assertRaises(KeyError, itr.get_next)
itr = CronParser('* * * * * *', base)
self.assertRaises(ValueError, itr.get_next)
itr = CronParser('* * * *', base)
self.assertRaises(ValueError, itr.get_next)
def testLastDayOfMonth(self):
base = datetime.datetime(2015, 9, 4)
itr = CronParser('0 0 L * *', base)
n1 = itr.get_next()
self.assertEqual(n1.month, 9)
self.assertEqual(n1.day, 30)
n2 = itr.get_next()
self.assertEqual(n2.month, 10)
self.assertEqual(n2.day, 31)
n3 = itr.get_next()
self.assertEqual(n3.month, 11)
self.assertEqual(n3.day, 30)
n4 = itr.get_next()
self.assertEqual(n4.month, 12)
self.assertEqual(n4.day, 31)
base = datetime.datetime(1996, 2, 27)
itr = CronParser('0 0 L * *', base)
n1 = itr.get_next()
self.assertEqual(n1.day, 29)
self.assertEqual(n1.month, 2)
n2 = itr.get_next()
self.assertEqual(n2.day, 31)
self.assertEqual(n2.month, 3)
def testSpecialExpr(self):
base = datetime.datetime(2000, 1, 1)
itr = CronParser('@yearly', base)
n1 = itr.get_next()
self.assertEqual(n1.day, 1)
self.assertEqual(n1.month, 1)
self.assertEqual(n1.year, base.year + 1)
self.assertEqual(n1.hour, 0)
self.assertEqual(n1.minute, 0)
itr = CronParser('@annually', base)
n1 = itr.get_next()
self.assertEqual(n1.day, 1)
self.assertEqual(n1.month, 1)
self.assertEqual(n1.year, base.year + 1)
self.assertEqual(n1.hour, 0)
self.assertEqual(n1.minute, 0)
itr = CronParser('@monthly', base)
n1 = itr.get_next()
self.assertEqual(n1.day, 1)
self.assertEqual(n1.month, base.month + 1)
self.assertEqual(n1.year, base.year)
self.assertEqual(n1.hour, 0)
self.assertEqual(n1.minute, 0)
itr = CronParser('@weekly', base)
n1 = itr.get_next()
self.assertEqual(n1.day, 2)
self.assertEqual(n1.month, base.month)
self.assertEqual(n1.year, base.year)
self.assertEqual(n1.hour, 0)
self.assertEqual(n1.minute, 0)
n2 = itr.get_next()
self.assertEqual(n2.day, 9)
self.assertEqual(n2.month, base.month)
self.assertEqual(n2.year, base.year)
self.assertEqual(n2.hour, 0)
self.assertEqual(n2.minute, 0)
n3 = itr.get_next()
self.assertEqual(n3.day, 16)
self.assertEqual(n3.month, base.month)
self.assertEqual(n3.year, base.year)
self.assertEqual(n3.hour, 0)
self.assertEqual(n3.minute, 0)
itr = CronParser('@daily', base)
n1 = itr.get_next()
self.assertEqual(n1.day, 2)
self.assertEqual(n1.month, base.month)
self.assertEqual(n1.year, base.year)
self.assertEqual(n1.hour, 0)
self.assertEqual(n1.minute, 0)
itr = CronParser('@midnight', base)
n1 = itr.get_next()
self.assertEqual(n1.day, 2)
self.assertEqual(n1.month, base.month)
self.assertEqual(n1.year, base.year)
self.assertEqual(n1.hour, 0)
self.assertEqual(n1.minute, 0)
itr = CronParser('@hourly', base)
n1 = itr.get_next()
self.assertEqual(n1.day, 1)
self.assertEqual(n1.month, base.month)
self.assertEqual(n1.year, base.year)
self.assertEqual(n1.hour, 1)
self.assertEqual(n1.minute, 0)
class TestsForJobGraph(BaseTestScheduler):
def testJobGraph(self):