Current Dir: /snap/core22/2955/usr/lib/python3/dist-packages/requests_unixsocket/__pycache__/
[FILE] adapters.cpython-310.pyc [ edit | delete | rename ]
[FILE] testutils.cpython-310.pyc [ edit | delete | rename ]
[FILE] __init__.cpython-310.pyc [ edit | delete | rename ]
Viewing: /snap/core22/2955/usr/lib/python3/dist-packages/requests_unixsocket/__pycache__/testutils.cpython-310.pyc
o
��V] � @ st d Z ddlZddlZddlZddlZddlZddlZe�e�Z G dd� dej
�Z G dd� d�Z G dd� dej
�Z
dS ) a
Utilities helpful for writing tests
Provides a UnixSocketServerThread that creates a running server, listening on a
newly created unix socket.
Example usage:
.. code-block:: python
def test_unix_domain_adapter_monkeypatch():
with UnixSocketServerThread() as usock_thread:
with requests_unixsocket.monkeypatch('http+unix://'):
urlencoded_usock = quote_plus(usock_process.usock)
url = 'http+unix://%s/path/to/page' % urlencoded_usock
r = requests.get(url)
� Nc s$ e Zd Z� fdd�Zdd� Z� ZS )�
KillThreadc s t t| �j|i |�� || _d S �N)�superr �__init__�server)�selfr �args�kwargs�� __class__� �?/usr/lib/python3/dist-packages/requests_unixsocket/testutils.pyr s
zKillThread.__init__c C s$ t �d� t�d� | jj�� d S )N� �Sleeping)�time�sleep�logger�debugr �_map�clear�r r r r
�run# s
zKillThread.run)�__name__�
__module__� __qualname__r r �
__classcell__r r r
r
r s r c @ s e Zd ZdZdd� ZdS )�WSGIAppNc C sh t �d|d � t �d|� d}dd|d fd|d fd
|d fg}d }|||� t �d |||� |gS )
Nz WSGIApp.__call__: Invoked for %s� PATH_INFOz WSGIApp.__call__: environ = %rz200 OK)z X-Transportzunix domain socketz
X-Socket-Path� SERVER_PORTzX-Requested-Query-String� QUERY_STRINGzX-Requested-Paths Hello world!zZWSGIApp.__call__: Responding with status_text = %r; response_headers = %r; body_bytes = %r)r r )r �environ�start_response� status_text�response_headers�
body_bytesr r r
�__call__, s
�
�zWSGIApp.__call__)r r r r r% r r r r
r ) s r c s< e Zd Z� fdd�Zdd� Zdd� Zdd� Zd d
� Z� ZS ) �UnixSocketServerThreadc s4 t t| �j|i |�� | �� | _d | _t�� | _d S r ) r r& r �get_tempfile_name�usockr � threading�Event�server_ready_event)r r r r
r r
r A s
z UnixSocketServerThread.__init__c C s, t �t�jt �� t�� jdd � f}d| S )Ni����z/tmp/test_requests.%s_%s_%s)�os�stat�__file__�st_ino�getpid�uuid�uuid4�hex�r r r r r
r' G s $z(UnixSocketServerThread.get_tempfile_namec C sD t �d| � t� }tj|| jd�}||_|| _| j�� |� � d S )Nz Call waitress.serve in %r ...)� unix_socket)
r r r �waitress�
create_serverr( r r+ �setr )r �wsgi_appr r r r
r N s
zUnixSocketServerThread.runc C s0 t �d| � | �� t �d| � | j�� | S )NzStarting %r ...z Started %r.)r r �startr+ �waitr r r r
� __enter__W s
z UnixSocketServerThread.__enter__c G s& | j �� | jrt| j��� d S d S r )r+ r; r r r: r4 r r r
�__exit__^ s
�z UnixSocketServerThread.__exit__) r r r r r' r r< r= r r r r
r
r& @ s r& )�__doc__�loggingr, r) r r1 r6 � getLoggerr r �Threadr r r& r r r r
�<module> s