Fix PubSub client health check handling for sync client implementation

redis/client.py

@@ -1009,10 +1009,22 @@ def is_health_check_response(self, response) -> bool:
         If there are no subscriptions redis responds to PING command with a
         bulk response, instead of a multi-bulk with "pong" and the response.
         """
-        return response in [
-            self.health_check_response,  # If there was a subscription
-            self.health_check_response_b,  # If there wasn't
-        ]
+        if self.encoder.decode_responses:
+            return (
+                response
+                in [
+                    self.health_check_response,  # If there is a subscription
+                    self.HEALTH_CHECK_MESSAGE,  # If there are no subscriptions and decode_responses=True
+                ]
+            )
+        else:
+            return (
+                response
+                in [
+                    self.health_check_response,  # If there is a subscription
+                    self.health_check_response_b,  # If there isn't a subscription and decode_responses=False
+                ]
+            )
 
     def check_health(self) -> None:
         conn = self.connection

tests/test_asyncio/test_pubsub.py

@@ -671,6 +671,63 @@ async def test_send_pubsub_ping_message(self, r: redis.Redis):
         await p.aclose()
 
 
+@pytest.mark.onlynoncluster
+class TestPubSubHealthCheckResponse:
+    """Tests for health check response validation with different decode_responses settings"""
+
+    async def test_health_check_response_decode_false_list_format(self, r: redis.Redis):
+        """Test health_check_response includes list format with decode_responses=False"""
+        p = r.pubsub()
+        # List format: [b"pong", b"redis-py-health-check"]
+        assert [b"pong", b"redis-py-health-check"] in p.health_check_response
+        await p.aclose()
+
+    async def test_health_check_response_decode_false_bytes_format(
+        self, r: redis.Redis
+    ):
+        """Test health_check_response includes bytes format with decode_responses=False"""
+        p = r.pubsub()
+        # Bytes format: b"redis-py-health-check"
+        assert b"redis-py-health-check" in p.health_check_response
+        await p.aclose()
+
+    async def test_health_check_response_decode_true_list_format(self, create_redis):
+        """Test health_check_response includes list format with decode_responses=True"""
+        r = await create_redis(decode_responses=True)
+        p = r.pubsub()
+        # List format: ["pong", "redis-py-health-check"]
+        assert ["pong", "redis-py-health-check"] in p.health_check_response
+        await p.aclose()
+        await r.aclose()
+
+    async def test_health_check_response_decode_true_string_format(self, create_redis):
+        """Test health_check_response includes string format with decode_responses=True"""
+        r = await create_redis(decode_responses=True)
+        p = r.pubsub()
+        # String format: "redis-py-health-check" (THE FIX!)
+        assert "redis-py-health-check" in p.health_check_response
+        await p.aclose()
+        await r.aclose()
+
+    async def test_health_check_response_decode_false_excludes_string(
+        self, r: redis.Redis
+    ):
+        """Test health_check_response excludes string format with decode_responses=False"""
+        p = r.pubsub()
+        # String format should NOT be in the list when decode_responses=False
+        assert "redis-py-health-check" not in p.health_check_response
+        await p.aclose()
+
+    async def test_health_check_response_decode_true_excludes_bytes(self, create_redis):
+        """Test health_check_response excludes bytes format with decode_responses=True"""
+        r = await create_redis(decode_responses=True)
+        p = r.pubsub()
+        # Bytes format should NOT be in the list when decode_responses=True
+        assert b"redis-py-health-check" not in p.health_check_response
+        await p.aclose()
+        await r.aclose()
+
+
 @pytest.mark.onlynoncluster
 class TestPubSubConnectionKilled:
     @skip_if_server_version_lt("3.0.0")

tests/test_pubsub.py

@@ -892,6 +892,67 @@ def test_send_pubsub_ping_message(self, r):
         )
 
 
+@pytest.mark.onlynoncluster
+class TestPubSubHealthCheckResponse:
+    """Tests for health check response validation with different decode_responses settings"""
+
+    def test_is_health_check_response_decode_false_list_format(self, r):
+        """Test is_health_check_response recognizes list format with decode_responses=False"""
+        p = r.pubsub()
+        # List format: [b"pong", b"redis-py-health-check"]
+        assert p.is_health_check_response([b"pong", b"redis-py-health-check"])
+
+    def test_is_health_check_response_decode_false_bytes_format(self, r):
+        """Test is_health_check_response recognizes bytes format with decode_responses=False"""
+        p = r.pubsub()
+        # Bytes format: b"redis-py-health-check"
+        assert p.is_health_check_response(b"redis-py-health-check")
+
+    def test_is_health_check_response_decode_false_rejects_string(self, r):
+        """Test is_health_check_response rejects string format with decode_responses=False"""
+        p = r.pubsub()
+        # String format should NOT be recognized when decode_responses=False
+        assert not p.is_health_check_response("redis-py-health-check")
+
+    def test_is_health_check_response_decode_true_list_format(self, request):
+        """Test is_health_check_response recognizes list format with decode_responses=True"""
+        r = _get_client(redis.Redis, request, decode_responses=True)
+        p = r.pubsub()
+        # List format: ["pong", "redis-py-health-check"]
+        assert p.is_health_check_response(["pong", "redis-py-health-check"])
+
+    def test_is_health_check_response_decode_true_string_format(self, request):
+        """Test is_health_check_response recognizes string format with decode_responses=True"""
+        r = _get_client(redis.Redis, request, decode_responses=True)
+        p = r.pubsub()
+        # String format: "redis-py-health-check" (THE FIX!)
+        assert p.is_health_check_response("redis-py-health-check")
+
+    def test_is_health_check_response_decode_true_rejects_bytes(self, request):
+        """Test is_health_check_response rejects bytes format with decode_responses=True"""
+        r = _get_client(redis.Redis, request, decode_responses=True)
+        p = r.pubsub()
+        # Bytes format should NOT be recognized when decode_responses=True
+        assert not p.is_health_check_response(b"redis-py-health-check")
+
+    def test_is_health_check_response_decode_true_rejects_invalid(self, request):
+        """Test is_health_check_response rejects invalid responses with decode_responses=True"""
+        r = _get_client(redis.Redis, request, decode_responses=True)
+        p = r.pubsub()
+        # Invalid responses should be rejected
+        assert not p.is_health_check_response("invalid-response")
+        assert not p.is_health_check_response(["pong", "invalid-response"])
+        assert not p.is_health_check_response(None)
+
+    def test_is_health_check_response_decode_false_rejects_invalid(self, r):
+        """Test is_health_check_response rejects invalid responses with decode_responses=False"""
+        p = r.pubsub()
+        # Invalid responses should be rejected
+        assert not p.is_health_check_response(b"invalid-response")
+        assert not p.is_health_check_response([b"pong", b"invalid-response"])
+        assert not p.is_health_check_response(None)
+
+
 @pytest.mark.onlynoncluster
 class TestPubSubConnectionKilled:
     @skip_if_server_version_lt("3.0.0")