|
Server : LiteSpeed System : Linux server51.dnsbootclub.com 4.18.0-553.62.1.lve.el8.x86_64 #1 SMP Mon Jul 21 17:50:35 UTC 2025 x86_64 User : nandedex ( 1060) PHP Version : 8.1.33 Disable Function : NONE Directory : /opt/cppython/lib/python3.8/test/__pycache__/ |
U
>��g�2 � @ s, d dl Z d dlZd dlZd dlZd dlZd dlmZ d dlZejZdd� Z G dd� dej
�Z G dd� de �Z
G d d
� d
ej�ZG d d � d ej�Zejd
d� �Zdefdd�ZG dd� dej
�ZG dd� de�ZG dd� de �ZeZG dd� dej
�ZG dd� dej
�ZG d d � d e�Ze d k�r(e� � dS ) � N)�supportc C sT |� � | �� z4z|�� \}}|�� W n tjk
r@ Y nX W 5 |�� X d S �N)�listen�set�close�accept�socket�timeout)�evtZserv�conn�addr� r
�2/opt/cppython/lib/python3.8/test/test_telnetlib.py�server s
r c @ sT e Zd Zdd� Zdd� Zdd� Zdd� Zd d
� Zd d � Zd
d� Z dd� Z
dd� Z dS )� GeneralTestsc C sr t �� | _t�tjtj�| _| j�d� t� | j�| _
t j t | j| jfd�| _
| j
�d� | j
�� | j�� d S )N�< )�target�argsT)� threading�Eventr
r �AF_INET� SOCK_STREAM�sock�
settimeoutr Z bind_port�port�Threadr �thread� setDaemon�start�wait��selfr
r
r �setUp s
zGeneralTests.setUpc C s | j �� | ` d S r )r �joinr r
r
r �tearDown# s
zGeneralTests.tearDownc C s t �t| j�}|j�� d S r )� telnetlib�Telnet�HOSTr r r �r! �telnetr
r
r � testBasic' s zGeneralTests.testBasicc C s: t �t| j��}| �|�� � W 5 Q R X | �|�� � d S r )r% r& r' r ZassertIsNotNone�
get_socketZ assertIsNone)r! �tnr
r
r �testContextManager, s z GeneralTests.testContextManagerc C s\ | � t�� d k� t�d� zt�t| j�}W 5 t�d � X | �|j �
� d� |j � � d S )N� ) �
assertTruer �getdefaulttimeout�setdefaulttimeoutr% r& r' r � assertEqualr �
gettimeoutr r( r
r
r �testTimeoutDefault1 s
z GeneralTests.testTimeoutDefaultc C sb | � t�� d k� t�d� ztjt| jd d�}W 5 t�d � X | � |j� � d k� |j�
� d S �Nr. )r ) r/ r r0 r1 r% r& r' r r r3 r r( r
r
r �testTimeoutNone; s
z GeneralTests.testTimeoutNonec C s2 t jt| jdd�}| �|j�� d� |j�� d S r5 )r% r& r' r r2 r r3 r r( r
r
r �testTimeoutValueF s z GeneralTests.testTimeoutValuec C s: t �� }|jt| jdd� | �|j�� d� |j�� d S r5 ) r% r& �openr' r r2 r r3 r r( r
r
r �testTimeoutOpenK s z GeneralTests.testTimeoutOpenc C sJ t jt| jdd�}|j}| �|�� |� | �|�� |�� � |j�� d S r5 ) r% r& r' r r r2 r+ �filenor )r! r) Zt_sockr
r
r � testGettersQ s
zGeneralTests.testGettersN) �__name__�
__module__� __qualname__r" r$ r* r- r4 r6 r7 r9 r; r
r
r
r r s
r c @ s* e Zd ZdZd
dd�Zdd� Zdd� Zd S ) �
SocketStubz* a socket proxy that re-defines sendall() r
c C s t |�| _g | _d| _d S )NF)�list�reads�writes�block)r! rA r
r
r �__init__[ s
zSocketStub.__init__c C s | j �|� d S r )rB �append)r! �datar
r
r �sendall_ s zSocketStub.sendallc C sZ d}| j r(t|�|k r(|| j �d�7 }qt|�|krV| j �d||d � � |d |� }|S )N� r )rA �len�pop�insert)r! �size�outr
r
r �recva s zSocketStub.recvN)r
)r< r= r> �__doc__rD rG rN r
r
r
r r? Y s
r? c @ s, e Zd Zdd� Zdd� Zdd� Zdd� Zd S )
� TelnetAlikec C s
t � �d S r )�NotImplementedErrorr r
r
r r: k s zTelnetAlike.filenoc C s d S r r
r r
r
r r m rH zTelnetAlike.closec C s
| j j S r )r rC r r
r
r �
sock_availn s zTelnetAlike.sock_availc G s> t �� �}tjj| |f|�� W 5 Q R X | j|�� 7 _d S r )r Zcaptured_stdoutr% r& �msg� _messages�getvalue)r! rS r rM r
r
r rS p s
zTelnetAlike.msgN)r< r= r> r: r rR rS r
r
r
r rP j s rP c @ sD e Zd Zdd� Zedd� �Zddd�Zdd � Zdd
d �Zd d
� Z dS )� MockSelectorc C s
i | _ d S r ��keysr r
r
r rD x s zMockSelector.__init__c C s dS )Ng����MbP?r
r r
r
r �
resolution{ s zMockSelector.resolutionNc C s t �|d||�}|| j|< |S )Nr )� selectors� SelectorKeyrX )r! �fileobj�eventsrF �keyr
r
r �register s
zMockSelector.registerc C s | j �|�S r )rX rJ )r! r\ r
r
r �
unregister� s zMockSelector.unregisterc C sF d}| j D ]}t|t�r
|jj} q&q
|r.g S dd� | j �� D �S d S )NFc S s g | ]}||j f�qS r
)r] )�.0r^ r
r
r �
<listcomp>� s z'MockSelector.select.<locals>.<listcomp>)rX �
isinstancerP r rC �values)r! r rC r\ r
r
r �select� s
zMockSelector.selectc C s | j S r rW r r
r
r �get_map� s zMockSelector.get_map)N)N)
r< r= r> rD �propertyrY r_ r` re rf r
r
r
r rV v s
rV c # s0 � fdd�}zt j}|t _d V W 5 |t _X d S )Nc s t � �S r )r? )Zignored�rA r
r �new_conn� s z test_socket.<locals>.new_conn)r �create_connection)rA ri Zold_connr
rh r � test_socket� s
rk r
c C s2 | D ]}qt | �� |dd�}d|_W 5 Q R X |S )za return a telnetlib.Telnet object that uses a SocketStub with
reads queued up to be read �dummyr � )rk rT )rA �cls�xr) r
r
r � test_telnet� s
rp c @ s e Zd Zdd� Zdd� ZdS )�ExpectAndReadTestCasec C s t j| _tt _d S r )r% �_TelnetSelector� old_selectorrV r r
r
r r" � s zExpectAndReadTestCase.setUpc C s | j t_d S r )rs r% rr r r
r
r r$ � s z ExpectAndReadTestCase.tearDownN)r< r= r> r" r$ r
r
r
r rq � s rq c @ sD e Zd Zdd� Zdd� Zdd� Zdd� Zd d
� Zd d � Zd
d� Z dS )� ReadTestsc C st dg}t |�}|�d�}| j|d|j|j|jjfd� dddg}d�|dd � �}t |�}|�d�}| �||� dS )
zc
read_until(expected, timeout=None)
test the blocking version of read_util
s xxxmatchyyy� matchs xxxmatch)rS s2 xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxs2 yyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyrH N���)rp Z
read_untilr2 ZcookedqZrawqr rA r# )r! �wantr) rF rA �expectr
r
r �test_read_until� s
zReadTests.test_read_untilc C s4 dddg}d� |�}t|�}|�� }| �||� dS )zJ
read_all()
Read all data until EOF; may block.
�� xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxs� yyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyys� zzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzrH N)r# rp �read_allr2 )r! rA rx r) rF r
r
r �
test_read_all� s
zReadTests.test_read_allc C sB t dg�}|�� }| �t|�dk� t � }|�� }| �d|� dS )zQ
read_some()
Read at least one byte or EOF; may block.
rz � rH N)rp Z read_somer/ rI r2 )r! r) rF r
r
r �test_read_some� s
zReadTests.test_read_somec C st d}t |g�}t||�}d|j_| �d|� � d|j_d}z||� 7 }W q: tk
r` Y qdY q:X q:| �||� dS )z
read_*_eager()
Read all data available already queued or on the socket,
without blocking.
�d xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxTrH FN)rp �getattrr rC r2 �EOFError)r! � func_namerw r) �funcrF r
r
r � _read_eager� s
zReadTests._read_eagerc C s | � d� | � d� d S )NZ
read_eagerZread_very_eager)r� r r
r
r �test_read_eager� s
zReadTests.test_read_eagerc C sV d}t |g�}| �d|�� � |jjr0|�� q |�� }| �||� | �t|j� d S �Nr rH )rp r2 �read_very_lazyr rA � fill_rawqZ assertRaisesr� )r! rw r) rF r
r
r r� � s
zReadTests.read_very_lazyc C s~ d}t |g�}| �d|�� � d}z |�� }||7 }|s@|�� W n tk
rZ Y qnY nX | �|�|�� q"| �||� d S r� )rp r2 Z read_lazyr� r� r/ �
startswith)r! rw r) rF Z read_datar
r
r �test_read_lazy s
zReadTests.test_read_lazyN)
r< r= r> ry r| r~ r� r� r� r� r
r
r
r rt � s
rt c @ s e Zd Zddd�Zdd� ZdS )�nego_collectorNc C s d| _ || _d| _d S )NrH )�seen� sb_getter�sb_seen)r! r� r
r
r rD s znego_collector.__init__c C s< | j || 7 _ |tjkr8| jr8| �� }| j|7 _d S r )r� �tl�SEr� r� )r! r �cmd�optZsb_datar
r
r �do_nego s znego_collector.do_nego)N)r< r= r> rD r� r
r
r
r r� s
r� c @ s e Zd ZdZdd� ZdS )�
WriteTestszKThe only thing that write does is replace each tl.IAC for
tl.IAC+tl.IACc C sz ddt j d dt j t j d t j t jdg}|D ]@}t� }|�|� d�|jj�}| �|�t jt jt j �|� q4d S )Ns data sample without IACs data sample withs one IACs a fews iacsrH ) r� �IACrp �writer# r rB r2 �replace)r! Z data_samplerF r) Zwrittenr
r
r �
test_write' s �
zWriteTests.test_writeN)r< r= r> rO r� r
r
r
r r� # s r� c @ s` e Zd Zejejejejejej ej
ej gZ dd� Z
dd� Zdd� Zdd� Zd d
� Zd d � Zd
S )� OptionTestsc C s� t |�}td�|��}t� }|�|j� |�� }|j}| �t|�dk� | � |dd� | j
� | � |dd� t j
� | � |t|| �� d|_dS )z helper for testing IAC + cmd rH r Nr} � )rp rI r# r� � set_option_negotiation_callbackr� r{ r� r/ �assertIn�cmdsr2 r� ZNOOPTr� )r! rF r) Zdata_len�nego�txtr� r
r
r �
_test_command7 s zOptionTests._test_commandc C s^ | j D ]<}| �tj|g� | �dtj|dg� | �dtj|dg� q| �dd� | j D �� d S )Nr sd yyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyy�
xxxxxxxxxx�
yyyyyyyyyyc S s g | ]}t j| �qS r
)r� r� )ra r� r
r
r rb K s z1OptionTests.test_IAC_commands.<locals>.<listcomp>)r� r� r� r� )r! r� r
r
r �test_IAC_commandsE s
z OptionTests.test_IAC_commandsc C s0 t jt j t j t j t jt j t j t j t j t j t jt j t j t j d t j t j t jt j d t j t j t j t j t jt j d t j t j d t j t j g}t|�}t|j�}|�|j� |� � }| �
|d� t jt j d t j d t j d }| �
|j |� | �
d|�� � d |_ d S )Ns aas bbs ccs ddrH s aabb)
r� r� ZSBr� rp r� Z read_sb_datar� r� r{ r2 r� r� )r! �sendr) r� r� Z want_sb_datar
r
r �test_SB_commandsM s "&&*�
$z OptionTests.test_SB_commandsc C s� dt jtdg� dft jt j tdg� dft jt j tdg� dft jt j tdg� dft jt j tdg� dfg}|D ]2\}}t|g�}|�d� |� � }| �
||j � q|d S ) N)� az : recv b''
�X z: IAC 88 not recognized
r} z : IAC DO 1
z
: IAC DONT 1
z
: IAC WILL 1
z
: IAC WONT 1
) r� r� �bytesZDOZDONTZWILLZWONTrp �set_debuglevelr{ r� rT )r! Zgiven_a_expect_b�a�br) r� r
r
r �test_debuglevel_reads_ s �
z!OptionTests.test_debuglevel_readsc C s0 t � }|�d� |�d� d}| �||j� d S )Nr} s xxxz send b'xxx'
)rp r� r� r� rT )r! r) Zexpectedr
r
r �test_debuglevel_writer s
z!OptionTests.test_debuglevel_writec C sJ t g �� tdd�}d|_W 5 Q R X |�d� |�d� | �|jd� d S )Nrl �0rm r} �testz0.*test)rk rP rT r� rS Z assertRegexr( r
r
r �test_debug_accepts_str_porty s
z'OptionTests.test_debug_accepts_str_portN)r< r= r> r� ZAOZAYTZBRKZECZELZGAZIPZNOPr� r� r� r� r� r� r� r
r
r
r r� 3 s $r� c @ s e Zd Zdd� ZdS )� ExpectTestsc C sB dddg}t |�}|�dg�\}}}| �|d�|dd� �� dS )z�
expect(expected, [timeout])
Read until the expected string has been seen, or a timeout is
hit (default is no timeout); may block.
r� ru r� rH Nrv )rp rx r2 r# )r! rw r) �_rF r
r
r � test_expect� s
zExpectTests.test_expectN)r< r= r> r� r
r
r
r r� � s r� �__main__) r rZ r% r �
contextlibr� r Zunittestr' r ZTestCaser �objectr? r& rP � BaseSelectorrV �contextmanagerrk rp rq rt r� r� r� r� r� r< �mainr
r
r
r �<module> s0 B
b P