o >hT @sZddlZddlmZddlmZddlmZeeGdddeZGdddej Z dS) N)patch) implementer) IReactorTimec@s<eZdZdZdZddZddZddZd d Zd d Z d S) FakeReactorzH This just fakes out enough reactor methods so .run() can work. FcCsd|_||_g|_dS)NF) stop_calledto_raisedelayed)selfrr /var/www/vedio/testing/chatpythonscript.ninositsolution.com/env/lib/python3.10/site-packages/autobahn/twisted/test/test_wamp_runner.py__init__)s zFakeReactor.__init__cOs|jN)rr argskwr r r run.szFakeReactor.runcCs d|_dS)NT)r)r r r r stop1s zFakeReactor.stopcOs|j||||fdSr )rappend)r delayfuncrkwargsr r r callLater4szFakeReactor.callLatercOstd)NzConnectTCP shouldn't get called) RuntimeErrorrr r r connectTCP7szFakeReactor.connectTCPN) __name__ __module__ __qualname____doc__rr rrrrr r r r r"s rc@s,eZdZedededddZdS)TestWampTwistedRunnerztxaio.use_twistedztxaio.start_loggingz txaio.configcGszddlm}ddlm}ddlm}Wn ty tdw|dd}|d}t d t |}|j ||j d d d d | |jWddS1sOwYdS)zz Ensure the runner doesn't swallow errors and that it exits the reactor properly if there is one. r)ApplicationRunner)ConnectionRefusedError)reactorz No twistedzws://localhost:1realmz It's a trap!ztwisted.internet.reactorcSsdSr r )_r r r Xsz:TestWampTwistedRunner.test_connect_error..T) start_reactorN)autobahn.twisted.wamprtwisted.internet.errorr twisted.internetr! ImportErrorunittestSkipTestrr assertRaisesr assertTruer)r rrr r!runner exception mockreactorr r r test_connect_errorAs       "z(TestWampTwistedRunner.test_connect_errorN)rrrrr1r r r r r;s r) r* unittest.mockrzope.interfacertwisted.internet.interfacesrobjectrTestCaserr r r r s