Improve tests

* Use context manager instead of pytest.fixture
* Add logging
* etc.

requests_unixsocket/tests/test_requests_unixsocket.py

@@ -3,6 +3,7 @@
 
 """Tests for requests_unixsocket"""
 
+import logging
 import multiprocessing
 import os
 import uuid
@@ -14,7 +15,9 @@ import waitress
 from requests_unixsocket import UnixAdapter
 
 
-@pytest.fixture
+logger = logging.getLogger(__name__)
+
+
 def wsgiapp():
     def _wsgiapp(environ, start_response):
         start_response(
@@ -27,49 +30,53 @@ def wsgiapp():
     return _wsgiapp
 
 
-@pytest.fixture
-def usock_process(wsgiapp):
 class UnixSocketServerProcess(multiprocessing.Process):
     def __init__(self, *args, **kwargs):
         super(UnixSocketServerProcess, self).__init__(*args, **kwargs)
-            self.unix_socket = self.get_tempfile_name()
+        self.usock = self.get_tempfile_name()
 
     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,
+        args = (os.stat(__file__).st_ino, os.getpid(), uuid.uuid4().hex[-8:])
-                    os.getpid(),
-                    uuid.uuid4().hex[-8:])
         return '/tmp/test_requests.%s_%s_%s' % args
 
     def run(self):
-            waitress.serve(wsgiapp, unix_socket=self.unix_socket)
+        logger.debug('Call waitress.serve in %r (pid %d) ...', self, self.pid)
+        waitress.serve(wsgiapp(), unix_socket=self.usock)
 
-    return UnixSocketServerProcess()
+    def __enter__(self):
+        logger.debug('Starting %r ...' % self)
+        self.start()
+        logger.debug('Started %r (pid %d)...', self, self.pid)
+        return self
+
+    def __exit__(self, *args):
+        logger.debug('Terminating %r (pid %d) ...', self, self.pid)
+        self.terminate()
+        logger.debug('Terminated %r (pid %d) ...', self, self.pid)
 
 
-def test_unix_domain_adapter_ok(usock_process):
+def test_unix_domain_adapter_ok():
-    from requests.compat import quote_plus
+    with UnixSocketServerProcess() as usock_process:
-
-    usock_process.start()
-
-    try:
         session = requests.Session()
         session.mount('http+unix://', UnixAdapter())
-        urlencoded_socket_name = quote_plus(usock_process.unix_socket)
+        urlencoded_usock = requests.compat.quote_plus(usock_process.usock)
-        url = 'http+unix://%s/path/to/page' % urlencoded_socket_name
+        url = 'http+unix://%s/path/to/page' % urlencoded_usock
+        logger.debug('Calling session.get(%r) ...', url)
         r = session.get(url)
+        logger.debug(
+            'Received response: %r with text: %r and headers: %r',
+            r, r.text, r.headers)
         assert r.status_code == 200
         assert r.headers['server'] == 'waitress'
         assert r.headers['X-Transport'] == 'unix domain socket'
         assert r.headers['X-Requested-Path'] == '/path/to/page'
-        assert r.headers['X-Socket-Path'] == usock_process.unix_socket
+        assert r.headers['X-Socket-Path'] == usock_process.usock
         assert isinstance(r.connection, UnixAdapter)
         assert r.url == url
         assert r.text == 'Hello world!'
-    finally:
-        usock_process.terminate()
 
 
 def test_unix_domain_adapter_connection_error():

test-requirements.txt

@@ -1,2 +1,3 @@
 pytest
+pytest-capturelog
 waitress