Implement http readiness checking

Send a HTTP request to the server if user specified readiness timeout.

Author: csernazs

pytest_httpserver/httpserver.py

@@ -8,6 +8,7 @@
 import threading
 import time
 import urllib.parse
+import urllib.request
 from collections import defaultdict
 from collections.abc import Callable
 from collections.abc import Generator
@@ -735,6 +736,7 @@ def thread_target(self) -> None:
         This should not be called directly, but can be overridden to tailor it to your needs.
         """
         assert self.server is not None
+        time.sleep(5)
         self.server.serve_forever()
 
     def is_running(self) -> bool:
@@ -938,6 +940,9 @@ class HTTPServer(HTTPServerBase):  # pylint: disable=too-many-instance-attribute
 
     :param threaded: whether to handle concurrent requests in separate threads
 
+    :param startup_timeout: maximum time in seconds to wait for server readiness.
+        Set to ``None`` to disable readiness waiting.
+
     .. py:attribute:: no_handler_status_code
 
         Attribute containing the http status code (int) which will be the response
@@ -956,6 +961,7 @@ def __init__(
         default_waiting_settings: WaitingSettings | None = None,
         *,
         threaded: bool = False,
+        startup_timeout: float | None = None,
     ) -> None:
         """
         Initializes the instance.
@@ -972,6 +978,32 @@ def __init__(
             self.default_waiting_settings = WaitingSettings()
         self._waiting_settings = copy(self.default_waiting_settings)
         self._waiting_result: queue.LifoQueue[bool] = queue.LifoQueue(maxsize=1)
+        self.startup_timeout = startup_timeout
+        self._readiness_check_pending = False
+
+    def start(self) -> None:
+        if self.startup_timeout is None:
+            self._readiness_check_pending = False
+        else:
+            self._readiness_check_pending = True
+
+        super().start()
+        self.wait_for_server_ready()
+
+    def wait_for_server_ready(self) -> None:
+        """
+        Waits until the server is ready to serve requests.
+        """
+        if not self._readiness_check_pending:
+            return
+
+        url = self.url_for("/")
+        if not url.startswith(("http://", "https://")):
+            raise ValueError(f"Invalid URL generated for readiness check check: {url}")  # noqa: EM102
+
+        with urllib.request.urlopen(url, timeout=self.startup_timeout) as resp:  # noqa: S310
+            if resp.status != 200 or resp.read() != b"OK":
+                raise HTTPServerError("Readiness check failed with status code: {}".format(resp.status))
 
     def clear(self) -> None:
         """
@@ -1272,6 +1304,9 @@ def dispatch(self, request: Request) -> Response:
         :param request: the request object from the werkzeug library
         :return: the response object what the handler responded, or a response which contains the error
         """
+        if self._readiness_check_pending:
+            self._readiness_check_pending = False
+            return Response("OK", status=200)
 
         if self.permanently_failed:
             return self.respond_permanent_failure()

tests/test_readiness.py

@@ -0,0 +1,23 @@
+from collections.abc import Generator
+
+import pytest
+import requests
+
+from pytest_httpserver.httpserver import HTTPServer
+
+
+@pytest.fixture
+def httpserver() -> Generator[HTTPServer, None, None]:
+    server = HTTPServer(startup_timeout=10)
+    server.start()
+    yield server
+    server.clear()
+    if server.is_running():
+        server.stop()
+
+
+def test_httpserver_readiness(httpserver: HTTPServer):
+    httpserver.expect_request("/").respond_with_data("Hello, world!")
+    resp = requests.get(httpserver.url_for("/"))
+    assert resp.status_code == 200
+    assert resp.text == "Hello, world!"

tests/test_release.py

@@ -231,6 +231,7 @@ def test_sdist_contents(build: Build, version: str):
             "test_port_changing.py",
             "test_querymatcher.py",
             "test_querystring.py",
+            "test_readiness.py",
             "test_release.py",
             "test_ssl.py",
             "test_thread_type.py",