Index: tulip/tasks_test.py |
=================================================================== |
--- a/tulip/tasks_test.py |
+++ b/tulip/tasks_test.py |
@@ -131,6 +131,27 @@ |
self.assertTrue(t.done()) |
self.assertFalse(t.cancel()) |
+ def test_stop_while_run_in_complete(self): |
+ x = 0 |
+ @tasks.coroutine |
+ def task(): |
+ nonlocal x |
+ while x < 10: |
+ yield from tasks.sleep(0.1) |
+ x += 1 |
+ if x == 2: |
+ self.event_loop.stop() |
+ |
+ t = tasks.Task(task()) |
+ t0 = time.monotonic() |
+ self.assertRaises( |
+ futures.InvalidStateError, |
+ self.event_loop.run_until_complete, t) |
+ t1 = time.monotonic() |
+ self.assertFalse(t.done()) |
+ self.assertTrue(0.2 <= t1-t0 <= 0.5) |
GvR
2013/02/11 22:57:35
Shouldn't this be 0.18 <= t1-t0 <= 0.22?
|
+ self.assertEqual(x, 2) |
+ |
def test_timeout(self): |
@tasks.task |
def task(): |
@@ -138,10 +159,13 @@ |
return 42 |
t = task() |
+ t0 = time.monotonic() |
self.assertRaises( |
futures.TimeoutError, |
self.event_loop.run_until_complete, t, 0.1) |
+ t1 = time.monotonic() |
self.assertFalse(t.done()) |
+ self.assertTrue(0.08 <= t1-t0 <= 0.12) |
def test_timeout_not(self): |
@tasks.task |
@@ -150,9 +174,12 @@ |
return 42 |
t = task() |
+ t0 = time.monotonic() |
r = self.event_loop.run_until_complete(t, 10.0) |
+ t1 = time.monotonic() |
self.assertTrue(t.done()) |
self.assertEqual(r, 42) |
+ self.assertTrue(0.08 <= t1-t0 <= 0.12) |
def test_wait(self): |
a = tasks.sleep(0.1) |