VhdZddlmZddlmZddlmZddlmZddl m Z ddl m Z m Z ddlmZee Gd d Zee eee Gd d eZee ey )z$ Utilities for Twisted.names tests. ) randrange) implementer) verifyClass) IPv4Address)succeed) IReactorUDP IUDPTransport)Clockc<eZdZdZdZdZdZd dZdZdZ d Z y) MemoryDatagramTransporta This L{IUDPTransport} implementation enforces the usual connection rules and captures sent traffic in a list for later inspection. @ivar _host: The host address to which this transport is bound. @ivar _protocol: The protocol connected to this transport. @ivar _sentPackets: A C{list} of two-tuples of the datagrams passed to C{write} and the addresses to which they are destined. @ivar _connectedTo: L{None} if this transport is unconnected, otherwise an address to which all traffic is supposedly sent. @ivar _maxPacketSize: An C{int} giving the maximum length of a datagram which will be successfully handled by C{write}. cJ||_||_g|_d|_||_yN)_host _protocol _sentPackets _connectedTo_maxPacketSize)selfhostprotocol maxPacketSizes L/home/dcms/DCMS/lib/python3.12/site-packages/twisted/names/test/test_util.py__init__z MemoryDatagramTransport.__init__&s( ! +c.tdg|jS)z_ Return the address which this transport is pretending to be bound to. UDP)rrrs rgetHostzMemoryDatagramTransport.getHost-s 5.4::..rcD|j td||f|_y)z> Connect this transport to the given address. NzAlready connected)r ValueError)rrports rconnectzMemoryDatagramTransport.connect4s(    (01 1!4LrNc| |j}| tdt||jkDr td|jj ||fy)z* Send the given datagram. NzNeed an addresszPacket too big)rr lenrrappend)rdatagramaddrs rwritezMemoryDatagramTransport.write<s\ <$$D <./ / x=4.. .-. .   (D!12rcL|jjtdS)z+ Shut down this transport. N)r stopProtocolrrs r stopListeningz%MemoryDatagramTransport.stopListeningHs ##%t}rcyzC Dummy implementation to satisfy L{IUDPTransport}. N)renableds rsetBroadcastAllowedz+MemoryDatagramTransport.setBroadcastAllowedO rcyr-r.rs rgetBroadcastAllowedz+MemoryDatagramTransport.getBroadcastAllowedUr1rr) __name__ __module__ __qualname____doc__rrr"r(r+r0r3r.rrr r s* ,/) 3  rr ceZdZdZdZddZy) MemoryReactoraO An L{IReactorTime} and L{IReactorUDP} provider. Time is controlled deterministically via the base class, L{Clock}. UDP is handled in-memory by connecting protocols to instances of L{MemoryDatagramTransport}. @ivar udpPorts: A C{dict} mapping port numbers to instances of L{MemoryDatagramTransport}. c<tj|i|_yr)r rudpPortsrs rrzMemoryReactor.__init__ls t rc|dk(r tdd}||jvrn||jvr tdt||f||}||j|<|j ||S)zR Pretend to bind a UDP port and connect the given protocol to it. rizAddress in use)rr;r r makeConnection)rr!r interfacer transports r listenUDPzMemoryReactor.listenUDPps} 19 E*t}}, 4== -. .+Y,=xW ' d *rN)i )r4r5r6r7rrAr.rrr9r9_s rr9N)r7randomrzope.interfacerzope.interface.verifyrtwisted.internet.addressrtwisted.internet.deferrtwisted.internet.interfacesrr twisted.internet.taskr r r9r.rrrJs} &-0*B' ]D D D N M23 [EB K'r