From 8865edc228e13f4421c80c8bab3da53d8f3854c9 Mon Sep 17 00:00:00 2001 From: Pranav Mulakala Date: Fri, 23 Oct 2026 15:21:07 -0400 Subject: [PATCH] fix(core): apply body predicates to accepted HTTP error responses --- src/testcontainers/core/wait_strategies.py | 5 ++ tests/core/test_wait_strategies.py | 83 +++++++++++++++++++ .../core/test_wait_strategies_integration.py | 48 ++++++++++- 3 files changed, 135 insertions(+), 1 deletion(-) diff --git a/src/testcontainers/core/wait_strategies.py b/src/testcontainers/core/wait_strategies.py index 4e9f5a778..13f5469cb 100644 --- a/src/testcontainers/core/wait_strategies.py +++ b/src/testcontainers/core/wait_strategies.py @@ -249,6 +249,9 @@ def for_response_predicate(self, predicate: Callable[[str], bool]) -> Self: """ Set a predicate to match response body against. + The predicate must match for every accepted status code, including + explicitly accepted HTTP error statuses. + Args: predicate: Function that takes response body and returns True if acceptable @@ -436,6 +439,8 @@ def _handle_http_error(self, error: Union[URLError, HTTPError]) -> bool: error.code in self._status_codes or (self._status_code_predicate and self._status_code_predicate(error.code)) ): + if self._response_predicate is not None: + return self._response_predicate(error.read().decode()) return True logger.debug(f"HTTP request failed: {error!s}") return False diff --git a/tests/core/test_wait_strategies.py b/tests/core/test_wait_strategies.py index 60c1be572..c1537d767 100644 --- a/tests/core/test_wait_strategies.py +++ b/tests/core/test_wait_strategies.py @@ -389,6 +389,89 @@ def test_try_http_request_closes_http_error(self, mock_urlopen, status_codes, ex assert result is expected_result assert fp.closed + @pytest.mark.parametrize("status_code", [404, 503]) + @pytest.mark.parametrize("match_status_with_predicate", [False, True]) + @pytest.mark.parametrize("body,expected_result", [(b"starting", False), (b"ready", True)]) + @patch("testcontainers.core.wait_strategies.urlopen") + def test_accepted_http_error_checks_response_body( + self, mock_urlopen, status_code, match_status_with_predicate, body, expected_result + ): + fp = io.BytesIO(body) + mock_urlopen.side_effect = HTTPError("http://localhost:8080/", status_code, "", Message(), fp) + response_predicate = Mock(side_effect=lambda text: text == "ready") + strategy = HttpWaitStrategy(8080).for_response_predicate(response_predicate) + if match_status_with_predicate: + strategy.for_status_code_matching(lambda code: code == status_code) + else: + strategy.for_status_code(status_code) + + result = strategy._try_http_request("http://localhost:8080/", {}, None) + + assert result is expected_result + response_predicate.assert_called_once_with(body.decode()) + assert fp.closed + + @patch("testcontainers.core.wait_strategies.urlopen") + def test_unaccepted_http_error_does_not_check_response_body(self, mock_urlopen): + fp = io.BytesIO(b"ready") + mock_urlopen.side_effect = HTTPError("http://localhost:8080/", 503, "", Message(), fp) + response_predicate = Mock(return_value=True) + strategy = HttpWaitStrategy(8080).for_response_predicate(response_predicate) + + assert not strategy._try_http_request("http://localhost:8080/", {}, None) + + response_predicate.assert_not_called() + assert fp.closed + + @patch("testcontainers.core.wait_strategies.urlopen") + def test_http_error_is_closed_when_response_predicate_raises(self, mock_urlopen): + fp = io.BytesIO(b"invalid response") + mock_urlopen.side_effect = HTTPError("http://localhost:8080/", 503, "", Message(), fp) + response_predicate = Mock(side_effect=ValueError("invalid response")) + strategy = HttpWaitStrategy(8080).for_status_code(503).for_response_predicate(response_predicate) + + with pytest.raises(ValueError, match="invalid response"): + strategy._try_http_request("http://localhost:8080/", {}, None) + + assert fp.closed + + @patch("testcontainers.core.wait_strategies.time.sleep") + @patch("testcontainers.core.wait_strategies.urlopen") + def test_http_wait_retries_until_error_response_body_matches(self, mock_urlopen, mock_sleep): + bodies = [io.BytesIO(b"starting"), io.BytesIO(b"ready")] + mock_urlopen.side_effect = [HTTPError("http://localhost:8080/", 503, "", Message(), fp) for fp in bodies] + container = Mock() + container.get_container_host_ip.return_value = "localhost" + container.get_exposed_port.return_value = 8080 + strategy = HttpWaitStrategy(8080).for_status_code(503).for_response_predicate(lambda body: body == "ready") + + strategy.wait_until_ready(container) + + assert mock_urlopen.call_count == 2 + mock_sleep.assert_called_once() + assert all(fp.closed for fp in bodies) + + @patch("testcontainers.core.wait_strategies.time.sleep") + @patch("testcontainers.core.wait_strategies.time.time", side_effect=[0, 0, 2]) + @patch("testcontainers.core.wait_strategies.urlopen") + def test_http_wait_times_out_when_error_response_body_never_matches(self, mock_urlopen, mock_time, mock_sleep): + fp = io.BytesIO(b"starting") + mock_urlopen.side_effect = HTTPError("http://localhost:8080/", 503, "", Message(), fp) + container = Mock() + container.get_container_host_ip.return_value = "localhost" + container.get_exposed_port.return_value = 8080 + strategy = ( + HttpWaitStrategy(8080) + .for_status_code(503) + .for_response_predicate(lambda body: body == "ready") + .with_startup_timeout(1) + ) + + with pytest.raises(TimeoutError, match="HTTP endpoint not ready"): + strategy.wait_until_ready(container) + + assert fp.closed + class TestHealthcheckWaitStrategy: """Test the HealthcheckWaitStrategy class.""" diff --git a/tests/core/test_wait_strategies_integration.py b/tests/core/test_wait_strategies_integration.py index e54fc7c9b..4ca1ae58a 100644 --- a/tests/core/test_wait_strategies_integration.py +++ b/tests/core/test_wait_strategies_integration.py @@ -1,14 +1,60 @@ from pathlib import Path +from textwrap import dedent +from urllib.request import urlopen import pytest from testcontainers.core.container import DockerContainer -from testcontainers.core.wait_strategies import LogMessageWaitStrategy +from testcontainers.core.wait_strategies import HttpWaitStrategy, LogMessageWaitStrategy class TestRealDockerIntegration: """Integration tests using real Docker containers.""" + @pytest.mark.parametrize("status_code", [200, 404, 503]) + def test_http_wait_checks_response_body_with_real_container(self, status_code): + # Keep the status constant so readiness depends on the response body. + # The first health request is not ready; the second one is. + server = dedent( + f""" + from http.server import BaseHTTPRequestHandler, HTTPServer + + class Handler(BaseHTTPRequestHandler): + health_requests = 0 + + def do_GET(self): + if self.path == "/requests": + self.send_response(200) + self.end_headers() + self.wfile.write(str(Handler.health_requests).encode()) + return + Handler.health_requests += 1 + body = b"starting" if Handler.health_requests == 1 else b"ready" + self.send_response({status_code}) + self.end_headers() + self.wfile.write(body) + + HTTPServer(("0.0.0.0", 8080), Handler).serve_forever() + """ + ) + strategy = ( + HttpWaitStrategy(8080, "/health") + .for_status_code(status_code) + .for_response_predicate(lambda body: body == "ready") + .with_startup_timeout(30) + .with_poll_interval(0.1) + ) + + with ( + DockerContainer("python:3.12-alpine") + .with_exposed_ports(8080) + .with_command(["python", "-u", "-c", server]) + .waiting_for(strategy) as container + ): + url = f"http://{container.get_container_host_ip()}:{container.get_exposed_port(8080)}/requests" + with urlopen(url, timeout=5) as response: + assert response.read() == b"2" + def test_log_message_wait_strategy_with_real_container(self): """Test LogMessageWaitStrategy with a real container that outputs known logs.""" strategy = LogMessageWaitStrategy("Hello from Docker!")