Repository: redis/redis-py
File: tests/test_maint_notifications_handling.py (Python)

Code before:
        finally:
            test_redis_client.close()

    def test_handshake_success_when_auto_and_command_not_supported(self):
        """Test that handshake is performed correctly."""
        maint_notifications_config = MaintNotificationsConfig(
            enabled="auto", endpoint_type=EndpointType.INTERNAL_IP
        )
        test_redis_client = self._get_client(
            ConnectionPool, maint_notifications_config=maint_notifications_config
        )

        try:
            # Perform Redis operations that should work with our improved mock responses
            result_set = test_redis_client.set("hello", "world")
            result_get = test_redis_client.get("hello")

            # Verify operations completed successfully
            assert result_set is True
            assert result_get == b"world"

        finally:
            test_redis_client.close()

    def test_handshake_failure_when_enabled(self):
        """Test that handshake is performed correctly."""
        maint_notifications_config = MaintNotificationsConfig(
            enabled=True, endpoint_type=EndpointType.INTERNAL_IP
        )
        test_redis_client = self._get_client(
            ConnectionPool, maint_notifications_config=maint_notifications_config
        )
        try:
            with pytest.raises(ResponseError):
                test_redis_client.set("hello", "world")

        finally:
            test_redis_client.close()


class TestMaintenanceNotificationsHandlingSingleProxy(TestMaintenanceNotificationsBase):
    """Integration tests for maintenance notifications handling with real connection pool."""

    def _validate_connection_handlers(self, conn, pool_handler, config):
        """Helper method to validate connection handlers are properly set."""
        # Test that the node moving handler function is correctly set
        parser_handler = conn._parser.node_moving_push_handler_func
        assert parser_handler is not None
        assert hasattr(parser_handler, "__self__")
        assert hasattr(parser_handler, "__func__")
        assert parser_handler.__self__ is pool_handler

Review comment (bug):
The docstring is generic and doesn't describe what this specific test is validating. It should explain that this test verifies enabled=True mode raises an exception when MAINT_NOTIFICATIONS command fails.
```suggestion
        """
        Test that when maintenance notifications are enabled (enabled=True),
        an exception (ResponseError) is raised if the MAINT_NOTIFICATIONS command fails.
        """
```

Code after:
        finally:
            test_redis_client.close()

    def test_handshake_success_when_auto_and_command_not_supported(self):
        """Test that when maintenance notifications are set to 'auto', the client gracefully handles unsupported MAINT_NOTIFICATIONS commands and normal Redis operations succeed."""
        maint_notifications_config = MaintNotificationsConfig(
            enabled="auto", endpoint_type=EndpointType.INTERNAL_IP
        )
        test_redis_client = self._get_client(
            ConnectionPool, maint_notifications_config=maint_notifications_config
        )

        try:
            # Perform Redis operations that should work with our improved mock responses
            result_set = test_redis_client.set("hello", "world")
            result_get = test_redis_client.get("hello")

            # Verify operations completed successfully
            assert result_set is True
            assert result_get == b"world"

        finally:
            test_redis_client.close()

    def test_handshake_failure_when_enabled(self):
        """Test that handshake is performed correctly."""
        maint_notifications_config = MaintNotificationsConfig(
            enabled=True, endpoint_type=EndpointType.INTERNAL_IP
        )
        test_redis_client = self._get_client(
            ConnectionPool, maint_notifications_config=maint_notifications_config
        )
        try:
            with pytest.raises(ResponseError):
                test_redis_client.set("hello", "world")

        finally:
            test_redis_client.close()


class TestMaintenanceNotificationsHandlingSingleProxy(TestMaintenanceNotificationsBase):
    """Integration tests for maintenance notifications handling with real connection pool."""

    def _validate_connection_handlers(self, conn, pool_handler, config):
        """Helper method to validate connection handlers are properly set."""
        # Test that the node moving handler function is correctly set
        parser_handler = conn._parser.node_moving_push_handler_func
        assert parser_handler is not None
        assert hasattr(parser_handler, "__self__")
        assert hasattr(parser_handler, "__func__")
        assert parser_handler.__self__ is pool_handler