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
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
|
import unittest
from tests import mock
import pyglet.clock
class ClockTestCase(unittest.TestCase):
"""Test clock using dummy time keeper
not tested:
positional and named arguments
"""
def setUp(self):
self.interval = .001
self.time = 0
self.callback_a = mock.Mock()
self.callback_b = mock.Mock()
self.callback_c = mock.Mock()
self.callback_d = mock.Mock()
self.clock = pyglet.clock.Clock(time_function=lambda: self.time)
def advance_clock(self, dt=1):
"""simulate the passage of time like a real clock would"""
frames = 0
end = self.time + dt
while self.time < end:
frames += 1
self.time += self.interval
self.clock.tick()
self.time = round(self.time, 0)
return frames
def test_schedule(self):
self.clock.schedule(self.callback_a)
frames = self.advance_clock()
self.assertEqual(self.callback_a.call_count, frames)
def test_schedule_once(self):
self.clock.schedule_once(self.callback_a, 1)
self.advance_clock(2)
self.assertEqual(self.callback_a.call_count, 1)
def test_schedule_once_multiple(self):
self.clock.schedule_once(self.callback_a, 1)
self.clock.schedule_once(self.callback_b, 2)
self.advance_clock(2)
self.assertEqual(self.callback_a.call_count, 1)
self.assertEqual(self.callback_b.call_count, 1)
def test_schedule_interval(self):
self.clock.schedule_interval(self.callback_a, 1)
self.advance_clock(2)
self.assertEqual(self.callback_a.call_count, 2)
def test_schedule_interval_for_duration(self):
self.clock.schedule_interval_for_duration(self.callback_a, 1, 5)
self.advance_clock(10)
self.assertEqual(self.callback_a.call_count, 4)
def test_schedule_interval_multiple(self):
self.clock.schedule_interval(self.callback_a, 1)
self.clock.schedule_interval(self.callback_b, 1)
self.advance_clock(2)
self.assertEqual(self.callback_a.call_count, 2)
self.assertEqual(self.callback_b.call_count, 2)
def test_schedule_interval_soft(self):
self.clock.schedule_interval_soft(self.callback_a, 1)
self.advance_clock(2)
self.assertEqual(self.callback_a.call_count, 2)
@unittest.skip('Requires changes to the clock')
def test_schedule_interval_soft_multiple(self):
self.clock.schedule_interval(self.callback_a, 1)
self.clock.schedule_interval_soft(self.callback_b, 1)
self.clock.schedule_interval_soft(self.callback_b, 1)
next_ts = set(i.next_ts for i in self.clock._scheduled_items)
self.assertEqual(len(next_ts), 3)
self.advance_clock()
self.assertEqual(self.callback_a.call_count, 1)
self.assertEqual(self.callback_b.call_count, 2)
def test_schedule_unschedule(self):
self.clock.schedule(self.callback_a)
self.clock.unschedule(self.callback_a)
self.advance_clock()
self.assertEqual(self.callback_a.call_count, 0)
def test_schedule_once_unschedule(self):
self.clock.schedule_once(self.callback_a, 1)
self.clock.unschedule(self.callback_a)
self.advance_clock()
self.assertEqual(self.callback_a.call_count, 0)
def test_schedule_interval_unschedule(self):
self.clock.schedule_interval(self.callback_a, 1)
self.clock.unschedule(self.callback_a)
self.advance_clock()
self.assertEqual(self.callback_a.call_count, 0)
def test_schedule_interval_soft_unschedule(self):
self.clock.schedule_interval_soft(self.callback_a, 1)
self.clock.unschedule(self.callback_a)
self.advance_clock()
self.assertEqual(self.callback_a.call_count, 0)
def test_unschedule_removes_all(self):
self.clock.schedule(self.callback_a)
self.clock.schedule_once(self.callback_a, 1)
self.clock.schedule_interval(self.callback_a, 1)
self.clock.schedule_interval_soft(self.callback_a, 1)
self.clock.schedule(self.callback_a)
self.clock.schedule(self.callback_b)
self.clock.unschedule(self.callback_a)
frames = self.advance_clock(10)
self.assertEqual(self.callback_a.call_count, 0)
# callback_b is used to verify that the entire event queue was not cleared
self.assertEqual(self.callback_b.call_count, frames)
def test_schedule_will_not_call_function(self):
self.clock.schedule(self.callback_a)
self.assertEqual(self.callback_a.call_count, 0)
self.clock.schedule_once(self.callback_a, 0)
self.assertEqual(self.callback_a.call_count, 0)
self.clock.schedule_interval(self.callback_a, 1)
self.assertEqual(self.callback_a.call_count, 0)
self.clock.schedule_interval_soft(self.callback_a, 1)
self.assertEqual(self.callback_a.call_count, 0)
def test_unschedule_will_not_call_function(self):
self.clock.schedule(self.callback_a)
self.clock.unschedule(self.callback_a)
self.assertEqual(self.callback_a.call_count, 0)
self.clock.schedule_once(self.callback_a, 0)
self.clock.unschedule(self.callback_a)
self.assertEqual(self.callback_a.call_count, 0)
self.clock.schedule_interval(self.callback_a, 1)
self.clock.unschedule(self.callback_a)
self.assertEqual(self.callback_a.call_count, 0)
self.clock.schedule_interval_soft(self.callback_a, 1)
self.clock.unschedule(self.callback_a)
self.assertEqual(self.callback_a.call_count, 0)
def test_unschedule_will_not_fail_if_already_unscheduled(self):
self.clock.schedule(self.callback_a)
self.clock.unschedule(self.callback_a)
self.clock.unschedule(self.callback_a)
self.clock.schedule_once(self.callback_a, 0)
self.clock.unschedule(self.callback_a)
self.clock.unschedule(self.callback_a)
self.clock.schedule_interval(self.callback_a, 1)
self.clock.unschedule(self.callback_a)
self.clock.unschedule(self.callback_a)
self.clock.schedule_interval_soft(self.callback_a, 1)
self.clock.unschedule(self.callback_a)
self.clock.unschedule(self.callback_a)
def test_call_sched_return_True_if_called_functions(self):
self.clock.schedule(self.callback_a)
self.assertTrue(self.clock.call_scheduled_functions(0))
@unittest.skip('Requires changes to the clock')
def test_call_sched_return_True_if_called_functions_interval(self):
self.clock.schedule_once(self.callback_a, 1)
self.assertFalse(self.clock.call_scheduled_functions(0))
self.clock.set_time(1)
self.assertTrue(self.clock.call_scheduled_functions(0))
def test_call_sched_return_False_if_no_called_functions(self):
self.assertFalse(self.clock.call_scheduled_functions(0))
def test_tick_return_last_delta(self):
self.assertEqual(self.clock.tick(), 0)
self.time = 1
self.assertEqual(self.clock.tick(), 1)
self.time = 3
self.assertEqual(self.clock.tick(), 2)
@unittest.skip('Requires changes to the clock')
def test_get_sleep_time_None_if_no_items(self):
self.assertIsNone(self.clock.get_sleep_time())
@unittest.skip('Requires changes to the clock')
def test_get_sleep_time_can_sleep(self):
self.clock.schedule_once(self.callback_a, 3)
self.clock.schedule_once(self.callback_b, 1)
self.clock.schedule_once(self.callback_c, 6)
self.clock.schedule_once(self.callback_d, 7)
self.assertEqual(self.clock.get_sleep_time(), 1)
self.advance_clock()
self.assertEqual(self.clock.get_sleep_time(), 2)
self.advance_clock(2)
self.assertEqual(self.clock.get_sleep_time(), 3)
self.advance_clock(3)
self.assertEqual(self.clock.get_sleep_time(), 1)
@unittest.skip('Requires changes to the clock')
def test_get_sleep_time_cannot_sleep(self):
self.clock.schedule(self.callback_a)
self.clock.schedule_once(self.callback_b, 1)
self.assertEqual(self.clock.get_sleep_time(), 0)
@unittest.skip
def test_schedule_item_during_tick(self):
def replicating_event(dt):
self.clock.schedule(replicating_event)
counter()
counter = mock.Mock()
self.clock.schedule(replicating_event)
# one tick for the original event
self.clock.tick()
self.assertEqual(counter.call_count, 1)
# requires access to private member
self.assertEqual(len(self.clock._schedule_items), 2)
# one tick from original, then two for new
# now event queue should have two items as well
self.clock.tick()
self.assertEqual(counter.call_count, 3)
# requires access to private member
self.assertEqual(len(self.clock._schedule_items), 4)
def test_unschedule_interval_item_during_tick(self):
def suicidal_event(dt):
counter()
self.clock.unschedule(suicidal_event)
counter = mock.Mock()
self.clock.schedule_interval(suicidal_event, 1)
self.advance_clock()
self.assertEqual(counter.call_count, 1)
@unittest.skip
def test_schedule_interval_item_during_tick(self):
def replicating_event(dt):
self.clock.schedule_interval(replicating_event, 1)
counter()
counter = mock.Mock()
self.clock.schedule_interval(replicating_event, 1)
# advance time for the original event
self.advance_clock()
self.assertEqual(counter.call_count, 1)
# requires access to private member
self.assertEqual(len(self.clock._schedule_interval_items), 2)
# one tick from original, then two for new
# now event queue should have two items as well
self.advance_clock()
self.assertEqual(counter.call_count, 3)
# requires access to private member
self.assertEqual(len(self.clock._schedule_interval_items), 4)
def test_scheduler_integrity(self):
"""most tests in this suite do not care about which order
scheduled items are executed. this test will verify that
the order things are executed is correct.
"""
expected_order = [self.callback_a, self.callback_b,
self.callback_c, self.callback_d]
# schedule backwards to verify that they are scheduled correctly,
# even if scheduled out-of-order.
for delay, func in reversed(list(enumerate(expected_order, start=1))):
self.clock.schedule_once(func, delay)
for index, func in enumerate(expected_order, start=1):
self.advance_clock()
self.assertTrue(func.called)
self.assertFalse(any(i.called for i in expected_order[index:]))
def test_slow_clock(self):
"""pyglet's clock will not make up for lost time. in this case, the
interval scheduled for callback_[bcd] is 1, and 2 seconds have passed.
since pyglet won't make up for lost time, they are only called once.
"""
self.clock.schedule(self.callback_a)
self.clock.schedule_once(self.callback_b, 1)
self.clock.schedule_interval(self.callback_c, 1)
self.clock.schedule_interval_soft(self.callback_d, 1)
# simulate a slow clock
self.time = 2
self.clock.tick()
self.assertEqual(self.callback_a.call_count, 1)
self.assertEqual(self.callback_b.call_count, 1)
self.assertEqual(self.callback_c.call_count, 1)
self.assertEqual(self.callback_d.call_count, 1)
def test_slow_clock_reschedules(self):
"""pyglet's clock will not make up for lost time. in this case, the
interval scheduled for callback_[bcd] is 1, and 2 seconds have passed.
since pyglet won't make up for lost time (call events that missed their
execution time), they are only called once. this test verifies that
missed events are rescheduled and executed later
"""
self.clock.schedule(self.callback_a)
self.clock.schedule_once(self.callback_b, 1)
self.clock.schedule_interval(self.callback_c, 1)
self.clock.schedule_interval_soft(self.callback_d, 1)
# simulate slow clock
self.time = 2
self.clock.tick()
# simulate a proper clock (advance clock time by one)
frames = self.advance_clock()
# make sure our clock is at 3 seconds
self.assertEqual(self.time, 3)
# the +1 is the call during the slow clock period
self.assertEqual(self.callback_a.call_count, frames + 1)
# only scheduled to happen once
self.assertEqual(self.callback_b.call_count, 1)
# 2 because they 'missed' a call when the clock lagged
# with a good clock, this would be 3
self.assertEqual(self.callback_c.call_count, 2)
self.assertEqual(self.callback_d.call_count, 2)
@unittest.skip('Requires changes to the clock')
def test_get_interval(self):
self.assertEqual(self.clock.get_interval(), 0)
self.advance_clock(100)
self.assertEqual(round(self.clock.get_interval(), 10), self.interval)
def test_soft_scheduling_stress_test(self):
"""test that the soft scheduler is able to correctly soft-schedule
several overlapping events.
this test delves into implementation of the clock, and may break
"""
# this value represents evenly scheduled items between 0 & 1
# and what is produced by the correct soft-scheduler
expected = [0.0625, 0.125, 0.1875, 0.25, 0.3125, 0.375, 0.4375, 0.5,
0.5625, 0.625, 0.6875, 0.75, 0.8125, 0.875, 0.9375, 1]
for i in range(16):
self.clock.schedule_interval_soft(None, 1)
# sort the clock items
items = sorted(i.next_ts for i in self.clock._schedule_interval_items)
self.assertEqual(items, expected)
|
