fixed bug with nested api.with_timeout interfering with each other; added a testcase

This commit is contained in:
Denis Bilenko
2008-12-10 01:57:29 +06:00
parent aec8ceb3f2
commit d14784eeb9
2 changed files with 10 additions and 4 deletions

View File

@@ -395,12 +395,13 @@ def with_timeout(seconds, func, *args, **kwds):
# pass timeout_value through to func().
has_timeout_value = "timeout_value" in kwds
timeout_value = kwds.pop("timeout_value", None)
timeout = exc_after(seconds, TimeoutError())
error = TimeoutError()
timeout = exc_after(seconds, error)
try:
try:
return func(*args, **kwds)
except TimeoutError:
if has_timeout_value:
except TimeoutError, ex:
if ex is error and has_timeout_value:
return timeout_value
raise
finally:

View File

@@ -1,5 +1,5 @@
import unittest
from eventlet.api import sleep, spawn, kill
from eventlet.api import sleep, spawn, kill, with_timeout, TimeoutError
DELAY = 0.1
@@ -29,6 +29,11 @@ class Test(unittest.TestCase):
sleep(DELAY)
assert state == ['start', 'except', 'finished'], state
def test_nested_with_timeout(self):
def func():
return with_timeout(0.2, sleep, 2, timeout_value=1)
self.assertRaises(TimeoutError, with_timeout, 0.1, func)
if __name__=='__main__':
unittest.main()