Файловый менеджер - Редактировать - /opt/imh-python/lib/python3.9/site-packages/tornado/test/__pycache__/tcpserver_test.cpython-39.pyc
Ðазад
a _�h � @ s� d dl Z d dlZd dlZd dlZd dlZd dlmZmZ d dlm Z d dl m Z d dl m Z d dlmZ d dlmZ d dlmZmZmZmZ G d d � d e�ZeG d d � d ej��ZdS ) � N)�utf8� to_unicode)�gen)�IOStream)�app_log)� TCPServer)� skipIfNonUnix)� AsyncTestCase� ExpectLog�bind_unused_port�gen_testc @ s8 e Zd Zedd� �Zedd� �Zdd� Zedd� �Zd S ) � TCPServerTestc c s� G dd� dt �}d }}z�t� \}}|� }|�|� tt�� �}ttd��> |�d|f�V |�d�V |� � V t j V W d � n1 s�0 Y W |d ur�|� � |d ur�|� � n"|d ur�|� � |d ur�|� � 0 d S )Nc @ s e Zd Zejdd� �ZdS )zFTCPServerTest.test_handle_stream_coroutine_logging.<locals>.TestServerc s s$ |� td��V |�� dd d S )N� hello� r )� read_bytes�len�close��self�stream�address� r ��/root/rpmbuild/BUILDROOT/imh-python39-modules-3.9.7-92.el8.x86_64/opt/imh-python/lib/python3.9/site-packages/tornado/test/tcpserver_test.py� handle_stream s zTTCPServerTest.test_handle_stream_coroutine_logging.<locals>.TestServer.handle_streamN��__name__� __module__� __qualname__r � coroutiner r r r r � TestServer s r zException in callback� localhostr )r r � add_socketr �socketr r �connect�write�read_until_closer Zmoment�stopr )r r �server�client�sock�portr r r �$test_handle_stream_coroutine_logging s( ( �z2TCPServerTest.test_handle_stream_coroutine_loggingc c sp G dd� dt �}t� \}}|� }|�|� tt�� �}|�d|f�V |�� V }| �|d� |�� |� � d S )Nc @ s e Zd Zdd� ZdS )zETCPServerTest.test_handle_stream_native_coroutine.<locals>.TestServerc � s |� d� |�� d S )N� data)r$ r r r r r r 2 s zSTCPServerTest.test_handle_stream_native_coroutine.<locals>.TestServer.handle_streamN)r r r r r r r r r 1 s r r r, ) r r r! r r" r# r% � assertEqualr& r )r r r) r* r' r( �resultr r r �#test_handle_stream_native_coroutine- s z1TCPServerTest.test_handle_stream_native_coroutinec C s. t � \}}t� }|�|� |�� |�� d S �N)r r r! r& )r r) r* r' r r r �test_stop_twice@ s z TCPServerTest.test_stop_twicec # s� G �fdd�dt �}t� \}}|� ���|� d|f�d}dd� t|�D �}g �tj��fdd��� � fd d�|D �V | �t��d d � z,t��|kr�| �d � W �D ] }|� � q�n�D ] }|� � q�0 d S ) Nc s e Zd Zej� fdd��ZdS )z7TCPServerTest.test_stop_in_callback.<locals>.TestServerc 3 s � � � |�� V d S r0 )r& r% r �r' r r r N s zETCPServerTest.test_stop_in_callback.<locals>.TestServer.handle_streamNr r r2 r r r M s r r �( c S s g | ]}t t�� ��qS r )r r" )�.0�ir r r � <listcomp>X � z7TCPServerTest.test_stop_in_callback.<locals>.<listcomp>c 3 s2 z| � ��V W n ty" Y n 0 � �| � d S r0 )r# �EnvironmentError�append)�c)�connected_clients� server_addrr r r# [ s z4TCPServerTest.test_stop_in_callback.<locals>.connectc s g | ] }� |��qS r r )r4 r: )r# r r r6 d r7 r z all clients failed connectingzHat least one client should fail connecting for the test to be meaningful) r r r! �ranger r Z assertGreaterr ZskipTestr )r r r) r* �NZclientsr: r )r# r; r' r< r �test_stop_in_callbackG s* � �z#TCPServerTest.test_stop_in_callbackN)r r r r r+ r/ r1 r? r r r r r s r c @ s, e Zd Zdd� Zdd� Zdd� Zdd� Zd S ) �TestMultiprocessc C sp t jtjt jt jd�}|j�t|�� |j�� |� � |j � � }|j �� |j dkrht d|j |f ��t|�S )N)�stdin�stdoutr z Process returned %d. stdout=%r)� subprocess�Popen�sys� executable�PIPErA r$ r r �waitrB �read� returncode� RuntimeErrorr )r �code�procrB r r r � run_subproc} s � �z TestMultiprocess.run_subprocc C s. t �d�}| �|�}| �d�t|��d� d S )Na from tornado.ioloop import IOLoop from tornado.tcpserver import TCPServer server = TCPServer() server.listen(0, address='127.0.0.1') IOLoop.current().run_sync(lambda: None) print('012', end='') � �012��textwrap�dedentrN r- �join�sorted�r rL �outr r r � test_single� s � z TestMultiprocess.test_singlec C s. t �d�}| �|�}| �d�t|��d� d S )Nab from tornado.ioloop import IOLoop from tornado.process import task_id from tornado.tcpserver import TCPServer server = TCPServer() server.bind(0, address='127.0.0.1') server.start(3) IOLoop.current().run_sync(lambda: None) print(task_id(), end='') rO rP rQ rV r r r � test_simple� s � z TestMultiprocess.test_simplec C s. t �d�}| �|�}| �d�t|��d� d S )Na from tornado.ioloop import IOLoop from tornado.netutil import bind_sockets from tornado.process import fork_processes, task_id from tornado.ioloop import IOLoop from tornado.tcpserver import TCPServer sockets = bind_sockets(0, address='127.0.0.1') fork_processes(3) server = TCPServer() server.add_sockets(sockets) IOLoop.current().run_sync(lambda: None) print(task_id(), end='') rO rP rQ rV r r r � test_advanced� s � z TestMultiprocess.test_advancedN)r r r rN rX rY rZ r r r r r@ v s r@ )r" rC rE rR ZunittestZtornado.escaper r �tornador Ztornado.iostreamr Z tornado.logr Ztornado.tcpserverr Ztornado.test.utilr Ztornado.testingr r r r r ZTestCaser@ r r r r �<module> s f
| ver. 1.1 | |
.
| PHP 8.3.30 | Ð“ÐµÐ½ÐµÑ€Ð°Ñ†Ð¸Ñ Ñтраницы: 0 |
proxy
|
phpinfo
|
ÐаÑтройка