File: //lib/python3/dist-packages/twisted/conch/test/__pycache__/test_recvline.cpython-38.pyc
U
W[>c � @ s� d Z ddlZddlZddlmZ ddlmZ ddlmZm Z m
Z
ddl m Z m
Z
ddlmZ ddlmZmZ dd lmZ dd
lmZ dd lmZ e�� r�eej�Z ej � ej �e d < n"e
� Z ej � ej �� e�!� �e d
< G dd� dej"�Z#ddlm$Z$ ddlm%Z% ddl&m'Z' G dd� dej(�Z)dZ*dZ+dZ,dZ-dZ.dZ/dZ0d Z1d Z2dd lm3Z3 z@dd l4m5Z5m6Z6m7Z7m8Z8m9Z9m:Z: dd l;m<Z<m=Z=m>Z>m?Z?m@Z@ W n eAk
�r� d!ZBY n�X d"ZBG d#d$� d$e7jC�ZDG d%d&� d&e8jE�ZFG d'd(� d(e5jG�ZHG d)d*� d*e6jI�ZJG d+d,� d,e?�ZKG d-d.� d.e=�ZLG d/d0� d0e<�ZMe �NeLeMe9jO� G d1d2� d2e%jP�ZQG d3d4� d4�ZRG d5d6� d6eR�ZSdd7lTmUZU G d8d9� d9ejVeUjW�ZXG d:d;� d;ejYeUjW�ZZG d<d=� d=eR�Z[zdd>lm\Z\ W n eAk
�r� dZ\Y nX G d?d@� d@eR�Z]G dAdB� dB�Z^G dCdD� dDe[ej"e^�Z_G dEdF� dFeSej"e^�Z`G dGdH� dHe]ej"e^�ZaG dIdJ� dJ�ZbG dKdL� dLe[ej"eb�ZcG dMdN� dNeSej"eb�ZdG dOdP� dPe]ej"eb�ZeG dQdR� dRej"�ZfdS )SzU
Tests for L{twisted.conch.recvline} and fixtures for testing related
functionality.
� N)�insults)�recvline)�reflect�
components�filepath)� iterbytes� bytesEnviron)�platform)�defer�error)�unittest)�portal)�StringTransport�
PYTHONPATHs
PYTHONPATHc @ sl e Zd Zdd� Zdd� Zdd� Zdd� Zd d
� Zd d � Zd
d� Z dd� Z
dd� Z dd� Z dd� Z
dd� ZdS )� ArrowsTestsc sF t � � _t�� � _t�� � _� fdd�� j_� � j_ � j�
� j� d S )Nc s � j S �N)�p� ��selfr �B/usr/lib/python3/dist-packages/twisted/conch/test/test_recvline.py�<lambda>' � z#ArrowsTests.setUp.<locals>.<lambda>) r ZunderlyingTransportr �ServerProtocol�ptr �HistoricRecvLiner �protocolFactory�factory�makeConnectionr r r r �setUp# s
zArrowsTests.setUpc C s@ | j �dd� | j �dd� | j �dd� | �| j �� d� dS )zy
When L{HistoricRecvLine} receives a printable character,
it adds it to the current line buffer.
� xN� y� z�� xyzr )r �keystrokeReceived� assertEqual�currentLineBufferr r r r �test_printableCharacters, s z$ArrowsTests.test_printableCharactersc sF � fdd�}t d�D ] }||� q� �� j�� d� |� jj� � �� j�� d� |� jj� � �� j�� d� |� jj� � �� j�� d� |� jj� � �� j�� d� |� jj� � �� j�� d� |� jj� � �� j�� d� |� jj� � �� j�� d� |� jj� � �� j�� d� |� jj� � �� j�� d� dS ) z�
When L{HistoricRecvLine} receives a LEFT_ARROW or
RIGHT_ARROW keystroke it moves the cursor left or right
in the current line buffer, respectively.
c s � j �| d �S r �r r% ��chr r r r > r z3ArrowsTests.test_horizontalArrows.<locals>.<lambda>r$ r# )� xyr" )r s yz)r r$ N)r r& r r' r Z RIGHT_ARROW�
LEFT_ARROW�r �kRr+ r r r �test_horizontalArrows8 s,
z!ArrowsTests.test_horizontalArrowsc s| � fdd�}t d�D ] }||� q� �� j�� d� |d� |d� |d� � �� j�� d� |d� � �� j�� d � d
S ) z�
When {HistoricRecvLine} receives a newline, it adds the current
line buffer to the end of its history buffer.
c s � j �| d �S r r) r* r r r r e r z*ArrowsTests.test_newline.<locals>.<lambda>� xyz
abc
123
��r$ � abc� 123r � c� b� a�
))r$ r4 r5 s cbar N)r r& r �currentHistoryBufferr. r r r � test_newline` s
� � �zArrowsTests.test_newlinec s8 � fdd�}t d�D ] }||� q� �� j�� d� � �� j�� d� |� jj� � �� j�� d� � �� j�� d� |� jj� � �� j�� d� � �� j�� d � |� jj� � �� j�� d
� � �� j�� d � |� jj� � �� j�� d
� � �� j�� d � td �D ]}|� jj� �q� �� j�� d� d
S )a
When L{HistoricRecvLine} receives UP_ARROW or DOWN_ARROW
keystrokes it move the current index in the current history
buffer up or down, and resets the current line buffer to the
previous or next line in history, respectively for each.
c s � j �| d �S r r) r* r r r r r z1ArrowsTests.test_verticalArrows.<locals>.<lambda>r1 r2 �r r ))r$ r4 )r5 )r5 r ))r$ )r4 r5 )r4 r )r r3 r# � N) r r& r r: r' r ZUP_ARROW�rangeZ
DOWN_ARROW)r r/ r+ �ir r r �test_verticalArrowsx s@
� � � � � �z ArrowsTests.test_verticalArrowsc sV � fdd�}t d�D ] }||� q� �� j�� d� |� jj� � �� j�� d� dS )z�
When L{HistoricRecvLine} receives a HOME keystroke it moves the
cursor to the beginning of the current line buffer.
c s � j �| d �S r r) r* r r r r � r z'ArrowsTests.test_home.<locals>.<lambda>� hello, world�rA r )r rA N)r r& r r' r �HOMEr. r r r � test_home� s
zArrowsTests.test_homec sb � fdd�}t d�D ] }||� q� �� j�� d� |� jj� |� jj� � �� j�� d� dS )z�
When L{HistoricRecvLine} receives an END keystroke it moves the cursor
to the end of the current line buffer.
c s � j �| d �S r r) r* r r r r � r z&ArrowsTests.test_end.<locals>.<lambda>rA rB N)r r& r r' r rC ZENDr. r r r �test_end� s
zArrowsTests.test_endc s� � fdd�}t d�D ] }||� q� �� j�� d� |� jj� � �� j�� d� |� jj� |� jj� � �� j�� d� |� jj� � �� j�� d� dS )z�
When L{HistoricRecvLine} receives a BACKSPACE keystroke it deletes
the character immediately before the cursor.
c s � j �| d �S r r) r* r r r r � r z,ArrowsTests.test_backspace.<locals>.<lambda>r$ r# �r, r )r r! N)r r& r r' r Z BACKSPACEr- r. r r r �test_backspace� s
zArrowsTests.test_backspacec s� � fdd�}t d�D ] }||� q� �� j�� d� |� jj� � �� j�� d� |� jj� |� jj� � �� j�� d� |� jj� |� jj� � �� j�� d� |� jj� |� jj� � �� j�� d� |� jj� � �� j�� d� dS ) z�
When L{HistoricRecvLine} receives a DELETE keystroke, it
delets the character immediately after the cursor.
c s � j �| d �S r r) r* r r r r � r z)ArrowsTests.test_delete.<locals>.<lambda>r$ r# rF )r r r< N)r r& r r' r ZDELETEr- r. r r r � test_delete� s"
zArrowsTests.test_deletec sr � fdd�}t d�D ] }||� q|� jj� |d� � �� j�� d� |� jj� |d� � �� j�� d� dS ) z�
When not in INSERT mode, L{HistoricRecvLine} inserts the typed
character at the cursor before the next character.
c s � j �| d �S r r) r* r r r r � r z)ArrowsTests.test_insert.<locals>.<lambda>r$ � A)� xyAr" � B)� xyBs AzN)r r r- r& r r' r. r r r � test_insert� s
zArrowsTests.test_insertc s~ � fdd�}t d�D ] }||� q|� jj� |� jj� |d� � �� j�� d� |� jj� |d� � �� j�� d� dS ) a
When in INSERT mode and upon receiving a keystroke with a printable
character, L{HistoricRecvLine} replaces the character at
the cursor with the typed character rather than inserting before.
Ah, the ironies of INSERT mode.
c s � j �| d �S r r) r* r r r r r z+ArrowsTests.test_typeover.<locals>.<lambda>r$ rI )rJ r rK )rL r N)r r ZINSERTr- r& r r' r. r r r �
test_typeover s
zArrowsTests.test_typeoverc sr � fdd�}� j }|j|j|j|j|j|j|j|j|j |j
|j |j |j
|jfD ] }||� � �� j�� d� qNdS )z�
When L{HistoricRecvLine} receives a keystroke for an unprintable
function key with no assigned behavior, the line buffer is unmodified.
c s � j �| d �S r r) r* r r r r $ r z8ArrowsTests.test_unprintableCharacters.<locals>.<lambda>r< N)r ZF1ZF2ZF3ZF4ZF5ZF6ZF7ZF8ZF9ZF10ZF11ZF12ZPGUPZPGDNr& r r' )r r/ r r+ r r r �test_unprintableCharacters s �z&ArrowsTests.test_unprintableCharactersN)�__name__�
__module__� __qualname__r r( r0 r; r@ rD rE rG rH rM rN rO r r r r r " s (* r )�telnet)�helper)�
LoopbackRelayc @ s e Zd Zdd� ZdS )�
EchoServerc C s | j �|d | j| j � d S )Nr9 )Zterminal�writeZpsZpn)r �liner r r � lineReceived3 s zEchoServer.lineReceivedN)rP rQ rR rY r r r r rV 2 s rV s [Ds [Cs [As [Bs [2~s [1~s [3~s [4~� )�checkers)�userauth� transport�channel�
connection�session�keys)� TerminalUser�TerminalSession�
TerminalRealm�TerminalSessionTransport� ConchFactoryFTc @ s0 e Zd ZdZdd� Zdd� Zdd� Zdd � Zd
S ) �SessionChannels sessionc O s6 t jj| f|�|� || _|| _|| _|| _|| _d S r )r^ �
SSHChannel�__init__r � protocolArgs�protocolKwArgs�width�height�r r rj rk rl rm �a�kwr r r ri N s zSessionChannel.__init__c C sd t �d| j| jddfd�}| j�| d|� | j�| dd� | j| j| j�| _ | | j _
| j � | � d S )Ns vt102r r s pty-reqs shell) r` ZpackRequest_pty_reqrm rl �connZ sendRequestr rj rk �_protocolInstancer r )r �dataZtermr r r � channelOpenY s zSessionChannel.channelOpenc C s | j �t�� � d S r )rr �connectionLostr ZConnectionDoner r r r �closedc s zSessionChannel.closedc C s | j �|� d S r )rr � dataReceived�r rs r r r rw g s zSessionChannel.dataReceivedN)rP rQ rR �nameri rt rv rw r r r r rg K s
rg c @ s$ e Zd Zdd� Zdd� Zdd� ZdS )�TestConnectionc O s6 t jj| f|�|� || _|| _|| _|| _|| _d S r )r_ �
SSHConnectionri r rj rk rl rm rn r r r ri l s zTestConnection.__init__c C s, t | j| j| j| j| j�| _| �| j� d S r )rg r rj rk rl rm �_TestConnection__channelZ openChannelr r r r �serviceStartedw s z TestConnection.serviceStartedc C s | j �|�S r )r| rW rx r r r rW | s zTestConnection.writeN)rP rQ rR ri r} rW r r r r rz k s rz c @ s e Zd Zdd� Zdd� ZdS )�TestAuthc O s t jj| |f|�|� || _d S r )r\ �SSHUserAuthClientri �password)r �usernamer� ro rp r r r ri � s zTestAuth.__init__c C s t �| j�S r )r
�succeedr� r r r r � getPassword� s zTestAuth.getPasswordN)rP rQ rR ri r� r r r r r~ � s r~ c @ s, e Zd Zdd� Zdd� Zdd� Zdd� Zd S )
�
TestTransportc
O s. || _ || _|| _|| _|| _|| _|| _d S r )r rj rk r� r� rl rm )
r r rj rk r� r� rl rm ro rp r r r ri � s zTestTransport.__init__c C s
t �d�S )NT)r
r� )r ZhostKeyZ fingerprintr r r �
verifyHostKey� s zTestTransport.verifyHostKeyc C s8 t | j| j| j| j| j�| _| �t| j | j
| j�� d S r ) rz r rj rk rl rm �_TestTransport__connectionZrequestServicer~ r� r� r r r r �connectionSecure� s �z TestTransport.connectionSecurec C s | j �|�S r )r� rW rx r r r rW � s zTestTransport.writeN)rP rQ rR ri r� r� rW r r r r r� � s
r� c @ s e Zd Zdd� ZdS )�TestSessionTransportc C s | j jjj�� S r )Zavatarrq r] r �serverProtocolr r r r r � s z$TestSessionTransport.protocolFactoryN)rP rQ rR r r r r r r� � s r� c @ s e Zd ZeZdS )� TestSessionN)rP rQ rR r� ZtransportFactoryr r r r r� � s r� c @ s e Zd ZdS )�TestUserN�rP rQ rR r r r r r� � s r� c @ s$ e Zd Zdd� Zdd� Zdd� ZdS )�NotifyingExpectableBufferc C s t �� | _t �� | _d S r )r
ZDeferred� onConnection�onDisconnectionr r r r ri � s
z"NotifyingExpectableBuffer.__init__c C s t j�| � | j�| � d S r )rT �ExpectableBuffer�connectionMader� �callbackr r r r r� � s z(NotifyingExpectableBuffer.connectionMadec C s | j �|� d S r )r� Zerrback)r �reasonr r r ru � s z(NotifyingExpectableBuffer.connectionLostN)rP rQ rR ri r� ru r r r r r� � s r� c @ s$ e Zd ZdZdZdd� Zdd� ZdS )�
_BaseMixin�P � c
C s� | j �� �� }|dg| jt|� d }| �t|�t|�� tt|��D ]Z}| �|| || d�|td|d �|d � �d d�|td|d �|d � � � qJd S )Nr � r s != ) �recvlineClient� __bytes__�
splitlines�HEIGHT�lenr& r> �join�max)r �linesZ
receivedLinesZ
expectedLinesr? r r r �
_assertBuffer� s � ��z_BaseMixin._assertBufferc s. �j �d�}��|� � �fdd�}|�|�S )Ns donec s �� � � d S r )r� )Zign��outputr r r �finished� s z)_BaseMixin._trivialTest.<locals>.finished)r� �expect�
_testwriteZ addCallback)r Z inputLiner� Zdoner� r r� r � _trivialTest� s
z_BaseMixin._trivialTestN)rP rQ rR �WIDTHr� r� r� r r r r r� � s r� c @ s e Zd Zdd� Zdd� ZdS )� _SSHMixinc s@ t st�d��d\}}t� }t|_�fdd�|_t�� }|� ||� t
� |�}|� |� t
|�}tjt�| �� �dd�}||jd< ||jd< | j|_|�� | �� �t��fdd���|�d �}t|�} t� �t� �fd d��� t � fd
d�d i ||| j | j �}
t|
�} |
� | � |� | � �| _!|
| _"|| _#| | _$| | _%�j&S ) NzMcryptography requirements missing, can't run historic recvline tests over ssh)s testusers testpassc s � S r r r ��
insultsServerr r r � r z!_SSHMixin.setUp.<locals>.<lambda>i )ZkeySizes ssh-rsac s � S r r r ��recvlineServerr r r � r c s � S r r r �r� r r r � r c s � S r r r ��
insultsClientr r r r r )'�sshr ZSkipTestrd r� Z userFactoryZchainedProtocolFactoryr[ Z'InMemoryUsernamePasswordDatabaseDontUseZaddUserr
ZPortalZregisterCheckerrf ra Z_getPersistentRSAKeyr ZFilePathZmktempZ
publicKeysZ privateKeysr� Z startFactoryr r Z
buildProtocolrU r� �ClientProtocolr� r� r� r r� � sshClient� sshServer�clientTransport�serverTransportr� ) r �ur ZrlmZcheckerZptlZ
sshFactoryZsshKeyr� r� r� r� r �r� r� r� r� r r � sH �
�
z_SSHMixin.setUpc C s | j �|� d S r )r� rW rx r r r r� s z_SSHMixin._testwriteN�rP rQ rR r r� r r r r r� � s /r� )� test_telnetc @ s e Zd ZdS )�TestInsultsClientProtocolNr� r r r r r� s r� c @ s e Zd ZdS )�TestInsultsServerProtocolNr� r r r r r� s r� c @ s e Zd Zdd� Zdd� ZdS )� _TelnetMixinc s� | � � �t�fdd���t��fdd��}t|�}t� �t�fdd��� t�� fdd��}t|�}|�|� |�|� |�� |�� �| _ || _
|| _ || _ �j
S )Nc s � S r r r r� r r r % r z$_TelnetMixin.setUp.<locals>.<lambda>c s � S r r r r� r r r & r c s � S r r r r� r r r * r c s � S r r r r� r r r + r )r� r� rS ZTelnetTransportrU r� r� r Z clearBufferr� � telnetClientr� r� r� )r Z telnetServerr� r� r� r r� r r # s"
z_TelnetMixin.setUpc C s | j �|� d S r )r� rW rx r r r r� < s z_TelnetMixin._testwriteNr� r r r r r� " s r� )�stdioc @ s$ e Zd Zdd� Zdd� Zdd� ZdS )� _StdioMixinc s� t � � t�� fdd��}t�|�}tj}tj}|�d�sB|�d�rN|d d� }||t �
| j �g}t �
� svdd� |D �}dd lm} |j|||td
d �}� | _| _|| _|| _t�td |j� �d �g��S )
Nc s � S r r r �� testTerminalr r r Q r z#_StdioMixin.setUp.<locals>.<lambda>z.pycz.pyo���c S s g | ]}|� t�� ��qS r )�encode�sys�getfilesystemencoding)�.0�argr r r �
<listcomp>a s z%_StdioMixin.setUp.<locals>.<listcomp>r )�reactorT)�envZusePTYs >>> )r� r r� r� ZTerminalProcessProtocolr� �
executable�__file__�endswithr Zqualr� r � isWindows�twisted.internetr� Z spawnProcess� properEnvr� r� �
processClientr� r
Z
gatherResults�filterr� r� )r r� r� �exe�module�argsr� r� r r� r r G s,
� �z_StdioMixin.setUpc sF z� j �d� W n tjtfk
r* Y nX � fdd�}� jj�|�S )NZKILLc s. | � tj� � �| jj� � �| jjd� d S )N� )�trapr ZProcessTerminatedZ assertIsNone�valueZexitCoder& Zstatus)Zfailurer r r r� y s z"_StdioMixin.tearDown.<locals>.trap)r� Z
signalProcessr ZProcessExitedAlready�OSErrorr� r� Z
addErrback)r r� r r r �tearDowns s z_StdioMixin.tearDownc C s | j �|� d S r )r� rW rx r r r r� � s z_StdioMixin._testwriteN)rP rQ rR r r� r� r r r r r� F s ,
r� c @ sX e Zd ZeZdd� Zdd� Zdd� Zdd� Zd d
� Z d d � Z
d
d� Z dd� Z dd� Z
dS )�RecvlineLoopbackMixinc C s | � ddddg�S )Ns first line
done� >>> first line�
first line� >>> done)r� r r r r �
testSimple� s ��z RecvlineLoopbackMixin.testSimplec C s" | � td td d dddg�S )Nr� r= � xxxx
dones >>> first xxxxs
first xxxxr� )r� �insert�leftr r r r �
testLeftArrow� s ��z#RecvlineLoopbackMixin.testLeftArrowc C s* | � td td td d dddg�S )Ns
right liner= � s xx
dones >>> right lixxs
right lixxr� )r� r� r� �rightr r r r �testRightArrow� s ��z$RecvlineLoopbackMixin.testRightArrowc C s | � dtd d dddg�S )N� second liner= r� s >>> second xxxxs second xxxxr� )r� � backspacer r r r �
testBackspace� s ��z#RecvlineLoopbackMixin.testBackspacec C s& | � dtd td d dddg�S )Ns delete xxxxr= � line
dones >>> delete lines delete liner� )r� r� �deleter r r r �
testDelete� s ��z RecvlineLoopbackMixin.testDeletec C s | � dtd d dddg�S )Ns third ine� � l
dones >>> third lines
third liner� )r� r� r r r r �
testInsert� s ��z RecvlineLoopbackMixin.testInsertc C s" | � dtd t d dddg�S )Ns fourth xiner= r� s >>> fourth lines fourth liner� )r� r� r� r r r r � testTypeover� s ��z"RecvlineLoopbackMixin.testTypeoverc C s | � td t d dddg�S )Ns blah lines home
dones
>>> home lines home liner� )r� r� �homer r r r �testHome� s ��z RecvlineLoopbackMixin.testHomec C s" | � dtd t d dddg�S )Ns end r= r� s >>> end lines end liner� )r� r� �endr r r r �testEnd� s ��z RecvlineLoopbackMixin.testEndN)rP rQ rR rV r� r� r� r� r� r� r� r� r� r� r r r r r� � s r� c @ s e Zd ZdS )�RecvlineLoopbackTelnetTestsNr� r r r r r� � s r� c @ s e Zd ZdS )�RecvlineLoopbackSSHTestsNr� r r r r r� � s r� c @ s e Zd ZedkrdZdS )�RecvlineLoopbackStdioTestsNzBTerminal requirements missing, can't run recvline tests over stdio�rP rQ rR r� �skipr r r r r� � s r� c @ s( e Zd ZeZdd� Zdd� Zdd� ZdS )� HistoricRecvlineLoopbackMixinc C s | � dt d dddddg�S )Ns first line
�
doner� r� r� )r� �upr r r r � testUpArrow� s
��z)HistoricRecvlineLoopbackMixin.testUpArrowc C s" | � dt t d dddddg�S )aH
Pressing down arrow to visit an entry that was added to the
history by pressing the up arrow instead of return does not
raise a L{TypeError}.
@see: U{http://twistedmatrix.com/trac/ticket/9031}
@return: A L{defer.Deferred} that fires when C{b"done"} is
echoed back.
s first line
partial liner� r� r� s >>> partial lines partial liner� �r� r� �downr r r r �$test_DownArrowToPartialLineInHistory� s ��zBHistoricRecvlineLoopbackMixin.test_DownArrowToPartialLineInHistoryc
C s* | � dtd t d dddddddg�S ) Ns first line
second line
r� r� r� r� s >>> second liner� r� r� r r r r �
testDownArrow s ��z+HistoricRecvlineLoopbackMixin.testDownArrowN)rP rQ rR rV r� r� r r r r r r r� � s
r� c @ s e Zd ZdS )�#HistoricRecvlineLoopbackTelnetTestsNr� r r r r r s r c @ s e Zd ZdS )� HistoricRecvlineLoopbackSSHTestsNr� r r r r r s r c @ s e Zd ZedkrdZdS )�"HistoricRecvlineLoopbackStdioTestsNzKTerminal requirements missing, can't run historic recvline tests over stdior� r r r r r s r c @ s e Zd ZdZdd� ZdS )�TransportSequenceTestsz5
L{twisted.conch.recvline.TransportSequence}
c C s | � ttj� dS )zh
Initializing a L{recvline.TransportSequence} with no args
raises an assertion.
N)Z assertRaises�AssertionErrorr ZTransportSequencer r r r �test_invalidSequence% s z+TransportSequenceTests.test_invalidSequenceN)rP rQ rR �__doc__r r r r r r s r )gr �osr� Ztwisted.conch.insultsr Z
twisted.conchr Ztwisted.pythonr r r Ztwisted.python.compatr r Ztwisted.python.runtimer r� r
r Z
twisted.trialr Z twisted.credr
Ztwisted.test.proto_helpersr r� �dict�environr� �pathsepr� �pathr� r� ZTestCaser rS rT Ztwisted.conch.test.loopbackrU r rV r� r� r� r� r� r� r� r� r� r[ Ztwisted.conch.sshr\ r] r^ r_ r` ra Ztwisted.conch.manhole_sshrb rc rd re rf � ImportErrorr� rh rg r{ rz r r~ ZSSHClientTransportr� r� r� r� ZregisterAdapterZISessionr� r� r� r� Ztwisted.conch.testr� r� Z TestProtocolr� r r� r� r� r� r� r� r� r� r� r r r r r r r r �<module> s�
�
5 ��
?L/