Implement http readiness checking

[csernazs]

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

[HayaoSuzuki]

I wrote a test.

from collections.abc import Generator

import pytest
import requests

from pytest_httpserver.httpserver import HTTPServer


@pytest.fixture
def httpserver_with_readiness() -> Generator[HTTPServer, None, None]:
    with HTTPServer(startup_timeout=10) as server:
        yield server


@pytest.fixture
def httpserver_without_readiness() -> Generator[HTTPServer, None, None]:
    with HTTPServer() as server:
        yield server


@pytest.mark.parametrize(
    ("startup_timeout", "expected_timeout"),
    [
        (10, 10),
        (None, None),
    ],
    ids=["with_timeout", "without_timeout"],
)
def test_httpserver_startup_timeout_attribute(startup_timeout: float | None, expected_timeout: float | None):
    with HTTPServer(startup_timeout=startup_timeout) as server:
        # Arrange
        server.expect_request("/test").respond_with_data("OK")

        # Act
        resp = requests.get(server.url_for("/test"))

        # Assert
        assert server.startup_timeout == expected_timeout
        assert resp.status_code == 200
        assert resp.text == "OK"


def test_httpserver_readiness_with_context_manager():
    with HTTPServer(startup_timeout=10) as server:
        # Arrange
        server.expect_request("/ctx").respond_with_data("context manager works")

        # Act
        resp = requests.get(server.url_for("/ctx"))

        # Assert
        assert resp.status_code == 200
        assert resp.text == "context manager works"


@pytest.mark.parametrize("num_cycles", [1, 3])
def test_httpserver_multiple_start_stop_cycles(num_cycles: int):
    # Arrange
    server = HTTPServer(startup_timeout=5)

    for i in range(num_cycles):
        # Arrange
        server.start()
        server.expect_request(f"/cycle{i}").respond_with_data(f"cycle {i}")

        # Act
        resp = requests.get(server.url_for(f"/cycle{i}"))

        # Assert
        assert resp.status_code == 200
        assert resp.text == f"cycle {i}"

        # Cleanup
        server.clear()
        server.stop()


@pytest.mark.parametrize(
    ("handler_type", "path", "expected_text"),
    [
        ("permanent", "/perm", "permanent handler"),
        ("oneshot", "/oneshot", "oneshot handler"),
    ],
    ids=["permanent", "oneshot"],
)
def test_httpserver_readiness_with_handler_types(
    httpserver_with_readiness: HTTPServer, handler_type: str, path: str, expected_text: str
):
    # Arrange
    if handler_type == "permanent":
        httpserver_with_readiness.expect_request(path).respond_with_data(expected_text)
    else:
        httpserver_with_readiness.expect_oneshot_request(path).respond_with_data(expected_text)

    # Act
    resp = requests.get(httpserver_with_readiness.url_for(path))

    # Assert
    assert resp.status_code == 200
    assert resp.text == expected_text


def test_httpserver_readiness_ordered_handlers(httpserver_with_readiness: HTTPServer):
    # Arrange
    httpserver_with_readiness.expect_ordered_request("/first").respond_with_data("1")
    httpserver_with_readiness.expect_ordered_request("/second").respond_with_data("2")

    # Act
    resp1 = requests.get(httpserver_with_readiness.url_for("/first"))
    resp2 = requests.get(httpserver_with_readiness.url_for("/second"))

    # Assert
    assert resp1.status_code == 200
    assert resp1.text == "1"
    assert resp2.status_code == 200
    assert resp2.text == "2"


def test_httpserver_readiness_does_not_interfere_with_handlers(httpserver_with_readiness: HTTPServer):
    # Arrange
    httpserver_with_readiness.expect_request("/").respond_with_data("user handler")

    # Act
    resp = requests.get(httpserver_with_readiness.url_for("/"))

    # Assert
    assert resp.status_code == 200
    assert resp.text == "user handler"


def test_httpserver_wait_for_server_ready_noop_when_no_timeout(httpserver_without_readiness: HTTPServer):
    # Act & Assert (no exception raised)
    httpserver_without_readiness.wait_for_server_ready()


@pytest.mark.parametrize("startup_timeout", [10, None], ids=["with_timeout", "without_timeout"])
def test_httpserver_readiness_check_pending_flag(startup_timeout: float | None):
    # Arrange
    server = HTTPServer(startup_timeout=startup_timeout)

    # Assert (before start)
    assert server._readiness_check_pending is False  # noqa: SLF001

    # Act
    with server:
        # Assert (after start)
        assert server._readiness_check_pending is False  # noqa: SLF001


def test_httpserver_oneshot_consumed_after_readiness(httpserver_with_readiness: HTTPServer):
    # Arrange
    httpserver_with_readiness.expect_oneshot_request("/oneshot").respond_with_data("once")

    # Act
    resp1 = requests.get(httpserver_with_readiness.url_for("/oneshot"))
    resp2 = requests.get(httpserver_with_readiness.url_for("/oneshot"))

    # Assert
    assert resp1.status_code == 200
    assert resp1.text == "once"
    assert resp2.status_code == 500

[csernazs]

hi @HayaoSuzuki ,

Thank you! I'll take a closer look. For the first sight I think some of your tests overlap with existing tests.

For example oneshot test at the end is very similar to the already defined oneshot test(s) in test/test_oneshot.py.

I'm thinking adding tests which records the calls for wait_for_server_ready() (perhaps via inheriting or "mocking" which calls the original method as well), to ensure that the server actually waited the startup.

[HayaoSuzuki]

Thank you for the feedback.
You're right that parts of the previous tests overlapped with existing ones.
I revised the tests to focus only on readiness behavior by introducing a RecordingHTTPServer subclass.
The subclass records wait_for_server_ready() calls and the _readiness_check_pending flag before and after each call while still calling the original method via super().
I also removed tests that overlapped with existing oneshot/ordered/handler-type coverage.

class RecordingHTTPServer(HTTPServer):
    """HTTPServer subclass that records wait_for_server_ready() calls."""

    def __init__(self, **kwargs: Any) -> None:
        super().__init__(**kwargs)
        self.wait_for_ready_call_count = 0
        self.readiness_pending_before_wait: list[bool] = []
        self.readiness_pending_after_wait: list[bool] = []

    def wait_for_server_ready(self) -> None:
        self.wait_for_ready_call_count += 1
        self.readiness_pending_before_wait.append(self._readiness_check_pending)
        super().wait_for_server_ready()
        self.readiness_pending_after_wait.append(self._readiness_check_pending)


@pytest.fixture
def recording_server_with_timeout() -> Generator[RecordingHTTPServer]:
    with RecordingHTTPServer(startup_timeout=10) as server:
        yield server


@pytest.fixture
def recording_server_without_timeout() -> Generator[RecordingHTTPServer]:
    with RecordingHTTPServer() as server:
        yield server


def test_wait_for_server_ready_called_with_timeout(
    recording_server_with_timeout: RecordingHTTPServer,
) -> None:
    assert recording_server_with_timeout.wait_for_ready_call_count == 1
    assert recording_server_with_timeout.readiness_pending_before_wait == [True]
    assert recording_server_with_timeout.readiness_pending_after_wait == [False]


def test_wait_for_server_ready_called_without_timeout(
    recording_server_without_timeout: RecordingHTTPServer,
) -> None:
    assert recording_server_without_timeout.wait_for_ready_call_count == 1
    assert recording_server_without_timeout.readiness_pending_before_wait == [False]
    assert recording_server_without_timeout.readiness_pending_after_wait == [False]


def test_wait_for_server_ready_called_each_start_stop_cycle() -> None:
    server = RecordingHTTPServer(startup_timeout=5)
    try:
        for i in range(3):
            server.start()
            assert server.wait_for_ready_call_count == i + 1
            server.clear()
            server.stop()
    finally:
        if server.is_running():
            server.clear()
            server.stop()

    assert server.readiness_pending_before_wait == [True, True, True]
    assert server.readiness_pending_after_wait == [False, False, False]

[csernazs]

This looks good for the first sight. Should I copy your code or do you want to make a commit? A commit with your name would be better I think as you deserve the credit.

Hopefully I'll have more time to work on this on the weekend.

Zsolt

[HayaoSuzuki]

Thanks, @csernazs .
I've pushed the changes I commented on to my fork and created a PR against your server-readiness-http-probe branch:
#464

Feel free to merge it when you have time.

[csernazs]

hi @HayaoSuzuki ,

This has been merged. Thanks for all the support and for the tests!

I'll open a separate PR with the docs & release notes update.

I also have a branch where configuring the server made easier via a fixture (as now users would need to create their own fixture which instantiates the class and passes the timeout parameter for readiness check).

Zsolt