https://t.me/RX1948
Server : LiteSpeed
System : Linux host 5.15.0-161-generic #171-Ubuntu SMP Sat Oct 11 08:17:01 UTC 2025 x86_64
User : idnco5810 ( 1093)
PHP Version : 8.2.29
Disable Function : NONE
Directory :  /lib/python3/dist-packages/twisted/names/test/__pycache__/

Upload File :
current_dir [ Writeable ] document_root [ Writeable ]

 

Current File : //lib/python3/dist-packages/twisted/names/test/__pycache__/test_util.cpython-310.pyc
o

�b��@s�dZddlmZddlmZddlmZddlmZddl	m
Z
ddlmZm
Z
ddlmZee
�Gd	d
�d
��Zee
e�ee�Gdd�de��Zeee�d
S)z$
Utilities for Twisted.names tests.
�)�	randrange)�implementer)�verifyClass)�IPv4Address)�succeed)�IReactorUDP�
IUDPTransport)�Clockc@sJeZdZdZdd�Zdd�Zdd�Zdd	d
�Zdd�Zd
d�Z	dd�Z
dS)�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}.
    cCs"||_||_g|_d|_||_dS�N)�_host�	_protocol�_sentPackets�_connectedTo�_maxPacketSize)�self�host�protocol�
maxPacketSize�r�>/usr/lib/python3/dist-packages/twisted/names/test/test_util.py�__init__&s

z MemoryDatagramTransport.__init__cCstdg|j�R�S)z_
        Return the address which this transport is pretending to be bound
        to.
        �UDP)rr�rrrr�getHost-szMemoryDatagramTransport.getHostcCs |jdur	td��||f|_dS)z>
        Connect this transport to the given address.
        NzAlready connected)r�
ValueError)rr�portrrr�connect4s
zMemoryDatagramTransport.connectNcCsH|dur|j}|durtd��t|�|jkrtd��|j�||f�dS)z*
        Send the given datagram.
        NzNeed an addresszPacket too big)rr�lenrr�append)r�datagram�addrrrr�write<szMemoryDatagramTransport.writecCs|j��td�S)z+
        Shut down this transport.
        N)r
�stopProtocolrrrrr�
stopListeningHs
z%MemoryDatagramTransport.stopListeningcC�dS�zC
        Dummy implementation to satisfy L{IUDPTransport}.
        Nr)r�enabledrrr�setBroadcastAllowedO�z+MemoryDatagramTransport.setBroadcastAllowedcCr%r&rrrrr�getBroadcastAllowedUr)z+MemoryDatagramTransport.getBroadcastAllowedr)�__name__�
__module__�__qualname__�__doc__rrrr"r$r(r*rrrrr
s
r
c@s"eZdZdZdd�Zd	dd�ZdS)
�
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}.
    cCst�|�i|_dSr)r	r�udpPortsrrrrrls

zMemoryReactor.__init__�� cCs\|dkr	tdd�}||jvrnq||jvrtd��t||f||�}||j|<|�|�|S)zR
        Pretend to bind a UDP port and connect the given protocol to it.
        rT�izAddress in use)rr0rr
�makeConnection)rrr�	interfacer�	transportrrr�	listenUDPps

�


zMemoryReactor.listenUDPN)r1r2)r+r,r-r.rr7rrrrr/_sr/N)r.�randomr�zope.interfacer�zope.interface.verifyr�twisted.internet.addressr�twisted.internet.deferr�twisted.internet.interfacesrr�twisted.internet.taskr	r
r/rrrr�<module>s
G!

https://t.me/RX1948 - 2025