ยปCore Development>Code coverage>Lib/test/test_sched.py

# Python code coverage for Lib/test/test_sched.py

#countcontent
1n/aimport queue
2n/aimport sched
3n/aimport time
4n/aimport unittest
5n/atry:
7n/aexcept ImportError:
9n/a
10n/aTIMEOUT = 10
11n/a
12n/a
13n/aclass Timer:
14n/a def __init__(self):
16n/a self._time = 0
17n/a self._stop = 0
18n/a
19n/a def time(self):
20n/a with self._cond:
21n/a return self._time
22n/a
23n/a # increase the time but not beyond the established limit
24n/a def sleep(self, t):
25n/a assert t >= 0
26n/a with self._cond:
27n/a t += self._time
28n/a while self._stop < t:
29n/a self._time = self._stop
30n/a self._cond.wait()
31n/a self._time = t
32n/a
33n/a # advance time limit for user code
35n/a assert t >= 0
36n/a with self._cond:
37n/a self._stop += t
38n/a self._cond.notify_all()
39n/a
40n/a
41n/aclass TestCase(unittest.TestCase):
42n/a
43n/a def test_enter(self):
44n/a l = []
45n/a fun = lambda x: l.append(x)
46n/a scheduler = sched.scheduler(time.time, time.sleep)
47n/a for x in [0.5, 0.4, 0.3, 0.2, 0.1]:
48n/a z = scheduler.enter(x, 1, fun, (x,))
49n/a scheduler.run()
50n/a self.assertEqual(l, [0.1, 0.2, 0.3, 0.4, 0.5])
51n/a
52n/a def test_enterabs(self):
53n/a l = []
54n/a fun = lambda x: l.append(x)
55n/a scheduler = sched.scheduler(time.time, time.sleep)
56n/a for x in [0.05, 0.04, 0.03, 0.02, 0.01]:
57n/a z = scheduler.enterabs(x, 1, fun, (x,))
58n/a scheduler.run()
59n/a self.assertEqual(l, [0.01, 0.02, 0.03, 0.04, 0.05])
60n/a
62n/a def test_enter_concurrent(self):
63n/a q = queue.Queue()
64n/a fun = q.put
65n/a timer = Timer()
66n/a scheduler = sched.scheduler(timer.time, timer.sleep)
67n/a scheduler.enter(1, 1, fun, (1,))
68n/a scheduler.enter(3, 1, fun, (3,))
70n/a t.start()
72n/a self.assertEqual(q.get(timeout=TIMEOUT), 1)
73n/a self.assertTrue(q.empty())
74n/a for x in [4, 5, 2]:
75n/a z = scheduler.enter(x - 1, 1, fun, (x,))
77n/a self.assertEqual(q.get(timeout=TIMEOUT), 2)
78n/a self.assertEqual(q.get(timeout=TIMEOUT), 3)
79n/a self.assertTrue(q.empty())
81n/a self.assertEqual(q.get(timeout=TIMEOUT), 4)
82n/a self.assertTrue(q.empty())
84n/a self.assertEqual(q.get(timeout=TIMEOUT), 5)
85n/a self.assertTrue(q.empty())
87n/a t.join(timeout=TIMEOUT)
88n/a self.assertFalse(t.is_alive())
89n/a self.assertTrue(q.empty())
90n/a self.assertEqual(timer.time(), 5)
91n/a
92n/a def test_priority(self):
93n/a l = []
94n/a fun = lambda x: l.append(x)
95n/a scheduler = sched.scheduler(time.time, time.sleep)
96n/a for priority in [1, 2, 3, 4, 5]:
97n/a z = scheduler.enterabs(0.01, priority, fun, (priority,))
98n/a scheduler.run()
99n/a self.assertEqual(l, [1, 2, 3, 4, 5])
100n/a
101n/a def test_cancel(self):
102n/a l = []
103n/a fun = lambda x: l.append(x)
104n/a scheduler = sched.scheduler(time.time, time.sleep)
105n/a now = time.time()
106n/a event1 = scheduler.enterabs(now + 0.01, 1, fun, (0.01,))
107n/a event2 = scheduler.enterabs(now + 0.02, 1, fun, (0.02,))
108n/a event3 = scheduler.enterabs(now + 0.03, 1, fun, (0.03,))
109n/a event4 = scheduler.enterabs(now + 0.04, 1, fun, (0.04,))
110n/a event5 = scheduler.enterabs(now + 0.05, 1, fun, (0.05,))
111n/a scheduler.cancel(event1)
112n/a scheduler.cancel(event5)
113n/a scheduler.run()
114n/a self.assertEqual(l, [0.02, 0.03, 0.04])
115n/a
117n/a def test_cancel_concurrent(self):
118n/a q = queue.Queue()
119n/a fun = q.put
120n/a timer = Timer()
121n/a scheduler = sched.scheduler(timer.time, timer.sleep)
122n/a now = timer.time()
123n/a event1 = scheduler.enterabs(now + 1, 1, fun, (1,))
124n/a event2 = scheduler.enterabs(now + 2, 1, fun, (2,))
125n/a event4 = scheduler.enterabs(now + 4, 1, fun, (4,))
126n/a event5 = scheduler.enterabs(now + 5, 1, fun, (5,))
127n/a event3 = scheduler.enterabs(now + 3, 1, fun, (3,))
129n/a t.start()
131n/a self.assertEqual(q.get(timeout=TIMEOUT), 1)
132n/a self.assertTrue(q.empty())
133n/a scheduler.cancel(event2)
134n/a scheduler.cancel(event5)
136n/a self.assertTrue(q.empty())
138n/a self.assertEqual(q.get(timeout=TIMEOUT), 3)
139n/a self.assertTrue(q.empty())
141n/a self.assertEqual(q.get(timeout=TIMEOUT), 4)
142n/a self.assertTrue(q.empty())
144n/a t.join(timeout=TIMEOUT)
145n/a self.assertFalse(t.is_alive())
146n/a self.assertTrue(q.empty())
147n/a self.assertEqual(timer.time(), 4)
148n/a
149n/a def test_empty(self):
150n/a l = []
151n/a fun = lambda x: l.append(x)
152n/a scheduler = sched.scheduler(time.time, time.sleep)
153n/a self.assertTrue(scheduler.empty())
154n/a for x in [0.05, 0.04, 0.03, 0.02, 0.01]:
155n/a z = scheduler.enterabs(x, 1, fun, (x,))
156n/a self.assertFalse(scheduler.empty())
157n/a scheduler.run()
158n/a self.assertTrue(scheduler.empty())
159n/a
160n/a def test_queue(self):
161n/a l = []
162n/a fun = lambda x: l.append(x)
163n/a scheduler = sched.scheduler(time.time, time.sleep)
164n/a now = time.time()
165n/a e5 = scheduler.enterabs(now + 0.05, 1, fun)
166n/a e1 = scheduler.enterabs(now + 0.01, 1, fun)
167n/a e2 = scheduler.enterabs(now + 0.02, 1, fun)
168n/a e4 = scheduler.enterabs(now + 0.04, 1, fun)
169n/a e3 = scheduler.enterabs(now + 0.03, 1, fun)
170n/a # queue property is supposed to return an order list of
171n/a # upcoming events
172n/a self.assertEqual(scheduler.queue, [e1, e2, e3, e4, e5])
173n/a
174n/a def test_args_kwargs(self):
175n/a seq = []
176n/a def fun(*a, **b):
177n/a seq.append((a, b))
178n/a
179n/a now = time.time()
180n/a scheduler = sched.scheduler(time.time, time.sleep)
181n/a scheduler.enterabs(now, 1, fun)
182n/a scheduler.enterabs(now, 1, fun, argument=(1, 2))
183n/a scheduler.enterabs(now, 1, fun, argument=('a', 'b'))
184n/a scheduler.enterabs(now, 1, fun, argument=(1, 2), kwargs={"foo": 3})
185n/a scheduler.run()
186n/a self.assertCountEqual(seq, [
187n/a ((), {}),
188n/a ((1, 2), {}),
189n/a (('a', 'b'), {}),
190n/a ((1, 2), {'foo': 3})
191n/a ])
192n/a
193n/a def test_run_non_blocking(self):
194n/a l = []
195n/a fun = lambda x: l.append(x)
196n/a scheduler = sched.scheduler(time.time, time.sleep)
197n/a for x in [10, 9, 8, 7, 6]:
198n/a scheduler.enter(x, 1, fun, (x,))
199n/a scheduler.run(blocking=False)
200n/a self.assertEqual(l, [])
201n/a
202n/a
203n/aif __name__ == "__main__":
204n/a unittest.main()