57 lines
1.7 KiB
Python
57 lines
1.7 KiB
Python
import structio
|
|
|
|
|
|
async def test_silent(i, j):
|
|
print(f"[test] Parent is alive, exiting after {i:.2f} seconds")
|
|
k = structio.clock()
|
|
with structio.skip_after(i) as scope:
|
|
print(f"[test] Sleeping for {j} seconds")
|
|
await structio.sleep(j)
|
|
print(f"[test] Finished in {structio.clock() - k:.2f} seconds (timed out: {scope.cancelled})")
|
|
|
|
|
|
async def test_loud(i, j):
|
|
print(f"[test] Parent is alive, exiting after {i:.2f} seconds")
|
|
k = structio.clock()
|
|
try:
|
|
with structio.with_timeout(i):
|
|
print(f"[test] Sleeping for {j} seconds")
|
|
await structio.sleep(j)
|
|
except structio.TimedOut:
|
|
print("[test] Timed out!")
|
|
print(f"[test] Finished in {structio.clock() - k:.2f} seconds")
|
|
|
|
|
|
async def deadlock():
|
|
await structio.Event().wait()
|
|
|
|
|
|
async def test_deadlock(i):
|
|
print(f"[test] Parent is alive, will exit in {i} seconds")
|
|
t = structio.clock()
|
|
with structio.skip_after(i):
|
|
print("[test] Entering deadlock")
|
|
await deadlock()
|
|
print(f"[test] Done in {structio.clock() - t:.2f} seconds")
|
|
|
|
|
|
async def test_nested(i):
|
|
print(f"[test] Parent is alive, will exit in {i} seconds")
|
|
t = structio.clock()
|
|
with structio.skip_after(i):
|
|
print("[test] Entered first scope")
|
|
with structio.skip_after(i * 2):
|
|
# Even though this scope's timeout is
|
|
# larger than its parent, structio will
|
|
# still cancel it when its containing
|
|
# scope expires
|
|
print("[test] Entered second scope")
|
|
await deadlock()
|
|
print(f"[test] Done in {structio.clock() - t:.2f} seconds")
|
|
|
|
|
|
structio.run(test_silent, 3, 5)
|
|
structio.run(test_loud, 3, 5)
|
|
structio.run(test_deadlock, 5)
|
|
structio.run(test_nested, 5)
|