Permalink
Switch branches/tags
Nothing to show
Find file
Fetching contributors…
Cannot retrieve contributors at this time
262 lines (192 sloc) 5.84 KB
import gc
import sys
import weakref
from contextlib import contextmanager
from twisted.internet.defer import Deferred, returnValue, fail, succeed, CancelledError
from txcoroutine import noreturn, coroutine
# from unnamedframework.util.testing import assert_raises, deferred_result, assert_not_raises
def test_pause_unpause():
after_yield_reached = [False]
ctrl_d = Deferred()
@coroutine
def fn():
yield ctrl_d
after_yield_reached[0] = True
d = fn()
assert not after_yield_reached[0]
d.pause()
ctrl_d.callback(None)
assert not after_yield_reached[0]
d.unpause()
assert after_yield_reached[0]
def test_cancel():
after_yield_reached = [False]
ctrl_d = Deferred()
@coroutine
def fn():
yield ctrl_d
after_yield_reached[0] = True
d = fn()
assert not after_yield_reached[0]
d.addErrback(lambda f: f.trap(CancelledError))
d.cancel()
ctrl_d.callback(None)
assert not after_yield_reached[0]
def test_cancel_stops_the_generator():
generator_exit_caught = [False]
ctrl_d = Deferred()
@coroutine
def fn():
try:
yield ctrl_d
except GeneratorExit:
generator_exit_caught[0] = True
d = fn()
d.addErrback(lambda f: f.trap(CancelledError))
d.cancel()
assert generator_exit_caught[0]
def test_main():
d_ctrl_1, d_ctrl_2 = Deferred(), Deferred()
wref = [None]
never_called = [True]
@coroutine
def test_coroutine():
dummy = Dummy()
wref[0] = weakref.ref(dummy)
yield d_ctrl_1 # pause for state inspection
noreturn(other_coroutine())
never_called[0] = False
other_coroutine_called = [False]
def other_coroutine():
other_coroutine_called[0] = True
yield d_ctrl_2 # pause for state inspection
final_d = test_coroutine()
gc.collect() # don't rely on automatic collection
assert wref[0]() # just to reinforce the test logic
d_ctrl_1.callback(None) # resume test_coroutine
assert other_coroutine_called[0], "the other coroutine should be executed"
gc.collect() # don't rely on automatic collection
assert not wref[0](), "objects in the caller should become garbage"
assert not final_d.called
d_ctrl_2.callback(None) # resume other_coroutine
assert final_d.called, "the whole procedure should have been completed"
assert never_called[0], "flow should never return to the noreturn-calling coroutine"
def test_deep_recursion():
def fact(n, result=1):
if n <= 1:
returnValue(result)
else:
noreturn(fact(n - 1, n * result))
yield
assert deferred_result(coroutine(fact)(1)) == 1
assert deferred_result(coroutine(fact)(10)) == safe_fact(10)
with recursion_limit(100):
try:
# +10 is actually too high here as we probably already have some stuff on the stack, but just to be sure
assert deferred_result(coroutine(fact)(110)) == safe_fact(110)
except RuntimeError:
assert False
# ...and now let's prove that the same (tail call optimizable) algorithm without noreturn will eat up the stack
def normal_fact(n, result=1):
if n <= 1:
returnValue(result)
else:
return normal_fact(n - 1, n * result)
with recursion_limit(100):
try:
normal_fact(110)
except RuntimeError:
pass
else:
assert False, "normal_fact(110)"
def test_with_previous_yield_result_not_none():
class MockException(Exception):
pass
fn_called = [False]
@coroutine
def fn():
fn_called[0] = True
try:
yield fail(MockException())
except MockException:
pass
noreturn(fn2())
def fn2():
yield succeed(None)
try:
fn()
except MockException:
assert False
assert fn_called[0]
def test_noreturn_of_other_inlineCallbacks_wrapped_callable():
after_noreturn_reached = [False]
@coroutine
def fn():
yield
noreturn(fn2())
after_noreturn_reached[0] = True
fn2_called = [False]
@coroutine
def fn2():
fn2_called[0] = True
yield
returnValue('someretval')
retval = deferred_result(fn())
assert fn2_called[0]
assert not after_noreturn_reached[0]
assert retval == 'someretval'
def test_noreturn_with_regular_function():
after_noreturn_reached = [False]
@coroutine
def fn():
yield
noreturn(fn2())
after_noreturn_reached[0] = True
def fn2():
return 'someretval'
retval = deferred_result(fn())
assert not after_noreturn_reached[0]
assert retval == 'someretval'
def test_pause_unpause_cancel_on_coroutine_with_no_depends_on():
@coroutine
def fn():
yield
c = fn()
c.pause()
c.unpause()
c.cancel()
def test_cancelled_coroutine_does_not_keep_its_last_depends_on_from_being_garbage_collected():
depends_on = Deferred()
@coroutine
def fn():
yield depends_on
c = fn()
depends_on_weakref = weakref.ref(depends_on)
depends_on = None
c.cancel()
assert not depends_on_weakref()
@contextmanager
def recursion_limit(n):
old = sys.getrecursionlimit()
sys.setrecursionlimit(n)
try:
yield
finally:
sys.setrecursionlimit(old)
# because object() cannot be weakly referenced
class Dummy:
pass
def safe_fact(n):
return reduce(lambda a, b: a * b, xrange(1, n + 1))
def deferred_result(d):
ret = [None]
exc = [None]
if isinstance(d, Deferred):
assert d.called
d.addCallback(lambda result: ret.__setitem__(0, result))
d.addErrback(lambda f: exc.__setitem__(0, f))
if exc[0]:
exc[0].raiseException()
return ret[0]
else:
return d