o >hQ'@sRzddlmZWn eyedwddlmZddlmZddlmZddlm Z m Z zddl m Z WneyEddl m Z YnwddlmZdd lmZmZdd lmZdd lmZdd lmZdd lmZdZeeGdddeZee GdddeZGddde eZGdddeZGdddeZddZ ddZ!dS))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/var/www/vedio/testing/chatpythonscript.ninositsolution.com/env/lib/python3.10/site-packages/autobahn/twisted/testing/__init__.pyresolveHostName@s   z#_StaticTestResolver.resolveHostNameN)r)__name__ __module__ __qualname__r!rrrr r>src@s(eZdZdZdZeddZddZdS)_TestNameResolverz9 A test version of IReactorPluggableNameResolver NcCs|jdur t|_|jSN) _resolverrrrrr nameResolverTs z_TestNameResolver.nameResolvercCs|j}||_|Sr&)r')rresolveroldrrr installNameResolverZsz%_TestNameResolver.installNameResolver)r"r#r$__doc__r'propertyr)r,rrrr r%Ls   r%c@seZdZdZdS)rzY Combine MemoryReactor, Clock and an IReactorPluggableNameResolver together. N)r"r#r$r-rrrr r`src@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_protocolrrr __init__xs   z_TwistedWebMemoryAgent.__init__Ncs|d}jj|||d}|rjjd\}}}} } } n jjd\}}}} } tdd|} tddd} tdddurDt _ t j d | | d }| d}t j |d | | d }|rgt|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_classrrrfactoryNT)isServer hostAddress peerAddressFcsj|<|Sr&)r7)protorr:rr add_mappings z0_TwistedWebMemoryAgent.open..add_mapping) startswithr3openr0 sslClients tcpClientsrr1getattrrr>r FakeTransport buildProtocolr rconnectr2add addCallback)rtransport_configoptionsr< is_securereal_client_protocolhostportr>context_factorytimeout bindAddressserver_addressclient_addressserver_transportclientProtocolclient_transportpumprDrrCr rFs@        z_TwistedWebMemoryAgent.openr&)r"r#r$r-r;rFrrrr r/hs r/c@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)r8)r4r5_pumping_waiting_for_stoptwisted.internetr8_global_reactor)rglobal_reactorrrr 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)r5rM)rprrr 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(rrr 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`rappendcallbackrdrrr 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`rjrkrrr rfs    z_Kalamazoo._pump_oncecCsD|j}t}|_|D]}||jjr|jjrq ||q dS)zE Flush all data between pending client/server pairs. N)r5r4flushclientIO disconnectedserverIOrM)r old_pumps new_pumpsrdrrr rns  z_Kalamazoo._flushN) r"r#r$r-r;rMrgrmrfrnrrrr r^s  r^cCstS)z4 return a new instance implementing IPumper )r^rrrr 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)rr/)r8r9r:rrr rs  rN)"twisted.internet.interfacesr ImportErrortwisted.internet.deferrtwisted.internet.addressrtwisted.internet._resolverrrrtwisted.internet.testingrtwisted.test.proto_helpers twisted.testr zope.interfacer r autobahn.websocket.interfacesr autobahn.twisted.websocketr rr__all__objectrr%rr/r^rrrrrr s@           SV