fix(core): apply body predicates to accepted HTTP error responses

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

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."""

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!")