From 73426b4f328b97c5ea91e0ce784d30d8e81071a4 Mon Sep 17 00:00:00 2001 From: Dmitry Kropachev Date: Tue, 3 Nov 2026 14:09:44 -0400 Subject: [PATCH 1/2] pool: make Host endpoint read-only Direct endpoint assignment can bypass coordinated host replacement and leave retained pools or requests attached to an inconsistent route. Expose the constructor-normalized endpoint through a getter-only property, keeping it available alongside host_id before conviction policy creation. Migrate endpoint-move regressions to replacement objects and coordinated lifecycle APIs. Cover assignment and deletion guards, stable identity, mutable health state, and request completion through the original pool. Check completion before reading the result so a regression fails promptly. Document the compatibility break in the API guide and changelog. Fixes: #924 --- CHANGELOG.rst | 8 ++ cassandra/pool.py | 17 ++-- docs/api/cassandra/pool.rst | 7 ++ tests/unit/test_cluster.py | 7 +- tests/unit/test_control_connection.py | 16 +-- tests/unit/test_host_connection_pool.py | 118 ++++++++++++++++------ tests/unit/test_lifecycle_coordination.py | 6 +- tests/unit/test_metadata.py | 7 +- tests/unit/test_response_future.py | 57 ++++++++--- tests/unit/utils.py | 25 ++++- 10 files changed, 202 insertions(+), 66 deletions(-) diff --git a/CHANGELOG.rst b/CHANGELOG.rst index 3f4c3dbdf4..61c9e79dd5 100644 --- a/CHANGELOG.rst +++ b/CHANGELOG.rst @@ -11,6 +11,14 @@ Features ``Session.execute_concurrent_async()``, which returns a ``concurrent.futures.Future`` without blocking the caller. +Compatibility Changes +--------------------- +* ``Host.endpoint`` is now read-only, like ``Host.host_id``. Assigning or + deleting either property raises ``AttributeError``. Endpoint moves publish + a new ``Host`` with the same UUID and retire the previous object, whose + endpoint stays unchanged. Use cluster metadata to find the current host + instead of assigning an endpoint directly (#924). + Bug Fixes --------- * Avoid loading optional NumPy during ordinary driver imports. NumPy is now diff --git a/cassandra/pool.py b/cassandra/pool.py index 8fdf493cf9..bc588cfe3e 100644 --- a/cassandra/pool.py +++ b/cassandra/pool.py @@ -53,11 +53,6 @@ class Host(object): Represents a single Cassandra node. """ - endpoint = None - """ - The :class:`~.connection.EndPoint` to connect to the node. - """ - broadcast_address = None """ broadcast address configured for the node, *if available*: @@ -181,13 +176,23 @@ def __init__(self, endpoint, conviction_policy_factory, datacenter=None, rack=No if host_id.int == 0: raise ValueError("host_id may not be the nil UUID") - self.endpoint = endpoint if isinstance(endpoint, EndPoint) else DefaultEndPoint(endpoint) + self._endpoint = endpoint if isinstance(endpoint, EndPoint) else DefaultEndPoint(endpoint) self._host_id = host_id self._is_removed = False self.conviction_policy = conviction_policy_factory(self) self.set_location_info(datacenter, rack) self.lock = RLock() + @property + def endpoint(self): + """ + The read-only :class:`~.connection.EndPoint` to connect to the node. + + When the endpoint changes, the driver replaces this Host with a new + object carrying the same :attr:`host_id` and retires the old object. + """ + return self._endpoint + @property def address(self): """ diff --git a/docs/api/cassandra/pool.rst b/docs/api/cassandra/pool.rst index f6a59ce58a..fec20f7f36 100644 --- a/docs/api/cassandra/pool.rst +++ b/docs/api/cassandra/pool.rst @@ -5,6 +5,13 @@ Hosts and Connection Pools .. automodule:: cassandra.pool +``Host.host_id`` and ``Host.endpoint`` are read-only. A node endpoint move +publishes a distinct ``Host`` with the same UUID and retires the previous +object. Retained references keep their original endpoint; consult cluster +metadata for the current ``Host``. Equality and hashing use the UUID, so old +and replacement objects compare equal even though they are distinct objects. +Host health, conviction policy, and reconnection state remain mutable. + .. autoclass:: Host () :members: :exclude-members: set_location_info, get_and_set_reconnection_handler diff --git a/tests/unit/test_cluster.py b/tests/unit/test_cluster.py index 34cb4e413d..c78671a200 100644 --- a/tests/unit/test_cluster.py +++ b/tests/unit/test_cluster.py @@ -43,7 +43,7 @@ from cassandra.pool import Host, _HostReconnectionHandler from cassandra.policies import ExponentialReconnectionPolicy, HostDistance, RetryPolicy, RoundRobinPolicy, DowngradingConsistencyRetryPolicy, SimpleConvictionPolicy from cassandra.query import SimpleStatement, named_tuple_factory, tuple_factory -from tests.unit.utils import (mock_session_pools, new_session_with_pool_state, +from tests.unit.utils import (mock_session_pools, move_host, new_session_with_pool_state, run_isolated_subprocess) from tests import connection_class import pytest @@ -3076,7 +3076,7 @@ def test_pool_creation_discards_result_after_endpoint_change(self): with patch('cassandra.cluster.HostConnection', side_effect=factory): future = session.add_or_renew_pool(host, is_host_addition=True) assert factory.creation_started.wait(5) - host.endpoint = DefaultEndPoint('127.0.0.2') + move_host(host, DefaultEndPoint('127.0.0.2')) factory.release_creation.set() assert future.result(timeout=5) is False @@ -5200,7 +5200,7 @@ def test_set_keyspace_for_all_pools_reports_all_errors(self, *_): callback.assert_called_once() assert callback.call_args.args[0] == {'host1': [keyspace_error]} - def test_remove_pool_after_host_endpoint_changes(self): + def test_remove_pool_by_equal_host_at_another_endpoint(self): host = Host("127.0.0.1", SimpleConvictionPolicy, host_id=uuid.uuid4()) pool = Mock(host=host) shutdown_future = Future() @@ -5209,7 +5209,6 @@ def test_remove_pool_after_host_endpoint_changes(self): session.cluster.executor.submit.return_value = shutdown_future session.is_shutdown = False - host.endpoint = DefaultEndPoint("127.0.0.2") same_host_at_another_endpoint = Host( "127.0.0.3", SimpleConvictionPolicy, host_id=host.host_id) diff --git a/tests/unit/test_control_connection.py b/tests/unit/test_control_connection.py index 73f6810730..846d3f002a 100644 --- a/tests/unit/test_control_connection.py +++ b/tests/unit/test_control_connection.py @@ -1891,9 +1891,9 @@ def test_skipped_local_row_does_not_identify_endpoint_owner_as_control_host(self def move_owner_before_local_row(lookup_endpoint): if lookup_endpoint == endpoint: - old_endpoint = endpoint_owner.endpoint - endpoint_owner.endpoint = endpoint - metadata.update_host(endpoint_owner, old_endpoint) + self.cluster.replace_host_endpoint( + endpoint_owner, endpoint, + endpoint_owner.datacenter, endpoint_owner.rack) return get_host(lookup_endpoint) metadata.get_host = Mock(side_effect=move_owner_before_local_row) @@ -2027,7 +2027,10 @@ def test_changed_endpoint_skip_keeps_current_host_tokens(self): def change_after_batch(moves): replacements = replace_hosts_endpoints(moves) - replacements[0].endpoint = DefaultEndPoint('192.168.1.10') + current = replacements[0] + self.cluster.replace_host_endpoint( + current, DefaultEndPoint('192.168.1.10'), + current.datacenter, current.rack) return replacements self.cluster.replace_hosts_endpoints = Mock(side_effect=change_after_batch) @@ -2072,10 +2075,9 @@ def discover_after_lookup(host_id): def test_new_host_does_not_displace_live_unix_endpoint_owner(self): self.control_connection.refresh_node_list_and_token_map() owner = self.cluster.metadata.get_host_by_host_id(HOST_ID_2) - old_endpoint = owner.endpoint unix_endpoint = UnixSocketEndPoint('/tmp/maintenance.sock') - owner.endpoint = unix_endpoint - self.cluster.metadata.update_host(owner, old_endpoint) + owner = self.cluster.replace_host_endpoint( + owner, unix_endpoint, owner.datacenter, owner.rack) self.connection.peer_results[1].append([ '192.168.1.9', '10.0.0.9', 'a', 'dc1', 'rack1', diff --git a/tests/unit/test_host_connection_pool.py b/tests/unit/test_host_connection_pool.py index f83772705e..cf248f7e09 100644 --- a/tests/unit/test_host_connection_pool.py +++ b/tests/unit/test_host_connection_pool.py @@ -24,14 +24,16 @@ from cassandra.cluster import Cluster, Session, ShardAwareOptions from cassandra._lifecycle import ConnectionFailureReservation, PoolOwnershipReservation -from cassandra.connection import Connection, ConnectionBusy, ConnectionException, ConnectionSendError, DefaultEndPoint +from cassandra.connection import (ClientRoutesEndPoint, Connection, ConnectionBusy, + ConnectionException, ConnectionSendError, + DefaultEndPoint, SniEndPoint, UnixSocketEndPoint) from cassandra.pool import HostConnection from cassandra.pool import Host, NoConnectionsAvailable from cassandra.policies import HostDistance, SimpleConvictionPolicy import pytest from tests.unit.util import HashableMock -from tests.unit.utils import new_session_with_pool_state +from tests.unit.utils import move_host, new_session_with_pool_state LOGGER = logging.getLogger(__name__) @@ -96,7 +98,9 @@ def test_connection_failure_respects_socket_ownership( elif ownership == 'retired_host': host._is_removed = True elif ownership == 'moved_endpoint': - host.endpoint = DefaultEndPoint('127.0.0.2') + moved = move_host(host, DefaultEndPoint('127.0.0.2')) + winner = HostConnection(moved, HostDistance.LOCAL, session) + session._pools = {moved: winner} elif ownership == 'shutdown_session': session.is_shutdown = True @@ -123,7 +127,7 @@ def test_connection_failure_respects_socket_ownership( session.cluster.on_down.assert_not_called() session.submit.assert_not_called() assert not current.is_shutdown - if ownership in ('draining', 'retired_pool', 'unpublished_pool'): + if ownership in ('draining', 'retired_pool', 'unpublished_pool', 'moved_endpoint'): assert current._connections.get(0) is replacement replacement.close.assert_not_called() @@ -149,7 +153,9 @@ def replace(): elif change == 'pool': session._pools[host] = winner elif change == 'endpoint': - host.endpoint = DefaultEndPoint('127.0.0.2') + moved = move_host(host, DefaultEndPoint('127.0.0.2')) + winner = HostConnection(moved, HostDistance.LOCAL, session) + session._pools = {moved: winner} else: host._is_removed = True except Exception as error: @@ -175,7 +181,7 @@ def convict(error): session.cluster.on_down.assert_not_called() session.submit.assert_not_called() assert not session._pools[host].is_shutdown - if change in ('socket', 'pool'): + if change in ('socket', 'pool', 'endpoint'): assert session._pools[host]._connections[0] is replacement replacement.close.assert_not_called() @@ -200,7 +206,7 @@ def distance(current): session._pools[host] = Mock( host=host, endpoint=host.endpoint, is_shutdown=False) elif change == 'endpoint': - host.endpoint = DefaultEndPoint('127.0.0.2') + move_host(host, DefaultEndPoint('127.0.0.2')) else: host._health_generation += 1 return HostDistance.IGNORED @@ -208,7 +214,7 @@ def distance(current): cluster.profile_manager.distance.side_effect = distance try: pool.return_connection(original) - assert host.is_up + assert host.is_up is (change != 'endpoint') assert pool.is_shutdown if change == 'pool': assert not session._pools[host].is_shutdown @@ -236,7 +242,7 @@ def supersede(): if change == 'pool': session._pools[host] = successor elif change == 'endpoint': - host.endpoint = DefaultEndPoint('127.0.0.2') + move_host(host, DefaultEndPoint('127.0.0.2')) elif change == 'host': host._is_removed = True elif change == 'session': @@ -758,24 +764,43 @@ def test_host_equality(self): assert a != c, 'Two Host instances with different host IDs should not be equal.' assert a != a.address, 'A Host should not compare equal to its address.' - def test_host_id_is_read_only(self): + def test_host_identity_and_endpoint_are_read_only(self): host_id = uuid.uuid4() host = Host('127.0.0.1', SimpleConvictionPolicy, host_id=host_id) - with pytest.raises(AttributeError): - host.host_id = uuid.uuid4() - - assert host.host_id == host_id - - def test_host_hash_is_stable_when_endpoint_changes(self): + endpoint = host.endpoint + original_hash = hash(host) + hosts_by_id = {host: 'pool'} + hosts = {host} + for name, value in (('host_id', uuid.uuid4()), + ('endpoint', DefaultEndPoint('127.0.0.2'))): + for operation in (setattr, delattr): + with self.subTest(attribute=name, operation=operation.__name__): + with pytest.raises(AttributeError): + if operation is setattr: + operation(host, name, value) + else: + operation(host, name) + assert hash(host) == original_hash + assert hosts_by_id[host] == 'pool' + assert host in hosts + assert host.host_id == host_id + assert host.endpoint is endpoint + + def test_host_hash_is_shared_by_endpoint_replacements(self): host = Host('127.0.0.1', SimpleConvictionPolicy, host_id=uuid.uuid4()) hosts_by_id = {host: "pool"} hosts = {host} - host.endpoint = DefaultEndPoint('127.0.0.2') + replacement = Host('127.0.0.2', SimpleConvictionPolicy, + host_id=host.host_id) assert hosts_by_id[host] == "pool" + assert hosts_by_id[replacement] == "pool" + assert hash(replacement) == hash(host) assert host in hosts + assert replacement in hosts + assert host.endpoint == DefaultEndPoint('127.0.0.1') def test_host_ordering_uses_host_id(self): first = Host('127.0.0.2', SimpleConvictionPolicy, host_id=uuid.UUID(int=1)) @@ -783,19 +808,47 @@ def test_host_ordering_uses_host_id(self): assert first < second - def test_host_id_is_set_before_conviction_policy_is_created(self): + def test_identity_and_endpoint_are_set_before_conviction_policy_is_created(self): host_id = uuid.UUID(int=1) hosts = set() - - def conviction_policy_factory(host): - assert host.host_id == host_id - hosts.add(host) - return SimpleConvictionPolicy(host) - - host = Host( - '127.0.0.1', conviction_policy_factory, host_id=host_id) - - assert host in hosts + endpoints = ('127.0.0.1', DefaultEndPoint('127.0.0.1', 9142), + UnixSocketEndPoint('/tmp/maintenance.sock'), + SniEndPoint('proxy.example', 'node', 9042), + ClientRoutesEndPoint(host_id, Mock(), '127.0.0.1', 9042)) + for endpoint in endpoints: + with self.subTest(endpoint=endpoint): + normalized = (DefaultEndPoint(endpoint) + if isinstance(endpoint, str) else endpoint) + + def conviction_policy_factory(host): + assert host.host_id == host_id + assert host.endpoint == normalized + if not isinstance(endpoint, str): + assert host.endpoint is endpoint + hosts.add(host) + return SimpleConvictionPolicy(host) + + host = Host(endpoint, conviction_policy_factory, host_id=host_id) + assert host in hosts + + def test_health_policy_and_reconnection_state_remain_mutable(self): + host = Host('127.0.0.1', SimpleConvictionPolicy, host_id=uuid.uuid4()) + policy = Mock() + host.conviction_policy = policy + host.set_up() + assert host.is_up + policy.reset.assert_called_once_with() + error = ConnectionException('socket failed') + result = host.signal_connection_failure(error) + assert result is policy.add_failure.return_value + policy.add_failure.assert_called_once_with(error) + host.set_down() + assert not host.is_up + handler = Mock() + assert host.get_and_set_reconnection_handler(handler) is None + assert host.is_currently_reconnecting() + assert host.get_and_set_reconnection_handler(None) is handler + assert not host.is_currently_reconnecting() class HostConnectionTests(_PoolTests): @@ -818,7 +871,7 @@ def test_replacement_skips_pool_after_endpoint_changes(self): pool = self.make_pool(host, session) original_endpoint = host.endpoint - host.endpoint = DefaultEndPoint('127.0.0.2') + move_host(host, DefaultEndPoint('127.0.0.2')) pool._replace(first) assert pool.endpoint == original_endpoint @@ -1155,7 +1208,7 @@ def sync_then_retire(pool, conn, **kwargs): assert pool._connections == {0: candidate} with session._lock, host.lock: if change == 'endpoint': - host.endpoint = DefaultEndPoint('127.0.0.2') + move_host(host, DefaultEndPoint('127.0.0.2')) elif change == 'host': host._is_removed = True elif change == 'session': @@ -1297,7 +1350,10 @@ def complete(): assert entered.wait(5) with session._lock, host.lock: if change == 'endpoint': - host.endpoint = DefaultEndPoint('127.0.0.2') + moved = move_host(host, DefaultEndPoint('127.0.0.2')) + winner = HostConnection(moved, HostDistance.IGNORED, session) + request.addfinalizer(winner.shutdown) + session._pools = {moved: winner} elif change == 'pool': winner = HostConnection(host, HostDistance.IGNORED, session) request.addfinalizer(winner.shutdown) diff --git a/tests/unit/test_lifecycle_coordination.py b/tests/unit/test_lifecycle_coordination.py index f2da4b1683..9ef4a63ca0 100644 --- a/tests/unit/test_lifecycle_coordination.py +++ b/tests/unit/test_lifecycle_coordination.py @@ -1232,7 +1232,9 @@ def distance(_host): if supersede == 'pool': session._pools[host] = successor elif supersede == 'endpoint': - host.endpoint = DefaultEndPoint('127.0.0.2') + with patch.object(cluster, '_notify_host_removed'), \ + patch.object(cluster, 'on_add'): + cluster.replace_host_endpoint(host, DefaultEndPoint('127.0.0.2')) elif supersede == 'host': host._is_removed = True elif supersede == 'session': @@ -1266,7 +1268,7 @@ def admit(): assert not worker.is_alive() assert not errors assert admitted == [supersede is None] - assert host.is_up is (supersede is not None) + assert host.is_up is (supersede not in (None, 'endpoint')) assert cluster.on_down_potentially_blocking.call_count == int(supersede is None) assert not successor.is_shutdown finally: diff --git a/tests/unit/test_metadata.py b/tests/unit/test_metadata.py index 542afdf543..1342d41b35 100644 --- a/tests/unit/test_metadata.py +++ b/tests/unit/test_metadata.py @@ -1326,10 +1326,11 @@ def test_endpoint_reindex_keeps_new_owner(self): metadata.add_or_return_host(moved) metadata.add_or_return_host(replacement) - moved.endpoint = new_endpoint - metadata.update_host(moved, old_endpoint) + current = Host(new_endpoint, SimpleConvictionPolicy, host_id=moved.host_id) + metadata.replace_host_endpoint(moved, current) assert metadata.get_host(old_endpoint) is replacement - assert metadata.get_host(new_endpoint) is moved + assert metadata.get_host(new_endpoint) is current + assert moved.endpoint == old_endpoint def test_iterate_all_hosts_and_modify(self): """ diff --git a/tests/unit/test_response_future.py b/tests/unit/test_response_future.py index ba1c5a2e71..80a9750360 100644 --- a/tests/unit/test_response_future.py +++ b/tests/unit/test_response_future.py @@ -39,6 +39,7 @@ from cassandra.policies import RetryPolicy, ExponentialBackoffRetryPolicy, SimpleConvictionPolicy from cassandra.pool import Host, NoConnectionsAvailable from cassandra.query import SimpleStatement, PreparedStatement, BoundStatement +from tests.unit.utils import move_host from tests.util import assertEqual, assertIsInstance import pytest @@ -91,8 +92,9 @@ def test_timeout_returns_stream_to_borrowed_pool_after_replacement(self): rf = self.make_response_future(session) assert rf._query(host) == 1 - new_pool, _ = self.make_owned_pool(host) - session._pools[host] = new_pool + replacement = move_host(host, DefaultEndPoint('127.0.0.2')) + new_pool, _ = self.make_owned_pool(replacement) + session._pools = {replacement: new_pool} rf._on_timeout() old_pool.return_connection.assert_called_once_with( @@ -110,10 +112,39 @@ def test_query_rejects_changed_endpoint_and_equal_replacement_host(self): replacement = Host('127.0.0.1', SimpleConvictionPolicy, host_id=host_id) assert rf._query(replacement) is None - host.endpoint = DefaultEndPoint('127.0.0.2') + moved = move_host(host, DefaultEndPoint('127.0.0.2')) + new_pool, new_connection = self.make_owned_pool(moved) + session._pools = {moved: new_pool} assert rf._query(host) is None pool.borrow_connection.assert_not_called() + new_pool.borrow_connection.assert_not_called() connection.send_msg.assert_not_called() + new_connection.send_msg.assert_not_called() + + def test_admitted_request_completes_through_original_pool_after_endpoint_move(self): + host = Host('127.0.0.1', SimpleConvictionPolicy, host_id=uuid.uuid4()) + session = self.make_owned_pool_session(host) + old_pool, connection = self.make_owned_pool(host) + session._pools[host] = old_pool + rf = self.make_response_future(session) + assert rf._query(host) == 1 + + replacement = move_host(host, DefaultEndPoint('127.0.0.2')) + new_pool, _ = self.make_owned_pool(replacement) + session._pools = {replacement: new_pool} + response = ResultMessage(RESULT_KIND_ROWS) + response.column_names = ['value'] + response.column_types = [] + response.parsed_rows = [(42,)] + connection._requests.pop(1)(response) + + assert rf._event.is_set() + assert rf.result().one() == (['value'], [(42,)]) + assert rf._request_pool is old_pool + assert connection.in_flight == 0 + old_pool.return_connection.assert_called_once_with(connection) + new_pool.return_connection.assert_not_called() + new_pool.borrow_connection.assert_not_called() def test_explicit_host_from_previous_cluster_uses_current_host(self): host_id = uuid.uuid4() @@ -171,7 +202,7 @@ def test_query_restores_borrowed_stream_when_endpoint_changes_during_borrow(self def change_endpoint(**kwargs): result = borrow(**kwargs) - host.endpoint = DefaultEndPoint('127.0.0.2') + move_host(host, DefaultEndPoint('127.0.0.2')) return result pool.borrow_connection.side_effect = change_endpoint @@ -198,7 +229,7 @@ def test_stale_borrow_preserves_previous_attempt_state(self): def change_endpoint(**kwargs): result = borrow(**kwargs) - second_host.endpoint = DefaultEndPoint('127.0.0.3') + move_host(second_host, DefaultEndPoint('127.0.0.3')) return result second_pool.borrow_connection.side_effect = change_endpoint @@ -217,14 +248,15 @@ def test_retry_uses_replacement_pool(self): rf = self.make_response_future(session) assert rf._query(host) == 1 - new_pool, new_connection = self.make_owned_pool(host, request_id=2) - session._pools[host] = new_pool - assert rf._query(host) == 2 + replacement = move_host(host, DefaultEndPoint('127.0.0.2')) + new_pool, new_connection = self.make_owned_pool(replacement, request_id=2) + session._pools = {replacement: new_pool} + assert rf._query(replacement) == 2 new_connection.send_msg.assert_called_once() old_connection.send_msg.assert_called_once() assert rf._request_pool is new_pool - def test_endpoint_change_waits_until_send_begins(self): + def test_endpoint_move_waits_to_retire_host_until_send_begins(self): host = Host('127.0.0.1', SimpleConvictionPolicy, host_id=uuid.uuid4()) session = self.make_owned_pool_session(host) pool, connection = self.make_owned_pool(host) @@ -237,8 +269,7 @@ def test_endpoint_change_waits_until_send_begins(self): def send_msg(message, stream_id, cb, **kwargs): def change_endpoint(): change_started.set() - with host.lock: - host.endpoint = DefaultEndPoint('127.0.0.2') + move_host(host, DefaultEndPoint('127.0.0.2')) change_done.set() thread = Thread(target=change_endpoint) @@ -247,6 +278,7 @@ def change_endpoint(): assert change_started.wait(1) assert not change_done.wait(0.01) assert host.endpoint == old_endpoint + assert not host._is_removed connection._requests[stream_id] = cb return 128 @@ -255,7 +287,8 @@ def change_endpoint(): assert rf._query(host) == 1 assert change_done.wait(1) threads[0].join(1) - assert host.endpoint != old_endpoint + assert host._is_removed + assert host.endpoint == old_endpoint def test_sends_to_different_hosts_can_progress_together(self): first_host = Host('127.0.0.1', SimpleConvictionPolicy, diff --git a/tests/unit/utils.py b/tests/unit/utils.py index bf9a67c530..c1455f8c17 100644 --- a/tests/unit/utils.py +++ b/tests/unit/utils.py @@ -21,7 +21,7 @@ from unittest.mock import Mock, patch from concurrent.futures import Future -from cassandra.cluster import Session +from cassandra.cluster import Cluster, Session from cassandra.driver_config import DriverConfigReporter @@ -54,6 +54,29 @@ def new_session_with_pool_state(pools=None): return session +def move_host(host, endpoint): + """Publish and retire a route while the caller controls pool cleanup. + + Socket and request race tests use mock sessions so they can finish stale + work separately from lifecycle notifications. Use the production move API + with those notifications stubbed to preserve that separation. + """ + old_endpoint = host.endpoint + with Cluster(contact_points=[]) as cluster: + cluster.metadata.add_or_return_host(host) + cluster._notify_host_removed = Mock() + cluster.on_add = Mock() + replacement = cluster.replace_host_endpoint(host, endpoint) + assert cluster.metadata.get_host_by_host_id(host.host_id) is replacement + assert cluster.metadata.get_host(endpoint) is replacement + assert cluster.metadata.get_host(old_endpoint) is None + assert replacement is not host + assert replacement.host_id == host.host_id + assert host.endpoint == old_endpoint + assert host._is_removed + return replacement + + def mock_session_pools(f): """ Helper decorator that allows tests to initialize :class:.`Session` objects From a8f715a934c3d0c4c272473324f86aef80ddf478 Mon Sep 17 00:00:00 2001 From: Dmitry Kropachev Date: Tue, 3 Nov 2026 14:10:03 -0400 Subject: [PATCH 2/2] tests: verify Host identity during a live endpoint move Keep the original Host reference across a node address change and verify that metadata publishes a distinct Host with the same UUID. Check that the retired reference retains its endpoint, both metadata indexes agree, and queries through the replacement address reach the expected node. --- tests/integration/standard/test_ip_change.py | 19 ++++++++++++++++--- 1 file changed, 16 insertions(+), 3 deletions(-) diff --git a/tests/integration/standard/test_ip_change.py b/tests/integration/standard/test_ip_change.py index 53debfa1f5..400ae6c160 100644 --- a/tests/integration/standard/test_ip_change.py +++ b/tests/integration/standard/test_ip_change.py @@ -3,6 +3,7 @@ import unittest from cassandra.cluster import ExecutionProfile +from cassandra.connection import DefaultEndPoint from cassandra.policies import WhiteListRoundRobinPolicy from tests.integration import use_cluster, get_node, get_cluster, local, TestCluster @@ -39,6 +40,10 @@ def teardown_class(cls): def test_change_address_during_live_session(self): node3 = get_node(3) + old_endpoint = DefaultEndPoint(*node3.network_interfaces['binary']) + old_host = self.cluster.metadata.get_host(old_endpoint) + assert old_host is not None + host_id = old_host.host_id LOGGER.debug("Stop node3") node3.stop() @@ -52,9 +57,16 @@ def test_change_address_during_live_session(self): node3.start(wait_for_binary_proto=True) def new_address_found(): - addresses = [str(host.endpoint.address) for host in self.cluster.metadata.all_hosts()] - LOGGER.debug(addresses) - assert new_ip in addresses + metadata = self.cluster.metadata + current = metadata.get_host_by_host_id(host_id) + assert current is not None + assert current is not old_host + assert current.host_id == host_id + assert current.endpoint == DefaultEndPoint(new_ip, old_endpoint.port) + assert old_host.endpoint == old_endpoint + assert old_host._is_removed + assert metadata.get_host(old_endpoint) is None + assert metadata.get_host(current.endpoint) is current wait_until_not_raised(new_address_found, 0.5, 100) @@ -66,5 +78,6 @@ def new_node_connectable(): local_info = self.session.execute("SELECT * FROM system.local WHERE key='local'", execution_profile="new_node").one() LOGGER.debug(local_info._asdict()) assert local_info.broadcast_address == new_ip + assert local_info.host_id == host_id wait_until_not_raised(new_node_connectable, 0.5, 100)