tests: Get rid of self.fail calls

Lorak-mmk · dkropachev · commit da874c25f346 · 2025-07-18T00:37:44.000-04:00
They can be replaced, depending on situation, with:
- assert
- pytest.raises
- pytest.fail
diff --git a/tests/integration/cqlengine/columns/test_counter_column.py b/tests/integration/cqlengine/columns/test_counter_column.py
@@ -13,6 +13,7 @@
 # limitations under the License.
 
 from uuid import uuid4
+import pytest
 
 from cassandra.cqlengine import columns
 from cassandra.cqlengine.management import sync_table, drop_table
@@ -32,37 +33,28 @@ class TestClassConstruction(BaseCassEngTestCase):
 
     def test_defining_a_non_counter_column_fails(self):
         """ Tests that defining a non counter column field in a model with a counter column fails """
-        try:
+        with pytest.raises(ModelDefinitionException):
             class model(Model):
                 partition = columns.UUID(primary_key=True, default=uuid4)
                 counter = columns.Counter()
                 text = columns.Text()
-            self.fail("did not raise expected ModelDefinitionException")
-        except ModelDefinitionException:
-            pass
 
 
     def test_defining_a_primary_key_counter_column_fails(self):
         """ Tests that defining primary keys on counter columns fails """
-        try:
+        with pytest.raises(TypeError):
             class model(Model):
                 partition = columns.UUID(primary_key=True, default=uuid4)
                 cluster = columns.Counter(primary_ley=True)
                 counter = columns.Counter()
-            self.fail("did not raise expected TypeError")
-        except TypeError:
-            pass
 
         # force it
-        try:
+        with pytest.raises(ModelDefinitionException):
             class model(Model):
                 partition = columns.UUID(primary_key=True, default=uuid4)
                 cluster = columns.Counter()
                 cluster.primary_key = True
                 counter = columns.Counter()
-            self.fail("did not raise expected ModelDefinitionException")
-        except ModelDefinitionException:
-            pass
 
 
 class TestCounterColumn(BaseCassEngTestCase):
diff --git a/tests/integration/long/test_consistency.py b/tests/integration/long/test_consistency.py
@@ -17,6 +17,7 @@
 import sys
 import time
 import traceback
+import pytest
 
 from cassandra import ConsistencyLevel, OperationTimedOut, ReadTimeout, WriteTimeout, Unavailable
 from cassandra.cluster import ExecutionProfile, EXEC_PROFILE_DEFAULT
@@ -51,12 +52,12 @@ def setUp(self):
         self.coordinator_stats = CoordinatorStats()
 
     def _cl_failure(self, consistency_level, e):
-        self.fail('Instead of success, saw %s for CL.%s:\n\n%s' % (
+        pytest.fail('Instead of success, saw %s for CL.%s:\n\n%s' % (
             e, ConsistencyLevel.value_to_name[consistency_level],
             traceback.format_exc()))
 
     def _cl_expected_failure(self, cl):
-        self.fail('Test passed at ConsistencyLevel.%s:\n\n%s' % (
+        pytest.fail('Test passed at ConsistencyLevel.%s:\n\n%s' % (
                   ConsistencyLevel.value_to_name[cl], traceback.format_exc()))
 
     def _insert(self, session, keyspace, count, consistency_level=ConsistencyLevel.ONE):
@@ -360,6 +361,3 @@ def test_pool_with_host_down(self):
             start(node_to_stop)
             wait_for_up(cluster, node_to_stop)
             cluster.shutdown()
-
-
-
diff --git a/tests/integration/long/test_loadbalancingpolicies.py b/tests/integration/long/test_loadbalancingpolicies.py
@@ -16,6 +16,7 @@
 import struct
 import sys
 import traceback
+import pytest
 from cassandra import cqltypes
 
 from cassandra import ConsistencyLevel, Unavailable, OperationTimedOut, ReadTimeout, ReadFailure, \
@@ -397,11 +398,8 @@ def test_dc_aware_roundrobin_one_remote_host(self):
         self.coordinator_stats.reset_counts()
         force_stop(2)
 
-        try:
+        with pytest.raises(NoHostAvailable):
             self._query(session, keyspace)
-            self.fail()
-        except NoHostAvailable:
-            pass
 
     def test_token_aware(self):
         keyspace = 'test_token_aware'
@@ -436,13 +434,11 @@ def token_aware(self, keyspace, use_prepared=False):
         force_stop(2)
         self._wait_for_nodes_down([2], cluster)
 
-        try:
+        with pytest.raises(Unavailable) as e:
             self._query(session, keyspace, use_prepared=use_prepared)
-            self.fail()
-        except Unavailable as e:
-            assert e.consistency == 1
-            assert e.required_replicas == 1
-            assert e.alive_replicas == 0
+        assert e.value.consistency == 1
+        assert e.value.required_replicas == 1
+        assert e.value.alive_replicas == 0
 
         self.coordinator_stats.reset_counts()
         start(2)
@@ -458,11 +454,8 @@ def token_aware(self, keyspace, use_prepared=False):
         stop(2)
         self._wait_for_nodes_down([2], cluster)
 
-        try:
+        with pytest.raises(Unavailable):
             self._query(session, keyspace, use_prepared=use_prepared)
-            self.fail()
-        except Unavailable:
-            pass
 
         self.coordinator_stats.reset_counts()
         start(2)
@@ -717,11 +710,8 @@ def test_white_list(self):
         force_stop(2)
         self._wait_for_nodes_down([2])
 
-        try:
+        with pytest.raises(NoHostAvailable):
             self._query(session, keyspace)
-            self.fail()
-        except NoHostAvailable:
-            pass
 
     def test_black_list_with_host_filter_policy(self):
         """
diff --git a/tests/integration/standard/test_cluster.py b/tests/integration/standard/test_cluster.py
@@ -24,6 +24,7 @@
 import warnings
 from packaging.version import Version
 import os
+import pytest
 
 import cassandra
 from cassandra.cluster import NoHostAvailable, ExecutionProfile, EXEC_PROFILE_DEFAULT, ControlConnection, Cluster
@@ -1490,7 +1491,7 @@ def test_invalid_protocol_version_beta_option(self):
             with self.assertRaises(NoHostAvailable):
                 cluster.connect()
         except Exception as e:
-            self.fail("Unexpected error encountered {0}".format(e.message))
+            pytest.fail("Unexpected error encountered {0}".format(e.message))
 
     @protocolv6
     def test_valid_protocol_version_beta_options_connect(self):
diff --git a/tests/integration/standard/test_connection.py b/tests/integration/standard/test_connection.py
@@ -22,6 +22,7 @@
 from threading import Thread, Event
 import time
 from unittest import SkipTest
+import pytest
 
 from cassandra import ConsistencyLevel, OperationTimedOut, DependencyException
 from cassandra.cluster import NoHostAvailable, ConnectionShutdown, ExecutionProfile, EXEC_PROFILE_DEFAULT
@@ -179,7 +180,7 @@ def wait_for_connections(self, host, cluster):
             if connections:
                 return connections
             time.sleep(.1)
-        self.fail("No new connections found")
+        pytest.fail("No new connections found")
 
     def wait_for_no_connections(self, host, cluster):
         retry = 0
@@ -189,7 +190,7 @@ def wait_for_no_connections(self, host, cluster):
             if not connections:
                 return
             time.sleep(.5)
-        self.fail("Connections never cleared")
+        pytest.fail("Connections never cleared")
 
 
 class ConnectionTests(object):
diff --git a/tests/integration/standard/test_metadata.py b/tests/integration/standard/test_metadata.py
@@ -1349,8 +1349,7 @@ def send_msg(self, msg, request_id, cb, encoder=ProtocolHandler.encode_message,
         for stmt in stmts:
             if "SELECT now() FROM system.local WHERE key='local'" in stmt:
                 continue
-            if "USING TIMEOUT 2000ms" not in stmt:
-                self.fail(f"query `{stmt}` does not contain `USING TIMEOUT 2000ms`")
+            assert "USING TIMEOUT 2000ms" in stmt, f"query `{stmt}` does not contain `USING TIMEOUT 2000ms`"
 
 
 class KeyspaceAlterMetadata(unittest.TestCase):
diff --git a/tests/integration/standard/test_query.py b/tests/integration/standard/test_query.py
@@ -937,13 +937,13 @@ def test_no_connection_refused_on_timeout(self):
                 # In this case result is an exception
                 exception_type = type(result).__name__
                 if exception_type == "NoHostAvailable":
-                    self.fail("PYTHON-91: Disconnected from Cassandra: %s" % result.message)
+                    pytest.fail("PYTHON-91: Disconnected from Cassandra: %s" % result.message)
                 if exception_type in ["WriteTimeout", "WriteFailure", "ReadTimeout", "ReadFailure", "ErrorMessageSub"]:
                     if type(result).__name__ in ["WriteTimeout", "WriteFailure"]:
                         received_timeout = True
                     continue
 
-                self.fail("Unexpected exception %s: %s" % (exception_type, result.message))
+                pytest.fail("Unexpected exception %s: %s" % (exception_type, result.message))
 
         # Make sure test passed
         assert received_timeout
diff --git a/tests/integration/standard/test_query_paging.py b/tests/integration/standard/test_query_paging.py
@@ -20,6 +20,7 @@
 
 from itertools import cycle, count
 from threading import Event
+import pytest
 
 from cassandra import ConsistencyLevel
 from cassandra.cluster import EXEC_PROFILE_DEFAULT, ExecutionProfile
@@ -289,7 +290,7 @@ def handle_page(rows, future, counter, number_of_calls):
 
             def handle_error(err):
                 event.set()
-                self.fail(err)
+                pytest.fail(err)
 
             future.add_callbacks(callback=handle_page, callback_args=(future, counter, number_of_calls),
                                  errback=handle_error)
diff --git a/tests/integration/standard/test_row_factories.py b/tests/integration/standard/test_row_factories.py
@@ -16,6 +16,7 @@
     BasicSharedKeyspaceUnitTestCaseWFunctionTable, BasicSharedKeyspaceUnitTestCase, execute_until_pass, TestCluster
 
 import unittest
+import pytest
 
 from cassandra.cluster import ResultSet, ExecutionProfile, EXEC_PROFILE_DEFAULT
 from cassandra.query import tuple_factory, named_tuple_factory, dict_factory, ordered_dict_factory
@@ -194,7 +195,7 @@ def test_no_exception_on_select(self):
         try:
             self.session.execute('SELECT * FROM test1rf.table_num_col')
         except ValueError as e:
-            self.fail("Unexpected ValueError exception: %s" % e.message)
+            pytest.fail("Unexpected ValueError exception: %s" % e.message)
 
     def test_can_select_using_alias(self):
         """
@@ -206,7 +207,7 @@ def test_can_select_using_alias(self):
         try:
             self.session.execute('SELECT key, "626972746864617465" AS my_col from test1rf.table_num_col')
         except ValueError as e:
-            self.fail("Unexpected ValueError exception: %s" % e.message)
+            pytest.fail("Unexpected ValueError exception: %s" % e.message)
 
     def test_can_select_with_dict_factory(self):
         """
@@ -218,4 +219,4 @@ def test_can_select_with_dict_factory(self):
             try:
                 cluster.connect().execute('SELECT * FROM test1rf.table_num_col')
             except ValueError as e:
-                self.fail("Unexpected ValueError exception: %s" % e.message)
+                pytest.fail("Unexpected ValueError exception: %s" % e.message)
diff --git a/tests/integration/standard/test_single_interface.py b/tests/integration/standard/test_single_interface.py
@@ -13,6 +13,7 @@
 # limitations under the License.
 
 import unittest
+import pytest
 
 from cassandra import ConsistencyLevel
 from cassandra.query import SimpleStatement
@@ -58,11 +59,9 @@ def test_single_interface(self):
             assert endpoint.address == host.broadcast_rpc_address
             assert endpoint.port == host.broadcast_rpc_port
 
-            if host.broadcast_rpc_port in broadcast_rpc_ports:
-                self.fail("Duplicate broadcast_rpc_port")
+            assert host.broadcast_rpc_port not in broadcast_rpc_ports, "Duplicate broadcast_rpc_port"
             broadcast_rpc_ports.append(host.broadcast_rpc_port)
-            if host.broadcast_port in broadcast_ports:
-                self.fail("Duplicate broadcast_port")
+            assert host.broadcast_port not in broadcast_ports, "Duplicate broadcast_port"
             broadcast_ports.append(host.broadcast_port)
 
         for _ in range(1, 100):
diff --git a/tests/integration/upgrade/__init__.py b/tests/integration/upgrade/__init__.py
@@ -26,7 +26,7 @@
 from ccmlib.node import TimeoutError
 import time
 import logging
-
+import pytest
 import unittest
 
 
@@ -78,7 +78,7 @@ def setUpClass(cls):
         cls.logger_handler = MockLoggingHandler()
         cls.logger = logging.getLogger(cluster.__name__)
         cls.logger.addHandler(cls.logger_handler)
-    
+
     @classmethod
     def tearDownClass(cls):
         cls.logger.removeHandler(cls.logger_handler)
@@ -166,7 +166,7 @@ def upgrade_node(self, node):
         try:
             node.start(wait_for_binary_proto=True, wait_other_notice=True)
         except TimeoutError:
-            self.fail("Error starting C* node while upgrading")
+            pytest.fail("Error starting C* node while upgrading")
 
         return True
 
diff --git a/tests/unit/advanced/test_metadata.py b/tests/unit/advanced/test_metadata.py
@@ -135,16 +135,14 @@ def wait_for_responses(self, *msgs, **kwargs):
             p._query_all()
 
             for q in conn.queries:
-                if "USING TIMEOUT" in q.query:
-                    self.fail(f"<{schemaClass.__name__}> query `{q.query}` contains `USING TIMEOUT`, while should not")
+                assert "USING TIMEOUT" not in q.query, f"<{schemaClass.__name__}> query `{q.query}` contains `USING TIMEOUT`, while should not"
 
             conn = FakeConnection()
             p = schemaClass(conn, 2.0, 1000, datetime.timedelta(seconds=2))
             p._query_all()
 
             for q in conn.queries:
-                if "USING TIMEOUT 2000ms" not in q.query:
-                    self.fail(f"{schemaClass.__name__} query `{q.query}` does not contain `USING TIMEOUT 2000ms`")
+                assert "USING TIMEOUT 2000ms" in q.query, f"{schemaClass.__name__} query `{q.query}` does not contain `USING TIMEOUT 2000ms`"
 
 
 def get_all_schema_parser_classes(cl):
diff --git a/tests/unit/test_parameter_binding.py b/tests/unit/test_parameter_binding.py
@@ -13,6 +13,7 @@
 # limitations under the License.
 
 import unittest
+import pytest
 
 from cassandra.encoder import Encoder
 from cassandra.protocol import ColumnMetadata
@@ -91,25 +92,19 @@ def setUpClass(cls):
 
     def test_invalid_argument_type(self):
         values = (0, 0, 0, 'string not int')
-        try:
+        with pytest.raises(TypeError) as e:
             self.bound.bind(values)
-        except TypeError as e:
-            assert 'v0' in str(e)
-            assert 'Int32Type' in str(e)
-            assert 'str' in str(e)
-        else:
-            self.fail('Passed invalid type but exception was not thrown')
+        assert 'v0' in str(e)
+        assert 'Int32Type' in str(e)
+        assert 'str' in str(e)
 
         values = (['1', '2'], 0, 0, 0)
 
-        try:
+        with pytest.raises(TypeError) as e:
             self.bound.bind(values)
-        except TypeError as e:
-            assert 'rk0' in str(e)
-            assert 'Int32Type' in str(e)
-            assert 'list' in str(e)
-        else:
-            self.fail('Passed invalid type but exception was not thrown')
+        assert 'rk0' in str(e)
+        assert 'Int32Type' in str(e)
+        assert 'list' in str(e)
 
     def test_inherit_fetch_size(self):
         keyspace = 'keyspace1'
diff --git a/tests/unit/test_policies.py b/tests/unit/test_policies.py
@@ -161,8 +161,7 @@ def host_down():
             else:
                 sys.setswitchinterval(original_interval)
 
-        if errors:
-            self.fail("Saw errors: %s" % (errors,))
+        assert not errors, "Saw errors: %s" % (errors,)
 
     def test_no_live_nodes(self):
         """
@@ -913,11 +912,8 @@ def test_schedule_negative_max_attempts(self):
         delay = 2
         max_attempts = -100
 
-        try:
+        with pytest.raises(ValueError):
             ConstantReconnectionPolicy(delay=delay, max_attempts=max_attempts)
-            self.fail('max_attempts should throw ValueError when negative')
-        except ValueError:
-            pass
 
     def test_schedule_infinite_attempts(self):
         delay = 2
diff --git a/tests/unit/test_sortedset.py b/tests/unit/test_sortedset.py
