o @Bi'@s^ddlmZzddlmZWn eyedwddlmZddlmZddl m Z ddlm Z m Z zddl mZWneyKddlmZYnwdd lmZdd lmZmZdd lmZdd lmZdd lmZddlmZdZeeGdddeZee GdddeZGdddeeZGdddeZ GdddeZ!ddZ"ddZ#dS))absolute_import)IHostnameResolverz=Twisted 17.1.0 or later required for autobahn.twisted.testing)Deferred) IPv4Address)HostResolution) ISSLTransportIReactorPluggableNameResolver)MemoryReactorClock)iosim)directlyProvides implementer)IWebSocketClientAgent)_TwistedWebSocketClientAgent)WebSocketServerProtocol)WebSocketServerFactory) create_pumpercreate_memory_agentMemoryReactorClockResolverc@seZdZdddZdS)_StaticTestResolverrcCs<t|}|||tdd|dkrdn||dS)zR Implement IHostnameResolver which always returns 127.0.0.1:31337 TCP 127.0.0.1rizN)rresolutionBeganaddressResolvedrresolutionComplete)selfreceiverhostName portNumber resolutionr a/var/www/html/Trade-python/venv/lib/python3.10/site-packages/autobahn/twisted/testing/__init__.pyresolveHostNameBs   z#_StaticTestResolver.resolveHostNameN)r)__name__ __module__ __qualname__r"r r r r!r@src@s(eZdZdZdZeddZddZdS)_TestNameResolverz9 A test version of IReactorPluggableNameResolver NcCs|jdur t|_|jSN) _resolverrrr r r! nameResolverVs z_TestNameResolver.nameResolvercCs|j}||_|Sr')r()rresolveroldr r r!installNameResolver\sz%_TestNameResolver.installNameResolver)r#r$r%__doc__r(propertyr*r-r r r r!r&Ns   r&c@seZdZdZdS)rzY Combine MemoryReactor, Clock and an IReactorPluggableNameResolver together. N)r#r$r%r.r r r r!rbsrc@s"eZdZdZddZdddZdS)_TwistedWebMemoryAgenta A testing agent which will hook up an instance of `server_protocol` for every client that is created via the `open` API call. :param reactor: the reactor to use for tests (usually an instance of MemoryReactorClockResolver) :param pumper: an implementation IPumper (e.g. as returned by `create_pumper`) :param server_protocol: the server-side WebSocket protocol class to instantiate (e.g. a subclass of `WebSocketServerProtocol` cCs2||_||_||_t|j|_t|_t|_dSr') _reactor_server_protocol_pumperr_agentset_pumpsdict_servers)rreactorpumperserver_protocolr r r!__init__zs   z_TwistedWebMemoryAgent.__init__Ncs|d}jj|||d}|rjjd\}}}} } } n jjd\}}}} } tdd|} tddd} t_ t j d| | d}| d }t j |d | | d}|r_t |tt |tt |||}j|fd d }|||S) a} Implement IWebSocketClientAgent with in-memory transports. :param transport_config: a string starting with 'wss://' or 'ws://' :param options: a dict containing options :param protocol_class: the client protocol class to instantiate (or `None` for defaults, which is to use `WebSocketClientProtocol`) zwss://)protocol_classrrrT)isServer hostAddress peerAddressNFcsj|<|Sr')r8)protorr;r r! add_mappings z0_TwistedWebMemoryAgent.open..add_mapping) startswithr4openr1 sslClients tcpClientsrr2rfactoryr FakeTransport buildProtocolr rconnectr3add addCallback)rtransport_configoptionsr= is_securereal_client_protocolhostportrIcontext_factorytimeout bindAddressserver_addressclient_addressserver_transportclientProtocolclient_transportpumprDr rCr!rFs>        z_TwistedWebMemoryAgent.openr')r#r$r%r.r<rFr r r r!r0js r0c@s@eZdZdZddZddZddZdd Zd d Zd d Z dS) _Kalamazooa Feeling whimsical about class names, see https://en.wikipedia.org/wiki/Handcar This is 'an IOPump pumper', an object which causes a series of IOPumps it is monitoring to do their I/O operations periodically. This needs the 'real' reactor which trial drives, because reasons: - so @inlineCallbacks / async-def functions work (if I could explain exactly why here, I would) - we need to 'break the loop' of synchronous calls somewhere and polluting the tests themselves with that is bad - get rid of e.g. .flush() calls in tests themselves (thus 'teaching' the tests about details of I/O scheduling that they shouldn't know). cCs*t|_d|_g|_ddlm}||_dS)NFr)r9)r5r6_pumping_waiting_for_stoptwisted.internetr9_global_reactor)rglobal_reactorr r r!r<s   z_Kalamazoo.__init__cCs|j|dS)zp Add a new IOPump. It will be removed when both its client and server are disconnected. N)r6rM)rpr r r!rMsz_Kalamazoo.addcCs$|jrdSd|_|jd|jdS)z Begin triggering I/O in all IOPump instances we have. We will keep periodically 'pumping' our IOPumps until `.stop()` is called. Call from `setUp()` for example. NTr)r_rb callLater _pump_oncer)r r r!startsz_Kalamazoo.startcCs@|jst|jrt}|j|d|_|St}|d|S)z :returns: a Deferred that fires when we have stopped pump()-ing Call from `tearDown()`, for example. FN)r_lenr`rappendcallbackrdr r r!stops  z_Kalamazoo.stopcCsB|jr||jd|jdS|jD]}|dqg|_dS)z{ flush all data from all our IOPump instances and schedule another iteration on the global reactor g?N)r__flushrbrerfr`rjrkr r r!rfs    z_Kalamazoo._pump_oncecCsD|j}t}|_|D]}||jjr|jjrq ||q dS)zE Flush all data between pending client/server pairs. N)r6r5flushclientIO disconnectedserverIOrM)r old_pumps new_pumpsrdr r r!rns  z_Kalamazoo._flushN) r#r$r%r.r<rMrgrmrfrnr r r r!r^s  r^cCstS)z4 return a new instance implementing IPumper )r^r r r r!rsrcCs|durt}t|||S)a) return a new instance implementing `IWebSocketClientAgent`. connection attempts will be satisfied by traversing the Upgrade request path starting at `resource` to find a `WebSocketResource` and then exchange data between client and server using purely in-memory buffers. N)rr0)r9r:r;r r r!rs  rN)$ __future__rtwisted.internet.interfacesr ImportErrortwisted.internet.deferrtwisted.internet.addressrtwisted.internet._resolverrrrtwisted.internet.testingr twisted.test.proto_helpers twisted.testr zope.interfacer r autobahn.websocket.interfacesr autobahn.twisted.websocketrrr__all__objectrr&rr0r^rrr r r r!sB            QV