← Back
Editing: test_udp.cpython-38.pyc
U W[�B � @ sb d Z ddlmZmZ eZddlZddlmZ ddl m Z ddl m Z ddl mZmZ ddlmZ dd lmZmZ dd lmZmZmZmZ dd lmZm Z dd l m Z dd l m Z m!Z! ddl"m#Z#m$Z$ ddl%m&Z&m'Z' ddl(m)Z) dd� Z*e*� Z+dd� Z,G dd� de �Z-G dd� de.�Z/G dd� dee/e-�Z0G dd � d ee/e-�Z1e2� �3e0�4� � e2� �3e1�4� � dS ) zU Tests for implementations of L{IReactorUDP} and the UDP parts of L{IReactorSocket}. � )�division�absolute_importN)� implementer)� verifyObject)�context)� ILogContext�err)�ReactorBuilder)�Deferred� maybeDeferred)�ILoggingContext�IListeningPort� IReactorUDP�IReactorSocket)� IPv4Address� IPv6Address)�DatagramProtocol)�LogObserverMixin� findFreePort)�defer�error)�Server� GoodClient)�SkipTestc C sN d} d}z t � t j�} | �d� d}W n t jk r< Y nX | rJ| �� |S )z5 Returns True if the system can bind an IPv6 address.NF)�::1r T)�socket�AF_INET6�bindr �close)ZsockZhas_ipv6� r �@/usr/lib/python3/dist-packages/twisted/internet/test/test_udp.py� _has_ipv6" s r! c C s t s d| _| S )Nz.Does not work on systems without IPv6 support.)�HAS_IPV6�skip)�fr r r �skipWithoutIPv66 s r% c @ s( e Zd ZdZdd� Zdd� Zdd� ZdS ) �DatagramTransportTestsMixinzP Mixin defining tests which apply to any port/datagram based transport. c C sd | � � }| �� }tt�G dd� dt��}|� }| �||�}d|�� jf }| �|f|d d � dS )zu When a port starts, a message including a description of the associated protocol is logged. c @ s e Zd Zdd� ZdS )zQDatagramTransportTestsMixin.test_startedListeningLogMessage.<locals>.SomeProtocolc S s dS )NzCrazy Protocolr ��selfr r r � logPrefixK s z[DatagramTransportTestsMixin.test_startedListeningLogMessage.<locals>.SomeProtocol.logPrefixN)�__name__� __module__� __qualname__r) r r r r � SomeProtocolI s r- z Crazy Protocol starting on %dr �messageN) �observe� buildReactorr r r �getListeningPort�getHost�port� assertEqual)r( �loggedMessages�reactorr- �protocol�p�expectedMessager r r � test_startedListeningLogMessageA s z;DatagramTransportTestsMixin.test_startedListeningLogMessagec sx | � � � | �� �| ��t� ��d��� jf }�fdd��� ��fdd�}��|� | ��� | �|f� d d � dS ) z� When a connection is lost a message is logged containing an address identifying the port and the fact that it was closed. z(UDP Port %s Closed)c s � � � d S �N��stop��ignored�r6 r r � stopReactor^ s zNDatagramTransportTestsMixin.test_connectionLostLogMessage.<locals>.stopReactorc s � d d �= t �j���� d S r; )r � stopListening� addCallbackr )r5 r8 rA r r �doStopListeninga s zRDatagramTransportTestsMixin.test_connectionLostLogMessage.<locals>.doStopListeningr r. N) r/ r0 r1 r r2 r3 ZcallWhenRunning� runReactorr4 )r( r9 rD r )r5 r8 r6 rA r � test_connectionLostLogMessageT s z9DatagramTransportTestsMixin.test_connectionLostLogMessagec s` G � fdd�dt �}| �� � |� }| �� |� | �� � | �|j� | �|j� | �|j� dS )z� L{DatagramProtocol.stopProtocol} is called asynchronously (ie, not re-entrantly) when C{stopListening} is used to stop the datagram transport. c s0 e Zd ZdZdZdZdZdd� Z� fdd�ZdS )zVDatagramTransportTestsMixin.test_stopProtocolScheduling.<locals>.DisconnectingProtocolFc S s d| _ d| _| j�� d| _d S )NTF)�started�inStartProtocol� transportrB r' r r r � startProtocolx s zdDatagramTransportTestsMixin.test_stopProtocolScheduling.<locals>.DisconnectingProtocol.startProtocolc s d| _ | j| _� �� d S )NT)�stoppedrH �stoppedInStartr= r' r@ r r � stopProtocol~ s zcDatagramTransportTestsMixin.test_stopProtocolScheduling.<locals>.DisconnectingProtocol.stopProtocolN) r* r+ r, rG rK rH rL rJ rM r r@ r r �DisconnectingProtocolq s rN N) r r0 r1 rE � assertTruerG rK Z assertFalserL )r( rN r7 r r@ r �test_stopProtocolSchedulingk s z7DatagramTransportTestsMixin.test_stopProtocolSchedulingN)r* r+ r, �__doc__r: rF rP r r r r r& = s r& c @ s� e Zd ZdZdd� Zdd� Zedd� �Zdd � Zd d � Z d d � Z dd� Z dd� Z edd� �Z edd� �Zdd� Zedd� �Zedd� �Zd d � Zd d � Zd S )!�UDPPortTestsMixinzY Tests for L{IReactorUDP.listenUDP} and L{IReactorSocket.adoptDatagramPort}. c C s* | � � }| �|t� �}| �tt|�� dS )zY L{IReactorUDP.listenUDP} returns an object providing L{IListeningPort}. N)r0 r1 r rO r r �r( r6 r3 r r r �test_interface� s z UDPPortTestsMixin.test_interfacec C sH t tjd�\}}| �� }| j|t� ||d�}| �|�� td||�� dS )z� L{IListeningPort.getHost} returns an L{IPv4Address} giving a dotted-quad of the IPv4 address the port is listening on as well as the port number. )�type)r3 � interfaceZUDPN) r r � SOCK_DGRAMr0 r1 r r4 r2 r )r( �hostZ portNumberr6 r3 r r r � test_getHost� s � �z UDPPortTestsMixin.test_getHostc C s@ | � � }| j|t� dd�}|�� }| �|jd� | �|t� dS )zr L{IListeningPort.getHost} returns an L{IPv6Address} when listening on an IPv6 interface. r �rV N)r0 r1 r r2 r4 rX ZassertIsInstancer )r( r6 r3 �addrr r r �test_getHostIPv6� s �z"UDPPortTestsMixin.test_getHostIPv6c C s& | � � }| jtj|jt� ddd� dS )z� An L{InvalidAddressError} is raised when trying to listen on an address that isn't a valid IPv4 or IPv6 address. r z example.comrZ N)r0 � assertRaisesr �InvalidAddressError� listenUDPr )r( r6 r r r �test_invalidInterface� s �z'UDPPortTestsMixin.test_invalidInterfacec s� G dd� dt �}��� � |d�}|j}��� |�}|�� }�fdd�}|�|� |�t� |�� fdd�� |�dd |j f� �� � � d S ) z� Datagram transports implement L{ILoggingContext.logPrefix} to return a message reflecting the protocol they are running. c @ s$ e Zd Zdd� Zdd� Zdd� ZdS )zIUDPPortTestsMixin.test_logPrefix.<locals>.CustomLogPrefixDatagramProtocolc S s || _ t� | _d S r; )�_prefixr �system)r( �prefixr r r �__init__� s zRUDPPortTestsMixin.test_logPrefix.<locals>.CustomLogPrefixDatagramProtocol.__init__c S s | j S r; )ra r' r r r r) � s zSUDPPortTestsMixin.test_logPrefix.<locals>.CustomLogPrefixDatagramProtocol.logPrefixc S s. | j d k r*| j }d | _ |�t�t�d � d S )Nrb )rb �callbackr �getr )r( �bytesr[ rb r r r �datagramReceived� s zZUDPPortTestsMixin.test_logPrefix.<locals>.CustomLogPrefixDatagramProtocol.datagramReceivedN)r* r+ r, rd r) rh r r r r � CustomLogPrefixDatagramProtocol� s ri zCustom Datagramsc s � � d| � d S )NzCustom Datagrams (UDP))r4 )rb r' r r � gotSystem� s z3UDPPortTestsMixin.test_logPrefix.<locals>.gotSystemc s � � � S r; r< r> r@ r r �<lambda>� � z2UDPPortTestsMixin.test_logPrefix.<locals>.<lambda>s some bytes� 127.0.0.1N) r r0 rb r1 r2 rC � addErrbackr �writer3 rE )r( ri r7 �dr3 �addressrj r )r6 r( r �test_logPrefix� s z UDPPortTestsMixin.test_logPrefixc s� G dd� dt �}��� �|� }|j}���|�}|�� }d� � �fdd�}|�|� |�t� |��fdd�� |�� d|j f� �� �� d S ) zH Write a sequence of L{bytes} to a L{DatagramProtocol}. c @ s e Zd Zdd� Zdd� ZdS )zDUDPPortTestsMixin.test_writeSequence.<locals>.SimpleDatagramProtocolc S s t � | _d S r; )r r r' r r r rd � s zMUDPPortTestsMixin.test_writeSequence.<locals>.SimpleDatagramProtocol.__init__c S s | j �|� d S r; )r re )r( �datar[ r r r rh � s zUUDPPortTestsMixin.test_writeSequence.<locals>.SimpleDatagramProtocol.datagramReceivedN)r* r+ r, rd rh r r r r �SimpleDatagramProtocol� s rt )s somes bytess tos writec s �� d�� �| � d S )Nrl )r4 �join)rs )� dataToWriter( r r �gotData� s z5UDPPortTestsMixin.test_writeSequence.<locals>.gotDatac s � � � S r; r< r> r@ r r rk � rl z6UDPPortTestsMixin.test_writeSequence.<locals>.<lambda>rm N) r r0 r r1 r2 rC rn r Z writeSequencer3 rE )r( rt r7 r r3 rq rw r )rv r6 r( r �test_writeSequence� s z$UDPPortTestsMixin.test_writeSequencec C s4 | � � }| �|t� �}| �t|�� j�t|�� dS )zQ C{str()} on the listening port object includes the port number. N)r0 r1 r �assertIn�strr2 r3 rS r r r �test_str s zUDPPortTestsMixin.test_strc C s4 | � � }| �|t� �}| �t|�� j�t|�� dS )zR C{repr()} on the listening port object includes the port number. N)r0 r1 r ry �reprr2 r3 rz rS r r r � test_repr s zUDPPortTestsMixin.test_reprc s� | � � �t� �t�� }�_| j��dd� t� � t�� }� _| j�� dd� � j�� }� �fdd�}�fdd�}t� ||g�}|� |� |� |� |� t � | � �� �jd }| �|d|j|jff� d S ) zS Writing to an IPv6 UDP socket on the loopback interface succeeds. r rZ c s, � j �dd�j �� jf� t�� }�_|S )�� Send a datagram from the client once it's started. @param ignored: a list of C{[None, None]}, which is ignored @returns: a deferred which fires when the server has received a datagram. � spamr )rI ro r2 r3 r r �packetReceived�r? ZserverReceived��client�serverr r �cbClientStarted# s �zDUDPPortTestsMixin.test_writeToIPv6Interface.<locals>.cbClientStartedc s � � � dS �z� Stop the reactor after a datagram is received. @param ignored: L{None}, which is ignored @returns: L{None} Nr< r> r@ r r �cbServerReceived0 s zEUDPPortTestsMixin.test_writeToIPv6Interface.<locals>.cbServerReceivedr r N�r0 r r r ZstartedDeferredr1 r rI r2 Z gatherResultsrC rn r rE Zpacketsr4 rX r3 �r( Z serverStartedZ clientStartedZcAddrr� r� rp Zpacketr �r� r6 r� r �test_writeToIPv6Interface s" z+UDPPortTestsMixin.test_writeToIPv6Interfacec s� | � � �t� �t�� }�_| j��dd� t� � t�� }� _| j�� dd� � j�� }� �fdd�}�fdd�}t� ||g�}|� |� |� |� |� t � | � �� �jd }| �|d|j|jff� d S ) z� An IPv6 address can be passed as the C{interface} argument to L{listenUDP}. The resulting Port accepts IPv6 datagrams. r rZ c s4 � j �d�j �� j� � j �d� t�� }�_|S )r~ r r )rI �connectr2 r3 ro r r r� r� r� r r r� S s zMUDPPortTestsMixin.test_connectedWriteToIPv6Interface.<locals>.cbClientStartedc s � � � dS r� r< r> r@ r r r� a s zNUDPPortTestsMixin.test_connectedWriteToIPv6Interface.<locals>.cbServerReceivedr r Nr� r� r r� r �"test_connectedWriteToIPv6InterfaceC s" z4UDPPortTestsMixin.test_connectedWriteToIPv6Interfacec C s. | � � }| �|t� �}| �tj|jdd� dS )zn Writing to a hostname instead of an IP address will raise an L{InvalidAddressError}. �spam)�example.invalid� N�r0 r1 r r] r r^ ro rS r r r �/test_writingToHostnameRaisesInvalidAddressErroru s �zAUDPPortTestsMixin.test_writingToHostnameRaisesInvalidAddressErrorc C s2 | � � }| j|t� dd�}| �tj|jdd� dS )�l Writing to an IPv6 address on an IPv4 socket will raise an L{InvalidAddressError}. rm rZ r� )r r� Nr� rS r r r �1test_writingToIPv6OnIPv4RaisesInvalidAddressError� s � �zCUDPPortTestsMixin.test_writingToIPv6OnIPv4RaisesInvalidAddressErrorc C s2 | � � }| j|t� dd�}| �tj|jdd� dS )r� r rZ r� )rm r� Nr� rS r r r �1test_writingToIPv4OnIPv6RaisesInvalidAddressError� s � �zCUDPPortTestsMixin.test_writingToIPv4OnIPv6RaisesInvalidAddressErrorc C s. | � � }| �|t� �}| �tj|jdd� dS )zq Connecting to a hostname instead of an IP address will raise an L{InvalidAddressError}. r� r� N)r0 r1 r r] r r^ r� rS r r r �2test_connectingToHostnameRaisesInvalidAddressError� s �zDUDPPortTestsMixin.test_connectingToHostnameRaisesInvalidAddressErrorc C s2 | � � }| �|t� �}|�d� | �|�� � dS )zk L{IListeningPort.setBroadcastAllowed} sets broadcast to be allowed on the socket. TN)r0 r1 r ZsetBroadcastAllowedrO ZgetBroadcastAllowedrS r r r �test_allowBroadcast� s z%UDPPortTestsMixin.test_allowBroadcastN)r* r+ r, rQ rT rY r% r\ r` rr rx r{ r} r� r� r� r� r� r� r� r r r r rR � s* # . 1 rR c @ s e Zd ZdZefZddd�ZdS ) �UDPServerTestsBuilderzM Run L{UDPPortTestsMixin} tests using newly created UDP sockets. r � � c C s |j ||||d�S )aB Get a UDP port from a reactor. @param reactor: A reactor used to build the returned L{IListeningPort} provider. @type reactor: L{twisted.internet.interfaces.IReactorUDP} @see: L{twisted.internet.IReactorUDP.listenUDP} for other argument and return types. )rV � maxPacketSize)r_ )r( r6 r7 r3 rV r� r r r r1 � s �z&UDPServerTestsBuilder.getListeningPortN)r r� r� )r* r+ r, rQ r �requiredInterfacesr1 r r r r r� � s �r� c @ s e Zd ZdZefZddd�ZdS ) �UDPFDServerTestsBuilderzC Run L{UDPPortTestsMixin} tests using adopted UDP sockets. r r� r� c C s� t �|�r�d|kr.tj}t�||�d d }ntj}||f}t�|tj�}|�|� |�d� z|� |� � |j ||�W �S |� � |� � X nt d��dS )a� Get a UDP port from a reactor, wrapping an already-initialized file descriptor. @param reactor: A reactor used to build the returned L{IListeningPort} provider. @type reactor: L{twisted.internet.interfaces.IReactorSocket} @param port: A port number to which the adopted socket will be bound. @type port: C{int} @param interface: The local IPv4 or IPv6 address to which the adopted socket will be bound. defaults to '', ie all IPv4 addresses. @type interface: C{str} @see: L{twisted.internet.IReactorSocket.adoptDatagramPort} for other argument and return types. �:r � Fz'Reactor does not provide IReactorSocketN)r Z providedByr r Z getaddrinfoZAF_INETrW r Z setblocking�filenor ZadoptDatagramPortZfamilyr ) r( r6 r7 r3 rV r� Zdomainrq ZportSockr r r r1 � s& � z(UDPFDServerTestsBuilder.getListeningPortN)r r� r� )r* r+ r, rQ r r� r1 r r r r r� � s �r� )5rQ Z __future__r r rU Z __metaclass__r Zzope.interfacer Zzope.interface.verifyr Ztwisted.pythonr Ztwisted.python.logr r Z#twisted.internet.test.reactormixinsr Ztwisted.internet.deferr r Ztwisted.internet.interfacesr r r r Ztwisted.internet.addressr r Ztwisted.internet.protocolr Z&twisted.internet.test.connectionmixinsr r Ztwisted.internetr r Ztwisted.test.test_udpr r Ztwisted.trial.unittestr r! r" r% r& �objectrR r� r� �globals�updateZmakeTestCaseClassesr r r r �<module> sB Q & � �7
Save File
Cancel