Da

✏️ Editing: test_util.cpython-310.pyc

�����b�����������������������@���s����d�Z�ddlmZ�ddlmZ�ddlmZ�ddlmZ�ddl	m
Z
�ddlmZm
Z
�ddlmZ�ee
�G�d	d
��d
��Zee
e��ee�G�dd��de��Zeee��d
S�)z$
Utilities for Twisted.names tests.
�����)�	randrange)�implementer)�verifyClass)�IPv4Address)�succeed)�IReactorUDP�
IUDPTransport)�Clockc�������������������@���sJ���e�Zd�ZdZdd��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}.
    c�����������������C���s"���||�_�||�_g�|�_d�|�_||�_d�S��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__c�����������������C���s���t�dg|�j�R���S�)z_
        Return the address which this transport is pretending to be bound
        to.
        �UDP)r���r����r���r���r���r����getHost-���s���zMemoryDatagramTransport.getHostc�����������������C���s ���|�j�dur	td��||f|�_�dS�)z>
        Connect this transport to the given address.
        NzAlready connected)r����
ValueError)r���r����portr���r���r����connect4���s���
zMemoryDatagramTransport.connectNc�����������������C���sH���|du�r|�j�}|du�rtd��t|�|�jkrtd��|�j�||f��dS�)z*
        Send the given datagram.
        NzNeed an addresszPacket too big)r���r����lenr���r����append)r����datagram�addrr���r���r����write<���s���zMemoryDatagramTransport.writec�����������������C���s���|�j�����td�S�)z+
        Shut down this transport.
        N)r
����stopProtocolr���r���r���r���r����
stopListeningH���s���
z%MemoryDatagramTransport.stopListeningc�����������������C�������dS��zC
        Dummy implementation to satisfy L{IUDPTransport}.
        Nr���)r����enabledr���r���r����setBroadcastAllowedO�������z+MemoryDatagramTransport.setBroadcastAllowedc�����������������C���r%���r&���r���r���r���r���r����getBroadcastAllowedU���r)���z+MemoryDatagramTransport.getBroadcastAllowedr���)�__name__�
__module__�__qualname__�__doc__r���r���r���r"���r$���r(���r*���r���r���r���r���r
������s����
r
���c�������������������@���s"���e�Zd�ZdZdd��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}.
    c�����������������C���s���t��|���i�|�_d�S�r���)r	���r����udpPortsr���r���r���r���r���l���s���

zMemoryReactor.__init__���� ��c�����������������C���s\���|dkr	�t�dd�}||�jvrnq||�jv�rtd��t||f||�}||�j|<�|�|��|S�)zR
        Pretend to bind a UDP port and connect the given protocol to it.
        r���T����i���zAddress in use)r���r0���r���r
����makeConnection)r���r���r����	interfacer����	transportr���r���r����	listenUDPp���s���

�

zMemoryReactor.listenUDPN)r1���r2���)r+���r,���r-���r.���r���r7���r���r���r���r���r/���_���s����r/���N)r.����randomr����zope.interfacer����zope.interface.verifyr����twisted.internet.addressr����twisted.internet.deferr����twisted.internet.interfacesr���r����twisted.internet.taskr	���r
���r/���r���r���r���r����<module>���s���
G!

📁 Sid Gifari File Manager