Merge pull request #71 from graingert/configure-strict-pytest

Author: graingert

pytest_httpbin/plugin.py

@@ -6,18 +6,14 @@
 
 @pytest.fixture(scope="session")
 def httpbin(request):
-    server = serve.Server(application=httpbin_app)
-    server.start()
-    request.addfinalizer(server.stop)
-    return server
+    with serve.Server(application=httpbin_app) as server:
+        yield server
 
 
 @pytest.fixture(scope="session")
 def httpbin_secure(request):
-    server = serve.SecureServer(application=httpbin_app)
-    server.start()
-    request.addfinalizer(server.stop)
-    return server
+    with serve.SecureServer(application=httpbin_app) as server:
+        yield server
 
 
 @pytest.fixture(scope="session", params=["http", "https"])

pytest_httpbin/serve.py

@@ -60,15 +60,16 @@ def finish_request(self, request, client_address):
         """
         request.settimeout(1.0)
         try:
-            ssock = ssl.wrap_socket(
-                request,
-                keyfile=os.path.join(CERT_DIR, "key.pem"),
-                certfile=os.path.join(CERT_DIR, "cert.pem"),
-                server_side=True,
-                suppress_ragged_eofs=False,
+            context = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
+            context.load_cert_chain(
+                os.path.join(CERT_DIR, "cert.pem"),
+                os.path.join(CERT_DIR, "key.pem"),
             )
-            self.base_environ["HTTPS"] = "yes"
-            self.RequestHandlerClass(ssock, client_address, self)
+            with context.wrap_socket(
+                request, server_side=True, suppress_ragged_eofs=False
+            ) as ssock:
+                self.base_environ["HTTPS"] = "yes"
+                self.RequestHandlerClass(ssock, client_address, self)
         except Exception as e:
             print("pytest-httpbin server hit an exception serving request: %s" % e)
             print("attempting to ignore so the rest of the tests can run")
@@ -106,6 +107,16 @@ def __del__(self):
     def start(self):
         self._thread.start()
 
+    def __enter__(self):
+        self.start()
+        return self
+
+    def __exit__(self, *args, **kwargs):
+        self.stop()
+        suppress_exc = self._server.__exit__(*args, **kwargs)
+        self._thread.join()
+        return suppress_exc
+
     def __add__(self, other):
         return self.url + other
 

setup.cfg

@@ -9,3 +9,9 @@ disable-noqa = True
 max-line-length = 88
 extend-ignore =
    E203,  # whitespace before : is not PEP8 compliant (& conflicts with black)
+
+
+[tool:pytest]
+addopts = --strict-config --strict-markers
+filterwarnings = error
+xfail_strict = true

tests/test_server.py

@@ -1,7 +1,10 @@
+import contextlib
 import os
+import re
+import socket
 
 import pytest
-import requests
+import requests.exceptions
 from httpbin import app as httpbin_app
 from util import get_raw_http_response
 
@@ -40,9 +43,33 @@ def test_server_should_be_http_1_1(httpbin):
 
 
 def test_dont_crash_on_certificate_problems(httpbin_secure):
-    with pytest.raises(Exception):
+    with pytest.raises(requests.exceptions.SSLError):
         # this request used to hang
         requests.get(httpbin_secure + "/get", verify=True, cert=__file__)
+
+    # and this request would never happen
+    requests.get(
+        httpbin_secure + "/get",
+        verify=True,
+    )
+
+
+def test_dont_crash_on_handshake_timeout(httpbin_secure, capsys):
+    with socket.socket() as sock:
+        sock.connect((httpbin_secure.host, httpbin_secure.port))
+        # this request used to hang
+        assert sock.recv(1) == b""
+
+    assert (
+        re.match(
+            r"pytest-httpbin server hit an exception serving request: .* The "
+            "handshake operation timed out\nattempting to ignore so the rest "
+            "of the tests can run\n",
+            capsys.readouterr().out,
+        )
+        is not None
+    )
+
     # and this request would never happen
     requests.get(
         httpbin_secure + "/get",
@@ -68,6 +95,7 @@ def test_fixed_port_environment_variables(protocol):
     # just have different port to avoid adrress already in use
     # if the second test run too fast after the first one (happens on pypy)
     port = 12345 + len(protocol)
+    server = contextlib.nullcontext()
 
     try:
         envvar_original = os.environ.get(envvar, None)
@@ -76,10 +104,7 @@ def test_fixed_port_environment_variables(protocol):
         assert server.port == port
     finally:
         # if we don't do this, it blocks:
-        try:
-            server.start()
-            server.stop()
-        except UnboundLocalError:
+        with server:
             pass
 
         # restore the original environ:

tests/util.py

@@ -14,17 +14,17 @@ def get_raw_http_response(host, port, path):
     ]
 
     # Connect to the server
-    s = socket.socket()
-    s.connect((host, port))
+    with socket.socket() as s:
+        s.connect((host, port))
 
-    # Send an HTTP request
-    s.send(CRLF.join(request))
+        # Send an HTTP request
+        s.send(CRLF.join(request))
 
-    # Get the response (in several parts, if necessary)
-    response = b""
-    buffer = s.recv(4096)
-    while buffer:
-        response += buffer
+        # Get the response (in several parts, if necessary)
+        response = b""
         buffer = s.recv(4096)
+        while buffer:
+            response += buffer
+            buffer = s.recv(4096)
 
-    return response
+        return response

tox.ini

@@ -12,7 +12,7 @@ envlist = py37, py38, py39, py310, pypy3
 wheel = True
 wheel_build_env = build
 extras = test
-commands = pytest -v -s
+commands = pytest -v -s {posargs}
 
 [testenv:build]
 # empty environment to build universal wheel once per tox invocation