← Back
Editing: test_util.cpython-38.pyc
U W[K � @ s� d Z ddlmZmZ ddlmZ ddlmZ ddlm Z ddl m Z ddl m Z ddlmZ dd lmZmZ ee�G d d � d e��Ze ee� ee�G d d � d e��Ze ee� dS )z$ Utilities for Twisted.names tests. � )�division�absolute_import)� randrange)� implementer)� verifyClass)� IPv4Address)�succeed)�Clock)� IReactorUDP� IUDPTransportc @ sJ e Zd ZdZdd� Zdd� Zdd� Zdd d �Zd d � 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 d| j�� S )z_ Return the address which this transport is pretending to be bound to. �UDP)r )r r �r r r r �getHost/ s z MemoryDatagramTransport.getHostc C s | j dk rtd��||f| _ dS )z> Connect this transport to the given address. NzAlready connected)r � ValueError)r r �portr r r �connect7 s z MemoryDatagramTransport.connectNc C sH |dkr| j }|dkr td��t|�| jkr4td��| j�||f� dS )z* Send the given datagram. NzNeed an addresszPacket too big)r r �lenr r �append)r ZdatagramZaddrr r r �write@ s z MemoryDatagramTransport.writec C s | j �� td�S )z+ Shut down this transport. N)r Z stopProtocolr r r r r � stopListeningM s z%MemoryDatagramTransport.stopListeningc C s dS �zC Dummy implementation to satisfy L{IUDPTransport}. Nr )r �enabledr r r �setBroadcastAllowedU s z+MemoryDatagramTransport.setBroadcastAllowedc C s dS r$ r r r r r �getBroadcastAllowed\ s z+MemoryDatagramTransport.getBroadcastAllowed)N) �__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 s s zMemoryReactor.__init__� � c C sZ |dkr t dd�}|| jkrq q|| jkr2td��t||f||�}|| j|< |�|� |S )zR Pretend to bind a UDP port and connect the given protocol to it. r � i zAddress in use)r r- r r ZmakeConnection)r r r Z interfacer Z transportr r r � listenUDPx s � zMemoryReactor.listenUDPN)r. r/ )r( r) r* r+ r r1 r r r r r, g s r, N)r+ Z __future__r r Zrandomr Zzope.interfacer Zzope.interface.verifyr Ztwisted.internet.addressr Ztwisted.internet.deferr Ztwisted.internet.taskr Ztwisted.internet.interfacesr r �objectr r, r r r r �<module> s L !
Save File
Cancel