blocking_httpserver: add howto and example

csernazs · csernazs · commit d2003ec16055 · 2022-09-05T22:16:45.000+02:00
diff --git a/doc/howto.rst b/doc/howto.rst
@@ -655,3 +655,19 @@ address to ``127.0.0.1`` (and thereby fixing the host part of the URL returned
 by the `url_for` call) solves the issue as the client will receive the address
 (``127.0.0.1``) instead of the name (``localhost``) so it won't move on to the
 IPv6 address.
+
+Running httpserver in blocking mode
+-----------------------------------
+
+In this mode, the code which is being tested (the client) is executed in a
+background thread, while the server events are synchronized to the main thread,
+so it looks like it is running in the main thread. This allows to catch the
+assertions occured on the server side synchronously, it is not needed to call
+the `check_assertions` method, assertions are raised to the main thread.
+
+This is an experimental feature so *pytest-httpserver* has no fixture for it.
+
+Example:
+
+.. literalinclude :: ../tests/test_blocking_httpserver_howto.py
+   :language: python
diff --git a/tests/test_blocking_httpserver_howto.py b/tests/test_blocking_httpserver_howto.py
@@ -0,0 +1,53 @@
+import threading
+from queue import Queue
+
+import pytest
+import requests
+
+from pytest_httpserver import BlockingHTTPServer
+
+# override httpserver fixture
+
+
+@pytest.fixture
+def httpserver():
+    server = BlockingHTTPServer(timeout=1)
+    server.start()
+
+    yield server
+
+    server.clear()
+    if server.is_running():
+        server.stop()
+
+
+def test_simplified(httpserver: BlockingHTTPServer):
+    def client(response_queue: Queue):
+        response = requests.get(httpserver.url_for("/foobar"), timeout=10)
+        response_queue.put(response)
+
+    # start the client, server is not yet configured
+    # it will block until we add a request handler to the server
+    # (see the timeout parameter of the http server)
+    response_queue: Queue[requests.models.Response] = Queue(maxsize=1)
+    thread = threading.Thread(target=client, args=(response_queue,))
+    thread.start()
+
+    try:
+        # check that the request is for /foobar and it is a GET method
+        # if this does not match, it will raise AssertionError and test will fail
+        client_connection = httpserver.assert_request(uri="/foobar", method="GET")
+
+        # with the received client_connection, we now need to send back the response
+        # this makes the request.get() call in client() to return
+        client_connection.respond_with_json({"foo": "bar"})
+
+    finally:
+        # wait for the client thread to complete
+        thread.join(timeout=1)
+        assert not thread.is_alive()  # check if join() has not timed out
+
+    # check the response the client received
+    response = response_queue.get(timeout=1)
+    assert response.status_code == 200
+    assert response.json() == {"foo": "bar"}
diff --git a/tests/test_release.py b/tests/test_release.py
@@ -187,6 +187,7 @@ def test_sdist_contents(build: Build, version: str):
             "assets",
             "conftest.py",
             "test_blocking_httpserver.py",
+            "test_blocking_httpserver_howto.py",
             "test_handler_errors.py",
             "test_headers.py",
             "test_json_matcher.py",
