Файловый менеджер - Редактировать - /opt/imh-python/lib/python3.9/site-packages/twisted/internet/test/__pycache__/test_udp.cpython-39.pyc
Ðазад
a k�h�A � @ sL d Z ddlZddlmZ ddlmZ ddlmZmZ ddl m Z m Z ddl m Z mZ ddlmZmZmZmZ dd lmZ dd lmZmZ dd lmZ dd lm Z dd l m Z m Z ddl m!Z!m"Z" ddl#m$Z$ dd� Z%e%� Z&dd� Z'G dd� de�Z(G dd� d�Z)G dd� dee)e(�Z*G dd� dee)e(�Z+e,� �-e*�.� � e,� �-e+�.� � dS ) zU Tests for implementations of L{IReactorUDP} and the UDP parts of L{IReactorSocket}. � N)� implementer)� verifyObject)�defer�error)� IPv4Address� IPv6Address)�Deferred� maybeDeferred)�IListeningPort�ILoggingContext�IReactorSocket� IReactorUDP)�DatagramProtocol)�LogObserverMixin� findFreePort)�ReactorBuilder)�context)� ILogContext�err)� GoodClient�Server)�SkipTestc C sJ d} d}z t � t j�} | �d� d}W n ty8 Y n0 | rF| �� |S )z4Returns True if the system can bind an IPv6 address.NF)�::1r T)�socket�AF_INET6�bind�OSError�close)�sock�has_ipv6� r ��/root/rpmbuild/BUILDROOT/imh-python39-modules-3.9.7-92.el8.x86_64/opt/imh-python/lib/python3.9/site-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! �skipWithoutIPv65 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! � logPrefixJ s z[DatagramTransportTestsMixin.test_startedListeningLogMessage.<locals>.SomeProtocol.logPrefixN)�__name__� __module__� __qualname__r* r r r r! � SomeProtocolH 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_startedListeningLogMessage@ s z;DatagramTransportTestsMixin.test_startedListeningLogMessagec sz | � � � | �� �| ��t� ��d��� j� d�}�fdd��� ��fdd�}��|� | ��� | �|f� d d � d S ) 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 z Closed)c s � � � d S �N��stop��ignored�r7 r r! � stopReactor] s zNDatagramTransportTestsMixin.test_connectionLostLogMessage.<locals>.stopReactorc s � d d �= t �j���� d S r< )r � stopListening� addCallbackr )r6 r9 rB r r! �doStopListening` s zRDatagramTransportTestsMixin.test_connectionLostLogMessage.<locals>.doStopListeningr r/ N) r0 r1 r2 r r3 r4 ZcallWhenRunning� runReactorr5 )r) r: rE r )r6 r9 r7 rB r! � test_connectionLostLogMessageS 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� transportrC r( r r r! � startProtocolw s zdDatagramTransportTestsMixin.test_stopProtocolScheduling.<locals>.DisconnectingProtocol.startProtocolc s d| _ | j| _� �� d S )NT)�stoppedrI �stoppedInStartr> r( rA r r! � stopProtocol} s zcDatagramTransportTestsMixin.test_stopProtocolScheduling.<locals>.DisconnectingProtocol.stopProtocolN) r+ r, r- rH rL rI rM rK rN r rA r r! �DisconnectingProtocolp s rO N) r r1 r2 rF � assertTruerH rL Z assertFalserM )r) rO r8 r rA r! �test_stopProtocolSchedulingi s z7DatagramTransportTestsMixin.test_stopProtocolSchedulingN)r+ r, r- �__doc__r; rG rQ 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)r1 r2 r rP r r �r) r7 r4 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)r4 � interfaceZUDPN) r r � SOCK_DGRAMr1 r2 r r5 r3 r )r) �hostZ portNumberr7 r4 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 �rW N)r1 r2 r r3 r5 rY ZassertIsInstancer )r) r7 r4 �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.comr[ N)r1 � assertRaisesr �InvalidAddressError� listenUDPr )r) r7 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< )rb r( r r r! r* � s zSUDPPortTestsMixin.test_logPrefix.<locals>.CustomLogPrefixDatagramProtocol.logPrefixc S s. | j d ur*| j }d | _ |�t�t�d � d S )Nrc )rc �callbackr �getr )r) �bytesr\ rc r r r! �datagramReceived� s zZUDPPortTestsMixin.test_logPrefix.<locals>.CustomLogPrefixDatagramProtocol.datagramReceivedN)r+ r, r- re r* ri r r r r! � CustomLogPrefixDatagramProtocol� s rj zCustom Datagramsc s � � d| � d S )NzCustom Datagrams (UDP))r5 )rc r( r r! � gotSystem� s z3UDPPortTestsMixin.test_logPrefix.<locals>.gotSystemc s � � � S r< r= r? rA r r! �<lambda>� � z2UDPPortTestsMixin.test_logPrefix.<locals>.<lambda>s some bytes� 127.0.0.1N) r r1 rc r2 r3 rD � addErrbackr �writer4 rF )r) rj r8 �dr4 �addressrk r )r7 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! re � s zMUDPPortTestsMixin.test_writeSequence.<locals>.SimpleDatagramProtocol.__init__c S s | j �|� d S r< )r rf )r) �datar\ r r r! ri � s zUUDPPortTestsMixin.test_writeSequence.<locals>.SimpleDatagramProtocol.datagramReceivedN)r+ r, r- re ri r r r r! �SimpleDatagramProtocol� s ru )s somes bytess tos writec s �� d�� �| � d S )Nrm )r5 �join)rt )� dataToWriter) r r! �gotData� s z5UDPPortTestsMixin.test_writeSequence.<locals>.gotDatac s � � � S r< r= r? rA r r! rl � rm z6UDPPortTestsMixin.test_writeSequence.<locals>.<lambda>rn N) r r1 r r2 r3 rD ro r Z writeSequencer4 rF )r) ru r8 r r4 rr rx r )rw r7 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)r1 r2 r �assertIn�strr3 r4 rT 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)r1 r2 r rz �reprr3 r4 r{ rT 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 r[ 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 )rJ rp r3 r4 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? rA r r! �cbServerReceived, s zEUDPPortTestsMixin.test_writeToIPv6Interface.<locals>.cbServerReceivedr r� N�r1 r r r ZstartedDeferredr2 r rJ r3 Z gatherResultsrD ro r rF Zpacketsr5 rY r4 �r) Z serverStartedZ clientStartedZcAddrr� r� rq �packetr �r� r7 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 r[ c s4 � j �d�j �� j� � j �d� t�� }�_|S )r r r� )rJ �connectr3 r4 rp r r r� r� r� r r! r� N s zMUDPPortTestsMixin.test_connectedWriteToIPv6Interface.<locals>.cbClientStartedc s � � � dS r� r= r? rA r r! r� \ s zNUDPPortTestsMixin.test_connectedWriteToIPv6Interface.<locals>.cbServerReceivedr r� Nr� r� r r� r! �"test_connectedWriteToIPv6Interface> 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�r1 r2 r r^ r r_ rp rT r r r! �/test_writingToHostnameRaisesInvalidAddressErroro 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}. rn r[ r� )r r� Nr� rT r r r! �1test_writingToIPv6OnIPv4RaisesInvalidAddressErrorz s zCUDPPortTestsMixin.test_writingToIPv6OnIPv4RaisesInvalidAddressErrorc C s2 | � � }| j|t� dd�}| �tj|jdd� dS )r� r r[ r� )rn r� Nr� rT 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)r1 r2 r r^ r r_ r� rT 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)r1 r2 r ZsetBroadcastAllowedrP ZgetBroadcastAllowedrT r r r! �test_allowBroadcast� s z%UDPPortTestsMixin.test_allowBroadcastN)r+ r, r- rR rU rZ r& r] ra rs ry r| r~ r� r� r� r� r� r� r� r r r r! rS � s* $ , 0 rS 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. )rW � maxPacketSize)r` )r) r7 r8 r4 rW r� r r r! r2 � s �z&UDPServerTestsBuilder.getListeningPortN)r r� r� )r+ r, r- rR r �requiredInterfacesr2 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|v r.tj}t�||�d d }ntj}||f}t�|tj�}|�|� |�d� z(|� |� � |j ||�W |� � |� � S |� � |� � 0 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 � getaddrinfo�AF_INETrX r � setblockingZadoptDatagramPort�fileno�familyr r ) r) r7 r8 r4 rW r� �domainrr ZportSockr r r! r2 � s* �� � z(UDPFDServerTestsBuilder.getListeningPortN)r r� r� )r+ r, r- rR r r� r2 r r r r! r� � s �r� )/rR r Zzope.interfacer Zzope.interface.verifyr Ztwisted.internetr r Ztwisted.internet.addressr r Ztwisted.internet.deferr r Ztwisted.internet.interfacesr r r r Ztwisted.internet.protocolr Z&twisted.internet.test.connectionmixinsr r Z#twisted.internet.test.reactormixinsr Ztwisted.pythonr Ztwisted.python.logr r Ztwisted.test.test_udpr r Ztwisted.trial.unittestr r" r# r& r' rS r� r� �globals�updateZmakeTestCaseClassesr r r r! �<module> s: Q � �9
| ver. 1.1 | |
.
| PHP 8.3.30 | Ð“ÐµÐ½ÐµÑ€Ð°Ñ†Ð¸Ñ Ñтраницы: 0 |
proxy
|
phpinfo
|
ÐаÑтройка