0x1949 Team - FAZEMRX - MANAGER
Edit File: 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