Make it compression=None a valid case

This regression was cased by 78f5542
And missed during review.
compression=None is the same as compression=False, but it is valid now.

Author: dkropachev

cassandra/cluster.py

@@ -685,7 +685,7 @@ class Cluster(object):
     Used for testing new protocol features incrementally before the new version is complete.
     """
 
-    compression: Union[bool, str] = True
+    compression: Union[bool, str, None] = True
     """
     Controls compression for communications between the driver and Cassandra.
     If left as the default of :const:`True`, either lz4 or snappy compression
@@ -1172,7 +1172,7 @@ def token_metadata_enabled(self, enabled):
     def __init__(self,
                  contact_points=_NOT_SET,
                  port=9042,
-                 compression: Union[bool, str] = True,
+                 compression: Union[bool, str, None] = True,
                  auth_provider=None,
                  load_balancing_policy=None,
                  reconnection_policy=None,
@@ -1285,7 +1285,7 @@ def __init__(self,
 
         self._resolve_hostnames()
 
-        if isinstance(compression, bool):
+        if isinstance(compression, bool) or compression is None:
             if compression and not locally_supported_compressions:
                 log.error(
                     "Compression is enabled, but no compression libraries are available. "

tests/unit/test_cluster.py

@@ -122,6 +122,55 @@ def test_port_range(self):
             with pytest.raises(ValueError):
                 cluster = Cluster(contact_points=['127.0.0.1'], port=invalid_port)
 
+    def test_compression_autodisabled_without_libraries(self):
+        with patch.dict('cassandra.cluster.locally_supported_compressions', {}, clear=True):
+            with patch('cassandra.cluster.log') as patched_logger:
+                cluster = Cluster(compression=True)
+
+        patched_logger.error.assert_called_once()
+        assert cluster.compression is False
+
+    def test_compression_validates_requested_algorithm(self):
+        with patch.dict('cassandra.cluster.locally_supported_compressions', {}, clear=True):
+            with pytest.raises(ValueError):
+                Cluster(compression='lz4')
+
+        with patch.dict('cassandra.cluster.locally_supported_compressions', {'lz4': ('c', 'd')}, clear=True):
+            with patch('cassandra.cluster.log') as patched_logger:
+                cluster = Cluster(compression='lz4')
+
+        patched_logger.error.assert_not_called()
+        assert cluster.compression == 'lz4'
+
+    def test_compression_type_validation(self):
+        with pytest.raises(TypeError):
+            Cluster(compression=123)
+
+    def test_connection_factory_passes_compression_kwarg(self):
+        endpoint = Mock(address='127.0.0.1')
+        scenarios = [
+            ({}, True, False),
+            ({'snappy': ('c', 'd')}, True, True),
+            ({'lz4': ('c', 'd')}, 'lz4', 'lz4'),
+            ({'lz4': ('c', 'd'), 'snappy': ('c', 'd')}, False, False),
+        ]
+
+        for supported, configured, expected in scenarios:
+            with patch.dict('cassandra.cluster.locally_supported_compressions', supported, clear=True):
+                with patch.object(Cluster.connection_class, 'factory', autospec=True, return_value='connection') as factory:
+                    cluster = Cluster(compression=configured)
+                    conn = cluster.connection_factory(endpoint)
+
+                assert conn == 'connection'
+                assert factory.call_count == 1
+                assert factory.call_args.kwargs['compression'] == expected
+                assert cluster.compression == expected
+
+    def test_compression_none_not_allowed(self):
+        with patch.dict('cassandra.cluster.locally_supported_compressions', {'lz4': ('c', 'd')}, clear=True):
+            with pytest.raises(TypeError):
+                Cluster(compression=None)
+
 
 class SchedulerTest(unittest.TestCase):
     # TODO: this suite could be expanded; for now just adding a test covering a ticket