← Back
Editing: test_base.cpython-38.pyc
U �q�[94 � @ sr d Z ddlZzddlmZ W n ek r< ddlmZ Y nX ddlmZ ddlmZ ddl m Z m Z m Z ddl mZ ddlmZ dd lmZ dd lmZmZmZ dd lmZ dd lmZmZ ee e �G d d� de ��Z G dd� de�Z dd� Z G dd� de �Z G dd� de e�Z!G dd� de e�Z"G dd� de�Z#G dd � d e�Z$z ddl%Z%W n ek �rl d e$_&Y nX dS ) z% Tests for L{twisted.internet.base}. � N)�Queue)� implementer)� ThreadPool)� IReactorTime�IReactorThreads�IResolverSimple)�DNSLookupError)� FirstOneWins)�Deferred)�ThreadedResolver� DelayedCall� ReactorBase)�Clock)�TestCase�SkipTestc @ s0 e Zd ZdZdd� Zdd� Zdd� Zdd � Zd S ) � FakeReactorzl A fake reactor implementation which just supports enough reactor APIs for L{ThreadedResolver}. c s> t � � _� jj� _t� � _� j�� � fdd�� _t� � _d S )Nc s � j S �N)� _threadpool� ��selfr �A/usr/lib/python3/dist-packages/twisted/internet/test/test_base.py�<lambda>( � z&FakeReactor.__init__.<locals>.<lambda>) r �_clockZ callLaterr r �startZ getThreadPoolr � _threadCallsr r r r �__init__" s zFakeReactor.__init__c O s | j �|||f� d S r )r �put�r �f�args�kwargsr r r �callFromThread- s zFakeReactor.callFromThreadc C s | j �� \}}}|||� d S r )r �getr r r r �_runThreadCalls1 s zFakeReactor._runThreadCallsc C s | j �� d S r )r �stopr r r r �_stop6 s zFakeReactor._stopN)�__name__� __module__� __qualname__�__doc__r r# r% r' r r r r r s r c @ s0 e Zd ZdZdd� Zdd� Zdd� Zdd � Zd S ) �ThreadedResolverTestsz( Tests for L{ThreadedResolver}. c s� d� d}d}t � }| �|j� g �g }� �fdd�}| �td|� t|�}|�||f�}|�|j� |� � | � �|g� | � |� g� |j � |d � | � |j j g � dS ) z� L{ThreadedResolver.getHostByName} returns a L{Deferred} which fires with the value returned by the call to L{socket.gethostbyname} in the threadpool of the reactor passed to L{ThreadedResolver.__init__}. z 10.0.0.17zfoo.bar.example.com� c s �� | � � S r )�append��name�ZipZlookedUpr r �fakeGetHostByNameN s z=ThreadedResolverTests.test_success.<locals>.fakeGetHostByName� gethostbyname� N)r � addCleanupr' �patch�socketr � getHostByName� addCallbackr. r% � assertEqualr �advance�calls)r r0 �timeout�reactorZ resolvedTor2 �resolver�dr r1 r � test_success? s" z"ThreadedResolverTests.test_successc C s� d}t � }| �|j� dd� }| �td|� g }t|�}|�d|f�}| �|t� |� |j � |� � | � t |�d� |j�|d � | � |jjg � dS )z� L{ThreadedResolver.getHostByName} returns a L{Deferred} which fires a L{Failure} if the call to L{socket.gethostbyname} raises an exception. r- c S s t d��d S )Nz ENOBUFS (this is a funny joke))�IOErrorr/ r r r r2 l s z=ThreadedResolverTests.test_failure.<locals>.fakeGetHostByNamer3 � some.namer4 N)r r5 r' r6 r7 r r8 � assertFailurer r9 r. r% r: �lenr r; r< �r r= r> r2 Z failedWithr? r@ r r r � test_failureb s z"ThreadedResolverTests.test_failurec s� d}t � }| �|j� t� � � fdd�}| �td|� g }t|�}|�d|f�}| �|t � |� |j � |j � |d � | �|g � |j � d� | �t|�d� � �td�� dS ) z� If L{socket.gethostbyname} does not complete before the specified timeout elapsed, the L{Deferred} returned by L{ThreadedResolver.getHostByName} fails with L{DNSLookupError}. � c s � � � �d S r )r$ r/ ��resultr r r2 � s z=ThreadedResolverTests.test_timeout.<locals>.fakeGetHostByNamer3 rC r4 zThe I/O was errorfulN)r r5 r' r r6 r7 r r8 rD r r9 r. r r; r: rE r rB rF r rI r � test_timeout� s z"ThreadedResolverTests.test_timeoutc s g � t t�G � fdd�dt��}G dd� dt�}|� }|� }|�|� tt� �}|j�|d� |j�|d� |j�|d� |j�|d� |j�|d� | � t � �d � | � t t t � ��tgd � | � d� d � | � d� d � | � d � d � | � d� d � | � d� d � dS )zm L{ThreadedResolver.getHostByName} is passed L{str}, encoded using IDNA if required. c s e Zd Zd� fdd� ZdS )zAThreadedResolverTests.test_resolverGivenStr.<locals>.FakeResolverr c s � � |� t� S r )r. r )r r0 Ztimeouts�r< r r r8 � s zOThreadedResolverTests.test_resolverGivenStr.<locals>.FakeResolver.getHostByNameN)r )r( r) r* r8 r rL r r � FakeResolver� s rM c @ s e Zd Zdd� ZdS )zFThreadedResolverTests.test_resolverGivenStr.<locals>.JustEnoughReactorc S s d S r r r r r r � installWaker� s zSThreadedResolverTests.test_resolverGivenStr.<locals>.JustEnoughReactor.installWakerN)r( r) r* rN r r r r �JustEnoughReactor� s rO zexample.exampleu vääntynyt.exampleu рф.examplez xn----7sbb4ac0ad0be6cf.xn--p1ai� r r4 zxn--vntynyt-5waa.example� zxn--p1ai.example� � N)r r �objectr ZinstallResolverr r Z nameResolverZresolveHostNamer: rE �list�map�type�str)r rM rO Zfaker> Zrecr rL r �test_resolverGivenStr� sF � � � � � z+ThreadedResolverTests.test_resolverGivenStrN)r( r) r* r+ rA rG rK rY r r r r r, ; s # !r, c C s dS )z8 Function used by L{DelayedCallTests.test_str}. Nr r r r r �nothing� s rZ c @ s` e Zd ZdZdd� Zdd� Zdd� Zdd � Zd d � Zd d � Z dd� Z dd� Z dd� Z dd� Z dS )�DelayedCallMixin� L{DelayedCall} c C s dd� }t |dd� di ||d�S )z� Get a L{DelayedCall} instance at a given C{time}. @param time: The absolute time at which the returned L{DelayedCall} will be scheduled. c S s d S r r )�callr r r �noop� s z0DelayedCallMixin._getDelayedCallAt.<locals>.noopc S s d S r r r r r r r � r z4DelayedCallMixin._getDelayedCallAt.<locals>.<lambda>r N)r )r �timer^ r r r �_getDelayedCallAt� s z"DelayedCallMixin._getDelayedCallAtc C s | � d�| _| � d�| _dS )zb Create two L{DelayedCall} instanced scheduled to run at different times. r r4 N)r` �zero�oner r r r �setUp� s zDelayedCallMixin.setUpc C s: t dtdddidddd� �}| �t|�dt|�f � dS ) �� The string representation of a L{DelayedCall} instance, as returned by L{str}, includes the unsigned id of the instance, as well as its state, the function to be called, and the function arguments. � �rR �ArP Nc S s dS �Ng �?r r r r r r � r z+DelayedCallMixin.test_str.<locals>.<lambda>z?<DelayedCall 0x%x [10.5s] called=0 cancelled=0 nothing(3, A=5)>)r rZ r: rX �id�r �dcr r r �test_str� s ��zDelayedCallMixin.test_strc C s4 t dtdddidddd� �}| �t|�t|�� dS )z� The string representation of a L{DelayedCall} instance, as returned by {repr}, is identical to that returned by L{str}. � )� rg � Nc S s dS )Ng�������?r r r r r r � r z,DelayedCallMixin.test_repr.<locals>.<lambda>)r rZ r: rX �reprrj r r r � test_repr� s zDelayedCallMixin.test_reprc C sJ | j | j }}| �||k � | �||k � | �||k � | �||k � dS )z� For two instances of L{DelayedCall} C{a} and C{b}, C{a < b} is true if and only if C{a} is scheduled to run before C{b}. N�ra rb � assertTrue� assertFalse�r ra rb r r r �test_lt s zDelayedCallMixin.test_ltc C sJ | j | j }}| �||k� | �||k� | �||k� | �||k� dS )z� For two instances of L{DelayedCall} C{a} and C{b}, C{a <= b} is true if and only if C{a} is scheduled to run before C{b} or at the same time as C{b}. Nrr ru r r r �test_le s zDelayedCallMixin.test_lec C sJ | j | j }}| �||k� | �||k� | �||k� | �||k� dS )z� For two instances of L{DelayedCall} C{a} and C{b}, C{a > b} is true if and only if C{a} is scheduled to run after C{b}. Nrr ru r r r �test_gt s zDelayedCallMixin.test_gtc C sJ | j | j }}| �||k� | �||k� | �||k� | �||k� dS )z� For two instances of L{DelayedCall} C{a} and C{b}, C{a > b} is true if and only if C{a} is scheduled to run after C{b} or at the same time as C{b}. Nrr ru r r r �test_ge' s zDelayedCallMixin.test_gec C s: | � | j| jk� | �| j| jk� | �| j| jk� dS )zD A L{DelayedCall} instance is only equal to itself. N)rt ra rb rs r r r r �test_eq4 s zDelayedCallMixin.test_eqc C s: | � | j| jk� | �| j| jk� | �| j| jk� dS )zM A L{DelayedCall} instance is not equal to any other object. N)rs ra rb rt r r r r �test_ne? s zDelayedCallMixin.test_neN)r( r) r* r+ r` rc rl rq rv rw rx ry rz r{ r r r r r[ � s r[ c @ s e Zd ZdZdd� Zdd� ZdS )�DelayedCallNoDebugTestsr\ c C s | � tdd� t�| � dS )z! Turn debug off. �debugFN�r6 r r[ rc r r r r rc O s z DelayedCallNoDebugTests.setUpc C s> t dtdddidddd� �}d�t|��}| �t|�|� dS ) rd re rf rg rP Nc S s dS rh r r r r r r ] r z2DelayedCallNoDebugTests.test_str.<locals>.<lambda>zA<DelayedCall 0x{:x} [10.5s] called=0 cancelled=0 nothing(3, A=5)>)r rZ �formatri r: rX )r rk Zexpectedr r r rl W s ��z DelayedCallNoDebugTests.test_strN�r( r) r* r+ rc rl r r r r r| K s r| c @ s e Zd ZdZdd� Zdd� ZdS )�DelayedCallDebugTestsr\ c C s | � tdd� t�| � dS )z Turn debug on. r} TNr~ r r r r rc i s zDelayedCallDebugTests.setUpc C s> t dtdddidddd� �}d�t|��}| �t|�|� dS ) rd re rf rg rP Nc S s dS rh r r r r r r w r z0DelayedCallDebugTests.test_str.<locals>.<lambda>z\<DelayedCall 0x{:x} \[10.5s\] called=0 cancelled=0 nothing\(3, A=5\) traceback at creation:)r rZ r ri Z assertRegexrX )r rk ZexpectedRegexpr r r rl q s �� �z DelayedCallDebugTests.test_strNr� r r r r r� e s r� c @ s e Zd ZdZdd� ZdS )� TestSpySignalCapturingReactorza Subclass of ReactorBase to capture signals delivered to the reactor for inspection. c C s dS )z* Required method, unused. Nr r r r r rN � s z*TestSpySignalCapturingReactor.installWakerN)r( r) r* r+ rN r r r r r� � s r� c @ s0 e Zd ZdZdd� Zdd� Zdd� Zdd � Zd S ) �ReactorBaseSignalTestszE Tests to exercise ReactorBase's signal exit reporting path. c C s t � }| �d|j� dS )zI The default value of the _exitSignal attribute is None. N)r� ZassertIs� _exitSignal�r r> r r r � test_exitSignalDefaultsToNone� s z4ReactorBaseSignalTests.test_exitSignalDefaultsToNonec C s( t � }|�tjd� | �tj|j� dS )zn ReactorBase's SIGINT handler saves the value of SIGINT to the _exitSignal attribute. N)r� ZsigInt�signal�SIGINT� assertEqualsr� r� r r r �test_captureSIGINT� s z)ReactorBaseSignalTests.test_captureSIGINTc C s( t � }|�tjd� | �tj|j� dS )zp ReactorBase's SIGTERM handler saves the value of SIGTERM to the _exitSignal attribute. N)r� ZsigTermr� �SIGTERMr� r� r� r r r �test_captureSIGTERM� s z*ReactorBaseSignalTests.test_captureSIGTERMc C s: t td�std��t� }|�tjd� | �tj|j� dS )zr ReactorBase's SIGBREAK handler saves the value of SIGBREAK to the _exitSignal attribute. �SIGBREAKz$signal module does not have SIGBREAKN)�hasattrr� r r� ZsigBreakr� r� r� r� r r r �test_captureSIGBREAK� s z+ReactorBaseSignalTests.test_captureSIGBREAKN)r( r) r* r+ r� r� r� r� r r r r r� � s r� zsignal module not available)'r+ r7 r � ImportErrorZqueueZzope.interfacer Ztwisted.python.threadpoolr Ztwisted.internet.interfacesr r r Ztwisted.internet.errorr Ztwisted.internet._resolverr Ztwisted.internet.deferr Ztwisted.internet.baser r r Ztwisted.internet.taskr Ztwisted.trial.unittestr r rT r r, rZ r[ r| r� r� r� r� �skipr r r r �<module> s8 y 0
Save File
Cancel