Add internal function to check if server is >=4.2.0

This will be useful when we implement SQL filter support.

Author: Zerpet

rabbitmq_amqp_python_client/connection.py

@@ -211,6 +211,38 @@ def _validate_server_properties(self) -> None:
 
         logger.debug(f"Connected to RabbitMQ server version {server_version}")
 
+    def _is_server_version_gte_4_2_0(self) -> bool:
+        """
+        Check if the server version is greater than or equal to 4.2.0.
+
+        This is an internal method that can be used to conditionally enable
+        features that require RabbitMQ 4.2.0 or higher.
+
+        Returns:
+            bool: True if server version >= 4.2.0, False otherwise
+
+        Raises:
+            ValidationCodeException: If connection is not established or
+                                   remote properties are not available
+        """
+        if self._conn is None or self._conn.conn is None:
+            raise ValidationCodeException("Connection not established")
+
+        remote_props = self._conn.conn.remote_properties
+        if remote_props is None:
+            raise ValidationCodeException("No remote properties received from server")
+
+        server_version = remote_props.get("version")
+        if server_version is None:
+            raise ValidationCodeException("Server version not provided")
+
+        try:
+            return version.parse(str(server_version)) >= version.parse("4.2.0")
+        except Exception as e:
+            raise ValidationCodeException(
+                f"Failed to parse server version '{server_version}': {e}"
+            )
+
     def dial(self) -> None:
         """
         Establish a connection to the AMQP server.

tests/test_server_validation.py

@@ -274,3 +274,171 @@ def test_validate_server_properties_with_additional_properties(self):
 
         # Should not raise any exception despite additional properties
         self.connection._validate_server_properties()
+
+
+class TestServerVersionGte420:
+    """Test cases for the _is_server_version_gte_4_2_0 method."""
+
+    def setup_method(self):
+        """Set up test fixtures."""
+        self.connection = Connection.__new__(Connection)
+        # Initialize required attributes that would normally be set in __init__
+        self.connection._conn = None
+        self.connection._addr = "amqp://localhost:5672"
+        self.connection._addrs = None
+        self.connection._conf_ssl_context = None
+        self.connection._managements = []
+        self.connection._recovery_configuration = None
+        self.connection._ssl_domain = None
+        self.connection._connections = []
+        self.connection._index = -1
+        self.connection._publishers = []
+        self.connection._consumers = []
+        self.connection._oauth2_options = None
+
+    def test_is_server_version_gte_4_2_0_exact_version(self):
+        """Test with exact version 4.2.0."""
+        mock_blocking_conn = Mock()
+        mock_proton_conn = Mock()
+        mock_remote_props = {"product": "RabbitMQ", "version": "4.2.0"}
+
+        mock_proton_conn.remote_properties = mock_remote_props
+        mock_blocking_conn.conn = mock_proton_conn
+        self.connection._conn = mock_blocking_conn
+
+        result = self.connection._is_server_version_gte_4_2_0()
+        assert result is True
+
+    def test_is_server_version_gte_4_2_0_higher_versions(self):
+        """Test with versions higher than 4.2.0."""
+        higher_versions = ["4.2.1", "4.3.0", "4.10.5", "5.0.0", "6.1.2"]
+
+        for version_str in higher_versions:
+            mock_blocking_conn = Mock()
+            mock_proton_conn = Mock()
+            mock_remote_props = {"product": "RabbitMQ", "version": version_str}
+
+            mock_proton_conn.remote_properties = mock_remote_props
+            mock_blocking_conn.conn = mock_proton_conn
+            self.connection._conn = mock_blocking_conn
+
+            result = self.connection._is_server_version_gte_4_2_0()
+            assert result is True, f"Version {version_str} should return True"
+
+    def test_is_server_version_gte_4_2_0_lower_versions(self):
+        """Test with versions lower than 4.2.0."""
+        lower_versions = ["4.1.9", "4.1.0", "4.0.0", "3.12.0", "3.9.5"]
+
+        for version_str in lower_versions:
+            mock_blocking_conn = Mock()
+            mock_proton_conn = Mock()
+            mock_remote_props = {"product": "RabbitMQ", "version": version_str}
+
+            mock_proton_conn.remote_properties = mock_remote_props
+            mock_blocking_conn.conn = mock_proton_conn
+            self.connection._conn = mock_blocking_conn
+
+            result = self.connection._is_server_version_gte_4_2_0()
+            assert result is False, f"Version {version_str} should return False"
+
+    def test_is_server_version_gte_4_2_0_no_connection(self):
+        """Test when connection is None."""
+        self.connection._conn = None
+
+        with pytest.raises(ValidationCodeException) as exc_info:
+            self.connection._is_server_version_gte_4_2_0()
+
+        assert "Connection not established" in str(exc_info.value)
+
+    def test_is_server_version_gte_4_2_0_no_proton_connection(self):
+        """Test when proton connection is None."""
+        mock_blocking_conn = Mock()
+        mock_blocking_conn.conn = None
+        self.connection._conn = mock_blocking_conn
+
+        with pytest.raises(ValidationCodeException) as exc_info:
+            self.connection._is_server_version_gte_4_2_0()
+
+        assert "Connection not established" in str(exc_info.value)
+
+    def test_is_server_version_gte_4_2_0_no_remote_properties(self):
+        """Test when remote_properties is None."""
+        mock_blocking_conn = Mock()
+        mock_proton_conn = Mock()
+        mock_proton_conn.remote_properties = None
+        mock_blocking_conn.conn = mock_proton_conn
+        self.connection._conn = mock_blocking_conn
+
+        with pytest.raises(ValidationCodeException) as exc_info:
+            self.connection._is_server_version_gte_4_2_0()
+
+        assert "No remote properties received from server" in str(exc_info.value)
+
+    def test_is_server_version_gte_4_2_0_missing_version(self):
+        """Test when version property is missing."""
+        mock_blocking_conn = Mock()
+        mock_proton_conn = Mock()
+        mock_remote_props = {
+            "product": "RabbitMQ"
+            # Missing "version" key
+        }
+
+        mock_proton_conn.remote_properties = mock_remote_props
+        mock_blocking_conn.conn = mock_proton_conn
+        self.connection._conn = mock_blocking_conn
+
+        with pytest.raises(ValidationCodeException) as exc_info:
+            self.connection._is_server_version_gte_4_2_0()
+
+        assert "Server version not provided" in str(exc_info.value)
+
+    def test_is_server_version_gte_4_2_0_invalid_version_format(self):
+        """Test with invalid version formats."""
+        invalid_versions = ["invalid-version", "", "not-a-version"]
+
+        for version_str in invalid_versions:
+            mock_blocking_conn = Mock()
+            mock_proton_conn = Mock()
+            mock_remote_props = {"product": "RabbitMQ", "version": version_str}
+
+            mock_proton_conn.remote_properties = mock_remote_props
+            mock_blocking_conn.conn = mock_proton_conn
+            self.connection._conn = mock_blocking_conn
+
+            with pytest.raises(ValidationCodeException) as exc_info:
+                self.connection._is_server_version_gte_4_2_0()
+
+            error_msg = str(exc_info.value)
+            assert "Failed to parse server version" in error_msg
+
+    def test_is_server_version_gte_4_2_0_edge_cases(self):
+        """Test with edge case versions."""
+        # Test cases around the boundary
+        test_cases = [
+            ("4.1.99", False),  # Just below
+            ("4.2.0", True),  # Exact match
+            ("4.2.0.0", True),  # With extra zeroes
+            ("v4.2.0", True),  # With v prefix
+            ("4.2.0-rc1", False),  # Pre-release should be less than 4.2.0
+        ]
+
+        for version_str, expected in test_cases:
+            mock_blocking_conn = Mock()
+            mock_proton_conn = Mock()
+            mock_remote_props = {"product": "RabbitMQ", "version": version_str}
+
+            mock_proton_conn.remote_properties = mock_remote_props
+            mock_blocking_conn.conn = mock_proton_conn
+            self.connection._conn = mock_blocking_conn
+
+            if version_str == "4.2.0-rc1":
+                # Pre-release versions should be handled correctly
+                result = self.connection._is_server_version_gte_4_2_0()
+                assert (
+                    result == expected
+                ), f"Version {version_str} should return {expected}"
+            else:
+                result = self.connection._is_server_version_gte_4_2_0()
+                assert (
+                    result == expected
+                ), f"Version {version_str} should return {expected}"