File: //usr/lib/python3/dist-packages/twisted/internet/test/__pycache__/test_resolver.cpython-310.pyc
o
�b�L � @ s� d Z ddlmZ ddlmZmZmZmZmZm Z m
Z
m Z m Z ddl
mZmZ ddlmZ ddlmZ ddlmZmZmZ ddlmZmZmZ dd l m Z m Z dd
l m Z m!Z! dd l"m#Z# dd l$m%Z% dd
l&m'Z'm(Z(m)Z)m*Z* ddl+m,Z, ddl-m.Z/ G dd� de,�Z0dd� Z1dd� Z2G dd� d�Z3ee)�G dd� d��Z4G dd� de/�Z5G d d � d e/�Z6ee*�G d d � d ��Z7G d d!� d!e/�Z8G d"d#� d#e!�Z9G d$d%� d%e/�Z:d&S )'zh
Tests for implementations of L{IHostnameResolver} and their interactions with
reactor implementations.
� )� defaultdict) �AF_INET�AF_INET6� AF_UNSPEC�
EAI_NONAME� IPPROTO_TCP�
SOCK_DGRAM� SOCK_STREAM�gaierror� getaddrinfo)�Lock�local)� implementer)� verifyObject)�
LockWorker�Team�createMemoryWorker)�ComplexResolverSimplifier� GAIResolver�SimpleResolverComplexifier)� IPv4Address� IPv6Address)�PluggableResolverMixin� ReactorBase)�Deferred)�DNSLookupError)�IHostnameResolver� IReactorPluggableNameResolver�IResolutionReceiver�IResolverSimple)�
ThreadPool)�SynchronousTestCasec @ � e Zd ZdZdd� ZdS )�DeterministicThreadPoolz6
Create a deterministic L{ThreadPool} object.
c C s" d| _ d| _d| _g | _|| _dS )zE
Create a L{DeterministicThreadPool} from a L{Team}.
� N)�min�max�name�threads�_team)�self�team� r, �E/usr/lib/python3/dist-packages/twisted/internet/test/test_resolver.py�__init__4 s
z DeterministicThreadPool.__init__N)�__name__�
__module__� __qualname__�__doc__r. r, r, r, r- r# / � r# c s4 t � \� } tttt� t� �� fdd�dd� ��| fS )z�
Create a deterministic threadpool.
@return: 2-tuple of L{ThreadPool}, 0-argument C{work} callable; when
C{work} is called, do the work.
c s � S �Nr, r, ��workerr, r- �<lambda>I � z#deterministicPool.<locals>.<lambda>c S s d S r4 r, r, r, r, r- r7 I r8 )r r# r r r r
)�doerr, r5 r- �deterministicPool? s
��r: c s& t � \� } G � fdd�d�}|� | fS )z�
Create a deterministic L{IReactorThreads}
@return: a 2-tuple consisting of an L{IReactorThreads}-like object and a
0-argument callable that will perform one unit of work invoked via that
object's C{callFromThread} method.
c s e Zd Z� fdd�ZdS )z(deterministicReactorThreads.<locals>.CFTc s �� � ��fdd�� d S )Nc s �� i ���S r4 r, r, ��a�f�kr, r- r7 [ s zIdeterministicReactorThreads.<locals>.CFT.callFromThread.<locals>.<lambda>)�do)r* r= r< r> r5 r; r- �callFromThreadZ s z7deterministicReactorThreads.<locals>.CFT.callFromThreadN)r/ r0 r1 r@ r, r5 r, r- �CFTY s rA )r )r9 rA r, r5 r- �deterministicReactorThreadsO s
rB c @ s4 e Zd ZdZdd� Zd dd�Zeeedfdd �Z d
S ) �FakeAddrInfoGetterz/
Test object implementing getaddrinfo.
c C s g | _ tt�| _dS )z1
Create a L{FakeAddrInfoGetter}.
N)�callsr �list�results�r* r, r, r- r. e s zFakeAddrInfoGetter.__init__r c C s4 | j �||||||f� | j| }|r|S ttd��)a�
Mock for L{socket.getaddrinfo}.
@param host: see L{socket.getaddrinfo}
@param port: see L{socket.getaddrinfo}
@param family: see L{socket.getaddrinfo}
@param socktype: see L{socket.getaddrinfo}
@param proto: see L{socket.getaddrinfo}
@param flags: see L{socket.getaddrinfo}
@return: L{socket.getaddrinfo}
z,nodename nor servname provided, or not known)rD �appendrF r
r )r* �host�port�family�socktype�proto�flagsrF r, r, r- r l s
z FakeAddrInfoGetter.getaddrinfo� c C s | j | �|||||f� dS )a�
Add a result for a given hostname. When this hostname is resolved, the
result will be a L{list} of all results C{addResultForHost} has been
called with using that hostname so far.
@param host: The hostname to give this result for. This will be the
next result from L{FakeAddrInfoGetter.getaddrinfo} when passed this
host.
@type canonname: native L{str}
@param sockaddr: The resulting socket address; should be a 2-tuple for
IPv4 or a 4-tuple for IPv6.
@param family: An C{AF_*} constant that will be returned from
C{getaddrinfo}.
@param socktype: A C{SOCK_*} constant that will be returned from
C{getaddrinfo}.
@param proto: An C{IPPROTO_*} constant that will be returned from
C{getaddrinfo}.
@param canonname: A canonical name that will be returned from
C{getaddrinfo}.
@type canonname: native L{str}
N)rF rH )r* rI �sockaddrrK rL rM � canonnamer, r, r- �addResultForHost� s $z#FakeAddrInfoGetter.addResultForHostN)r r r r )
r/ r0 r1 r2 r. r r r r rR r, r, r, r- rC ` s
�rC c @ s8 e Zd ZdZdZdZdd� Zdd� Zdd� Zd d
� Z d S ) � ResultHolderzI
A resolution receiver which holds onto the results it received.
Fc C s
|| _ dS )z>
Create a L{ResultHolder} with a L{UnitTest}.
N)� _testCase)r* �testCaser, r, r- r. � �
zResultHolder.__init__c C s d| _ || _g | _dS )zg
Hostname resolution began.
@param hostResolution: see L{IResolutionReceiver}
TN)�_started� _resolution�
_addresses)r* �hostResolutionr, r, r- �resolutionBegan� s
z ResultHolder.resolutionBeganc C s | j �|� dS )z^
An address was resolved.
@param address: see L{IResolutionReceiver}
N)rY rH )r* �addressr, r, r- �addressResolved� s z ResultHolder.addressResolvedc C s
d| _ dS )z2
Hostname resolution is complete.
TN)�_endedrG r, r, r- �resolutionComplete� rV z ResultHolder.resolutionCompleteN)
r/ r0 r1 r2 rW r^ r. r[ r] r_ r, r, r, r- rS � s
rS c @ r" )� HelperTestsz?
Tests for error cases of helpers used in this module.
c C sD t � \| _| _dd� }| j�|� | �� | �t| �t��d� dS )zq
L{DeterministicThreadPool} will log any exceptions that its "thread"
workers encounter.
c S s dd S )Nr$ r r, r, r, r, r- � divideByZero� s z9HelperTests.test_logErrorsInThreads.<locals>.divideByZeror$ N)r: �pool� doThreadWork� callInThread� assertEqual�len�flushLoggedErrors�ZeroDivisionError)r* ra r, r, r- �test_logErrorsInThreads� s
z#HelperTests.test_logErrorsInThreadsN)r/ r0 r1 r2 ri r, r, r, r- r` � r3 r` c @ s` e Zd ZdZdd� Zdd� Zdd� Zdd � Zd
d � Zd d
� Z dd� Z
dd� Z dd� Z dd� Z
dS )�HostnameResolutionTestsz(
Tests for hostname resolution.
c sD t � \� _� _t� \� _� _t� � _t� j� fdd�� jj �� _
dS )z*
Set up a L{GAIResolver}.
c � � j S r4 �rb r, rG r, r- r7 � � z/HostnameResolutionTests.setUp.<locals>.<lambda>N) r: rb rc rB �reactor�
doReactorWorkrC �getterr r �resolverrG r, rG r- �setUp� s
�z HostnameResolutionTests.setUpc C s� t | �}| j�dd� | j�|d�}| �|j|� | �|jd� | �|j d� | �
� | � � | �|j d� | �|j t
ddd�g� dS ) z�
Resolving an individual hostname that results in one address from
getaddrinfo results in a single call each to C{resolutionBegan},
C{addressResolved}, and C{resolutionComplete}.
�sample.example.com)�4.3.2.1r TF�TCPrt r N)rS rp rR rq �resolveHostName�assertIsrX re rW r^ rc ro rY r �r* �receiver�
resolutionr, r, r- �test_resolveOneHost� s z+HostnameResolutionTests.test_resolveOneHostc C s� t | �}d}d}| jjddd||ftd� | j�|d�}| �|j|� | �|j d� | �|j
d� | � � | � � | �|j
d� | �|j
td dd||�g� d
S ) a
Resolving an individual hostname that results in one address from
getaddrinfo results in a single call each to C{resolutionBegan},
C{addressResolved}, and C{resolutionComplete}; C{addressResolved} will
receive an L{IPv6Address}.
r$ � rs �::1r )rK TFru N)rS rp rR r rq rv rw rX re rW r^ rc ro rY r )r* ry �flowInfo�scopeIDrz r, r, r- �test_resolveOneIPv6Host s ��z/HostnameResolutionTests.test_resolveOneIPv6Hostc C sb t | �}| j�|d�}| �|j|� | �� | �� | �|jd� | �|j d� | �|j
g � dS )a
Resolving a hostname that results in C{getaddrinfo} raising a
L{gaierror} will result in the L{IResolutionReceiver} receiving a call
to C{resolutionComplete} with no C{addressResolved} calls in between;
no failure is logged.
rs TN) rS rq rv rw rX rc ro re rW r^ rY rx r, r, r- �
test_gaierror! s z%HostnameResolutionTests.test_gaierrorc C s` t | �}| jj|d|d�}| �|j|� | �� | �� | jjd \}}}}} }
| � ||� dS )z�
Verify that the given set of address types results in the given C{AF_}
constant being passed to C{getaddrinfo}.
@param addrTypes: iterable of L{IAddress} implementers
@param expectedAF: an C{AF_*} constant
rs )� addressTypesr N)
rS rq rv rw rX rc ro rp rD re ) r* � addrTypes�
expectedAFry rz rI rJ rK rL rM rN r, r, r- �_resolveOnlyTest1 s �z(HostnameResolutionTests._resolveOnlyTestc C � | � tgt� dS )z�
When passed an C{addressTypes} parameter containing only
L{IPv4Address}, L{GAIResolver} will pass C{AF_INET} to C{getaddrinfo}.
N)r� r r rG r, r, r- �test_resolveOnlyIPv4D � z,HostnameResolutionTests.test_resolveOnlyIPv4c C r� )z�
When passed an C{addressTypes} parameter containing only
L{IPv6Address}, L{GAIResolver} will pass C{AF_INET6} to C{getaddrinfo}.
N)r� r r rG r, r, r- �test_resolveOnlyIPv6K r� z,HostnameResolutionTests.test_resolveOnlyIPv6c C s | � ttgt� | � dt� dS )z�
When passed an C{addressTypes} parameter containing both L{IPv4Address}
and L{IPv6Address} (or the default of C{None}, which carries the same
meaning), L{GAIResolver} will pass C{AF_UNSPEC} to C{getaddrinfo}.
N)r� r r r rG r, r, r- �test_resolveBothR s z(HostnameResolutionTests.test_resolveBothc
C s� t | �}| jj|ddd� t | �}| jj|ddd� | �� | �� | �� | �� | jjd \}}}}}}| jjd \}}}} }}| �|t� | �| t � dS )z�
When passed a C{transportSemantics} paramter, C{'TCP'} (the value
present in L{IPv4Address.type} to indicate a stream transport) maps to
C{SOCK_STREAM} and C{'UDP'} maps to C{SOCK_DGRAM}.
� example.comru )�transportSemantics�UDPr r$ N)
rS rq rv rc ro rp rD re r r )
r* ry � receiver2rI rJ rK � socktypeTrM rN � socktypeUr, r, r- �#test_transportSemanticsToSocketType[ s � z;HostnameResolutionTests.test_transportSemanticsToSocketTypec C s� t | �}d}d}ttfD ]}| jjddd||ft|d� | jjddt|d� q | j�|d� | � � | �
� |j \}}}}| � |j
d� | � |j
d� | � |j
d � | � |j
d � d
S ) z�
When L{GAIResolver} receives a C{SOCK_DGRAM} result from
C{getaddrinfo}, it returns a C{'TCP'} L{IPv4Address} or L{IPv6Address};
if it receives C{SOCK_STREAM} then it returns a C{'UDP'} type of same.
r$ r| r� r} r )rK rL )z 127.0.0.3r ru r� N)rS r r rp rR r r rq rv rc ro rY re �type) r* ry r~ r rL �stream4�stream6�dgram4�dgram6r, r, r- � test_socketTypeToAddressTypep s*
��z4HostnameResolutionTests.test_socketTypeToAddressTypeN)r/ r0 r1 r2 rr r{ r� r� r� r� r� r� r� r� r, r, r, r- rj � s rj c @ s" e Zd ZdZdd� Zddd�ZdS ) �SillyResolverSimplez6
Trivial implementation of L{IResolverSimple}
c C s
g | _ dS )zd
Create a L{SillyResolverSimple} with a queue of requests it is working
on.
N)� _requestsrG r, r, r- r. � s
z SillyResolverSimple.__init__r, c C s | j �t� � | j d S )z�
Implement L{IResolverSimple.getHostByName}.
@param name: see L{IResolverSimple.getHostByName}.
@param timeout: see L{IResolverSimple.getHostByName}.
@return: see L{IResolverSimple.getHostByName}.
���)r� rH r )r* r' �timeoutr, r, r- �
getHostByName� s
z!SillyResolverSimple.getHostByNameN)r, )r/ r0 r1 r2 r. r� r, r, r, r- r� � s r� c @ s8 e Zd ZdZdd� Zdd� Zdd� Zdd � Zd
d � Zd S )
�LegacyCompatibilityTestsz�
Older applications may supply an object to the reactor via
C{installResolver} that only provides L{IResolverSimple}.
L{SimpleResolverComplexifier} is a wrapper for an L{IResolverSimple}.
c C s� t � }t|�}t| �}| �|jd� |�|d� | �|jd� | �|jd� | �|jg � |jd � d� | �|jt
ddd�g� | �|jd� dS )z�
L{SimpleResolverComplexifier} translates C{resolveHostName} into
L{IResolutionReceiver.addressResolved}.
Fr� Tr � 192.168.1.1ru N) r� r rS re rW rv r^ rY r� �callbackr �r* �simple�complexry r, r, r- � test_success� s z%LegacyCompatibilityTests.test_successc C s� t � }t|�}t| �}| �|jd� |�|d� | �|jd� | �|jd� | �|jg � |jd � t
d�� | �|jd� | �|jg � dS )z�
L{SimpleResolverComplexifier} translates a known error result from
L{IResolverSimple.resolveHostName} into an empty result.
Fr� Tr �nopeN) r� r rS re rW rv r^ rY r� �errbackr r� r, r, r- � test_failure� s z%LegacyCompatibilityTests.test_failurec C s� t � }t|�}t| �}| �|jd� |�|d� | �|jd� | �|jd� | �|jg � |jd � t
d�� | �t | � t
��d� | �|jd� | �|jg � dS )z�
L{SimpleResolverComplexifier} translates an unknown error result from
L{IResolverSimple.resolveHostName} into an empty result and a logged
error.
Fr� Tr �zowr$ N)
r� r rS re rW rv r^ rY r� r� rh rf rg r� r, r, r- �
test_error� s z#LegacyCompatibilityTests.test_errorc s� t � \� _� _t� \� _� _t� � _t� j� fdd�� jj �� _
t � j
�}� j� dd� |�
d�}|�
d�}� �� � �� � �� � �� � �� �|�jt� � �� �|�d� dS )z�
L{ComplexResolverSimplifier} translates an L{IHostnameResolver} into an
L{IResolverSimple} for applications that still expect the old
interfaces to be in place.
c rk r4 rl r, rG r, r- r7 � rm z:LegacyCompatibilityTests.test_simplifier.<locals>.<lambda>r� )� 192.168.3.4�� znx.example.comr� N)r: rb rc rB rn ro rC rp r r rq r rR r� re �failureResultOfr� r �successResultOf)r* �simpleResolver�success�failurer, rG r- �test_simplifier� s �
z(LegacyCompatibilityTests.test_simplifierc C s� t � }t|�}t| �}|�|dd� | �|jd� | �|jd� | �|jg � |jd � d� | �|jt
ddd�g� | �|jd� dS ) z�
L{SimpleResolverComplexifier} preserves the C{port} argument passed to
C{resolveHostName} in its returned addresses.
r� r� TFr r� ru N) r� r rS rv re rW r^ rY r� r� r r� r, r, r- �test_portNumber� s z(LegacyCompatibilityTests.test_portNumberN) r/ r0 r1 r2 r� r� r� r� r� r, r, r, r- r� � s r� c @ r" )�JustEnoughReactorzT
Just enough subclass implementation to be a valid L{ReactorBase} subclass.
c C s dS )z
Do nothing.
Nr, rG r, r, r- � installWaker r8 z JustEnoughReactor.installWakerN)r/ r0 r1 r2 r� r, r, r, r- r� r3 r� c @ s( e Zd ZdZdd� Zdd� Zdd� ZdS ) �ReactorInstallationTestsz�
Tests for installing old and new resolvers onto a
L{PluggableResolverMixin} and L{ReactorBase} (from which all of Twisted's
reactor implementations derive).
c C s, t � }tt|� tt|j� tt|j� dS )z�
L{PluggableResolverMixin} (and its subclasses) implement both
L{IReactorPluggableNameResolver} and L{IReactorPluggableResolver}.
N)r r r r rq r � nameResolver�r* rn r, r, r- �test_interfaceCompliance s
z1ReactorInstallationTests.test_interfaceCompliancec C s> t � }t� }tt|�|�� | �|jt� | �|jj |� dS )zf
L{PluggableResolverMixin} will wrap an L{IResolverSimple} in a
complexifier.
N)
r r� r r �installResolver�assertIsInstancer� r rw �_simpleResolver)r* rn �itr, r, r- � test_installingOldStyleResolver( s
z8ReactorInstallationTests.test_installingOldStyleResolverc C sX t � }| �|jt� | �|jjt� | �|jt� | �|jj |� | �|jj
|j� dS )zD
L{ReactorBase} defaults to using a L{GAIResolver}.
N) r� r� r� r rw � _getaddrinfor rq r �_reactor�
_nameResolverr� r, r, r- �test_defaultToGAIResolver3 s z2ReactorInstallationTests.test_defaultToGAIResolverN)r/ r0 r1 r2 r� r� r� r, r, r, r- r� s
r� N);r2 � collectionsr �socketr r r r r r r r
r � threadingr r
�zope.interfacer �zope.interface.verifyr �twisted._threadsr r r �twisted.internet._resolverr r r �twisted.internet.addressr r �twisted.internet.baser r �twisted.internet.deferr �twisted.internet.errorr �twisted.internet.interfacesr r r r �twisted.python.threadpoolr �twisted.trial.unittestr! �UnitTestr# r: rB rC rS r` rj r� r� r� r� r, r, r, r- �<module> s: , L' &d