Skip to content

P
pylibs
Commit 7fb41afc authored by root's avatar root
Browse files
Options

Python3

parent ad97104d
Hide whitespace changes
Inline Side-by-side
Showing with 1327 additions and 90 deletions
TxScheduling/txscheduling/cron.py
View file @ 7fb41afc
......@@ -295,7 +295,7 @@ def parseCronLine(line):
if not line:
raise InvalidCronLine('Empty cron line provided')
if not isinstance(line, basestring):
if not isinstance(line, str):
raise InvalidCronLine('Cron line must be a string')
line = re.split('\s+',line.strip())
......@@ -449,7 +449,7 @@ def parseCronEntry(entry,min,max):
if not entry:
raise InvalidCronEntry('Empty cron entry')
if not isinstance(entry, basestring):
if not isinstance(entry, str):
raise InvalidCronEntry('Cron entry must be a string')
try:
......@@ -523,7 +523,7 @@ def parseCronEntry(entry,min,max):
if begin == end and begin % step != 0:
raise InvalidCronEntry('Invalid range or step specified: %s' % (e))
total.update(range(begin,end,step))
total.update(list(range(begin,end,step)))
elif not begin is None:
raise InvalidCronEntry('Invalid range or step specified: %s' % (e))
......
TxScheduling/txscheduling/cron.py.bak 0 → 100644
View file @ 7fb41afc
import time
import datetime
import re
import zope.interface
from txscheduling.interfaces import ISchedule
""" This module provides an implementation of the
txscheduling.interface.ISchedule interface along with two useful
functions for parsing cron lines. """
class CronSchedule(object):
zope.interface.implements(ISchedule)
_minutes = None
_hours = None
_doms = None # days of the month
_months = None
_dows = None # days of the week
def __init__(self,cron_line):
kwargs = parseCronLine(cron_line)
self._minutes = kwargs.get('minutes')
self._hours = kwargs.get('hours')
self._doms = kwargs.get('doms')
self._months = kwargs.get('months')
self._dows = kwargs.get('dows')
def __eq__(self,other):
if not isinstance(other,CronSchedule):
return False
return (self._minutes == other._minutes and
self._hours == other._hours and
self._doms == other._doms and
self._months == other._months and
self._dows == other._dows)
def _getNextMonth(self, current):
# If the current month is a valid option, try to parse for the next valid day
if current.month in self._months:
try:
return self._getNextDay(current)
except NoMatch:
pass
# Find the next month if it occurs in the current year
for month in self._months:
if month > current.month:
return self._getFirstDay(current.replace(month=month,
day=1,
hour=self._hours[0],
minute=self._minutes[0]))
# No remaining months this year, go to next year
return self._getFirstDay(current.replace(year=current.year+1,
month=self._months[0],
day=1,
hour=self._hours[0],
minute=self._minutes[0]))
def _getFirstDay(self,current):
all_doms = False
all_dows = False
fdom = current.replace(day=1)
if len(self._doms) == 31:
all_doms = True
if len(self._dows) == 7:
all_dows = True
# All days of the week and days of the month, return day = 1
if all_doms and all_dows:
return fdom
# All days of the week, return first day of the month
if all_dows:
return current.replace(day=self._doms[0])
dows = self._dows
if dows[0] == 0:
import copy
dows = copy.copy(dows)
del dows[0]
dows.append(7)
current_dow = fdom.isoweekday()
# If the current day of the week is in the days of the week, return
if not all_dows and current_dow in dows:
return fdom
# If the first day of the month is specifically listed, return it
if not all_doms and self._doms[0] == 1:
return fdom
dow_distance = None
dom_distance = None
distance = None
if not all_dows:
for dow in dows:
if dow > current_dow:
dow_distance = dow-current_dow
break
if dow_distance is None:
dow_distance = dows[0]+7-current_dow
if not all_doms:
dom_distance = self._doms[0] - 1
if not dow_distance is None and not dom_distance is None:
if dow_distance <= dom_distance:
distance = dow_distance
else:
distance = dom_distance
if distance is None and not dow_distance is None:
distance = dow_distance
if distance is None and not dom_distance is None:
distance = dom_distance
return current.replace(day=distance + 1)
def _getNextDay(self, current):
all_doms = False
all_dows = False
if len(self._doms) == 31:
all_doms = True
if len(self._dows) == 7:
all_dows = True
dows = self._dows
if dows[0] == 0:
import copy
dows = copy.copy(dows)
del dows[0]
dows.append(7)
# If the current day is a valid option, try to parse for the next valid hour
if ((all_doms and all_dows) or
(current.day in self._doms and all_dows) or
(current.isoweekday() in dows and not all_dows)):
try:
return self._getNextHour(current)
except NoMatch:
pass
distance = None
if all_doms and all_dows:
distance = 1
else:
dow_distance = None
dom_distance = None
if not all_dows:
current_dow = current.isoweekday()
for dow in dows:
if dow > current_dow:
dow_distance = dow-current_dow
break
if dow_distance is None:
dow_distance = dows[0]+7-current_dow
if not all_doms:
for dom in self._doms:
if dom > current.day:
dom_distance = dom - current.day
break
if not dow_distance is None and dom_distance is not None:
if dow_distance <= dom_distance:
distance = dow_distance
else:
distance = dom_distance
if distance is None and dow_distance is not None:
distance = dow_distance
if distance is None and dom_distance is not None:
distance = dom_distance
if distance is None or distance <= 0:
raise NoMatch('no matching days in month')
next_day = (current.replace(hour=self._hours[0],
minute=self._minutes[0]) +
distance * datetime.timedelta(days=1))
if next_day.month != current.month:
raise NoMatch('no remaining days in the current month')
return next_day
def _getNextHour(self, current):
if current.hour in self._hours:
try:
return self._getNextMinute(current)
except NoMatch:
pass
for hour in self._hours:
if hour > current.hour:
return current.replace(hour=hour,minute=self._minutes[0])
raise NoMatch('no remaining hours in the current day')
def _getNextMinute(self, current):
for minute in self._minutes:
if minute > current.minute:
return current.replace(minute=minute)
raise NoMatch('no remaining minutes in the current hour')
def getNextEntry(self,current=None):
if current is None:
current = datetime.datetime.now()
if not isinstance(current,datetime.datetime):
raise ValueError('current value must be a datetime.datetime object')
return self._getNextMonth(current.replace(second=0,microsecond=0))
def getDelayForNext(self):
next = self.getNextEntry()
return time.mktime(next.timetuple()) - time.time()
class InvalidCronLine(Exception):
pass
class InvalidCronEntry(Exception):
pass
class NoMatch(Exception):
pass
_cronStepRe = re.compile('^\*/(?P<step>\d{1,2})$')
_cronRangeRe = re.compile('^(?P<begin>\d{1,2})-(?P<end>\d{1,2})$')
_cronRangeStepRe = re.compile('^(?P<begin>\d{1,2})-(?P<end>\d{1,2})/(?P<step>\d{1,2})$')
def parseCronLine(line):
"""
Parse a standard cron string (minus the command) and return them as a
dictionary. The syntax for this was pulled from the
Wikipedia page: http://en.wikipedia.org/wiki/Cron
Currently, there is no support for textual days of the week
(i.e. Monday,Tuesday).
Examples:
>>> parseCronLine('* * * * *') == {
... 'hours': range(0,24),
... 'doms': range(1,32),
... 'minutes': range(0,60),
... 'dows': range(0,7),
... 'months': range(1,13)}
True
Cron lines must have 5 whitespace separated entries
>>> parseCronLine('* * * *')
Traceback (most recent call last):
...
InvalidCronLine: Improper number of elements encountered: 4
Cron lines cannot be empty
>>> parseCronLine('')
Traceback (most recent call last):
...
InvalidCronLine: Empty cron line provided
Cron lines must be a string
>>> parseCronLine(True)
Traceback (most recent call last):
...
InvalidCronLine: Cron line must be a string
"""
if not line:
raise InvalidCronLine('Empty cron line provided')
if not isinstance(line, basestring):
raise InvalidCronLine('Cron line must be a string')
line = re.split('\s+',line.strip())
if len(line) != 5:
raise InvalidCronLine('Improper number of elements encountered: %s' % len(line))
schedule = {}
schedule['minutes'] = parseCronEntry(line[0],0,59)
schedule['hours'] = parseCronEntry(line[1],0,23)
schedule['doms'] = parseCronEntry(line[2],1,31)
schedule['months'] = parseCronEntry(line[3],1,12)
schedule['dows'] = parseCronEntry(line[4],0,6)
return schedule
def parseCronEntry(entry,min,max):
"""Parse a single cron entry for something like hours or minutes from a cron
scheduling line. The given min and max are used to verify that results are
in the proper range. The following formats are supports:
* => All in the available range
*/5 => Only those values in the available range that are divisible by five
1-5 => The range of 1-5
And any combination of the above using commas to separate the entries.
Examples:
A single entry
>>> parseCronEntry('0',0,12)
[0]
A star entry
>>> parseCronEntry('*',0,3)
[0, 1, 2, 3]
A range entry
>>> parseCronEntry('3-5',0,12)
[3, 4, 5]
A frequency entry
>>> parseCronEntry('*/3',0,12)
[0, 3, 6, 9, 12]
A frequency that doesn't match the minimum
>>> parseCronEntry('*/3',1,12)
[3, 6, 9, 12]
A ranged frequency entry
>>> parseCronEntry('3-9/3',0,12)
[3, 6, 9]
A ranged frequency entry where frequency doesn't match the range
>>> parseCronEntry('2-10/3',0,12)
[3, 6, 9]
A set of entries
>>> parseCronEntry('1,3,5',1,12)
[1, 3, 5]
A set of ranges
>>> parseCronEntry('1-3,6-9',1,12)
[1, 2, 3, 6, 7, 8, 9]
A complex entry
>>> parseCronEntry('*/5,1,12-14,22-28/3',1,30)
[1, 5, 10, 12, 13, 14, 15, 20, 24, 25, 27, 30]
Minimum argument must be convertible to an integer
>>> parseCronEntry('1','a', 5000)
Traceback (most recent call last):
...
ValueError: minimum and maximum should be convertible to integers
Maximum argument must be convertible to an integer
>>> parseCronEntry('1',5, 'a')
Traceback (most recent call last):
...
ValueError: minimum and maximum should be convertible to integers
Minimum argument must not be negative
>>> parseCronEntry('1',-100,100)
Traceback (most recent call last):
...
ValueError: minimum must be non-negative
Minimum argument must be less than the maximum argument
>>> parseCronEntry('1',1,0)
Traceback (most recent call last):
...
ValueError: minimum must be less than or equal to maximum
Entries are required
>>> parseCronEntry(None, 1, 5)
Traceback (most recent call last):
...
InvalidCronEntry: Empty cron entry
>>> parseCronEntry('', 1, 5)
Traceback (most recent call last):
...
InvalidCronEntry: Empty cron entry
Entries must be a string
>>> parseCronEntry([1], 1, 5)
Traceback (most recent call last):
...
InvalidCronEntry: Cron entry must be a string
Entries must be inside of the allowed range
>>> parseCronEntry('0', 1, 5)
Traceback (most recent call last):
...
InvalidCronEntry: Value, 0-0, out of allowed range: 1-5
>>> parseCronEntry('6', 1, 5)
Traceback (most recent call last):
...
InvalidCronEntry: Value, 6-6, out of allowed range: 1-5
>>> parseCronEntry('-6', 1, 5)
Traceback (most recent call last):
...
InvalidCronEntry: Value, -6--6, out of allowed range: 1-5
Ranges must be in minimum-maximum order
>>> parseCronEntry('3-1', 1, 5)
Traceback (most recent call last):
...
InvalidCronEntry: Invalid range or step specified: 3-1
Range frequency must be valid for range size
>>> parseCronEntry('1-5/40', 1, 5)
Traceback (most recent call last):
...
InvalidCronEntry: Invalid range or step specified: 1-5/40
Negative frequencies are not allowed
>>> parseCronEntry('1-5/-2', 1, 5)
Traceback (most recent call last):
...
InvalidCronEntry: Invalid cron entry
>>> parseCronEntry('this is garbage',1,5)
Traceback (most recent call last):
...
InvalidCronEntry: Invalid cron entry
"""
if not entry:
raise InvalidCronEntry('Empty cron entry')
if not isinstance(entry, basestring):
raise InvalidCronEntry('Cron entry must be a string')
try:
min = int(min)
max = int(max)
except ValueError:
raise ValueError('minimum and maximum should be convertible to integers')
if min > max:
raise ValueError('minimum must be less than or equal to maximum')
if min < 0:
raise ValueError('minimum must be non-negative')
try:
entry = entry.split(',')
except AttributeError:
raise InvalidCronEntry('Cron entry must be a string')
total = set()
for e in entry:
try:
int_val = int(e)
except ValueError:
pass
else:
total.add(int_val)
continue
begin = None
end = None
step = 1
if e == '*':
begin = min
end = max + 1
if begin is None:
#If this match works, then it is of the form */int
match = _cronStepRe.search(e)
if not match is None:
begin = min
end = max + 1
step = int(match.group('step'))
if begin is None:
match = _cronRangeRe.search(e)
if not match is None:
begin = int(match.group('begin'))
end = int(match.group('end')) + 1
step = 1
if begin is None:
match = _cronRangeStepRe.search(e)
if not match is None:
begin = int(match.group('begin'))
end = int(match.group('end')) + 1
step = int(match.group('step'))
if (begin is not None and begin < end and step > 0 and
begin >= min and end <= max + 1):
# need to align the start properly
while begin % step != 0 and begin < end:
begin += 1
if begin == end and begin % step != 0:
raise InvalidCronEntry('Invalid range or step specified: %s' % (e))
total.update(range(begin,end,step))
elif not begin is None:
raise InvalidCronEntry('Invalid range or step specified: %s' % (e))
if len(total) == 0:
raise InvalidCronEntry('Invalid cron entry')
total = list(total)
total.sort()
if total[0] < min or total[len(total)-1] > max:
raise InvalidCronEntry('Value, %s-%s, out of allowed range: %s-%s' % (total[0],total[len(total)-1],min,max))
return total
__all__ = [
'CronSchedule',
'InvalidCronLine'
]
TxScheduling/txscheduling/task.py
View file @ 7fb41afc
......@@ -65,7 +65,7 @@ class ScheduledCall(object):
self._reschedule()
return self.deferred
except Exception, e:
except Exception as e:
log.error('Exception while starting %r: %s' % (self, str(e),))
self.running = False
self.deferred = None
......@@ -119,9 +119,9 @@ class ScheduledCall(object):
def __repr__(self):
if hasattr(self.f, 'func_name'):
func = self.f.func_name
func = self.f.__name__
if hasattr(self.f, 'im_class'):
func = self.f.im_class.__name__ + '.' + func
func = self.f.__self__.__class__.__name__ + '.' + func
else:
func = reflect.safe_repr(self.f)
......
TxScheduling/txscheduling/task.py.bak 0 → 100644
View file @ 7fb41afc
from logging import getLogger
from twisted.python import reflect
from twisted.internet import defer
from txscheduling.interfaces import ISchedule
log = getLogger('twisted.schedule.task')
class ScheduledCall(object):
"""Call a function repeatedly.
If C{f} returns a deferred, rescheduling will not take place until the
deferred has fired. The result value is ignored.
@ivar f: The function to call.
@ivar a: A tuple of arguments to pass the function.
@ivar kw: A dictionary of keyword arguments to pass to the function.
@ivar clock: A provider of
L{twisted.internet.interfaces.IReactorTime}. The default is
L{twisted.internet.reactor}. Feel free to set this to
something else, but it probably ought to be set *before*
calling L{start}.
@type _lastTime: C{float}
@ivar _lastTime: The time at which this instance most recently scheduled
itself to run.
"""
def __init__(self, f, *a, **kw):
self.call = None
self.running = False
self.scheduled = None
self._lastTime = 0.0
self.starttime = None
self.f = f
self.a = a
self.kw = kw
from twisted.internet import reactor
self.clock = reactor
def start(self, schedule):
"""Start running function based on the provided schedule.
@param schedule: An object that provides or can be adapted to an
ISchedule interface.
@return: A Deferred whose callback will be invoked with
C{self} when C{self.stop} is called, or whose errback will be
invoked when the function raises an exception or returned a
deferred that has its errback invoked.
"""
assert not self.running, ("Tried to start an already running "
"ScheduledCall.")
self.schedule = ISchedule(schedule)
try:
self.running = True
self.deferred = defer.Deferred()
self.starttime = self.clock.seconds()
self._lastTime = None
self._reschedule()
return self.deferred
except Exception, e:
log.error('Exception while starting %r: %s' % (self, str(e),))
self.running = False
self.deferred = None
if self.call is not None:
try:
self.call.cancel()
except:
pass
raise e
finally:
log.debug('%r.start(%r) started' % (self, schedule))
def stop(self):
""" Stop running function. """
assert self.running, ("Tried to stop a ScheduledCall that was "
"not running.")
self.running = False
if self.call is not None:
self.call.cancel()
self.call = None
d, self.deferred = self.deferred, None
d.callback(self)
def __call__(self):
def cb(result):
if self.running:
self._reschedule()
else:
d, self.deferred = self.deferred, None
d.callback(self)
def eb(failure):
self.running = False
d, self.deferred = self.deferred, None
d.errback(failure)
self.call = None
d = defer.maybeDeferred(self.f, *self.a, **self.kw)
d.addCallback(cb)
d.addErrback(eb)
def _reschedule(self):
""" Schedule the next iteration of this scheduled call. """
if self.call is None:
delay = self.schedule.getDelayForNext()
self._lastTime = self.clock.seconds() + delay
self.call = self.clock.callLater(delay, self)
def __repr__(self):
if hasattr(self.f, 'func_name'):
func = self.f.func_name
if hasattr(self.f, 'im_class'):
func = self.f.im_class.__name__ + '.' + func
else:
func = reflect.safe_repr(self.f)
return 'ScheduledCall<%s>(%s, *%s, **%s)' % (
self.schedule, func, reflect.safe_repr(self.a),
reflect.safe_repr(self.kw))
\ No newline at end of file
TxScheduling/txscheduling/tests/cron.py
View file @ 7fb41afc
......@@ -15,38 +15,38 @@ class StarTestCase(TestCase):
def testNextMinute(self):
""" Next runtime is next minute """
for i in range(0,59):
self.assertEqual(self.schedule.getNextEntry(datetime(2008, 01, 01,
self.assertEqual(self.schedule.getNextEntry(datetime(2008, 0o1, 0o1,
00, i, 00,
00)),
datetime(2008,01,01,00,i+1,00,00))
datetime(2008,0o1,0o1,00,i+1,00,00))
def testNextHour(self):
""" Next runtime is next hour """
for i in range(0,23):
self.assertEqual(self.schedule.getNextEntry(datetime(2008, 01, 01,
self.assertEqual(self.schedule.getNextEntry(datetime(2008, 0o1, 0o1,
i, 59, 00,
00)),
datetime(2008,01,01,i+1,00,00,00))
datetime(2008,0o1,0o1,i+1,00,00,00))
def testNextDay(self):
""" Next runtime is next day """
for i in range(1,31):
self.assertEqual(self.schedule.getNextEntry(datetime(2008, 01, i,
self.assertEqual(self.schedule.getNextEntry(datetime(2008, 0o1, i,
23, 59, 00,
00)),
datetime(2008,01,i+1,00,00,00,00))
datetime(2008,0o1,i+1,00,00,00,00))
def testNextMonth(self):
""" Next runtime is next month """
self.assertEqual(self.schedule.getNextEntry(datetime(2008, 01, 31, 23,
self.assertEqual(self.schedule.getNextEntry(datetime(2008, 0o1, 31, 23,
59, 00, 00)),
datetime(2008,02,01,00,00,00,00))
datetime(2008,0o2,0o1,00,00,00,00))
def testNextYear(self):
""" Next runtime is next year """
self.assertEqual(self.schedule.getNextEntry(datetime(2008, 12, 31, 23,
59, 00, 00)),
datetime(2009,01,01,00,00,00,00))
datetime(2009,0o1,0o1,00,00,00,00))
class RangeTestCase(TestCase):
def setUp(self):
......@@ -56,96 +56,96 @@ class RangeTestCase(TestCase):
""" Test range get next minute """
# Test minutes that should end up in the current hour at 15 minutes
for i in range(0,14):
self.assertEqual(self.schedule.getNextEntry(datetime(2008, 05, 05,
03, i, 00,
self.assertEqual(self.schedule.getNextEntry(datetime(2008, 0o5, 0o5,
0o3, i, 00,
00)),
datetime(2008,05,05,03,15,00,00))
datetime(2008,0o5,0o5,0o3,15,00,00))
# Test minutes that should end up in the current hour at i+1 minutes
for i in range(14,20):
self.assertEqual(self.schedule.getNextEntry(datetime(2008, 05, 05,
03, i, 00,
self.assertEqual(self.schedule.getNextEntry(datetime(2008, 0o5, 0o5,
0o3, i, 00,
00)),
datetime(2008,05,05,03,i+1,00,00))
datetime(2008,0o5,0o5,0o3,i+1,00,00))
# Test minutes that should end up at hour+1 and 15 minutes
for i in range(20,60):
self.assertEqual(self.schedule.getNextEntry(datetime(2008, 05, 05,
03, i, 00,
self.assertEqual(self.schedule.getNextEntry(datetime(2008, 0o5, 0o5,
0o3, i, 00,
00)),
datetime(2008,05,05,04,15,00,00))
datetime(2008,0o5,0o5,0o4,15,00,00))
def testNextHour(self):
""" Test range get next hour """
for i in range(0,3):
self.assertEqual(self.schedule.getNextEntry(datetime(2008, 05, 05,
self.assertEqual(self.schedule.getNextEntry(datetime(2008, 0o5, 0o5,
i, 59, 00,
00)),
datetime(2008,05,05,3,15,00,00))
datetime(2008,0o5,0o5,3,15,00,00))
for i in range(3,6):
self.assertEqual(self.schedule.getNextEntry(datetime(2008, 05, 05,
self.assertEqual(self.schedule.getNextEntry(datetime(2008, 0o5, 0o5,
i, 59, 00,
00)),
datetime(2008,05,05,i+1,15,00,00))
datetime(2008,0o5,0o5,i+1,15,00,00))
for i in range(6,24):
self.assertEqual(self.schedule.getNextEntry(datetime(2008, 05, 05,
self.assertEqual(self.schedule.getNextEntry(datetime(2008, 0o5, 0o5,
i, 59, 00,
00)),
datetime(2008,05,06,03,15,00,00))
datetime(2008,0o5,0o6,0o3,15,00,00))
def testNextDay(self):
""" Test range get next day """
for i in range(1,5):
self.assertEqual(self.schedule.getNextEntry(datetime(2008, 05, i,
self.assertEqual(self.schedule.getNextEntry(datetime(2008, 0o5, i,
23, 59, 00,
00)),
datetime(2008,05,5,3,15,00,00))
datetime(2008,0o5,5,3,15,00,00))
for i in range(5,10):
self.assertEqual(self.schedule.getNextEntry(datetime(2008, 05, i,
self.assertEqual(self.schedule.getNextEntry(datetime(2008, 0o5, i,
23, 59, 00,
00)),
datetime(2008,05,i+1,3,15,00,00))
datetime(2008,0o5,i+1,3,15,00,00))
for i in range(10,13):
self.assertEqual(self.schedule.getNextEntry(datetime(2008, 05, i,
self.assertEqual(self.schedule.getNextEntry(datetime(2008, 0o5, i,
23, 59, 00,
00)),
datetime(2008,05,13,3,15,00,00))
datetime(2008,0o5,13,3,15,00,00))
self.assertEqual(self.schedule.getNextEntry(datetime(2008, 05, 13, 23,
self.assertEqual(self.schedule.getNextEntry(datetime(2008, 0o5, 13, 23,
59, 00, 00)),
datetime(2008,05,14,3,15,00,00))
datetime(2008,0o5,14,3,15,00,00))
for i in range(14,20):
self.assertEqual(self.schedule.getNextEntry(datetime(2008, 05, i,
self.assertEqual(self.schedule.getNextEntry(datetime(2008, 0o5, i,
23, 59, 00,
00)),
datetime(2008,05,20,3,15,00,00))
datetime(2008,0o5,20,3,15,00,00))
self.assertEqual(self.schedule.getNextEntry(datetime(2008, 05, 20, 23,
self.assertEqual(self.schedule.getNextEntry(datetime(2008, 0o5, 20, 23,
59, 00, 00)),
datetime(2008,05,21,3,15,00,00))
datetime(2008,0o5,21,3,15,00,00))
def testNextMonth(self):
""" Test range get next month """
self.assertEqual(self.schedule.getNextEntry(datetime(2008, 01, 01, 00,
self.assertEqual(self.schedule.getNextEntry(datetime(2008, 0o1, 0o1, 00,
00, 00, 00)),
datetime(2008,05,05,03,15,00,00))
datetime(2008,0o5,0o5,0o3,15,00,00))
self.assertEqual(self.schedule.getNextEntry(datetime(2008, 05, 31, 23,
self.assertEqual(self.schedule.getNextEntry(datetime(2008, 0o5, 31, 23,
59, 00, 00)),
datetime(2008,06,03,03,15,00,00))
datetime(2008,0o6,0o3,0o3,15,00,00))
self.assertEqual(self.schedule.getNextEntry(datetime(2008, 07, 30, 23,
self.assertEqual(self.schedule.getNextEntry(datetime(2008, 0o7, 30, 23,
59, 00, 00)),
datetime(2008,8,05,03,15,00,00))
datetime(2008,8,0o5,0o3,15,00,00))
self.assertEqual(self.schedule.getNextEntry(datetime(2008, 8, 31, 23,
59, 00, 00)),
datetime(2009,05,05,03,15,00,00))
datetime(2009,0o5,0o5,0o3,15,00,00))
class AllDOWTestCase(TestCase):
def setUp(self):
......@@ -154,16 +154,16 @@ class AllDOWTestCase(TestCase):
def testNextDay(self):
""" Test all days of the week get next day """
for i in range(1,30):
self.assertEqual(self.schedule.getNextEntry(datetime(2008, 01, i,
self.assertEqual(self.schedule.getNextEntry(datetime(2008, 0o1, i,
23, 59, 00,
00)),
datetime(2008,01,(i/5)*5+5,00,00,00,00))
datetime(2008,0o1,(i/5)*5+5,00,00,00,00))
def testNextMonth(self):
""" Test all days of the week get next month """
self.assertEqual(self.schedule.getNextEntry(datetime(2008, 01, 31, 23,
self.assertEqual(self.schedule.getNextEntry(datetime(2008, 0o1, 31, 23,
59, 00, 00)),
datetime(2008,02,05,00,00,00,00))
datetime(2008,0o2,0o5,00,00,00,00))
class FillingCoverageTestCase(TestCase):
def test_getFirstDayWithSundayDOW(self):
......
TxScheduling/txscheduling/tests/cron.py.bak 0 → 100644
View file @ 7fb41afc
from datetime import datetime
from unittest import TestCase, TextTestRunner, TestSuite, TestLoader
from doctest import DocTestSuite
from txscheduling import cron
from txscheduling.cron import CronSchedule, InvalidCronLine
class StarTestCase(TestCase):
def setUp(self):
self.schedule = CronSchedule('* * * * *')
def testNextMinute(self):
""" Next runtime is next minute """
for i in range(0,59):
self.assertEqual(self.schedule.getNextEntry(datetime(2008, 01, 01,
00, i, 00,
00)),
datetime(2008,01,01,00,i+1,00,00))
def testNextHour(self):
""" Next runtime is next hour """
for i in range(0,23):
self.assertEqual(self.schedule.getNextEntry(datetime(2008, 01, 01,
i, 59, 00,
00)),
datetime(2008,01,01,i+1,00,00,00))
def testNextDay(self):
""" Next runtime is next day """
for i in range(1,31):
self.assertEqual(self.schedule.getNextEntry(datetime(2008, 01, i,
23, 59, 00,
00)),
datetime(2008,01,i+1,00,00,00,00))
def testNextMonth(self):
""" Next runtime is next month """
self.assertEqual(self.schedule.getNextEntry(datetime(2008, 01, 31, 23,
59, 00, 00)),
datetime(2008,02,01,00,00,00,00))
def testNextYear(self):
""" Next runtime is next year """
self.assertEqual(self.schedule.getNextEntry(datetime(2008, 12, 31, 23,
59, 00, 00)),
datetime(2009,01,01,00,00,00,00))
class RangeTestCase(TestCase):
def setUp(self):
self.schedule = CronSchedule('15-20 3-6 5-10 5-8 2-3')
def testNextMinute(self):
""" Test range get next minute """
# Test minutes that should end up in the current hour at 15 minutes
for i in range(0,14):
self.assertEqual(self.schedule.getNextEntry(datetime(2008, 05, 05,
03, i, 00,
00)),
datetime(2008,05,05,03,15,00,00))
# Test minutes that should end up in the current hour at i+1 minutes
for i in range(14,20):
self.assertEqual(self.schedule.getNextEntry(datetime(2008, 05, 05,
03, i, 00,
00)),
datetime(2008,05,05,03,i+1,00,00))
# Test minutes that should end up at hour+1 and 15 minutes
for i in range(20,60):
self.assertEqual(self.schedule.getNextEntry(datetime(2008, 05, 05,
03, i, 00,
00)),
datetime(2008,05,05,04,15,00,00))
def testNextHour(self):
""" Test range get next hour """
for i in range(0,3):
self.assertEqual(self.schedule.getNextEntry(datetime(2008, 05, 05,
i, 59, 00,
00)),
datetime(2008,05,05,3,15,00,00))
for i in range(3,6):
self.assertEqual(self.schedule.getNextEntry(datetime(2008, 05, 05,
i, 59, 00,
00)),
datetime(2008,05,05,i+1,15,00,00))
for i in range(6,24):
self.assertEqual(self.schedule.getNextEntry(datetime(2008, 05, 05,
i, 59, 00,
00)),
datetime(2008,05,06,03,15,00,00))
def testNextDay(self):
""" Test range get next day """
for i in range(1,5):
self.assertEqual(self.schedule.getNextEntry(datetime(2008, 05, i,
23, 59, 00,
00)),
datetime(2008,05,5,3,15,00,00))
for i in range(5,10):
self.assertEqual(self.schedule.getNextEntry(datetime(2008, 05, i,
23, 59, 00,
00)),
datetime(2008,05,i+1,3,15,00,00))
for i in range(10,13):
self.assertEqual(self.schedule.getNextEntry(datetime(2008, 05, i,
23, 59, 00,
00)),
datetime(2008,05,13,3,15,00,00))
self.assertEqual(self.schedule.getNextEntry(datetime(2008, 05, 13, 23,
59, 00, 00)),
datetime(2008,05,14,3,15,00,00))
for i in range(14,20):
self.assertEqual(self.schedule.getNextEntry(datetime(2008, 05, i,
23, 59, 00,
00)),
datetime(2008,05,20,3,15,00,00))
self.assertEqual(self.schedule.getNextEntry(datetime(2008, 05, 20, 23,
59, 00, 00)),
datetime(2008,05,21,3,15,00,00))
def testNextMonth(self):
""" Test range get next month """
self.assertEqual(self.schedule.getNextEntry(datetime(2008, 01, 01, 00,
00, 00, 00)),
datetime(2008,05,05,03,15,00,00))
self.assertEqual(self.schedule.getNextEntry(datetime(2008, 05, 31, 23,
59, 00, 00)),
datetime(2008,06,03,03,15,00,00))
self.assertEqual(self.schedule.getNextEntry(datetime(2008, 07, 30, 23,
59, 00, 00)),
datetime(2008,8,05,03,15,00,00))
self.assertEqual(self.schedule.getNextEntry(datetime(2008, 8, 31, 23,
59, 00, 00)),
datetime(2009,05,05,03,15,00,00))
class AllDOWTestCase(TestCase):
def setUp(self):
self.schedule = CronSchedule('*/15 * */5 * *')
def testNextDay(self):
""" Test all days of the week get next day """
for i in range(1,30):
self.assertEqual(self.schedule.getNextEntry(datetime(2008, 01, i,
23, 59, 00,
00)),
datetime(2008,01,(i/5)*5+5,00,00,00,00))
def testNextMonth(self):
""" Test all days of the week get next month """
self.assertEqual(self.schedule.getNextEntry(datetime(2008, 01, 31, 23,
59, 00, 00)),
datetime(2008,02,05,00,00,00,00))
class FillingCoverageTestCase(TestCase):
def test_getFirstDayWithSundayDOW(self):
""" Test next day with Sunday DOW entry """
schedule = CronSchedule('* * * * 0,3,5')
self.assertEqual(schedule.getNextEntry(datetime(2008,8,31,23,59,00,00)),
datetime(2008,9,3,00,00,00,00))
class AllDOMTestCase(TestCase):
def setUp(self):
self.schedule = CronSchedule('*/15 * * * 1,3,5')
def testNextDay(self):
""" Test all days of the month get next day """
self.assertEqual(self.schedule.getNextEntry(datetime(2008, 9, 1, 23,
59, 00, 00)),
datetime(2008,9,3,00,00,00,00))
self.assertEqual(self.schedule.getNextEntry(datetime(2008, 9, 15, 23,
59, 00, 00)),
datetime(2008,9,17,00,00,00,00))
self.assertEqual(self.schedule.getNextEntry(datetime(2008, 9, 16, 23,
59, 00, 00)),
datetime(2008,9,17,00,00,00,00))
self.assertEqual(self.schedule.getNextEntry(datetime(2008, 9, 17, 23,
59, 00, 00)),
datetime(2008,9,19,00,00,00,00))
def testNextMonth(self):
""" Test all days of the month get next month """
self.assertEqual(self.schedule.getNextEntry(datetime(2008, 6, 30, 23,
59, 00, 00)),
datetime(2008,7,2,00,00,00,00))
self.assertEqual(self.schedule.getNextEntry(datetime(2008, 8, 29, 23,
59, 00, 00)),
datetime(2008,9,1,00,00,00,00))
self.assertEqual(self.schedule.getNextEntry(datetime(2008, 9, 29, 23,
59, 00, 00)),
datetime(2008,10,1,00,00,00,00))
class SimpleTests(TestCase):
def setUp(self):
self.schedule = CronSchedule('* * * * *')
def test_invalid_cronline(self):
""" CronSchedule invalid cron line """
self.assertRaises(InvalidCronLine,CronSchedule,' ')
def test_equality(self):
""" CronSchedule equality testing """
self.assertEqual(self.schedule,CronSchedule('* * * * *'))
self.assertEqual(self.schedule == 'blah',False)
def test_invalid_call_to_get_next(self):
""" CronSchedule getNextEntry with invalid arguments """
self.assertRaises(ValueError,self.schedule.getNextEntry,'')
def test_suite():
suite = TestSuite()
suite.addTest(TestLoader().loadTestsFromTestCase(SimpleTests))
suite.addTest(TestLoader().loadTestsFromTestCase(AllDOMTestCase))
suite.addTest(TestLoader().loadTestsFromTestCase(AllDOWTestCase))
suite.addTest(TestLoader().loadTestsFromTestCase(FillingCoverageTestCase))
suite.addTest(TestLoader().loadTestsFromTestCase(RangeTestCase))
suite.addTest(TestLoader().loadTestsFromTestCase(StarTestCase))
suite.addTest(DocTestSuite(cron))
return suite
if __name__ == '__main__':
TextTestRunner(verbosity=2).run(test_suite())
\ No newline at end of file
TxScheduling/txscheduling/tests/task.py
View file @ 7fb41afc
......@@ -92,8 +92,8 @@ class CallableTests(TestCase):
self.sc = TestableScheduledCall(self.clock, f)
def errback(fail):
self.assert_(fail.check(TestException),
u'Expecting a TestException failure')
self.assertTrue(fail.check(TestException),
'Expecting a TestException failure')
def callback(result):
self.fail('Callback should not be called')
......@@ -110,8 +110,8 @@ class CallableTests(TestCase):
self.sc = TestableScheduledCall(self.clock, f)
def errback(fail):
self.assert_(fail.check(TypeError),
u'Expecting a TypeError failure')
self.assertTrue(fail.check(TypeError),
'Expecting a TypeError failure')
def callback(result):
self.fail('Callback should not be called')
......@@ -179,36 +179,36 @@ class SimpleTimingTests(TestCase):
""" No calls before scheduled delay """
self.sc.start(SimpleSchedule(1))
self.assertEqual(self.callable.count, 0,
u'Callable should not be called before time has passed')
'Callable should not be called before time has passed')
self.clock.pump([0.9])
self.assertEqual(self.callable.count, 0,
u'Callable should not be called before sufficient time '
'Callable should not be called before sufficient time '
'has passed')
self.sc.stop()
self.assertEqual(self.callable.count, 0,
u'Callable should not be called after stopping')
'Callable should not be called after stopping')
self.clock.pump([0.1]*50)
self.assertEqual(self.callable.count, 0,
u'Callable should not be called after stopping')
'Callable should not be called after stopping')
def test_calls(self):
""" Verify calls at proper times """
self.sc.start(SimpleSchedule(1))
self.assertEqual(self.callable.count, 0,
u'Callable should not be called before time has passed')
'Callable should not be called before time has passed')
self.clock.pump([0.1]*11)
self.assertEqual(self.callable.count, 1,
u'Callable should be called once now')
'Callable should be called once now')
self.clock.pump([0.1]*10)
self.assertEqual(self.callable.count, 2,
u'Callable should be called twice now')
'Callable should be called twice now')
self.sc.stop()
self.assertEqual(self.callable.count, 2,
u'Callable should not be called after stopping')
'Callable should not be called after stopping')
self.clock.pump([0.1]*50)
self.assertEqual(self.callable.count, 2,
u'Callable should not be called after stopping')
'Callable should not be called after stopping')
class LongRunningTimingTests(TestCase):
......@@ -227,86 +227,86 @@ class LongRunningTimingTests(TestCase):
""" No calls before scheduled delay on long running task """
self.sc.start(SimpleSchedule(1))
self.assertEqual(self.callable.count, 0,
u'Callable should not be called before time has passed')
'Callable should not be called before time has passed')
self.assertEqual(self.started.count, 0,
u'Callable should not be called before time has passed')
'Callable should not be called before time has passed')
self.clock.pump([0.9])
self.assertEqual(self.callable.count, 0,
u'Callable should not be called before sufficient time '
'Callable should not be called before sufficient time '
'has passed')
self.assertEqual(self.started.count, 0,
u'Callable should not be called before sufficient time '
'Callable should not be called before sufficient time '
'has passed')
self.sc.stop()
self.assertEqual(self.callable.count, 0,
u'Callable should not be called after stopping')
'Callable should not be called after stopping')
self.assertEqual(self.started.count, 0,
u'Callable should not be called after stopping')
'Callable should not be called after stopping')
self.clock.pump([0.1]*50)
self.assertEqual(self.callable.count, 0,
u'Callable should not be called after stopping')
'Callable should not be called after stopping')
self.assertEqual(self.started.count, 0,
u'Callable should not be called after stopping')
'Callable should not be called after stopping')
def test_calls(self):
""" Verify calls at proper times on long running task """
self.sc.start(SimpleSchedule(2))
self.assertEqual(self.callable.count, 0,
u'Callable should not be called before time has passed')
'Callable should not be called before time has passed')
self.assertEqual(self.started.count, 0,
u'Callable should not be called before time has passed')
'Callable should not be called before time has passed')
# Time will be approx. 2.1
self.clock.pump([0.1]*21)
self.assertEqual(self.callable.count, 0,
u'Callable should be running now: %f' % (self.clock.rightNow,))
'Callable should be running now: %f' % (self.clock.rightNow,))
self.assertEqual(self.started.count, 1,
u'Callable should be running now: %f' % (self.clock.rightNow,))
'Callable should be running now: %f' % (self.clock.rightNow,))
# Time will be approx. 3.2
self.clock.pump([0.1]*11)
self.assertEqual(self.callable.count, 1,
u'Callable should be called once now: %f' % (self.clock.rightNow,))
'Callable should be called once now: %f' % (self.clock.rightNow,))
self.assertEqual(self.started.count, 1,
u'Callable should be called once now: %f' % (self.clock.rightNow,))
'Callable should be called once now: %f' % (self.clock.rightNow,))
# Time will be approx. 3.9
self.clock.pump([0.1]*7)
self.assertEqual(self.callable.count, 1,
u'Callable should be called once now: %f' % (self.clock.rightNow,))
'Callable should be called once now: %f' % (self.clock.rightNow,))
self.assertEqual(self.started.count, 1,
u'Callable should be called once now: %f' % (self.clock.rightNow,))
'Callable should be called once now: %f' % (self.clock.rightNow,))
# Time will be approx 5.2
self.clock.pump([0.1]*13)
self.assertEqual(self.callable.count, 1,
u'Callable should be running now: %f' % (self.clock.rightNow,))
'Callable should be running now: %f' % (self.clock.rightNow,))
self.assertEqual(self.started.count, 2,
u'Callable should be running now: %f' % (self.clock.rightNow,))
'Callable should be running now: %f' % (self.clock.rightNow,))
# Time will be approx. 6.4
self.clock.pump([0.1]*12)
self.assertEqual(self.callable.count, 2,
u'Callable should be called twice now: %f' % (self.clock.rightNow,))
'Callable should be called twice now: %f' % (self.clock.rightNow,))
self.assertEqual(self.started.count, 2,
u'Callable should be called twice now: %f' % (self.clock.rightNow,))
'Callable should be called twice now: %f' % (self.clock.rightNow,))
self.sc.stop()
self.assertEqual(self.callable.count, 2,
u'Callable should not be called after stopping: %f' % (self.clock.rightNow,))
'Callable should not be called after stopping: %f' % (self.clock.rightNow,))
self.assertEqual(self.started.count, 2,
u'Callable should not be called after stopping: %f' % (self.clock.rightNow,))
'Callable should not be called after stopping: %f' % (self.clock.rightNow,))
self.clock.pump([0.1]*50)
self.assertEqual(self.callable.count, 2,
u'Callable should not be called after stopping: %f' % (self.clock.rightNow,))
'Callable should not be called after stopping: %f' % (self.clock.rightNow,))
self.assertEqual(self.started.count, 2,
u'Callable should not be called after stopping: %f' % (self.clock.rightNow,))
'Callable should not be called after stopping: %f' % (self.clock.rightNow,))
def test_suite():
suite = unittest.TestSuite()
......
TxScheduling/txscheduling/tests/task.py.bak 0 → 100644
View file @ 7fb41afc
import unittest
import zope.interface
from twisted.trial.unittest import TestCase
from twisted.internet import interfaces, task, reactor, defer, error
from twisted.python import failure
from txscheduling.task import ScheduledCall
from txscheduling.interfaces import ISchedule
class TestableScheduledCall(ScheduledCall):
""" A ScheduledCall with a different clock """
def __init__(self, clock, *a, **kw):
ScheduledCall.__init__(self,*a, **kw)
self.clock = clock
class SimpleSchedule(object):
""" A schedule that always returns the same delay for the next call """
zope.interface.implements(ISchedule)
def __init__(self,delay=1):
self._delay = delay
def getDelayForNext(self):
return self._delay
class IncrementingCallable(object):
def __init__(self):
self.count = 0
def __call__(self):
self.count += 1
class TestException(Exception):
pass
class SimpleTests(TestCase):
""" Tests to verify basic behavior """
def setUp(self):
super(SimpleTests, self).setUp()
self.sc = ScheduledCall(f=lambda: None)
def test_clock(self):
""" Test that the clock used is the reactor """
self.assertEqual(self.sc.clock, reactor)
def test_stoppingStopped(self):
""" Test an invalid stop call to a stopped ScheduledCall """
self.assertRaises(AssertionError, self.sc.stop)
def test_startingStarted(self):
""" Test an invalid start call to a running ScheduledCall """
self.sc.start(SimpleSchedule())
self.assertRaises(AssertionError,self.sc.start,SimpleSchedule())
self.sc.stop()
def test_startingInvalidSchedule(self):
""" Test calling start with an invalid schedule """
self.assertRaises(TypeError, self.sc.start,'this is garbage')
def test_testStopCallback(self):
""" Test start deferred callback on stop """
def cb(result):
self.assertIdentical(result, self.sc)
d = self.sc.start(SimpleSchedule())
d.addCallback(cb)
self.sc.stop()
class CallableTests(TestCase):
""" Run tests to verify that the error handling works properly when the
callable has problems and that arguments are passed through properly """
def setUp(self):
super(CallableTests, self).setUp()
self.clock = task.Clock()
def test_callable_exception(self):
""" Test error back call raises exception """
def f(*args, **kwargs):
raise TestException('broken')
self.sc = TestableScheduledCall(self.clock, f)
def errback(fail):
self.assert_(fail.check(TestException),
u'Expecting a TestException failure')
def callback(result):
self.fail('Callback should not be called')
d = self.sc.start(SimpleSchedule(0.1))
d.addCallbacks(callback, errback)
self.clock.pump([0.1]*3)
def test_callable_invalid_arguments(self):
""" Test error back call invalid arguments """
def f(required, *args, **kwargs):
raise TestException('broken')
self.sc = TestableScheduledCall(self.clock, f)
def errback(fail):
self.assert_(fail.check(TypeError),
u'Expecting a TypeError failure')
def callback(result):
self.fail('Callback should not be called')
d = self.sc.start(SimpleSchedule(0.1))
d.addCallbacks(callback, errback)
self.clock.pump([0.1]*3)
def test_postional_arguments(self):
""" Test that positional arguments are passed through properly """
def f(*args, **kwargs):
self.assertEqual(len(args),3)
self.assertEqual(len(kwargs),0)
self.assertEqual(args[0],'a')
self.assertEqual(args[1],True)
self.assertEqual(args[2],'c')
self.sc = TestableScheduledCall(self.clock, f, 'a', True, 'c')
d = self.sc.start(SimpleSchedule(0.1))
self.clock.pump([0.1,0.05])
self.sc.stop()
def test_named_arguments(self):
""" Test that named arguments are passed through properly """
def f(*args, **kwargs):
self.assertEqual(len(args),0)
self.assertEqual(len(kwargs),3)
self.assertEqual(kwargs['kw1'],'a')
self.assertEqual(kwargs['kw2'],True)
self.assertEqual(kwargs['kw3'],'c')
self.sc = TestableScheduledCall(self.clock, f, kw1='a', kw3='c', kw2=True,)
d = self.sc.start(SimpleSchedule(0.1))
self.clock.pump([0.1,0.05])
self.sc.stop()
def test_mixed_arguments(self):
""" Test that mixed arguments are passed through properly """
def f(*args, **kwargs):
self.assertEqual(len(args),2)
self.assertEqual(len(kwargs),3)
self.assertEqual(args[0], 'p1')
self.assertEqual(args[1], False)
self.assertEqual(kwargs['kw1'],'a')
self.assertEqual(kwargs['kw2'],True)
self.assertEqual(kwargs['kw3'],'c')
self.sc = TestableScheduledCall(self.clock, f, 'p1', False, kw1='a',
kw3='c', kw2=True)
d = self.sc.start(SimpleSchedule(0.1))
self.clock.pump([0.1,0.05])
self.sc.stop()
class SimpleTimingTests(TestCase):
""" Tests to verify simple timing behavior """
def setUp(self):
super(SimpleTimingTests, self).setUp()
self.clock = task.Clock()
self.callable = IncrementingCallable()
self.sc = TestableScheduledCall(self.clock, self.callable)
def test_no_calls(self):
""" No calls before scheduled delay """
self.sc.start(SimpleSchedule(1))
self.assertEqual(self.callable.count, 0,
u'Callable should not be called before time has passed')
self.clock.pump([0.9])
self.assertEqual(self.callable.count, 0,
u'Callable should not be called before sufficient time '
'has passed')
self.sc.stop()
self.assertEqual(self.callable.count, 0,
u'Callable should not be called after stopping')
self.clock.pump([0.1]*50)
self.assertEqual(self.callable.count, 0,
u'Callable should not be called after stopping')
def test_calls(self):
""" Verify calls at proper times """
self.sc.start(SimpleSchedule(1))
self.assertEqual(self.callable.count, 0,
u'Callable should not be called before time has passed')
self.clock.pump([0.1]*11)
self.assertEqual(self.callable.count, 1,
u'Callable should be called once now')
self.clock.pump([0.1]*10)
self.assertEqual(self.callable.count, 2,
u'Callable should be called twice now')
self.sc.stop()
self.assertEqual(self.callable.count, 2,
u'Callable should not be called after stopping')
self.clock.pump([0.1]*50)
self.assertEqual(self.callable.count, 2,
u'Callable should not be called after stopping')
class LongRunningTimingTests(TestCase):
def deferredCallable(self, *args, **kwargs):
self.started()
return task.deferLater(self.clock, 1.1, self.callable, *args, **kwargs)
def setUp(self):
super(LongRunningTimingTests, self).setUp()
self.clock = task.Clock()
self.callable = IncrementingCallable()
self.started = IncrementingCallable()
self.sc = TestableScheduledCall(self.clock, self.deferredCallable)
def test_no_calls(self):
""" No calls before scheduled delay on long running task """
self.sc.start(SimpleSchedule(1))
self.assertEqual(self.callable.count, 0,
u'Callable should not be called before time has passed')
self.assertEqual(self.started.count, 0,
u'Callable should not be called before time has passed')
self.clock.pump([0.9])
self.assertEqual(self.callable.count, 0,
u'Callable should not be called before sufficient time '
'has passed')
self.assertEqual(self.started.count, 0,
u'Callable should not be called before sufficient time '
'has passed')
self.sc.stop()
self.assertEqual(self.callable.count, 0,
u'Callable should not be called after stopping')
self.assertEqual(self.started.count, 0,
u'Callable should not be called after stopping')
self.clock.pump([0.1]*50)
self.assertEqual(self.callable.count, 0,
u'Callable should not be called after stopping')
self.assertEqual(self.started.count, 0,
u'Callable should not be called after stopping')
def test_calls(self):
""" Verify calls at proper times on long running task """
self.sc.start(SimpleSchedule(2))
self.assertEqual(self.callable.count, 0,
u'Callable should not be called before time has passed')
self.assertEqual(self.started.count, 0,
u'Callable should not be called before time has passed')
# Time will be approx. 2.1
self.clock.pump([0.1]*21)
self.assertEqual(self.callable.count, 0,
u'Callable should be running now: %f' % (self.clock.rightNow,))
self.assertEqual(self.started.count, 1,
u'Callable should be running now: %f' % (self.clock.rightNow,))
# Time will be approx. 3.2
self.clock.pump([0.1]*11)
self.assertEqual(self.callable.count, 1,
u'Callable should be called once now: %f' % (self.clock.rightNow,))
self.assertEqual(self.started.count, 1,
u'Callable should be called once now: %f' % (self.clock.rightNow,))
# Time will be approx. 3.9
self.clock.pump([0.1]*7)
self.assertEqual(self.callable.count, 1,
u'Callable should be called once now: %f' % (self.clock.rightNow,))
self.assertEqual(self.started.count, 1,
u'Callable should be called once now: %f' % (self.clock.rightNow,))
# Time will be approx 5.2
self.clock.pump([0.1]*13)
self.assertEqual(self.callable.count, 1,
u'Callable should be running now: %f' % (self.clock.rightNow,))
self.assertEqual(self.started.count, 2,
u'Callable should be running now: %f' % (self.clock.rightNow,))
# Time will be approx. 6.4
self.clock.pump([0.1]*12)
self.assertEqual(self.callable.count, 2,
u'Callable should be called twice now: %f' % (self.clock.rightNow,))
self.assertEqual(self.started.count, 2,
u'Callable should be called twice now: %f' % (self.clock.rightNow,))
self.sc.stop()
self.assertEqual(self.callable.count, 2,
u'Callable should not be called after stopping: %f' % (self.clock.rightNow,))
self.assertEqual(self.started.count, 2,
u'Callable should not be called after stopping: %f' % (self.clock.rightNow,))
self.clock.pump([0.1]*50)
self.assertEqual(self.callable.count, 2,
u'Callable should not be called after stopping: %f' % (self.clock.rightNow,))
self.assertEqual(self.started.count, 2,
u'Callable should not be called after stopping: %f' % (self.clock.rightNow,))
def test_suite():
suite = unittest.TestSuite()
suite.addTest(unittest.TestLoader().loadTestsFromTestCase(SimpleTests))
suite.addTest(unittest.TestLoader().loadTestsFromTestCase(CallableTests))
suite.addTest(unittest.TestLoader().loadTestsFromTestCase(SimpleTimingTests))
suite.addTest(unittest.TestLoader().loadTestsFromTestCase(LongRunningTimingTests))
return suite
if __name__ == '__main__':
unittest.TextTestRunner(verbosity=2).run(test_suite())
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment