Skip to content

Commit aa9bbe8

Browse files
committed
uasyncio.core: Add test for full wait for a time-scheduled coro.
Specifically, that a coroutine scheduled to run at some time (after some delay) waits requested time before it's run and not run prematurely in case an I/O completion happens before it.
1 parent 48ead94 commit aa9bbe8

File tree

1 file changed

+34
-0
lines changed

1 file changed

+34
-0
lines changed

uasyncio.core/test_full_wait.py

Lines changed: 34 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,34 @@
1+
# Test that coros scheduled to run at some time don't run prematurely
2+
# in case of I/O completion before that.
3+
import uasyncio.core as uasyncio
4+
5+
class MockEventLoop(uasyncio.EventLoop):
6+
7+
def __init__(self):
8+
super().__init__()
9+
self.t = 0
10+
self.msgs = []
11+
12+
def time(self):
13+
return self.t
14+
15+
def pass_time(self, delta):
16+
self.t += delta
17+
18+
def wait(self, delay):
19+
self.pass_time(100)
20+
if self.t == 100:
21+
self.call_soon(lambda: self.msgs.append("I should be run first, time: %s" % self.time()))
22+
if self.t == 1000:
23+
raise StopIteration
24+
25+
26+
loop = MockEventLoop()
27+
loop.call_later_ms_(500, lambda: loop.msgs.append("I should be run second, time: %s" % loop.time()))
28+
try:
29+
loop.run_forever()
30+
except StopIteration:
31+
pass
32+
33+
print(loop.msgs)
34+
assert loop.msgs == ['I should be run first, time: 100', 'I should be run second, time: 500']

0 commit comments

Comments
 (0)