SP/web2py/gluon/tests/test_scheduler.py

897 lines
33 KiB
Python
Raw Permalink Normal View History

2018-10-25 15:33:07 +00:00
#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
Unit tests for gluon.scheduler
"""
import os
import unittest
import glob
import datetime
import sys
from gluon.storage import Storage
from gluon.languages import translator
from gluon.scheduler import JobGraph, Scheduler, CronParser
from gluon.dal import DAL
class BaseTestScheduler(unittest.TestCase):
def setUp(self):
self.db = None
self.cleanfolder()
from gluon.globals import current
s = Storage({'application': 'welcome',
'folder': 'applications/welcome',
'controller': 'default'})
current.request = s
T = translator('', 'en')
current.T = T
self.db = DAL('sqlite://dummy2.db', check_reserved=['all'])
def cleanfolder(self):
if self.db:
self.db.close()
try:
os.unlink('dummy2.db')
except:
pass
tfiles = glob.glob('*_scheduler*.table')
for a in tfiles:
os.unlink(a)
def tearDown(self):
self.cleanfolder()
try:
self.inner_teardown()
except:
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):
s = Scheduler(self.db)
myjob = JobGraph(self.db, 'job_1')
fname = 'foo'
# We have a few items to wear, and there's an "order" to respect...
# Items are: watch, jacket, shirt, tie, pants, undershorts, belt, shoes, socks
# Now, we can't put on the tie without wearing the shirt first, etc...
watch = s.queue_task(fname, task_name='watch')
jacket = s.queue_task(fname, task_name='jacket')
shirt = s.queue_task(fname, task_name='shirt')
tie = s.queue_task(fname, task_name='tie')
pants = s.queue_task(fname, task_name='pants')
undershorts = s.queue_task(fname, task_name='undershorts')
belt = s.queue_task(fname, task_name='belt')
shoes = s.queue_task(fname, task_name='shoes')
socks = s.queue_task(fname, task_name='socks')
# before the tie, comes the shirt
myjob.add_deps(tie.id, shirt.id)
# before the belt too comes the shirt
myjob.add_deps(belt.id, shirt.id)
# before the jacket, comes the tie
myjob.add_deps(jacket.id, tie.id)
# before the belt, come the pants
myjob.add_deps(belt.id, pants.id)
# before the shoes, comes the pants
myjob.add_deps(shoes.id, pants.id)
# before the pants, comes the undershorts
myjob.add_deps(pants.id, undershorts.id)
# before the shoes, comes the undershorts
myjob.add_deps(shoes.id, undershorts.id)
# before the jacket, comes the belt
myjob.add_deps(jacket.id, belt.id)
# before the shoes, comes the socks
myjob.add_deps(shoes.id, socks.id)
## results in the following topological sort
# 9,3,6 --> 4,5 --> 8,7 --> 2
# socks, shirt, undershorts
# tie, pants
# shoes, belt
# jacket
known_toposort = [
set([socks.id, shirt.id, undershorts.id]),
set([tie.id, pants.id]),
set([shoes.id, belt.id]),
set([jacket.id])
]
toposort = myjob.validate('job_1')
self.assertEqual(toposort, known_toposort)
# add a cyclic dependency, jacket to undershorts
myjob.add_deps(undershorts.id, jacket.id)
# no exceptions raised, but result None
self.assertEqual(myjob.validate('job_1'), None)
def testJobGraphFailing(self):
s = Scheduler(self.db)
myjob = JobGraph(self.db, 'job_1')
fname = 'foo'
# We have a few items to wear, and there's an "order" to respect...
# Items are: watch, jacket, shirt, tie, pants, undershorts, belt, shoes, socks
# Now, we can't put on the tie without wearing the shirt first, etc...
watch = s.queue_task(fname, task_name='watch')
jacket = s.queue_task(fname, task_name='jacket')
shirt = s.queue_task(fname, task_name='shirt')
tie = s.queue_task(fname, task_name='tie')
pants = s.queue_task(fname, task_name='pants')
undershorts = s.queue_task(fname, task_name='undershorts')
belt = s.queue_task(fname, task_name='belt')
shoes = s.queue_task(fname, task_name='shoes')
socks = s.queue_task(fname, task_name='socks')
# before the tie, comes the shirt
myjob.add_deps(tie.id, shirt.id)
# before the belt too comes the shirt
myjob.add_deps(belt.id, shirt.id)
# before the jacket, comes the tie
myjob.add_deps(jacket.id, tie.id)
# before the belt, come the pants
myjob.add_deps(belt.id, pants.id)
# before the shoes, comes the pants
myjob.add_deps(shoes.id, pants.id)
# before the pants, comes the undershorts
myjob.add_deps(pants.id, undershorts.id)
# before the shoes, comes the undershorts
myjob.add_deps(shoes.id, undershorts.id)
# before the jacket, comes the belt
myjob.add_deps(jacket.id, belt.id)
# before the shoes, comes the socks
myjob.add_deps(shoes.id, socks.id)
# add a cyclic dependency, jacket to undershorts
myjob.add_deps(undershorts.id, jacket.id)
# no exceptions raised, but result None
self.assertEqual(myjob.validate('job_1'), None)
# and no deps added
deps_inserted = self.db(self.db.scheduler_task_deps.id>0).count()
self.assertEqual(deps_inserted, 0)
def testJobGraphDifferentJobs(self):
s = Scheduler(self.db)
myjob1 = JobGraph(self.db, 'job_1')
myjob2 = JobGraph(self.db, 'job_2')
fname = 'foo'
# We have a few items to wear, and there's an "order" to respect...
# Items are: watch, jacket, shirt, tie, pants, undershorts, belt, shoes, socks
# Now, we can't put on the tie without wearing the shirt first, etc...
watch = s.queue_task(fname, task_name='watch')
jacket = s.queue_task(fname, task_name='jacket')
shirt = s.queue_task(fname, task_name='shirt')
tie = s.queue_task(fname, task_name='tie')
pants = s.queue_task(fname, task_name='pants')
undershorts = s.queue_task(fname, task_name='undershorts')
belt = s.queue_task(fname, task_name='belt')
shoes = s.queue_task(fname, task_name='shoes')
socks = s.queue_task(fname, task_name='socks')
# before the tie, comes the shirt
myjob1.add_deps(tie.id, shirt.id)
# before the belt too comes the shirt
myjob1.add_deps(belt.id, shirt.id)
# before the jacket, comes the tie
myjob1.add_deps(jacket.id, tie.id)
# before the belt, come the pants
myjob1.add_deps(belt.id, pants.id)
# before the shoes, comes the pants
myjob2.add_deps(shoes.id, pants.id)
# before the pants, comes the undershorts
myjob2.add_deps(pants.id, undershorts.id)
# before the shoes, comes the undershorts
myjob2.add_deps(shoes.id, undershorts.id)
# before the jacket, comes the belt
myjob2.add_deps(jacket.id, belt.id)
# before the shoes, comes the socks
myjob2.add_deps(shoes.id, socks.id)
# every job by itself can be completed
self.assertNotEqual(myjob1.validate('job_1'), None)
self.assertNotEqual(myjob1.validate('job_2'), None)
# and, implicitly, every queued task can be too
self.assertNotEqual(myjob1.validate(), None)
# add a cyclic dependency, jacket to undershorts
myjob2.add_deps(undershorts.id, jacket.id)
# every job can still be completed by itself
self.assertNotEqual(myjob1.validate('job_1'), None)
self.assertNotEqual(myjob1.validate('job_2'), None)
# but trying to see if every task will ever be completed fails
self.assertEqual(myjob2.validate(), None)
class TestsForSchedulerAPIs(BaseTestScheduler):
def testQueue_Task(self):
def isnotqueued(result):
self.assertEqual(result.id, None)
self.assertEqual(result.uuid, None)
self.assertEqual(len(result.errors.keys()) > 0, True)
def isqueued(result):
self.assertNotEqual(result.id, None)
self.assertNotEqual(result.uuid, None)
self.assertEqual(len(result.errors.keys()), 0)
s = Scheduler(self.db)
fname = 'foo'
watch = s.queue_task(fname, task_name='watch')
# queuing a task returns id, errors, uuid
self.assertEqual(set(watch.keys()), set(['id', 'uuid', 'errors']))
# queueing nothing isn't allowed
self.assertRaises(TypeError, s.queue_task, *[])
# passing pargs and pvars wrongly
# # pargs as dict
isnotqueued(s.queue_task(fname, dict(a=1), dict(b=1)))
# # pvars as list
isnotqueued(s.queue_task(fname, ['foo', 'bar'], ['foo', 'bar']))
# two tasks with the same uuid won't be there
isqueued(s.queue_task(fname, uuid='a'))
isnotqueued(s.queue_task(fname, uuid='a'))
# # #FIXME add here every parameter
def testTask_Status(self):
s = Scheduler(self.db)
fname = 'foo'
watch = s.queue_task(fname, task_name='watch')
# fetch status by id
by_id = s.task_status(watch.id)
# fetch status by uuid
by_uuid = s.task_status(watch.uuid)
# fetch status by query
by_query = s.task_status(self.db.scheduler_task.function_name == 'foo')
self.assertEqual(by_id, by_uuid)
self.assertEqual(by_id, by_query)
# fetch status by anything else throws
self.assertRaises(SyntaxError, s.task_status, *[[1, 2]])
# adding output returns the joined set, plus "result"
rtn = s.task_status(watch.id, output=True)
self.assertEqual(set(rtn.keys()), set(['scheduler_run', 'scheduler_task', 'result']))
class testForSchedulerRunnerBase(BaseTestScheduler):
def inner_teardown(self):
from gluon import current
fdest = os.path.join(current.request.folder, 'models', 'scheduler.py')
os.unlink(fdest)
additional_files = [
os.path.join(current.request.folder, 'private', 'demo8.pholder'),
os.path.join(current.request.folder, 'views', 'issue_1485_2.html'),
]
for f in additional_files:
try:
os.unlink(f)
except:
pass
def writeview(self, content, dest=None):
from gluon import current
fdest = os.path.join(current.request.folder, 'views', dest)
with open(fdest, 'w') as q:
q.write(content)
def writefunction(self, content, initlines=None):
from gluon import current
fdest = os.path.join(current.request.folder, 'models', 'scheduler.py')
if initlines is None:
initlines = """
import os
import time
from gluon.scheduler import Scheduler
db_dal = os.path.abspath(os.path.join(request.folder, '..', '..', 'dummy2.db'))
sched_dal = DAL('sqlite://%s' % db_dal, folder=os.path.dirname(db_dal))
sched = Scheduler(sched_dal, max_empty_runs=15, migrate=False, heartbeat=1)
def termination():
sched.terminate()
sched_dal.commit()
"""
with open(fdest, 'w') as q:
q.write(initlines)
q.write(content)
def exec_sched(self):
import subprocess
call_args = [sys.executable, 'web2py.py', '--no-banner', '-D', '20','-K', 'welcome']
ret = subprocess.call(call_args, env=dict(os.environ))
return ret
def fetch_results(self, sched, task):
info = sched.task_status(task.id)
task_runs = self.db(self.db.scheduler_run.task_id == task.id).select()
return info, task_runs
def exec_asserts(self, stmts, tag):
for stmt in stmts:
self.assertEqual(stmt[1], True, msg="%s - %s" % (tag, stmt[0]))
class TestsForSchedulerRunner(testForSchedulerRunnerBase):
def testRepeats_and_Expired_and_Prio(self):
s = Scheduler(self.db)
repeats = s.queue_task('demo1', ['a', 'b'], dict(c=1, d=2), repeats=2, period=5)
a_while_ago = datetime.datetime.now() - datetime.timedelta(seconds=60)
expired = s.queue_task('demo4', stop_time=a_while_ago)
prio1 = s.queue_task('demo1', ['scheduled_first'])
prio2 = s.queue_task('demo1', ['scheduled_second'], next_run_time=a_while_ago)
self.db.commit()
self.writefunction(r"""
def demo1(*args,**vars):
print('you passed args=%s and vars=%s' % (args, vars))
return args[0]
def demo4():
time.sleep(15)
print("I'm printing something")
return dict(a=1, b=2)
""")
ret = self.exec_sched()
self.assertEqual(ret, 0)
# repeats check
task, task_run = self.fetch_results(s, repeats)
res = [
("task status completed", task.status == 'COMPLETED'),
("task times_run is 2", task.times_run == 2),
("task ran 2 times only", len(task_run) == 2),
("scheduler_run records are COMPLETED ", (task_run[0].status == task_run[1].status == 'COMPLETED')),
("period is respected", (task_run[1].start_time > task_run[0].start_time + datetime.timedelta(seconds=task.period)))
]
self.exec_asserts(res, 'REPEATS')
# expired check
task, task_run = self.fetch_results(s, expired)
res = [
("task status expired", task.status == 'EXPIRED'),
("task times_run is 0", task.times_run == 0),
("task didn't run at all", len(task_run) == 0)
]
self.exec_asserts(res, 'EXPIRATION')
# prio check
task1 = s.task_status(prio1.id, output=True)
task2 = s.task_status(prio2.id, output=True)
res = [
("tasks status completed", task1.scheduler_task.status == task2.scheduler_task.status == 'COMPLETED'),
("priority2 was executed before priority1" , task1.scheduler_run.id > task2.scheduler_run.id)
]
self.exec_asserts(res, 'PRIORITY')
def testNoReturn_and_Timeout_and_Progress(self):
s = Scheduler(self.db)
noret1 = s.queue_task('demo5')
noret2 = s.queue_task('demo3')
timeout1 = s.queue_task('demo4', timeout=5)
timeout2 = s.queue_task('demo4')
progress = s.queue_task('demo6', sync_output=2)
termination = s.queue_task('termination')
self.db.commit()
self.writefunction(r"""
def demo3():
time.sleep(3)
print(1/0)
return None
def demo4():
time.sleep(15)
print("I'm printing something")
return dict(a=1, b=2)
def demo5():
time.sleep(3)
print("I'm printing something")
rtn = dict(a=1, b=2)
def demo6():
time.sleep(5)
print('50%')
time.sleep(5)
print('!clear!100%')
return 1
""")
ret = self.exec_sched()
self.assertEqual(ret, 0)
# noreturn check
task1, task_run1 = self.fetch_results(s, noret1)
task2, task_run2 = self.fetch_results(s, noret2)
res = [
("tasks no_returns1 completed", task1.status == 'COMPLETED'),
("tasks no_returns2 failed", task2.status == 'FAILED'),
("no_returns1 doesn't have a scheduler_run record", len(task_run1) == 0),
("no_returns2 has a scheduler_run record FAILED", (len(task_run2) == 1 and task_run2[0].status == 'FAILED')),
]
self.exec_asserts(res, 'NO_RETURN')
# timeout check
task1 = s.task_status(timeout1.id, output=True)
task2 = s.task_status(timeout2.id, output=True)
res = [
("tasks timeouts1 timeoutted", task1.scheduler_task.status == 'TIMEOUT'),
("tasks timeouts2 completed", task2.scheduler_task.status == 'COMPLETED')
]
self.exec_asserts(res, 'TIMEOUT')
# progress check
task1 = s.task_status(progress.id, output=True)
res = [
("tasks percentages completed", task1.scheduler_task.status == 'COMPLETED'),
("output contains only 100%", task1.scheduler_run.run_output.strip() == "100%")
]
self.exec_asserts(res, 'PROGRESS')
def testDrift_and_env_and_immediate(self):
s = Scheduler(self.db)
immediate = s.queue_task('demo1', ['a', 'b'], dict(c=1, d=2), immediate=True)
env = s.queue_task('demo7')
drift = s.queue_task('demo1', ['a', 'b'], dict(c=1, d=2), period=93, prevent_drift=True)
termination = s.queue_task('termination')
self.db.commit()
self.writefunction(r"""
def demo1(*args,**vars):
print('you passed args=%s and vars=%s' % (args, vars))
return args[0]
import random
def demo7():
time.sleep(random.randint(1,5))
print(W2P_TASK, request.now)
return W2P_TASK.id, W2P_TASK.uuid, W2P_TASK.run_id
""")
ret = self.exec_sched()
self.assertEqual(ret, 0)
# immediate check, can only check that nothing breaks
task1 = s.task_status(immediate.id)
res = [
("tasks status completed", task1.status == 'COMPLETED'),
]
self.exec_asserts(res, 'IMMEDIATE')
# drift check
task, task_run = self.fetch_results(s, drift)
res = [
("task status completed", task.status == 'COMPLETED'),
("next_run_time is exactly start_time + period", (task.next_run_time == task.start_time + datetime.timedelta(seconds=task.period)))
]
self.exec_asserts(res, 'DRIFT')
# env check
task1 = s.task_status(env.id, output=True)
res = [
("task %s returned W2P_TASK correctly" % (task1.scheduler_task.id), task1.result == [task1.scheduler_task.id, task1.scheduler_task.uuid, task1.scheduler_run.id]),
]
self.exec_asserts(res, 'ENV')
def testRetryFailed(self):
s = Scheduler(self.db)
failed = s.queue_task('demo2', retry_failed=1, period=1)
failed_consecutive = s.queue_task('demo8', retry_failed=2, repeats=2, period=1)
self.db.commit()
self.writefunction(r"""
def demo2():
1/0
def demo8():
placeholder = os.path.join(request.folder, 'private', 'demo8.pholder')
with open(placeholder, 'a') as g:
g.write('\nplaceholder for demo8 created')
num_of_lines = 0
with open(placeholder) as f:
num_of_lines = len([a for a in f.read().split('\n') if a])
print('number of lines', num_of_lines)
if num_of_lines <= 2:
1/0
else:
os.unlink(placeholder)
return 1
""")
ret = self.exec_sched()
# process finished just fine
self.assertEqual(ret, 0)
# failed - checks
task, task_run = self.fetch_results(s, failed)
res = [
("task status failed", task.status == 'FAILED'),
("task times_run is 0", task.times_run == 0),
("task times_failed is 2", task.times_failed == 2),
("task ran 2 times only", len(task_run) == 2),
("scheduler_run records are FAILED", (task_run[0].status == task_run[1].status == 'FAILED')),
("period is respected", (task_run[1].start_time > task_run[0].start_time + datetime.timedelta(seconds=task.period)))
]
self.exec_asserts(res, 'FAILED')
# failed consecutive - checks
task, task_run = self.fetch_results(s, failed_consecutive)
res = [
("task status completed", task.status == 'COMPLETED'),
("task times_run is 2", task.times_run == 2),
("task times_failed is 0", task.times_failed == 0),
("task ran 6 times", len(task_run) == 6),
("scheduler_run records for COMPLETED is 2", len([run.status for run in task_run if run.status == 'COMPLETED']) == 2),
("scheduler_run records for FAILED is 4", len([run.status for run in task_run if run.status == 'FAILED']) == 4),
]
self.exec_asserts(res, 'FAILED_CONSECUTIVE')
def testRegressions(self):
s = Scheduler(self.db)
huge_result = s.queue_task('demo10', retry_failed=1, period=1)
issue_1485 = s.queue_task('issue_1485')
termination = s.queue_task('termination')
self.db.commit()
self.writefunction(r"""
def demo10():
res = 'a' * 99999
return dict(res=res)
def issue_1485():
return response.render('issue_1485.html', dict(variable='abc'))
""")
self.writeview(r"""<span>{{=variable}}</span>""", 'issue_1485.html')
ret = self.exec_sched()
# process finished just fine
self.assertEqual(ret, 0)
# huge_result - checks
task_huge = s.task_status(huge_result.id, output=True)
res = [
("task status completed", task_huge.scheduler_task.status == 'COMPLETED'),
("task times_run is 1", task_huge.scheduler_task.times_run == 1),
("result is the correct one", task_huge.result == dict(res='a' * 99999))
]
self.exec_asserts(res, 'HUGE_RESULT')
task_issue_1485 = s.task_status(issue_1485.id, output=True)
res = [
("task status completed", task_issue_1485.scheduler_task.status == 'COMPLETED'),
("task times_run is 1", task_issue_1485.scheduler_task.times_run == 1),
("result is the correct one", task_issue_1485.result == '<span>abc</span>')
]
self.exec_asserts(res, 'issue_1485')
if __name__ == '__main__':
unittest.main()