diff --git a/awx/main/tasks/system.py b/awx/main/tasks/system.py index 7f83f03a3e..08419a80fe 100644 --- a/awx/main/tasks/system.py +++ b/awx/main/tasks/system.py @@ -588,23 +588,21 @@ def inspect_established_receptor_connections(mesh_status): InstanceLink.objects.bulk_update(update_links, ['link_state']) -def inspect_execution_and_hop_nodes(instance_list, receptor_ctl): +def inspect_execution_and_hop_nodes(instance_list, mesh_status): with advisory_lock('inspect_execution_and_hop_nodes_lock', wait=False) as acquired: if not acquired: logger.debug("Not running inspect_execution_and_hop_nodes, another instance holds lock") return + if mesh_status is None: + logger.debug("Not running inspect_execution_and_hop_nodes, mesh status unavailable") + return start = time.monotonic() node_lookup = {inst.hostname: inst for inst in instance_list} - try: - mesh_status = receptor_ctl.simple_command('status') - except ValueError as exc: - logger.error(f'Error running receptorctl status command, error: {str(exc)}') - return inspect_established_receptor_connections(mesh_status) nowtime = now() - workers = mesh_status['Advertisements'] + workers = mesh_status.get('Advertisements') or [] updated_count = 0 for ad in workers: @@ -727,6 +725,27 @@ def _get_active_task_ids_from_dispatcherd(binder): return None +def _mesh_all_ready_nodes_visible(mesh_status): + """Return False if the receptor mesh is still re-establishing after a controller restart. + + Uses KnownConnectionCosts as the stability signal: receptor's routing protocol + populates this table as connections establish via gossip. An empty table means no + routing has propagated yet (Window A — typically the first ~10s after restart). + + No DB state is consulted. KnownConnectionCosts is maintained entirely by the receptor + Go process, making it a reliable mesh-state signal free of stale DB records. + + Fails open (returns True) when mesh_status is None so existing error paths are + not bypassed. + """ + if mesh_status is None: + return True # fail open: status unavailable, let normal peer-judgment proceed + if not (mesh_status.get('KnownConnectionCosts') or {}): + logger.info('Mesh stability gate: routing table empty, deferring peer-judgment (receptor re-establishing)') + return False + return True + + def _heartbeat_instance_management(): """Common logic for heartbeat instance management.""" logger.debug("Cluster node heartbeat task.") @@ -749,7 +768,13 @@ def _heartbeat_instance_management(): this_inst.mark_offline(errors='Receptor not available') return None, None, None - inspect_execution_and_hop_nodes(instance_list, ctl) + try: + mesh_status = ctl.simple_command('status') + except (OSError, RuntimeError, ValueError) as exc: + logger.warning(f'Receptor status unavailable: {exc}') + mesh_status = None + + inspect_execution_and_hop_nodes(instance_list, mesh_status) for inst in list(instance_list): if inst == this_inst: @@ -779,6 +804,9 @@ def _heartbeat_instance_management(): logger.error("Cluster Host Not Found: {}".format(settings.CLUSTER_HOST_ID)) return None, None, None + if lost_instances and not _mesh_all_ready_nodes_visible(mesh_status): + return this_inst, instance_list, [] + return this_inst, instance_list, lost_instances @@ -801,39 +829,50 @@ def _heartbeat_check_versions(this_inst, instance_list): raise RuntimeError("Shutting down.") +def _reap_and_mark_lost_instance(other_inst): + """Reap a lost instance's running jobs and mark it offline (or deprovision it).""" + try: + # Only reap jobs never dispatched to receptor; dispatched jobs are left for adoption (AAP-89602) + explanation = "Job reaped due to instance shutdown" + reaper.reap(other_inst, job_explanation=explanation) + # Any jobs that were waiting to be processed by this node will be handed back to task manager + UnifiedJob.objects.filter(status='waiting', controller_node=other_inst.hostname).update(status='pending', controller_node='', execution_node='') + except Exception: + logger.exception('failed to re-process jobs for lost instance {}'.format(other_inst.hostname)) + try: + if settings.AWX_AUTO_DEPROVISION_INSTANCES and other_inst.node_type == "control": + deprovision_hostname = other_inst.hostname + other_inst.delete() # FIXME: what about associated inbound links? + logger.info("Host {} Automatically Deprovisioned.".format(deprovision_hostname)) + elif other_inst.node_state == Instance.States.READY: + other_inst.mark_offline(errors=_('Another cluster node has determined this instance to be unresponsive')) + logger.error("Host {} last checked in at {}, marked as lost.".format(other_inst.hostname, other_inst.last_seen)) + + except DatabaseError as e: + cause = e.__cause__ + if cause and hasattr(cause, 'sqlstate'): + sqlstate = cause.sqlstate + sqlstate_str = psycopg.errors.lookup(sqlstate) + logger.debug('SQL Error state: {} - {}'.format(sqlstate, sqlstate_str)) + + if sqlstate == psycopg.errors.NoData: + logger.debug('Another instance has marked {} as lost'.format(other_inst.hostname)) + else: + logger.exception("Error marking {} as lost.".format(other_inst.hostname)) + else: + logger.exception('No SQL state available. Error marking {} as lost'.format(other_inst.hostname)) + + def _heartbeat_handle_lost_instances(lost_instances, this_inst): """Handle lost instances by reaping their running jobs and marking them offline.""" for other_inst in lost_instances: - try: - # Any jobs marked as running will be marked as error - explanation = "Job reaped due to instance shutdown" - reaper.reap(other_inst, job_explanation=explanation) - # Any jobs that were waiting to be processed by this node will be handed back to task manager - UnifiedJob.objects.filter(status='waiting', controller_node=other_inst.hostname).update(status='pending', controller_node='', execution_node='') - except Exception: - logger.exception('failed to re-process jobs for lost instance {}'.format(other_inst.hostname)) - try: - if settings.AWX_AUTO_DEPROVISION_INSTANCES and other_inst.node_type == "control": - deprovision_hostname = other_inst.hostname - other_inst.delete() # FIXME: what about associated inbound links? - logger.info("Host {} Automatically Deprovisioned.".format(deprovision_hostname)) - elif other_inst.node_state == Instance.States.READY: - other_inst.mark_offline(errors=_('Another cluster node has determined this instance to be unresponsive')) - logger.error("Host {} last checked in at {}, marked as lost.".format(other_inst.hostname, other_inst.last_seen)) - - except DatabaseError as e: - cause = e.__cause__ - if cause and hasattr(cause, 'sqlstate'): - sqlstate = cause.sqlstate - sqlstate_str = psycopg.errors.lookup(sqlstate) - logger.debug('SQL Error state: {} - {}'.format(sqlstate, sqlstate_str)) - - if sqlstate == psycopg.errors.NoData: - logger.debug('Another instance has marked {} as lost'.format(other_inst.hostname)) - else: - logger.exception("Error marking {} as lost.".format(other_inst.hostname)) - else: - logger.exception('No SQL state available. Error marking {} as lost'.format(other_inst.hostname)) + # Serialize offline handling against the task manager so we never reap jobs + # mid-schedule; if the lock is held, retry this instance on the next heartbeat. + with advisory_lock('task_manager_lock', wait=False) as acquired: + if not acquired: + logger.info(f'task_manager_lock held, deferring offline handling for {other_inst.hostname} to next heartbeat cycle') + continue + _reap_and_mark_lost_instance(other_inst) @task(queue=get_task_queuename, timeout=1800, on_duplicate='queue_one') diff --git a/awx/main/tests/functional/tasks/test_tasks_system.py b/awx/main/tests/functional/tasks/test_tasks_system.py index 94a1ebedc7..edd3178305 100644 --- a/awx/main/tests/functional/tasks/test_tasks_system.py +++ b/awx/main/tests/functional/tasks/test_tasks_system.py @@ -4,6 +4,7 @@ import logging import os import tempfile import shutil +from contextlib import contextmanager from unittest import mock import pytest @@ -14,9 +15,15 @@ from awx.main.tasks.system import ( inspect_established_receptor_connections, clear_setting_cache, _batched_delete_inventory, + _mesh_all_ready_nodes_visible, + _heartbeat_handle_lost_instances, + _reap_and_mark_lost_instance, + inspect_execution_and_hop_nodes, + _heartbeat_instance_management, ) from awx.main.management.commands.dispatcherd import Command from django.db import DatabaseError +from django.utils.timezone import now, timedelta from awx.main.models import Instance, Inventory, Job, Organization, ReceptorAddress, InstanceLink from awx.main.models.inventory import Group, Host @@ -250,3 +257,313 @@ def test_configure_dispatcher_logging_updates_level(settings): assert logging.getLogger('dispatcherd').level == logging.WARNING settings.LOGGING = original_logging_settings + + +# ── Helpers ─────────────────────────────────────────────────────────────────── + +_UNSET = object() + + +def _mesh_status(known_costs=_UNSET, advertisements=_UNSET): + return { + 'KnownConnectionCosts': {} if known_costs is _UNSET else known_costs, + 'Advertisements': [] if advertisements is _UNSET else [{'NodeID': n} for n in (advertisements or [])], + } + + +def _make_lock(acquired): + @contextmanager + def _cm(name, wait=True, **kwargs): + yield acquired + + return _cm + + +def _make_lock_sequence(sequence): + it = iter(sequence) + + @contextmanager + def _cm(name, wait=True, **kwargs): + yield next(it) + + return _cm + + +# ── Tests: _mesh_all_ready_nodes_visible ────────────────────────────────────── + + +@pytest.mark.django_db +class TestMeshAllReadyNodesVisible: + """Gate uses KnownConnectionCosts (receptor routing table) as the stability signal. + + No DB state is consulted — KnownConnectionCosts is maintained entirely by + receptor's routing protocol. Empty table = routing not yet established (Window A). + """ + + def test_defers_when_routing_table_empty(self): + status = _mesh_status(known_costs={}) + assert _mesh_all_ready_nodes_visible(status) is False + + def test_defers_when_routing_table_null(self): + """Receptor is a Go service; nil map marshals to JSON null → Python None.""" + status = _mesh_status(known_costs=None) + assert _mesh_all_ready_nodes_visible(status) is False + + def test_passes_when_routing_established(self): + status = _mesh_status(known_costs={'ctrl-0': {'ee-0': 1}}) + assert _mesh_all_ready_nodes_visible(status) is True + + def test_passes_when_routing_established_dead_ee_not_in_routing(self): + """Gate checks only whether routing exists, not which nodes appear.""" + status = _mesh_status(known_costs={'ctrl-0': {'live-ee': 1}}) + assert _mesh_all_ready_nodes_visible(status) is True + + def test_fails_open_when_mesh_status_none(self): + """Fail open so existing peer-judgment error paths are not bypassed.""" + assert _mesh_all_ready_nodes_visible(None) is True + + +# ── Tests: _heartbeat_handle_lost_instances (task manager lock) ─────────────── + + +@pytest.mark.django_db +class TestHeartbeatHandleLostInstancesLock: + def _run(self, lost_instances, lock_behavior): + with ( + mock.patch('awx.main.tasks.system.reaper'), + mock.patch('awx.main.tasks.system.advisory_lock', lock_behavior), + ): + _heartbeat_handle_lost_instances(lost_instances, None) + + def test_processes_instance_when_lock_acquired(self, settings): + settings.AWX_AUTO_DEPROVISION_INSTANCES = False + inst = Instance.objects.create(hostname='ctrl-1', node_type='control', node_state='ready') + self._run([inst], _make_lock(True)) + inst.refresh_from_db() + assert inst.node_state == Instance.States.UNAVAILABLE + + def test_skips_instance_when_lock_unavailable(self, settings): + settings.AWX_AUTO_DEPROVISION_INSTANCES = False + inst = Instance.objects.create(hostname='ctrl-1', node_type='control', node_state='ready') + self._run([inst], _make_lock(False)) + inst.refresh_from_db() + assert inst.node_state == Instance.States.READY + + def test_logs_when_instance_deferred(self, settings): + settings.AWX_AUTO_DEPROVISION_INSTANCES = False + inst = Instance.objects.create(hostname='ctrl-1', node_type='control', node_state='ready') + with ( + mock.patch('awx.main.tasks.system.reaper'), + mock.patch('awx.main.tasks.system.advisory_lock', _make_lock(False)), + mock.patch('awx.main.tasks.system.logger') as mock_log, + ): + _heartbeat_handle_lost_instances([inst], None) + assert mock_log.info.called + assert 'ctrl-1' in mock_log.info.call_args[0][0] + + def test_per_instance_lock_first_skipped_second_processed(self, settings): + settings.AWX_AUTO_DEPROVISION_INSTANCES = False + inst1 = Instance.objects.create(hostname='ctrl-1', node_type='control', node_state='ready') + inst2 = Instance.objects.create(hostname='ctrl-2', node_type='control', node_state='ready') + self._run([inst1, inst2], _make_lock_sequence([False, True])) + inst1.refresh_from_db() + inst2.refresh_from_db() + assert inst1.node_state == Instance.States.READY + assert inst2.node_state == Instance.States.UNAVAILABLE + + def test_all_instances_processed_when_lock_always_acquired(self, settings): + settings.AWX_AUTO_DEPROVISION_INSTANCES = False + inst1 = Instance.objects.create(hostname='ctrl-1', node_type='control', node_state='ready') + inst2 = Instance.objects.create(hostname='ctrl-2', node_type='control', node_state='ready') + self._run([inst1, inst2], _make_lock(True)) + inst1.refresh_from_db() + inst2.refresh_from_db() + assert inst1.node_state == Instance.States.UNAVAILABLE + assert inst2.node_state == Instance.States.UNAVAILABLE + + +# ── Tests: _reap_and_mark_lost_instance (exception / branch coverage) ──────── + + +@pytest.mark.django_db +class TestReapAndMarkLostInstance: + def _inst(self, **kwargs): + defaults = {'hostname': 'ctrl-1', 'node_type': 'control', 'node_state': 'ready'} + defaults.update(kwargs) + return Instance.objects.create(**defaults) + + def test_reap_exception_does_not_prevent_mark_offline(self, settings): + settings.AWX_AUTO_DEPROVISION_INSTANCES = False + inst = self._inst() + with mock.patch('awx.main.tasks.system.reaper') as mock_reaper: + mock_reaper.reap.side_effect = Exception('receptor timeout') + _reap_and_mark_lost_instance(inst) + inst.refresh_from_db() + assert inst.node_state == Instance.States.UNAVAILABLE + + def test_auto_deprovision_deletes_control_node(self, settings): + settings.AWX_AUTO_DEPROVISION_INSTANCES = True + inst = self._inst(hostname='ctrl-dep') + with mock.patch('awx.main.tasks.system.reaper'): + _reap_and_mark_lost_instance(inst) + assert not Instance.objects.filter(hostname='ctrl-dep').exists() + + def test_skips_mark_offline_when_node_already_unavailable(self, settings): + settings.AWX_AUTO_DEPROVISION_INSTANCES = False + inst = self._inst(node_state='unavailable') + with mock.patch('awx.main.tasks.system.reaper'): + _reap_and_mark_lost_instance(inst) + inst.refresh_from_db() + assert inst.node_state == Instance.States.UNAVAILABLE + + def test_database_error_without_sqlstate_logs_exception(self, settings): + settings.AWX_AUTO_DEPROVISION_INSTANCES = False + inst = self._inst() + err = DatabaseError('constraint violation') + err.__cause__ = None + with ( + mock.patch('awx.main.tasks.system.reaper'), + mock.patch.object(inst, 'mark_offline', side_effect=err), + mock.patch('awx.main.tasks.system.logger') as mock_log, + ): + _reap_and_mark_lost_instance(inst) + assert mock_log.exception.called + assert 'No SQL state' in mock_log.exception.call_args[0][0] + + def test_database_error_with_sqlstate_logs_details(self, settings): + settings.AWX_AUTO_DEPROVISION_INSTANCES = False + inst = self._inst() + err = DatabaseError('unique violation') + + class _FakePsycopgError(Exception): + sqlstate = 'some_state' + + err.__cause__ = _FakePsycopgError('underlying pg error') + with ( + mock.patch('awx.main.tasks.system.reaper'), + mock.patch.object(inst, 'mark_offline', side_effect=err), + mock.patch('awx.main.tasks.system.logger') as mock_log, + mock.patch('awx.main.tasks.system.psycopg') as mock_psycopg, + ): + mock_psycopg.errors.lookup.return_value = 'SomeError' + mock_psycopg.errors.NoData = 'other_state' + _reap_and_mark_lost_instance(inst) + mock_psycopg.errors.lookup.assert_called_once_with('some_state') + assert mock_log.exception.called + + def test_database_error_with_nodata_sqlstate_logs_debug(self, settings): + settings.AWX_AUTO_DEPROVISION_INSTANCES = False + inst = self._inst() + err = DatabaseError('nodata') + + class _FakePsycopgError(Exception): + sqlstate = 'nodata_state' + + err.__cause__ = _FakePsycopgError('nodata cause') + with ( + mock.patch('awx.main.tasks.system.reaper'), + mock.patch.object(inst, 'mark_offline', side_effect=err), + mock.patch('awx.main.tasks.system.logger') as mock_log, + mock.patch('awx.main.tasks.system.psycopg') as mock_psycopg, + ): + mock_psycopg.errors.lookup.return_value = 'NoData' + mock_psycopg.errors.NoData = 'nodata_state' + _reap_and_mark_lost_instance(inst) + mock_log.exception.assert_not_called() + debug_messages = [str(call) for call in mock_log.debug.call_args_list] + assert any('marked' in m for m in debug_messages) + + +# ── Tests: inspect_execution_and_hop_nodes ──────────────────────────────────── + + +@pytest.mark.django_db +class TestInspectExecutionAndHopNodes: + @mock.patch('awx.main.tasks.system.inspect_established_receptor_connections') + def test_skips_when_lock_not_acquired(self, mock_inspect_conns): + with mock.patch('awx.main.tasks.system.advisory_lock', _make_lock(False)): + inspect_execution_and_hop_nodes([], _mesh_status()) + mock_inspect_conns.assert_not_called() + + @mock.patch('awx.main.tasks.system.inspect_established_receptor_connections') + def test_skips_when_mesh_status_none(self, mock_inspect_conns): + with mock.patch('awx.main.tasks.system.advisory_lock', _make_lock(True)): + inspect_execution_and_hop_nodes([], None) + mock_inspect_conns.assert_not_called() + + @mock.patch('awx.main.tasks.system.inspect_established_receptor_connections') + def test_runs_when_lock_acquired(self, mock_inspect_conns): + with mock.patch('awx.main.tasks.system.advisory_lock', _make_lock(True)): + inspect_execution_and_hop_nodes([], _mesh_status()) + mock_inspect_conns.assert_called_once() + + @mock.patch('awx.main.tasks.system.inspect_established_receptor_connections') + def test_updates_last_seen_for_execution_nodes(self, mock_inspect_conns): + exec_node = Instance.objects.create(hostname='exec-1', node_type='execution', node_state='ready') + control_node = Instance.objects.create(hostname='control-1', node_type='control', node_state='ready') + status = { + 'KnownConnectionCosts': {}, + 'Advertisements': [ + {'NodeID': 'exec-1', 'Time': '2026-01-01T00:00:00+00:00'}, + {'NodeID': 'control-1', 'Time': '2026-01-01T00:00:00+00:00'}, + ], + } + with mock.patch('awx.main.tasks.system.advisory_lock', _make_lock(True)): + inspect_execution_and_hop_nodes([exec_node, control_node], status) + exec_node.refresh_from_db() + assert exec_node.last_seen is not None # was None; function set it from Advertisements Time + control_node.refresh_from_db() + assert control_node.last_seen is None # control nodes not updated by this function + + +# ── Integration tests: _heartbeat_instance_management ──────────────────────── + + +@pytest.mark.django_db +def test_heartbeat_defers_lost_instances_when_mesh_gate_blocks(settings): + """Gate returns False (empty KnownConnectionCosts) → lost_instances suppressed.""" + settings.CLUSTER_HOST_ID = 'ctrl-0' + settings.AWX_AUTO_DEPROVISION_INSTANCES = False + settings.CLUSTER_NODE_HEARTBEAT_PERIOD = 60 + settings.CLUSTER_NODE_MISSED_HEARTBEAT_TOLERANCE = 2 + + this_inst = Instance.objects.create(hostname='ctrl-0', node_type='control', node_state='ready') + this_inst.last_seen = now() - timedelta(seconds=30) + this_inst.save(update_fields=['last_seen']) + + lost_peer = Instance.objects.create(hostname='ctrl-1', node_type='control', node_state='ready') + lost_peer.last_seen = now() - timedelta(seconds=200) # > 120s grace → is_lost() True + lost_peer.save(update_fields=['last_seen']) + + mock_ctl = mock.MagicMock() + mock_ctl.simple_command.return_value = {'KnownConnectionCosts': {}, 'Advertisements': []} + + with ( + mock.patch('awx.main.tasks.system.get_receptor_ctl', return_value=mock_ctl), + mock.patch('awx.main.tasks.system.inspect_execution_and_hop_nodes'), + mock.patch.object(Instance, 'local_health_check'), + ): + _, _, lost_result = _heartbeat_instance_management() + + assert lost_result == [] + + +@pytest.mark.django_db +def test_heartbeat_marks_offline_when_receptor_unavailable(settings): + """FileNotFoundError from get_receptor_ctl → this_inst marked offline, returns (None, None, None).""" + settings.CLUSTER_HOST_ID = 'ctrl-0' + settings.AWX_AUTO_DEPROVISION_INSTANCES = False + + this_inst = Instance.objects.create(hostname='ctrl-0', node_type='control', node_state='ready') + this_inst.last_seen = now() - timedelta(seconds=30) + this_inst.save(update_fields=['last_seen']) + + with ( + mock.patch('awx.main.tasks.system.get_receptor_ctl', side_effect=FileNotFoundError), + mock.patch.object(Instance, 'local_health_check'), + ): + result = _heartbeat_instance_management() + + assert result == (None, None, None) + this_inst.refresh_from_db() + assert this_inst.node_state == Instance.States.UNAVAILABLE diff --git a/awx/main/tests/unit/tasks/test_system.py b/awx/main/tests/unit/tasks/test_system.py index cdbf722bee..c567dc4833 100644 --- a/awx/main/tests/unit/tasks/test_system.py +++ b/awx/main/tests/unit/tasks/test_system.py @@ -1,7 +1,7 @@ import pytest from unittest.mock import MagicMock, patch -from awx.main.tasks.system import _heartbeat_instance_management, update_inventory_computed_fields, inspect_execution_and_hop_nodes -from awx.main.models import Instance, Inventory +from awx.main.tasks.system import update_inventory_computed_fields +from awx.main.models import Inventory from django.db import DatabaseError @@ -62,82 +62,3 @@ def test_update_inventory_computed_fields_database_error_nosqlstate(mock_logger, mock_filter.assert_called_once_with(id=1) mock_update_computed_fields.assert_called_once() mock_inventory.update_computed_fields.assert_called_once() - - -@patch('awx.main.tasks.system.get_receptor_ctl', side_effect=FileNotFoundError) -@patch('awx.main.tasks.system.Instance.objects.filter') -def test_heartbeat_marks_offline_when_receptor_unavailable(mock_filter, mock_get_ctl): - this_inst = MagicMock(spec=Instance) - this_inst.hostname = 'test-host' - mock_filter.return_value = [this_inst] - - with patch('awx.main.tasks.system.settings') as mock_settings: - mock_settings.CLUSTER_HOST_ID = 'test-host' - result = _heartbeat_instance_management() - - assert result == (None, None, None) - this_inst.local_health_check.assert_called_once() - this_inst.mark_offline.assert_called_once_with(errors='Receptor not available') - - -class TestInspectExecutionAndHopNodes: - """Tests for the advisory_lock guard in inspect_execution_and_hop_nodes.""" - - @patch('awx.main.tasks.system.inspect_established_receptor_connections') - @patch('awx.main.tasks.system.advisory_lock') - def test_skips_when_lock_not_acquired(self, mock_lock, mock_inspect_conns): - mock_lock.return_value.__enter__ = MagicMock(return_value=False) - mock_lock.return_value.__exit__ = MagicMock(return_value=False) - receptor_ctl = MagicMock() - - inspect_execution_and_hop_nodes([], receptor_ctl) - - receptor_ctl.simple_command.assert_not_called() - mock_inspect_conns.assert_not_called() - - @patch('awx.main.tasks.system.inspect_established_receptor_connections') - @patch('awx.main.tasks.system.advisory_lock') - def test_runs_when_lock_acquired(self, mock_lock, mock_inspect_conns): - mock_lock.return_value.__enter__ = MagicMock(return_value=True) - mock_lock.return_value.__exit__ = MagicMock(return_value=False) - receptor_ctl = MagicMock() - receptor_ctl.simple_command.return_value = {'Advertisements': [], 'KnownConnectionCosts': {}} - - inspect_execution_and_hop_nodes([], receptor_ctl) - - receptor_ctl.simple_command.assert_called_once_with('status') - mock_inspect_conns.assert_called_once() - - @patch('awx.main.tasks.system.inspect_established_receptor_connections') - @patch('awx.main.tasks.system.advisory_lock') - def test_updates_last_seen_for_execution_nodes(self, mock_lock, mock_inspect_conns): - mock_lock.return_value.__enter__ = MagicMock(return_value=True) - mock_lock.return_value.__exit__ = MagicMock(return_value=False) - - exec_node = MagicMock(spec=Instance) - exec_node.hostname = 'exec-1' - exec_node.node_type = 'execution' - exec_node.node_state = 'ready' - exec_node.last_seen = None - exec_node.capacity = 100 - exec_node.enabled = True - exec_node.cpu = 4 - exec_node.memory = 8000000000 - - control_node = MagicMock(spec=Instance) - control_node.hostname = 'control-1' - control_node.node_type = 'control' - - receptor_ctl = MagicMock() - receptor_ctl.simple_command.return_value = { - 'Advertisements': [ - {'NodeID': 'exec-1', 'Time': '2026-01-01T00:00:00+00:00'}, - {'NodeID': 'control-1', 'Time': '2026-01-01T00:00:00+00:00'}, - ], - 'KnownConnectionCosts': {}, - } - - inspect_execution_and_hop_nodes([exec_node, control_node], receptor_ctl) - - exec_node.save.assert_called_once_with(update_fields=['last_seen']) - control_node.save.assert_not_called()