-
Notifications
You must be signed in to change notification settings - Fork 2
/
executiontime_calculator.py
122 lines (93 loc) · 4.02 KB
/
executiontime_calculator.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
from datetime import datetime
from datetime import timedelta
from job import WEEKDAYS
from job import Job
def calculateNextExecution(job, now=datetime.now()):
executionTime = now.replace()
if job.executionType == "weekly":
diff = WEEKDAYS.index(job.executionDay) - now.weekday()
if diff < 0 and now.day < (-1 * diff):
diff += now.day
executionTime.replace(month=executionTime.month - 1)
executionTime = executionTime.replace(day=now.day + diff)
elif job.executionType == "monthly":
executionTime = executionTime.replace(day=job.executionDay)
# add the calculated difference
executionTime = executionTime.replace(hour=job.executionTime.hour,
minute=job.executionTime.minute,
second=job.executionTime.second,
microsecond=job.executionTime.microsecond)
addition = timedelta()
if now > executionTime:
#add interval
if job.executionType == "daily":
addition = timedelta(days=1)
if job.executionType == "weekly":
addition = timedelta(weeks=1)
elif job.executionType == "monthly":
if executionTime.month < 12:
executionTime = executionTime.replace(month=executionTime.month + 1)
else:
executionTime = executionTime.replace(month=1)
# add the delta
executionTime = executionTime + addition
# set the next execution date on the job
job.nextExecution = executionTime
def test():
test_monthly()
test_weekly()
testDaily()
return
def test_monthly():
print "TEST Monthly"
job = Job("testRef")
now = datetime(2014, 2, 2, 10)
# execute later
addition = timedelta(hours=1)
job.setExecution("monthly", (now + addition).time(), now.day)
calculateNextExecution(job, now)
assert datetime(2014, 2, 2, 11) == job.nextExecution, "Calculated wrong execution date: %s"\
% str(job.nextExecution)
# execute tomorrow
addition = timedelta(hours=-1)
job.setExecution("monthly", (now + addition).time(), now.day)
calculateNextExecution(job, now)
assert datetime(2014, 3, 2, 9) == job.nextExecution, "Calculated wrong execution date: %s"\
% str(job.nextExecution)
print "OK"
def test_weekly():
print "TEST Weekly"
job = Job("testRef")
now = datetime(2014, 2, 2, 10)
# execute later
addition = timedelta(hours=1)
job.setExecution("weekly", (now + addition).time(), "So")
calculateNextExecution(job, now)
assert datetime(2014, 2, 2, 11) == job.nextExecution, "Calculated wrong execution date: %s"\
% str(job.nextExecution)
# execute tomorrow
addition = timedelta(hours=-1)
job.setExecution("weekly", (now + addition).time(), "So")
calculateNextExecution(job, now)
assert datetime(2014, 2, 9, 9) == job.nextExecution, "Calculated wrong execution date: %s"\
% str(job.nextExecution)
print "OK"
def testDaily():
print "TEST Daily"
job = Job("testRef")
now = datetime(2014, 2, 2, 10)
# execute later
addition = timedelta(hours=1)
job.setExecution("daily", (now + addition).time())
calculateNextExecution(job, now)
assert datetime(2014, 2, 2, 11) == job.nextExecution, "Calculated wrong execution date: %s"\
% str(job.nextExecution)
# execute tomorrow
addition = timedelta(hours=-1)
job.setExecution("daily", (now + addition).time())
calculateNextExecution(job, now)
assert datetime(2014, 2, 3, 9) == job.nextExecution, "Calculated wrong execution date: %s"\
% str(job.nextExecution)
print "OK"
if __name__ == '__main__':
test()