HEX
Server: Apache/2.4.52 (Ubuntu)
System: Linux spn-python 5.15.0-89-generic #99-Ubuntu SMP Mon Oct 30 20:42:41 UTC 2023 x86_64
User: arjun (1000)
PHP: 8.1.2-1ubuntu2.20
Disabled: NONE
$ pwd: /lib/python3/dist-packages/twisted/names/test/__pycache__/
Upload Files
File: //lib/python3/dist-packages/twisted/names/test/__pycache__/test_rootresolve.cpython-310.pyc
o

�b4d�@s�dZddlmZddlmZddlmZmZmZm	Z	ddl
mZddlm
Z
mZddlmZmZmZmZmZmZmZmZmZmZmZmZmZmZddlmZm Z dd	l!m"Z"dd
l#m$Z$ddl%m&Z&ddl'm(Z(dd
l)m*Z*m+Z+dd�Z,dd�Z-Gdd�de+�Z.Gdd�de/�Z0dd�Z1Gdd�de+�Z2gd�Z3ee�Gdd�d��Z4eee4�Gdd�de*�Z5Gdd �d �Z6e(j7e8d!d"�Z9d#S)$z.
Test cases for Twisted.names' root resolver.
�)�implementer)�verifyClass)�Deferred�TimeoutError�
gatherResults�succeed)�IResolverSimple)�client�root)�CNAME�ENAME�HS�IN�NS�OK�A�Message�Name�Query�Record_A�Record_CNAME�	Record_NS�RRHeader)�DNSNameError�
ResolverError)�Resolver)�
MemoryReactor)�msg)�util)�SynchronousTestCase�TestCasecCs|\}}}|djS)z�
    From the result of a L{Deferred} returned by L{IResolver.lookupAddress},
    return the payload of the first record in the answer section.
    r��payload��results�ans�auth�add�r(�E/usr/lib/python3/dist-packages/twisted/names/test/test_rootresolve.py�
getOnePayload&s

r*cCst|���S)z�
    From the result of a L{Deferred} returned by L{IResolver.lookupAddress},
    return the first IPv4 address from the answer section.
    )r*�
dottedQuad�r$r(r(r)�
getOneAddress/sr-c@s�eZdZdZdd�Zdd�Zdd�Zgggefdd	�Zd&dd�Z	d
d�Z
dd�Zdd�Zdd�Z
dd�Zdd�Zdd�Zdd�Zdd�Zdd �Zd!d"�Zd#d$�Zd%S)'�RootResolverTestsz3
    Tests for L{twisted.names.root.Resolver}.
    cCs�t�}tg|d�}|�tdtt�dgd|�}|j��\}}|j\\}}t	�}	|	�
|�|�|	jtdtt�g�|�|	j
g�|�|	jg�|�|	jg�g}
|�|
j�|�|
g�|	jdd�=d|	_|	j
�tdtd�d��|j�|	��d�|
d	S)
aK
        Invoke L{Resolver._query} and verify that it sends the correct DNS
        query.  Deliver a canned response to the query and return whatever the
        L{Deferred} returned by L{Resolver._query} fires with.

        @param filter: The value to pass for the C{filter} parameter to
            L{Resolver._query}.
        )�reactor�foo.example.com)�1.1.2.3i)�N��	5.8.13.21r!r)rr�_queryrrr�udpPorts�popitem�_sentPacketsr�fromStr�assertEqual�queries�answers�	authority�
additional�addCallback�append�answerrr�	_protocol�datagramReceived�toStr)�self�filterr/�resolver�d�
portNumber�	transport�packet�address�message�responser(r(r)�
_queryTest<s.	�
�zRootResolverTests._queryTestc	CsJ|�d�\}}}|�|tdtddd�d�g�|�|g�|�|g�dS)a)
        L{Resolver._query} accepts a L{Query} instance and an address, issues
        the query, and returns a L{Deferred} which fires with the response to
        the query.  If a true value is passed for the C{filter} parameter, the
        result is a three-tuple of lists of records.
        Tr0r4r��ttlr!N)rOr:rr)rErAr=r>r(r(r)�test_filteredQuerygs�z$RootResolverTests.test_filteredQueryc	Csd|�d�}|�|t�|�|jg�|�|jtdtddd�d�g�|�|jg�|�|j	g�dS)z�
        Similar to L{test_filteredQuery}, but for the case where a false value
        is passed for the C{filter} parameter.  In this case, the result is a
        L{Message} instance.
        Fr0r4rrPr!N)
rO�assertIsInstancerr:r;r<rrr=r>)rErMr(r(r)�test_unfilteredQueryus
�z&RootResolverTests.test_unfilteredQuerycCsHt|d�}|j|f|j|f|j|ffD]\}}|�dd�|D��q|S)a�
        Create a L{Message} suitable for use as a response to a query.

        @param answers: A C{list} of two-tuples giving data for the answers
            section of the message.  The first element of each tuple is a name
            for the L{RRHeader}.  The second element is the payload.
        @param authority: A C{list} like C{answers}, but for the authority
            section of the response.
        @param additional: A C{list} like C{answers}, but for the
            additional section of the response.
        @param rCode: The response code the message will be created with.

        @return: A new L{Message} initialized with the given values.
        )�rCodec	Ss*g|]\}}t||jt|dt�|d��qS)�CLASSr!)r�TYPE�getattrr)�.0�name�recordr(r(r)�
<listcomp>�s���z.RootResolverTests._respond.<locals>.<listcomp>)rr<r=r>�extend)rEr<r=r>rUrN�section�datar(r(r)�_respond�s
���zRootResolverTests._respond�
cs(dg}t||�}��fdd�}||_|S)a�
        Create and return a new L{root.Resolver} modified to resolve queries
        against the record data represented by C{servers}.

        @param serverResponses: A mapping from dns server addresses to
            mappings.  The inner mappings are from query two-tuples (name,
            type) to dictionaries suitable for use as **arguments to
            L{_respond}.  See that method for details.
        r1c	sltd|j�d|���|D]&}z�|}Wn	tyYq
w||jj|jf}t�jdi|���SdS)NzQuery for QNAME z at r()rrZ�KeyError�typerr`)�query�serverAddresses�timeoutrF�addr�server�records�rE�serverResponsesr(r)rd�s��z-RootResolverTests._getResolver.<locals>.query)rr5)rErk�maximumQueries�rootsrGrdr(rjr)�_getResolver�s



zRootResolverTests._getResolvercCsrdtfdtd�fgdtd�fgd�idtfddtd�fgiid�}|�|�}|�d�}|�t�|�|jd�|S)a
        L{root.Resolver.lookupAddress} looks up the I{A} records for the
        specified hostname by first querying one of the root servers the
        resolver was created with and then following the authority delegations
        until a result is received.
        r0�ns1.example.com�34.55.89.144�r=r>r<�10.0.0.1)�r1�5)rprt�rrrrn�
lookupAddressr?r-r:�rE�serversrGrHr(r(r)�test_lookupAddress�s	�����


z$RootResolverTests.test_lookupAddresscCs�td�}t|_dtfd|fgdtd�fgdtd�fgd�idtfddtd�fgiid�}|�|�}|�d�}|�t�|�|j	td��|S)	z�
        If a response includes a record with a class different from the one
        in the query, it is ignored and lookup continues until a record with
        the right class is found.
        rrr0ro�10.0.0.2)r<r=r>r<�10.0.0.3)rs�rzrt)
rr
rVrrrnrvr?r*r:)rE�badClassrxrGrHr(r(r)�test_lookupChecksClass�s$�����


z(RootResolverTests.test_lookupChecksClasscCs|dtfddtd�fgidtfddtd�fgiidtfddtd�fgiid�}|�|�}|�d�}|�t�|�|jd�|S)z�
        If an intermediate response includes no glue records for the
        authorities, separate queries are made to find those addresses.
        r0r=sns1.example.orgr<rrrz)rs)rrrtrurwr(r(r)�test_missingGlue�s"���
���


z"RootResolverTests.test_missingGluecCs4ddtfdtiii}|�|�}|�d�}|�|t�S)z�
        If a name is missing, L{Resolver.lookupAddress} returns a L{Deferred}
        which fails with L{DNSNameError}.
        rsr0rU)rrrnrv�
assertFailurerrwr(r(r)�test_missingNames���

z"RootResolverTests.test_missingNamecCs0ddtfiii}|�|�}|�d�}|�|t�S)z�
        If a query is responded to with no answers or nameserver records, the
        L{Deferred} returned by L{Resolver.lookupAddress} fires with
        L{ResolverError}.
        rs�example.com)rrnrvr�rrwr(r(r)�test_answerlesss��

z!RootResolverTests.test_answerlesscCsJddtfddtd�fgidtfdtiii}|�|�}|�d�}|�|t�S)z�
        If there is an error resolving the nameserver in a delegation response,
        the L{Deferred} returned by L{Resolver.lookupAddress} fires with that
        error.
        rsr�r=rorU)rrrrnrvr�rrwr(r(r)�test_delegationLookupError.s����


z,RootResolverTests.test_delegationLookupErrorcCsFddtfddtd�fgidtfiii}|�|�}|�d�}|�|t�S)z�
        If there are no records in the response to a lookup of a delegation
        nameserver, the L{Deferred} returned by L{Resolver.lookupAddress} fires
        with L{ResolverError}.
        rsr�r=ro)rrrnrvr�rrwr(r(r)�test_delegationLookupEmptyBs���

z,RootResolverTests.test_delegationLookupEmptycCsfddtfdtidtfddtd�fgiii}|�|�}|�d�}dd�}|�|�|�|jtd��|S)z�
        L{Resolver.lookupNameservers} is like L{Resolver.lookupAddress}, except
        it queries for I{NS} records instead of I{A} records.
        rsr�rUr<rocSs|\}}}|djjS�Nr)r"rZr#r(r(r)�
getOneNamefs
z<RootResolverTests.test_lookupNameservers.<locals>.getOneName)	rrrrrn�lookupNameserversr?r:r)rErxrGrHr�r(r(r)�test_lookupNameserversTs����



z(RootResolverTests.test_lookupNameserversc	Cs|ddtfddtd�fdtd�fgiii}|�|�}|�d�}|�dd��|�|jtdttd�d�tdttd�d�g�|S)	z�
        If a I{CNAME} record is encountered as the answer to a query for
        another record type, that record is returned as the answer.
        rsr�r<�example.netz10.0.0.7cS�|dSr�r(r,r(r(r)�<lambda>�z<RootResolverTests.test_returnCanonicalName.<locals>.<lambda>r!�	rrrrnrvr?r:rrrwr(r(r)�test_returnCanonicalNamens&

����


��z*RootResolverTests.test_returnCanonicalNamec	Cs�ddtfddtd�fgidtfddtd�fgiii}|�|�}|�d�}|�dd��|�|jtdttd�d�tdttd�d�g�|S)	z�
        If no record of the requested type is included in a response, but a
        I{CNAME} record for the query name is included, queries are made to
        resolve the value of the I{CNAME}.
        rsr�r<r�z10.0.0.5cSr�r�r(r,r(r(r)r��r�z<RootResolverTests.test_followCanonicalName.<locals>.<lambda>r!r�rwr(r(r)�test_followCanonicalName�s&����


��z*RootResolverTests.test_followCanonicalNamecCsHddtfddtd�fdtd�fgiii}|�|�}|�d�}|�|t�S)z�
        If there is a cycle between I{CNAME} records in a response, this is
        detected and the L{Deferred} returned by the lookup method fails
        with L{ResolverError}.
        rsr�r<r�)rrrnrvr�rrwr(r(r)�test_detectCanonicalNameLoop�s

����


z.RootResolverTests.test_detectCanonicalNameLoopcCs�dtfddtd�fgidtfddtd�fgiidtfdtd�fgdtd�fgd�idtfddtd	�fgiid
�}|�|d�}|�|�d�t�}|�|d�}|�d�}|�t�|�|j	td	��t
||g�S)
z�
        L{Resolver.lookupAddress} won't issue more queries following
        delegations than the limit passed to its initializer.
        r�r=ror<rzsns2.example.comr{rqz10.0.0.4)rsr|)r{rt��)rrrrnr�rvrr?r*r:r)rErx�failer�failD�	succeeder�succeedDr(r(r)�test_boundedQueries�s0	��������!

z%RootResolverTests.test_boundedQueriesN)ra)�__name__�
__module__�__qualname__�__doc__rOrRrTrr`rnryr~rr�r�r�r�r�r�r�r�r�r(r(r(r)r.7s&+
r.c@�eZdZdZdd�ZdS)�ResolverFactoryArgumentszf
    Raised by L{raisingResolverFactory} with the *args and **kwargs passed to
    that function.
    cCs||_||_dS)z�
        Store the supplied args and kwargs as attributes.

        @param args: Positional arguments.
        @param kwargs: Keyword arguments.
        N��args�kwargs)rEr�r�r(r(r)�__init__��
z!ResolverFactoryArguments.__init__N)r�r�r�r�r�r(r(r(r)r���r�cOs
t||��)a=
    Raise a L{ResolverFactoryArguments} exception containing the
    positional and keyword arguments passed to resolverFactory.

    @param args: A L{list} of all the positional arguments supplied by
        the caller.

    @param kwargs: A L{list} of all the keyword arguments supplied by
        the caller.
    )r�r�r(r(r)�raisingResolverFactory�s
r�c@s(eZdZdZdd�Zdd�Zdd�ZdS)	� RootResolverResolverFactoryTestsz6
    Tests for L{root.Resolver._resolverFactory}.
    cCs tdgtd�}|�|jt�dS)z�
        L{root.Resolver.__init__} accepts a C{resolverFactory}
        argument and assigns it to C{self._resolverFactory}.
        N)�hints�resolverFactory)rr��assertIs�_resolverFactory�rE�rr(r(r)�#test_resolverFactoryArgumentPresentszDRootResolverResolverFactoryTests.test_resolverFactoryArgumentPresentcCs tdgd�}|�|jtj�dS)z�
        L{root.Resolver.__init__} sets L{client.Resolver} as the
        C{_resolverFactory} if a C{resolverFactory} argument is not
        supplied.
        N)r�)rr�r�r	r�r(r(r)�"test_resolverFactoryArgumentAbsentszCRootResolverResolverFactoryTests.test_resolverFactoryArgumentAbsentcCsJt�}tdgt|d�}|�t|jd�}|�d|dgd�f|j|jf�dS)zy
        L{root.Resolver._resolverFactory} is supplied with C{reactor} and
        C{servers} keyword arguments.
        �192.0.2.101)r�r�r/zexample.comr()r�rt)r/rxN)	�objectrr��assertRaisesr�rvr:r�r�)rE�dummyReactorr��er(r(r)�)test_resolverFactoryOnlyExpectedArgumentss�
�zJRootResolverResolverFactoryTests.test_resolverFactoryOnlyExpectedArgumentsN)r�r�r�r�r�r�r�r(r(r(r)r�	s
	r�)
za.root-servers.netzb.root-servers.netzc.root-servers.netzd.root-servers.netze.root-servers.netzf.root-servers.netzg.root-servers.netzh.root-servers.netzi.root-servers.netzj.root-servers.netzk.root-servers.netzl.root-servers.netzm.root-servers.netc@s eZdZdZdd�Zdd�ZdS)�StubResolverz�
    An L{IResolverSimple} implementer which traces all getHostByName
    calls and their deferred results. The deferred results can be
    accessed and fired synchronously.
    cCsg|_g|_dS)z�
        @type calls: L{list} of L{tuple} containing C{args} and
            C{kwargs} supplied to C{getHostByName} calls.
        @type pendingResults: L{list} of L{Deferred} returned by
            C{getHostByName}.
        N)�calls�pendingResults)rEr(r(r)r�Lr�zStubResolver.__init__cOs&|j�||f�t�}|j�|�|S)al
        A fake implementation of L{IResolverSimple.getHostByName}

        @param args: A L{list} of all the positional arguments supplied by
           the caller.

        @param kwargs: A L{list} of all the keyword arguments supplied by
           the caller.

        @return: A L{Deferred} which may be fired later from the test
            fixture.
        )r�r@rr�)rEr�r�rHr(r(r)�
getHostByNameVs
zStubResolver.getHostByNameN)r�r�r�r�r�r�r(r(r(r)r�Ds
r�c@sHeZdZdZdd�Zdd�Zdd�Zdd	�Zd
d�Zdd
�Z	dd�Z
dS)�BootstrapTestsz%
    Tests for L{root.bootstrap}
    cCst�t��}|�|tj�dS)zl
        L{root.bootstrap} returns an object which is initially a
        L{root.DeferredResolver}.
        N)r
�	bootstrapr�rS�DeferredResolver)rE�deferredResolverr(r(r)�test_returnsDeferredResolverqsz+BootstrapTests.test_returnsDeferredResolvercCs,t�}t�|�|�|jdd�tD��dS)z�
        The L{IResolverSimple} supplied to L{root.bootstrap} is used to lookup
        the IP addresses of the 13 root name servers.
        cSsg|]}|fif�qSr(r()rY�sr(r(r)r\�sz=BootstrapTests.test_resolves13RootServers.<locals>.<listcomp>N)r�r
r�r:r��ROOT_SERVERS)rE�stubResolverr(r(r)�test_resolves13RootServersys
z)BootstrapTests.test_resolves13RootServerscCs6t�}t�|�}|jD]}|�d�q|�|t�dS)z�
        The L{root.DeferredResolver} initially returned by L{root.bootstrap}
        becomes a L{root.Resolver} when the supplied resolver has successfully
        looked up all root hints.
        r�N)r�r
r�r��callbackrSr�rEr�r�rHr(r(r)�test_becomesResolver�s


z#BootstrapTests.test_becomesResolvercCs>t�}t�|�}|jD]}|�d�q|�|jdgd�dS)z�
        The L{root.Resolver} which eventually replaces L{root.DeferredResolver}
        is supplied with the IP addresses of the 13 root servers.
        r��
N)r�r
r�r�r�r:r�r�r(r(r)�test_resolverReceivesRootHints�s


z-BootstrapTests.test_resolverReceivesRootHintscs^t�}t�|��t|j�}t|�}|D]}|�d�q|�t����fdd�}|�	|�dS)z�
        The L{root.Resolver} is eventually created, even if some of the root
        hint lookups fail. Only the working root hint IP addresses are supplied
        to the L{root.Resolver}.
        r�cs���jdgd�dS)Nr���r:r���res�r�rEr(r)�
checkHints�szFBootstrapTests.test_continuesWhenSomeRootHintsFail.<locals>.checkHintsN)
r�r
r��iterr��nextr��errbackr�addBoth�rEr�r$�d1rHr�r(r�r)�#test_continuesWhenSomeRootHintsFail�s

z2BootstrapTests.test_continuesWhenSomeRootHintsFailcsnt�}t�|��t|j�}t|�}|D]}|�t��q|�t����fdd�}|�|���	�j
t�dS)z�
        The L{root.Resolver} is eventually created, even if all of the root hint
        lookups fail. Pending and new lookups will then fail with
        AttributeError.
        cs���jg�dS�Nr�r�r�r(r)r��szEBootstrapTests.test_continuesWhenAllRootHintsFail.<locals>.checkHintsN)r�r
r�r�r�r�r�rr��
addCleanup�flushLoggedErrorsr�r(r�r)�"test_continuesWhenAllRootHintsFail�s


z1BootstrapTests.test_continuesWhenAllRootHintsFailcCs<t�}tj|td�}|jD]}|�d�q
|�|jt�dS)z�
        L{root.bootstrap} accepts a C{resolverFactory} argument which is passed
        as an argument to L{root.Resolver} when it has successfully looked up
        root hints.
        )r�r�N)r�r
r�r�r�r�r�r�r�r(r(r)�test_passesResolverFactory�s�
z)BootstrapTests.test_passesResolverFactoryN)r�r�r�r�r�r�r�r�r�r�r�r(r(r(r)r�ls	r�c@r�)�StubDNSDatagramProtocolz�
    A do-nothing stand-in for L{DNSDatagramProtocol} which can be used to avoid
    network traffic in tests where that kind of thing doesn't matter.
    cOst�Sr�)r)rE�a�kwr(r(r)rd�szStubDNSDatagramProtocol.queryN)r�r�r�r�rdr(r(r(r)r��r�r�zbtwisted.names.root.retry is deprecated since Twisted 10.0.  Use a Resolver object for retry logic.)�categoryrMN):r��zope.interfacer�zope.interface.verifyr�twisted.internet.deferrrrr�twisted.internet.interfacesr�
twisted.namesr	r
�twisted.names.dnsrrr
rrrrrrrrrrr�twisted.names.errorrr�twisted.names.rootr�twisted.names.test.test_utilr�twisted.python.logr�
twisted.trialr�twisted.trial.unittestrr r*r-r.�	Exceptionr�r�r�r�r�r�r��suppress�DeprecationWarning�_retrySuppressionr(r(r(r)�<module>s@@	6*
$f

�
@ Telegram