File: //usr/lib/python3/dist-packages/twisted/conch/test/__pycache__/test_channel.cpython-310.pyc
o
�b�/ � @ s� d Z ddlmZ ddlmZ z.ddlmZ ddlmZ ddl m
Z
ddl m Z ddl
mZ dd lmZ dd
lmZ d ZW n
eyI d ZeZ
Y nw dd
lmZ G dd� de
�Zddd�Zeee�G dd� de��ZdS )z
Test ssh/channel.py.
� )�skipIf)� verifyObject)�channel)�SSHTransportAddress)�
SSHService)�SSHServerTransport)�
interfaces)� IPv4Address)�StringTransport� zConch SSH not supported.)�TestCasec @ s8 e Zd ZdZdd� Zdd� Zdd� Zdd � Zd
d � Zd S )
�MockConnectiona�
A mock for twisted.conch.ssh.connection.SSHConnection. Record the data
that channels send, and when they try to close the connection.
@ivar data: a L{dict} mapping channel id #s to lists of data sent by that
channel.
@ivar extData: a L{dict} mapping channel id #s to lists of 2-tuples
(extended data type, data) sent by that channel.
@ivar closes: a L{dict} mapping channel id #s to True if that channel sent
a close message.
c C s i | _ i | _i | _d S )N)�data�extData�closes��self� r �A/usr/lib/python3/dist-packages/twisted/conch/test/test_channel.py�__init__* s
zMockConnection.__init__c C s dS )z,
Return our logging prefix.
r
r r r r r � logPrefix/ s zMockConnection.logPrefixc C s | j �|g ��|� dS )z'
Record the sent data.
N)r �
setdefault�append)r r r r r r �sendData5 � zMockConnection.sendDatac C s | j �|g ��||f� dS )z0
Record the sent extended data.
N)r r r )r r �typer r r r �sendExtendedData; s z MockConnection.sendExtendedDatac C s d| j |< dS )z?
Record that the channel sent a close message.
TN)r )r r r r r � sendCloseA s zMockConnection.sendCloseN) �__name__�
__module__� __qualname__�__doc__r r r r r r r r r r
s r
Nc C s& t � }|�t||d�� |�| � dS )a�
Connect a SSHTransport which is already connected to a remote peer to
the channel under test.
@param service: Service used over the connected transport.
@type service: L{SSHService}
@param hostAddress: Local address of the connected transport.
@type hostAddress: L{interfaces.IAddress}
@param peerAddress: Remote address of the connected transport.
@type peerAddress: L{interfaces.IAddress}
)� hostAddress� peerAddressN)r �makeConnectionr
�
setService)�servicer" r# � transportr r r �connectSSHTransportH s
�r( c @ s� e Zd ZdZdd� Zdd� Zdd� Zdd � Zd
d � Zd d
� Z dd� Z
dd� Z dd� Z dd� Z
dd� Zdd� Zdd� Zd d � Zd d � Zd S )!� ChannelTestsz"
Tests for L{SSHChannel}.
c C s& t � | _tj| jdd�| _d| j_dS )z�
Initialize the channel. remoteMaxPacket is 10 so that data is able
to be sent (the default of 0 means no data is sent because no packets
are made).
�
)�conn�remoteMaxPackets channelN)r
r+ r �
SSHChannel�namer r r r �setUpc s zChannelTests.setUpc C s | � ttj| j�� dS )zK
L{SSHChannel} instances provide L{interfaces.ITransport}.
N)�
assertTruer r �
ITransportr r r r r �test_interfacem r zChannelTests.test_interfacec C s t j| jd�}| �|jd� | �|jd� | �|jd� | �|jd� | �|jd� | �|j| j� | � |j
� | � |j � t �ddddd d
d �}| �|jd� | �|jd� | �|jd� | �|jd� | �|jd� | �|jd � | �|j
d
� | �|j d � d S )
a�
Test that SSHChannel initializes correctly. localWindowSize defaults
to 131072 (2**17) and localMaxPacket to 32768 (2**15) as reasonable
defaults (what OpenSSH uses for those variables).
The values in the second set of assertions are meaningless; they serve
only to verify that the instance variables are assigned in the correct
order.
)r+ i i � r � � � � � � � N) r r- r+ � assertEqual�localWindowSize�localWindowLeft�localMaxPacket�remoteWindowLeftr, � assertIsNoner �avatar)r �c�c2r r r � test_inits s$
zChannelTests.test_initc C s. | � t| j�d� | � ttjdd��d� dS )z{
Test that str(SSHChannel) works gives the channel name and local and
remote windows at a glance..
z%<SSHChannel channel (lw 131072 rw 0)>r3 �� localWindowz <SSHChannel None (lw 1 rw 0)>N)r: �strr r- r r r r �test_str� s �zChannelTests.test_strc C s. | � | j�� d� | � tjdd��� d� dS )z
Test that bytes(SSHChannel) works, gives the channel name and
local and remote windows at a glance..
s% <SSHChannel channel (lw 131072 rw 0)>r3 rD s <SSHChannel None (lw 1 rw 0)>N)r: r � __bytes__r- r r r r �
test_bytes� s
��zChannelTests.test_bytesc C s | � | j�� d� dS )z�
Test that SSHChannel.logPrefix gives the name of the channel, the
local channel ID and the underlying connection.
z.SSHChannel channel (unknown) on MockConnectionN)r: r r r r r r �test_logPrefix� s
�zChannelTests.test_logPrefixc s dg� � fdd�}|| j _| j �d� | j �dd� | j �d� | �| j jd� | �| j j� | �� d � | �| j j d � | �| j
j | j dg� | �| j j g � | �| j
j
| j d
g� d� d< | j �d � | �� d � | j �d � | j �� | j �d � | �� d � d
S )zq
Test that addWindowBytes adds bytes to the window and resumes writing
if it was paused.
Fc � d� d< d S �NTr r r ��cbr r �stubStartWriting� � z:ChannelTests.test_addWindowBytes.<locals>.stubStartWriting� testr3 �2 �* r � )r3 rQ � sP aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaN)r � startWriting�write�
writeExtended�addWindowBytesr: r> r0 �
areWriting�bufr+ r �extBufr � assertFalse�loseConnection)r rO r rM r �test_addWindowBytes� s(
z ChannelTests.test_addWindowBytesc C sL dd� | j _| �| j �dd�� | �| j �dd�� | �| j �dd�� dS )zn
Test that requestReceived handles requests by dispatching them to
request_* methods.
c S s | dkS )NrT r )r r r r �<lambda>� s z3ChannelTests.test_requestReceived.<locals>.<lambda>s test-methodrT � as
bad-methodN)r �request_test_methodr0 �requestReceivedr] r r r r �test_requestReceived� s z!ChannelTests.test_requestReceivedc C s* | � | jj� | j�� | �| jj� dS )zM
Test that the default closeReceieved closes the connection.
N)r] r �closing�
closeReceivedr0 r r r r �test_closeReceieved� s
z ChannelTests.test_closeReceievedc s$ dg� � fdd�}|| j _| j �d� | j �d� | �| j j� | �� d � | j �d� | j �d� | jj| j }| � |d dg� | � | j j
d
� | j �d � | � |g d �� | � | j j
d
� d� d< | j �d� | �| j j� | �� d � | � |g d�� | � | j j d� | � | j j
d� dS )z�
Test that write handles data correctly. Send data up to the size
of the remote window, splitting the data into packets of length
remoteMaxPacket.
Fc rK rL r r rM r r �stubStopWriting� rP z0ChannelTests.test_write.<locals>.stubStopWriting� dra r rU � ta� da� � 12345678901)rk rj �
1234567890� 1r7 � 123456)rk rj rn ro � 12345� 6N) r � stopWritingrW r] rZ r0 rY r+ r r: r> r[ �r rh r r rM r �
test_write� s, zChannelTests.test_writec sB dg� � fdd�}|| j _| j �dd� | j �dd� | j �dd� | �| j j� | �� d � | j �d
� | j �dd� | jj| j }| � |g d �� | � | j j
d � | j �d
d� | � |g d�� | � | j j
d� d� d < | j �dd� | �| j j� | �� d � | � |g d�� | � | j j ddgg� | � | j j
d � dS )z�
Test that writeExtended handles data correctly. Send extended data
up to the size of the window, splitting the extended data into packets
of length remoteMaxPacket.
Fc rK rL r r rM r r rh rP z8ChannelTests.test_writeExtended.<locals>.stubStopWritingr3 ri ra r4 � tr rU )�r3 rk �r4 rv �r4 ra rl r5 rm )rw rx ry �r5 rn �r5 ro r7 r6 rp )rw rx ry rz r{ )r6 rq rr N) r rs rX r] rZ r0 rY r+ r r: r> r\ rt r rM r �test_writeExtended s8 �� z ChannelTests.test_writeExtendedc C sB | j �d� | j �dd� td�D �� | �| jj| j dg� dS )zS
Test that writeSequence is equivalent to write(''.join(sequece)).
rU c s s � | ]}d |f V qdS )s %dNr )�.0�ir r r � <genexpr>= s � z2ChannelTests.test_writeSequence.<locals>.<genexpr>r* s
0123456789N)r rY �
writeSequence�ranger: r+ r r r r r �test_writeSequence8 s z ChannelTests.test_writeSequencec C s� | j �d� | j �dd� | j �� | �| jj�| j �� | j �d� | �| jj�| j �� | j �d� | � | jj�| j �� dS )zk
Tesyt that loseConnection() doesn't close the channel until all
the data is sent.
s datar3 s datadatar6 � N)
r rW rX r^ r? r+ r �getrY r0 r r r r �test_loseConnection@ s
z ChannelTests.test_loseConnectionc C �6 t ddd�}t| jj|d� | �t|�| j�� � dS )z�
L{SSHChannel.getPeer} returns the same object as the underlying
transport's C{getPeer} method returns.
�TCPz 192.168.0.1i1� )r& r# N)r r( r r+ r: r �getPeer)r �peerr r r � test_getPeerN � zChannelTests.test_getPeerc C r� )z�
L{SSHChannel.getHost} returns the same object as the underlying
transport's C{getHost} method returns.
r� z 127.0.0.1i90 )r& r" N)r r( r r+ r: r �getHost)r �hostr r r � test_getHostX r� zChannelTests.test_getHostN)r r r r! r/ r2 rC rG rI rJ r_ rd rg ru r| r� r� r� r� r r r r r) ] s"
$1
r) )NN)r! �unittestr �zope.interface.verifyr �twisted.conch.sshr �twisted.conch.ssh.addressr �twisted.conch.ssh.servicer �twisted.conch.ssh.transportr �twisted.internetr �twisted.internet.addressr �twisted.test.proto_helpersr
�skipTest� ImportError�object�twisted.trial.unittestr r
r( r) r r r r �<module> s* �
+