File: //usr/lib/python3/dist-packages/requests_unixsocket/__pycache__/testutils.cpython-38.pyc
U
��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||� || _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� ZSleeping)�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 %sZ PATH_INFOz WSGIApp.__call__: environ = %rz200 OK)z X-Transportzunix domain socketz
X-Socket-PathZ SERVER_PORTzX-Requested-Query-StringZ QUERY_STRINGzX-Requested-Paths Hello world!zZWSGIApp.__call__: Responding with status_text = %r; response_headers = %r; body_bytes = %r)r r )r �environZstart_responseZ status_textZresponse_headersZ
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 s0 t t| �j||� | �� | _d | _t�� | _d S r ) r r r �get_tempfile_name�usockr � threadingZEvent�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�uuidZuuid4�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 ...)Z unix_socket)
r r r �waitressZ
create_serverr r r" �setr )r Zwsgi_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 �� | jr t| j��� d S r )r" r. r r r- r* r r r
�__exit__^ s
z UnixSocketServerThread.__exit__) r r r r r r r/ r0 r r r r
r
r @ s
r )�__doc__Zloggingr# r! r r( r+ Z getLoggerr r ZThreadr r r r r r r
�<module> s