<s> from mock import patch <EOL> import pytest <EOL> import txaio <EOL> from txaio . testutil import replace_loop <EOL> def test_default_reactor ( framework_tx ) : <EOL> """<STR_LIT>""" <EOL> pytest . importorskip ( '<STR_LIT>' ) <EOL> assert txaio . config . loop is None <EOL> try : <EOL> txaio . call_later ( <NUM_LIT:1> , lambda : None ) <EOL> from twisted . internet import reactor <EOL> assert txaio . config . loop is reactor <EOL> finally : <EOL> txaio . config . loop = None <EOL> def test_explicit_reactor_future ( framework ) : <EOL> """<STR_LIT>""" <EOL> pytest . importorskip ( '<STR_LIT>' ) <EOL> if txaio . using_twisted : <EOL> pytest . skip ( ) <EOL> with patch . object ( txaio . config , '<STR_LIT>' ) as fake_loop : <EOL> f = txaio . create_future ( '<STR_LIT:result>' ) <EOL> f . add_done_callback ( lambda _ : None ) <EOL> assert len ( fake_loop . method_calls ) == <NUM_LIT:2> <EOL> c = fake_loop . method_calls [ <NUM_LIT:1> ] <EOL> assert c [ <NUM_LIT:0> ] == '<STR_LIT>' <EOL> def test_explicit_reactor_coroutine ( framework ) : <EOL> """<STR_LIT>""" <EOL> pytest . importorskip ( '<STR_LIT>' ) <EOL> if txaio . using_twisted : <EOL> pytest . skip ( ) <EOL> from asyncio import coroutine <EOL> @ coroutine <EOL> def some_coroutine ( ) : <EOL> yield '<STR_LIT>' <EOL> with patch . object ( txaio . config , '<STR_LIT>' ) as fake_loop : <EOL> txaio . as_future ( some_coroutine ) <EOL> assert len ( fake_loop . method_calls ) == <NUM_LIT:2> <EOL> c = fake_loop . method_calls [ <NUM_LIT:1> ] <EOL> assert c [ <NUM_LIT:0> ] == '<STR_LIT>' <EOL> def test_call_later_tx ( framework_tx ) : <EOL> '''<STR_LIT>''' <EOL> from twisted . internet . task import Clock <EOL> new_loop = Clock ( ) <EOL> calls = [ ] <EOL> with replace_loop ( new_loop ) as fake_loop : <EOL> def foo ( * args , ** kw ) : <EOL> calls . append ( ( args , kw ) ) <EOL> delay = txaio . call_later ( <NUM_LIT:1> , foo , <NUM_LIT:5> , <NUM_LIT:6> , <NUM_LIT:7> , foo = "<STR_LIT:bar>" ) <EOL> assert len ( calls ) == <NUM_LIT:0> <EOL> assert hasattr ( delay , '<STR_LIT>' ) <EOL> fake_loop . advance ( <NUM_LIT:2> ) <EOL> assert len ( calls ) == <NUM_LIT:1> <EOL> assert calls [ <NUM_LIT:0> ] [ <NUM_LIT:0> ] == ( <NUM_LIT:5> , <NUM_LIT:6> , <NUM_LIT:7> ) <EOL> assert calls [ <NUM_LIT:0> ] [ <NUM_LIT:1> ] == dict ( foo = "<STR_LIT:bar>" ) <EOL> def test_call_later_aio ( framework_aio ) : <EOL> '''<STR_LIT>''' <EOL> pytest . importorskip ( '<STR_LIT>' ) <EOL> def time_gen ( ) : <EOL> when = yield <EOL> assert when == <NUM_LIT:1> <EOL> when = yield <NUM_LIT:0> <EOL> print ( "<STR_LIT>" , when ) <EOL> from asyncio . test_utils import TestLoop <EOL> new_loop = TestLoop ( time_gen ) <EOL> calls = [ ] <EOL> with replace_loop ( new_loop ) as fake_loop : <EOL> def foo ( * args , ** kw ) : <EOL> calls . append ( ( args , kw ) ) <EOL> delay = txaio . call_later ( <NUM_LIT:1> , foo , <NUM_LIT:5> , <NUM_LIT:6> , <NUM_LIT:7> , foo = "<STR_LIT:bar>" ) <EOL> assert len ( calls ) == <NUM_LIT:0> <EOL> assert hasattr ( delay , '<STR_LIT>' ) <EOL> fake_loop . advance_time ( <NUM_LIT:2> ) <EOL> fake_loop . _run_once ( ) <EOL> assert len ( calls ) == <NUM_LIT:1> <EOL> assert calls [ <NUM_LIT:0> ] [ <NUM_LIT:0> ] == ( <NUM_LIT:5> , <NUM_LIT:6> , <NUM_LIT:7> ) <EOL> assert calls [ <NUM_LIT:0> ] [ <NUM_LIT:1> ] == dict ( foo = "<STR_LIT:bar>" ) </s>
