-
Notifications
You must be signed in to change notification settings - Fork 2
/
timing.py
153 lines (118 loc) · 6.15 KB
/
timing.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
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
# coding: utf8
# Copyright 2013-2018 Vincent Jacques <vincent@vincent-jacques.net>
from __future__ import division, absolute_import, print_function
import unittest
from ActionTree import *
from . import *
class TimingTestCase(ActionTreeTestCase):
def test_success(self):
a = self._action("a")
report = execute(a)
self.assertIsInstance(report.get_action_status(a).pending_time, datetime.datetime)
self.assertEqual(report.get_action_status(a).ready_time, report.get_action_status(a).pending_time)
self.assertIsNone(report.get_action_status(a).cancel_time)
self.assertEqual(report.get_action_status(a).start_time, report.get_action_status(a).ready_time)
self.assertGreater(report.get_action_status(a).success_time, report.get_action_status(a).start_time)
self.assertIsNone(report.get_action_status(a).failure_time)
def test_failure(self):
a = self._action("a", exception=Exception())
report = execute(a, do_raise=False)
self.assertIsInstance(report.get_action_status(a).pending_time, datetime.datetime)
self.assertEqual(report.get_action_status(a).ready_time, report.get_action_status(a).pending_time)
self.assertIsNone(report.get_action_status(a).cancel_time)
self.assertEqual(report.get_action_status(a).start_time, report.get_action_status(a).ready_time)
self.assertIsNone(report.get_action_status(a).success_time)
self.assertGreater(report.get_action_status(a).failure_time, report.get_action_status(a).start_time)
def test_cancelation_before_ready(self):
a = self._action("a")
b = self._action("b", exception=Exception())
a.add_dependency(b)
report = execute(a, do_raise=False)
self.assertIsInstance(report.get_action_status(b).pending_time, datetime.datetime)
self.assertEqual(report.get_action_status(b).ready_time, report.get_action_status(b).pending_time)
self.assertIsNone(report.get_action_status(b).cancel_time)
self.assertEqual(report.get_action_status(b).start_time, report.get_action_status(b).ready_time)
self.assertIsNone(report.get_action_status(b).success_time)
self.assertGreater(report.get_action_status(b).failure_time, report.get_action_status(b).start_time)
self.assertIsInstance(report.get_action_status(a).pending_time, datetime.datetime)
self.assertIsNone(report.get_action_status(a).ready_time)
self.assertEqual(report.get_action_status(a).cancel_time, report.get_action_status(b).failure_time)
self.assertIsNone(report.get_action_status(a).start_time)
self.assertIsNone(report.get_action_status(a).success_time)
self.assertIsNone(report.get_action_status(a).failure_time)
def test_cancelation_with_keep_going(self):
a = self._action("a")
b = self._action("b")
a.add_dependency(b)
c = self._action("c", exception=Exception())
b.add_dependency(c)
report = execute(a, keep_going=True, do_raise=False)
self.assertEqual(report.get_action_status(b).cancel_time, report.get_action_status(c).failure_time)
self.assertEqual(report.get_action_status(a).cancel_time, report.get_action_status(b).cancel_time)
def test_leaves_have_same_ready_time(self):
a = self._action("a")
b = self._action("b")
c = self._action("c")
d = self._action("d")
a.add_dependency(b)
a.add_dependency(c)
a.add_dependency(d)
report = execute(a)
self.assertEqual(report.get_action_status(c).ready_time, report.get_action_status(b).ready_time)
self.assertEqual(report.get_action_status(d).ready_time, report.get_action_status(b).ready_time)
def test_many_dependencies_with_unlimited_cpu_cores(self):
MANY = 20
a = self._action("a")
deps = [self._action(str(i)) for i in range(MANY)]
for dep in deps:
a.add_dependency(dep)
report = execute(a, cpu_cores=UNLIMITED)
for dep in deps[1:]:
self.assertEqual(report.get_action_status(dep).start_time, report.get_action_status(deps[0]).start_time)
def test_many_dependencies_with_one_cpu_cores(self):
MANY = 20
a = self._action("a")
deps = [self._action(str(i)) for i in range(MANY)]
for dep in deps:
a.add_dependency(dep)
report = execute(a, cpu_cores=1)
# No two actions have started at the same time
start_times = set(report.get_action_status(dep).start_time for dep in deps)
self.assertEqual(len(start_times), MANY)
def test_many_dependencies_with_limited_cpu_cores(self):
MANY = 20
a = self._action("a")
deps = [self._action(str(i)) for i in range(MANY)]
for dep in deps:
a.add_dependency(dep)
report = execute(a, cpu_cores=3)
# Only the first three actions have started at the same time
start_times = set(report.get_action_status(dep).start_time for dep in deps)
self.assertEqual(len(start_times), MANY - 2)
def test_scarce_resource_with_many_cpu_cores(self):
r = Resource(1)
a = self._action("a")
b = self._action("b")
b.require_resource(r, 1)
c = self._action("c")
c.require_resource(r, 1)
a.add_dependency(b)
a.add_dependency(c)
report = execute(a, cpu_cores=6)
# @todo Start next action at the same timestamp
self.assertTrue(
report.get_action_status(b).start_time > report.get_action_status(c).success_time or
report.get_action_status(c).start_time > report.get_action_status(b).success_time
)
def test_abundant_resource_with_many_cpu_cores(self):
r = Resource(2)
a = self._action("a")
b = self._action("b")
b.require_resource(r, 1)
c = self._action("c")
c.require_resource(r, 1)
a.add_dependency(b)
a.add_dependency(c)
report = execute(a, cpu_cores=6)
self.assertEqual(report.get_action_status(c).ready_time, report.get_action_status(b).ready_time)
self.assertEqual(report.get_action_status(c).start_time, report.get_action_status(b).start_time)