Fix IntStat comparison operators and metrics cleanup on shutdown

- Add comparison operators (__eq__, __ne__, __lt__, __le__, __gt__, __ge__)
  to IntStat class for direct comparison with integers
- Add __hash__ method to IntStat for use in sets/dicts
- Add Metrics.shutdown() method to remove stats from global registry
- Call metrics.shutdown() from Cluster.shutdown() to prevent stale
  weakref errors when accessing getStats() after cluster shutdown
- Fix test using 'is' instead of '==' for IntStat comparison
- Add unit tests for new comparison operators and shutdown functionality

Author: dkropachev

cassandra/cluster.py

@@ -1795,6 +1795,9 @@ def shutdown(self):
 
         self.executor.shutdown()
 
+        if self.metrics_enabled and self.metrics:
+            self.metrics.shutdown()
+
         _discard_cluster_shutdown(self)
 
     def __enter__(self):

cassandra/metrics.py

@@ -61,6 +61,37 @@ def __iadd__(self, other):
     def __int__(self):
         return self._value
 
+    def __eq__(self, other):
+        if isinstance(other, IntStat):
+            return self._value == other._value
+        return self._value == other
+
+    def __ne__(self, other):
+        return not self.__eq__(other)
+
+    def __lt__(self, other):
+        if isinstance(other, IntStat):
+            return self._value < other._value
+        return self._value < other
+
+    def __le__(self, other):
+        if isinstance(other, IntStat):
+            return self._value <= other._value
+        return self._value <= other
+
+    def __gt__(self, other):
+        if isinstance(other, IntStat):
+            return self._value > other._value
+        return self._value > other
+
+    def __ge__(self, other):
+        if isinstance(other, IntStat):
+            return self._value >= other._value
+        return self._value >= other
+
+    def __hash__(self):
+        return hash(self._value)
+
     def __repr__(self):
         return f"IntStat({self.name}={self._value})"
 
@@ -477,3 +508,15 @@ def set_stats_name(self, stats_name):
             del _stats_registry[self.stats_name]
             self.stats_name = stats_name
             _stats_registry[self.stats_name] = stats
+
+    def shutdown(self):
+        """
+        Remove this metrics instance from the global registry.
+        Called when the cluster is shutdown to prevent stale references.
+        """
+        with _registry_lock:
+            if self.stats_name in _stats_registry:
+                del _stats_registry[self.stats_name]
+            # Also clean up the legacy 'cassandra' entry if it points to our stats
+            if _stats_registry.get('cassandra') is self.stats:
+                del _stats_registry['cassandra']

tests/integration/standard/test_metrics.py

@@ -355,10 +355,10 @@ def setUpClass(cls):
     def wait_for_count(self, ra, expected_count, error=False):
         for _ in range(10):
             if not error:
-                if ra.successful is expected_count:
+                if ra.successful == expected_count:
                     return True
             else:
-                if ra.errors is expected_count:
+                if ra.errors == expected_count:
                     return True
             time.sleep(.01)
         return False

tests/unit/test_metrics.py

@@ -63,6 +63,35 @@ def test_repr(self):
         stat += 42
         self.assertEqual(repr(stat), "IntStat(my_counter=42)")
 
+    def test_equality(self):
+        stat = IntStat('test')
+        stat += 5
+        self.assertEqual(stat, 5)
+        self.assertEqual(5, stat)
+        self.assertNotEqual(stat, 3)
+        self.assertNotEqual(stat, 10)
+
+    def test_comparison_operators(self):
+        stat = IntStat('test')
+        stat += 5
+        self.assertTrue(stat > 0)
+        self.assertTrue(stat >= 5)
+        self.assertTrue(stat < 10)
+        self.assertTrue(stat <= 5)
+        self.assertFalse(stat > 5)
+        self.assertFalse(stat < 5)
+
+    def test_comparison_with_intstat(self):
+        stat1 = IntStat('test1')
+        stat2 = IntStat('test2')
+        stat1 += 5
+        stat2 += 5
+        self.assertEqual(stat1, stat2)
+        stat2 += 1
+        self.assertNotEqual(stat1, stat2)
+        self.assertTrue(stat1 < stat2)
+        self.assertTrue(stat2 > stat1)
+
 
 class StatTest(unittest.TestCase):
     """Tests for Stat (gauge) class."""
@@ -300,5 +329,58 @@ def __init__(self):
             self.assertNotIn('/test_path', _stats_registry)
 
 
+class MetricsShutdownTest(unittest.TestCase):
+    """Tests for Metrics shutdown functionality."""
+
+    def setUp(self):
+        # Clean up registry before each test
+        with _registry_lock:
+            keys_to_remove = [k for k in _stats_registry.keys()
+                              if k.startswith('cassandra')]
+            for k in keys_to_remove:
+                del _stats_registry[k]
+
+    def test_shutdown_removes_from_registry(self):
+        import weakref
+        from cassandra.metrics import Metrics
+
+        class MockCluster:
+            def __init__(self):
+                self.metadata = type('obj', (object,), {'all_hosts': lambda: []})()
+                self.sessions = []
+
+        cluster = MockCluster()
+        proxy = weakref.proxy(cluster)
+
+        metrics = Metrics(proxy)
+        stats_name = metrics.stats_name
+
+        with _registry_lock:
+            self.assertIn(stats_name, _stats_registry)
+
+        metrics.shutdown()
+
+        with _registry_lock:
+            self.assertNotIn(stats_name, _stats_registry)
+
+    def test_shutdown_is_idempotent(self):
+        import weakref
+        from cassandra.metrics import Metrics
+
+        class MockCluster:
+            def __init__(self):
+                self.metadata = type('obj', (object,), {'all_hosts': lambda: []})()
+                self.sessions = []
+
+        cluster = MockCluster()
+        proxy = weakref.proxy(cluster)
+
+        metrics = Metrics(proxy)
+
+        # Should not raise even if called multiple times
+        metrics.shutdown()
+        metrics.shutdown()
+
+
 if __name__ == '__main__':
     unittest.main()