o @Bi@$@sddlmZmZddlZddlZddlmZejddrFddl m Z ddl m Z ddl mZe eGd d d eZGd d d ejZdSzddlZdd lm Z mZWneyiddlZdd l m Z mZYnwddlmZGdddejZdS))absolute_importprint_functionN) replace_loop USE_TWISTEDF)patch) implementer) IReactorTimec@s<eZdZdZdZddZddZddZd d Zd d Z d S) FakeReactorzP This just fakes out enough reactor methods so .run() can work. FcCsd|_||_g|_dS)NF) stop_calledto_raisedelayed)selfr r^/var/www/html/Trade-python/venv/lib/python3.10/site-packages/autobahn/wamp/test/test_runner.py__init__-s zFakeReactor.__init__cOs|jN)r r argskwrrrrun2szFakeReactor.runcCs d|_dS)NT)r )r rrrstop5s zFakeReactor.stopcOs|j||||fdSr)r append)r delayfuncrkwargsrrr callLater8szFakeReactor.callLatercOstd)NzConnectTCP shouldn't get called) RuntimeErrorrrrr connectTCP;szFakeReactor.connectTCPN) __name__ __module__ __qualname____doc__r rrrrrrrrrr &s r c@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)z Ensure the runner doesn't swallow errors and that it exits the reactor properly if there is one. rApplicationRunner)ConnectionRefusedError)reactorz No twistedzws://localhost:1realmz It's a trap!ztwisted.internet.reactorcSsdSrr)_rrr[sz:TestWampTwistedRunner.test_connect_error..T) start_reactorN)autobahn.twisted.wampr$twisted.internet.errorr%twisted.internetr& ImportErrorunittestSkipTestrr assertRaisesr assertTruer )r rr$r%r&runner exception mockreactorrrrtest_connect_errorDs       "z(TestWampTwistedRunner.test_connect_errorN)rrr rr6rrrrr">s r")rMockr#c@s@eZdZdZddZddZddZdd Zd d Zd d Z dS)TestApplicationRunnerzI Test the autobahn.asyncio.wamp.ApplicationRunner class. cOsFz|jWn ty|j}Ynw|j}|||g|Ri|dSr)assertRaisesRegexAttributeErrorassertRaisesRegexp)r r4errorrrfrrr_assertRaisesRegexns   z(TestApplicationRunner._assertRaisesRegexc CsttM}tjtd|d,tttfd|_i}tdd|d}|d|||j j ddWd n1s=wYWd d SWd d S1sUwYd S) z Ensure that loop.create_connection is called with the exact SSL context object that is passed (as ssl) to the __init__ method of ApplicationRunner. get_event_loop return_valuews://127.0.0.1:8080/wsr'ssl_unused_rDN rr7robjectasynciorun_until_completer$rassertIscreate_connection call_args)r looprDr3rrrtest_explicit_SSLContextws  "z.TestApplicationRunner.test_explicit_SSLContextc CsttI}tjtd|d(tttfd|_tdd}|d|d|j j ddWd n1s9wYWd d SWd d S1sQwYd S) z Ensure that loop.create_connection is called with ssl=False if no ssl argument is passed to the __init__ method of ApplicationRunner and the websocket URL starts with "ws:". r?r@rBr'rEFrFrDNrGr rNr3rrr test_omitted_SSLContext_insecures   "z6TestApplicationRunner.test_omitted_SSLContext_insecurec CsttJ}tjtd|d)tttfd|_tdd}||j| d|j j ddWdn1s:wYWddSWddS1sRwYdS) z Ensure that loop.create_connection is called with ssl=True if no ssl argument is passed to the __init__ method of ApplicationRunner and the websocket URL starts with "wss:". r?r@zwss://127.0.0.1:8080/wssr'TrFrDN) rr7rrHrIrJr$rfailrKrLrMrPrrrtest_omitted_SSLContext_secures   "z4TestApplicationRunner.test_omitted_SSLContext_securecCshtt%}tttfd|_tdddd}d}|t||jdWddS1s-wYdS) z ApplicationRunner must raise an exception if given an ssl value of True but only a "ws:" URL. r@ws://127.0.0.1:8080/wssr'TrC^ssl argument value passed to ApplicationRunner conflicts with the "ws:" prefix of the url argument\. Did you mean to use "wss:"\?$rEN)rr7rJr$r> Exceptionr)r rNr3r<rrr"test_conflict_SSL_True_with_ws_urls "z8TestApplicationRunner.test_conflict_SSL_True_with_ws_urlcCsddl}z|jWn tyd}Ynw|}tt%}tttfd|_tdd|d}d}|t||j d WddS1sFwYdS) z ApplicationRunner must raise an exception if given an ssl value that is an instance of SSLContext, but only a "ws:" URL. rNTr@rTr'rCrUrE) rDcreate_default_contextr:rr7rJr$r>rVr)r rDcontextrNr3r<rrr$test_conflict_SSLContext_with_ws_urls  "z:TestApplicationRunner.test_conflict_SSLContext_with_ws_urlN) rrr r!r>rOrQrSrWrZrrrrr8js   r8) __future__rrosr/txaio.testutilrenvirongetmockrzope.interfacertwisted.internet.interfacesrrHr TestCaser"rI unittest.mockr7r.trolliusautobahn.asyncio.wampr$r8rrrrs(    "