ikan Uploader :
Directory : /usr/lib/python3/dist-packages/requests_unixsocket/
Upload File : |
| Current File : //usr/lib/python3/dist-packages/requests_unixsocket/testutils.py |
"""
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)
"""
import logging
import os
import threading
import time
import uuid
import waitress
logger = logging.getLogger(__name__)
class KillThread(threading.Thread):
def __init__(self, server, *args, **kwargs):
super(KillThread, self).__init__(*args, **kwargs)
self.server = server
def run(self):
time.sleep(1)
logger.debug('Sleeping')
self.server._map.clear()
class WSGIApp:
server = None
def __call__(self, environ, start_response):
logger.debug('WSGIApp.__call__: Invoked for %s', environ['PATH_INFO'])
logger.debug('WSGIApp.__call__: environ = %r', environ)
status_text = '200 OK'
response_headers = [
('X-Transport', 'unix domain socket'),
('X-Socket-Path', environ['SERVER_PORT']),
('X-Requested-Query-String', environ['QUERY_STRING']),
('X-Requested-Path', environ['PATH_INFO'])]
body_bytes = b'Hello world!'
start_response(status_text, response_headers)
logger.debug(
'WSGIApp.__call__: Responding with '
'status_text = %r; '
'response_headers = %r; '
'body_bytes = %r',
status_text, response_headers, body_bytes)
return [body_bytes]
class UnixSocketServerThread(threading.Thread):
def __init__(self, *args, **kwargs):
super(UnixSocketServerThread, self).__init__(*args, **kwargs)
self.usock = self.get_tempfile_name()
self.server = None
self.server_ready_event = threading.Event()
def get_tempfile_name(self):
# I'd rather use tempfile.NamedTemporaryFile but IDNA limits
# the hostname to 63 characters and we'll get a "InvalidURL:
# URL has an invalid label" error if we exceed that.
args = (os.stat(__file__).st_ino, os.getpid(), uuid.uuid4().hex[-8:])
return '/tmp/test_requests.%s_%s_%s' % args
def run(self):
logger.debug('Call waitress.serve in %r ...', self)
wsgi_app = WSGIApp()
server = waitress.create_server(wsgi_app, unix_socket=self.usock)
wsgi_app.server = server
self.server = server
self.server_ready_event.set()
server.run()
def __enter__(self):
logger.debug('Starting %r ...' % self)
self.start()
logger.debug('Started %r.', self)
self.server_ready_event.wait()
return self
def __exit__(self, *args):
self.server_ready_event.wait()
if self.server:
KillThread(self.server).start()
Kontol Shell Bypass