import os from pathlib import Path import sys from types import SimpleNamespace from unittest import mock import pytest APP_DIR = Path(__file__).resolve().parents[1] / 'app' sys.path.insert(0, str(APP_DIR)) import supervisor class Guard: def __init__(self): self.depth = 0 self.entries = 0 def __enter__(self): self.depth += 1 self.entries += 1 return self def __exit__(self, *args): self.depth -= 1 def event(): state = [False] value = mock.Mock() value.is_set.side_effect = lambda: state[0] value.set.side_effect = lambda: state.__setitem__(0, True) value.clear.side_effect = lambda: state.__setitem__(0, False) value.wait.side_effect = lambda timeout=None: state[0] return value def lifecycle_context(**overrides): context = { 'lifecycle_phase': supervisor.PHASE_ACTIVE, 'activation_state': supervisor.PHASE_ACTIVE, 'start_gate_open': True, 'shutdown_requested': False, 'shutdown_event': event(), 'shutdown_retry_event': event(), 'control_lock': Guard(), 'supervisor_config': {'source_handoff_drain_timeout_sec': 0}, } context.update(overrides) return context @pytest.fixture(autouse=True) def isolated_runtime(monkeypatch): blocked = mock.Mock(side_effect=AssertionError('unexpected real runtime I/O')) for name in ('OwnedProcess', 'open_process', 'connect_postgres', 'load_postgres_env'): monkeypatch.setattr(supervisor, name, blocked) monkeypatch.setattr(supervisor, 'open', blocked, raising=False) monkeypatch.setattr(supervisor.sqlite3, 'connect', blocked) monkeypatch.setattr(supervisor.socket, 'socket', blocked) monkeypatch.setattr(supervisor.socket, 'create_connection', blocked) monkeypatch.setattr(supervisor.subprocess, 'Popen', blocked) monkeypatch.setattr(supervisor, 'signal', SimpleNamespace(SIGTERM=15, signal=mock.Mock())) monkeypatch.setattr(supervisor, 'threading', SimpleNamespace( Event=event, RLock=Guard, Thread=mock.Mock(), )) clock = SimpleNamespace(now=0.0) def sleep(seconds): clock.now += seconds clock.sleep = mock.Mock(side_effect=sleep) monkeypatch.setattr(supervisor, 'time', SimpleNamespace( monotonic=lambda: clock.now, time=lambda: clock.now, sleep=clock.sleep, )) for name in ( 'update_instance_activation', 'write_instance_metadata', 'write_shutdown_receipt', 'remove_shutdown_receipt', 'write_status_file', 'print_table', ): monkeypatch.setattr(supervisor, name, mock.Mock()) monkeypatch.setattr(supervisor, 'remove_instance_if_matches', mock.Mock(return_value=True)) monkeypatch.setattr(supervisor, 'scan_worker_snapshot', mock.Mock(return_value={'active': 0})) monkeypatch.setattr(supervisor, 'current_scan_worker_snapshot', mock.Mock(return_value={ 'active': 0, 'limit': 0, 'trufflehog': 0, 'sources': {}, 'detail': '', })) monkeypatch.setattr(supervisor, 'pipeline_status_snapshot', mock.Mock(return_value={})) monkeypatch.setattr(supervisor, 'table_signature', mock.Mock(return_value=())) monkeypatch.setattr(supervisor, 'build_runtime_table_lines', mock.Mock(return_value=[])) monkeypatch.setattr(supervisor, 'runtime_authority_error', mock.Mock(return_value='')) monkeypatch.setattr(supervisor, 'enable_ansi_terminal', mock.Mock(return_value=True)) monkeypatch.setattr(supervisor, 'read_single_key', mock.Mock(return_value=None)) monkeypatch.setattr(supervisor, 'sys', SimpleNamespace( stdin=mock.Mock(), stdout=mock.Mock(), stderr=mock.Mock(), executable=sys.executable, )) return clock @pytest.fixture def runtime(monkeypatch, tmp_path, isolated_runtime): h = SimpleNamespace( trace=[], children=[], metadata={}, guard=Guard(), previous_handler=object(), handler=None, context=None, pipeline_enabled=False, on_install=lambda: None, clock=isolated_runtime, ) h.args = SimpleNamespace( config=str(tmp_path / 'config.yaml'), sources=None, with_postgres=True, background=False, background_child=False, stop_background=False, background_status=False, attach=False, cmd=None, dry_run=False, autostart=True, non_interactive=True, launch_nonce='mock-launch', runtime_bootstrap_entrypoint=str(APP_DIR / 'supervisor.py'), expected_config_sha256='a' * 64, expected_supervisor_sha256='b' * 64, expected_code_manifest_sha256='c' * 64, ) h.options = {'dashboard': {'enabled': False}, 'source_handoff_drain_timeout_sec': 0} h.authority = { 'config_sha256': 'a' * 64, 'supervisor_sha256': 'b' * 64, 'code_manifest_sha256': 'c' * 64, 'code_manifest': {}, } config = {'global': {'queue_dir': str(tmp_path / 'queues')}, 'sources': {'fixture': {}}, 'keychecks': {}} monkeypatch.setattr(supervisor, 'parse_args', mock.Mock(return_value=h.args)) monkeypatch.setattr(supervisor, 'load_supervisor_runtime', mock.Mock(return_value=( config, str(tmp_path), str(tmp_path / 'results'), h.options, ['fixture'], ))) for name in ('preflight_lifecycle_paths', 'load_postgres_env'): monkeypatch.setattr(supervisor, name, mock.Mock()) monkeypatch.setattr(supervisor, 'validate_managed_runtime_startup', mock.Mock( return_value=SimpleNamespace(config=config, config_sha256='a' * 64), )) monkeypatch.setattr(supervisor, 'supervisor_runtime_from_config', mock.Mock(return_value=( config, str(tmp_path), str(tmp_path / 'results'), h.options, ['fixture'], ))) monkeypatch.setattr(supervisor, 'runtime_authority', mock.Mock(return_value=h.authority)) monkeypatch.setattr(supervisor, 'sha256_file', mock.Mock(return_value='a' * 64)) monkeypatch.setattr(supervisor, 'canonical_database_url', mock.Mock(return_value='postgresql://mock/mock')) monkeypatch.setattr(supervisor, 'supervised_child_environment', mock.Mock(return_value={})) monkeypatch.setattr(supervisor, 'DependencyGate', mock.Mock()) h.instance_file = str(tmp_path / 'instance.json') monkeypatch.setattr(supervisor, 'background_paths', mock.Mock(return_value=( h.instance_file, str(tmp_path / 'log'), str(tmp_path / 'status'), str(tmp_path / 'pid'), ))) monkeypatch.setattr(supervisor, 'background_lock_path', mock.Mock(return_value=str(tmp_path / 'lock'))) path = mock.Mock(wraps=os.path) path.exists.return_value = False environment = {supervisor.RUNTIME_BOOTSTRAP_ENV: supervisor.RUNTIME_BOOTSTRAP_VALUE} monkeypatch.setattr(supervisor, 'os', SimpleNamespace( name='posix', path=path, getenv=mock.Mock(side_effect=environment.get), getpid=lambda: 123, )) h.instance_lock = mock.Mock() h.cluster_lock = mock.Mock() for name, held, label in ( ('SupervisorInstanceLock', h.instance_lock, 'instance.release'), ('ClusterAuthorityLock', h.cluster_lock, 'cluster.release'), ): constructor = mock.Mock() constructor.return_value.acquire.return_value = held held.release.side_effect = lambda label=label: h.trace.append(label) monkeypatch.setattr(supervisor, name, constructor) supervisor.threading.RLock = lambda: h.guard def child(obj, name, enabled=True): obj.source, obj.enabled, obj.manual_only = name, enabled, False obj.status, obj.desired_state = 'stopped', 'stopped' obj.start = mock.Mock(side_effect=lambda **kw: h.trace.append(name + '.start') or True) obj.poll = mock.Mock() def stop(**kw): h.trace.append(name + '.stop') obj.status = 'stopped' return True obj.stop = mock.Mock(side_effect=stop) obj.is_running = mock.Mock(return_value=False) obj.build_command = mock.Mock(return_value=['mock-child']) obj.log_path = obj.state_path = str(tmp_path / name) obj.use_per_source_state = obj.once = obj.repeat = obj.restart = False obj.interval = 0 h.children.append(obj) if name == 'fixture': h.source = obj return obj class Pipeline: def __init__(self, name, *args): child(self, name, h.pipeline_enabled) monkeypatch.setattr(supervisor, 'ManagedPipelineWorker', Pipeline) monkeypatch.setattr(supervisor, 'ManagedSource', mock.Mock(side_effect=lambda name, *args: child(SimpleNamespace(), name))) class ManagedKeychecks: def __init__(self, *args): child(self, 'keychecks') monkeypatch.setattr(supervisor, 'ManagedKeychecks', ManagedKeychecks) monkeypatch.setattr(supervisor, 'ManagedDockerShadow', mock.Mock(side_effect=lambda *args: child(SimpleNamespace(), 'docker-shadow', False))) h.dashboard = mock.Mock(desired_state='stopped', process=None) h.controller = mock.Mock( terminal=True, stop_succeeded=True, authority_release_safe=True, lifecycle_action_required=True, ready=True, detail='', state=supervisor.PostgresState.STOPPED, ) h.controller.snapshot.return_value = {'state': 'STOPPED', 'failures': 0, 'detail': ''} h.dashboard.snapshot.return_value = {'status': 'disabled'} for obj, name, result, label in ( (h.dashboard, 'start', False, 'dashboard.start'), (h.dashboard, 'stop', True, 'dashboard.stop'), (h.dashboard, 'close', None, 'dashboard.close'), (h.controller, 'request_stop', None, 'postgres.stop'), (h.controller, 'close', True, 'postgres.close'), ): getattr(obj, name).side_effect = lambda *a, result=result, label=label, **kw: h.trace.append(label) or result monkeypatch.setattr(supervisor, 'ManagedDashboard', mock.Mock(return_value=h.dashboard)) monkeypatch.setattr(supervisor, 'controller_from_config', mock.Mock(return_value=h.controller)) h.server = mock.Mock(server_address=('127.0.0.1', 12345)) h.server.shutdown.side_effect = lambda: h.trace.append('control.shutdown') h.server.server_close.side_effect = lambda: h.trace.append('control.close') def control_server(*args, **kwargs): h.context = args[2] return h.server monkeypatch.setattr(supervisor, 'start_control_server', mock.Mock(side_effect=control_server)) def metadata(*args, **kwargs): h.metadata.update(h.authority, **kwargs) return dict(h.metadata) monkeypatch.setattr(supervisor, 'build_instance_metadata', mock.Mock(side_effect=metadata)) monkeypatch.setattr(supervisor, 'load_instance_metadata', mock.Mock(side_effect=lambda path: dict(h.metadata))) def publish(path, instance_id, phase): assert h.guard.depth > 0 h.trace.append('metadata.' + phase) h.metadata['activation_state'] = phase supervisor.update_instance_activation.side_effect = publish supervisor.write_shutdown_receipt.side_effect = lambda *args: h.trace.append('receipt') def register(signum, callback): if callback is h.previous_handler: h.trace.append('signal.restore') else: h.trace.append('signal.install') h.handler = callback h.context = callback.__closure__[0].cell_contents h.on_install() return h.previous_handler supervisor.signal.signal.side_effect = register def start_thread(): if h.args.background_child: with h.guard: h.context['activation_callback']() supervisor.threading.Thread.return_value.start.side_effect = start_thread h.loop = mock.Mock() monkeypatch.setattr(supervisor, 'non_interactive_loop', h.loop) monkeypatch.setattr(supervisor, 'interactive_loop', h.loop) h.shutdown = mock.Mock(wraps=supervisor.coordinated_shutdown) h.retain = mock.Mock(wraps=supervisor.retain_unsafe_authority) monkeypatch.setattr(supervisor, 'coordinated_shutdown', h.shutdown) monkeypatch.setattr(supervisor, 'retain_unsafe_authority', h.retain) h.writer = mock.Mock() h.writer.close.side_effect = lambda: h.trace.append('log.close') monkeypatch.setattr(supervisor, 'install_bounded_background_output', mock.Mock(return_value=h.writer)) monkeypatch.setattr(supervisor, 'print', mock.Mock(side_effect=lambda *a, **kw: h.trace.append('log.print')), raising=False) supervisor.sys.stdout.flush.side_effect = lambda: h.trace.append('stdout.flush') supervisor.sys.stderr.flush.side_effect = lambda: h.trace.append('stderr.flush') return h def test_signal_callback_is_only_an_idempotent_flag_assignment(runtime): h = runtime def check_callback(): before = dict(h.context) trace = list(h.trace) entries = h.guard.entries h.handler(15, object()) h.handler(15, None) assert h.context == {**before, 'shutdown_requested': True} assert h.trace == trace assert h.guard.entries == entries h.context['shutdown_event'].set.assert_not_called() h.context['shutdown_retry_event'].set.assert_not_called() supervisor.update_instance_activation.assert_not_called() supervisor.write_shutdown_receipt.assert_not_called() h.on_install = check_callback assert supervisor.main() == 0 assert h.context['shutdown_requested'] is True assert supervisor.signal.signal.call_args_list[-1] == mock.call(15, h.previous_handler) for source in h.children: source.start.assert_not_called() @pytest.mark.parametrize('background', [False, True]) def test_receipt_parity_and_finalization_order(runtime, background): h = runtime h.args.background_child = background assert supervisor.main() == 0 supervisor.write_shutdown_receipt.assert_called_once_with(h.instance_file, h.metadata['instance_id'], 0) supervisor.remove_instance_if_matches.assert_not_called() supervisor.remove_shutdown_receipt.assert_not_called() before_receipt = ['fixture.stop', 'postgres.close', 'dashboard.close', 'control.shutdown', 'control.close', 'stdout.flush', 'stderr.flush'] if background: before_receipt.append('log.close') assert all(h.trace.index(label) < h.trace.index('receipt') for label in before_receipt) assert h.trace.index('receipt') < h.trace.index('instance.release') < h.trace.index('cluster.release') < h.trace.index('signal.restore') assert h.loop.call_args.kwargs['stay_alive'] is background def test_windows_does_not_register_signals(runtime): supervisor.os.name = 'nt' assert supervisor.main() == 0 supervisor.signal.signal.assert_not_called() @pytest.mark.parametrize('missing', ['bootstrap', 'managed_postgres']) def test_mutating_launch_gates_precede_config_locks_children_and_signals(runtime, missing): if missing == 'bootstrap': supervisor.os.getenv.side_effect = lambda name, default=None: default else: runtime.args.with_postgres = False with pytest.raises(SystemExit, match='canonical runtime bootstrap|unmanaged PostgreSQL'): supervisor.main() supervisor.sha256_file.assert_not_called() supervisor.load_supervisor_runtime.assert_not_called() supervisor.preflight_lifecycle_paths.assert_not_called() supervisor.ClusterAuthorityLock.assert_not_called() supervisor.SupervisorInstanceLock.assert_not_called() supervisor.signal.signal.assert_not_called() assert not runtime.children @pytest.mark.parametrize('flag,dispatch', [ ('background', 'start_background'), ('stop_background', 'stop_background'), ('background_status', 'background_status'), ('attach', 'attach_background'), ('cmd', 'send_background_command'), ('dry_run', None), ]) def test_non_owning_dispatch_does_not_register_signals(runtime, monkeypatch, flag, dispatch): setattr(runtime.args, flag, 'status' if flag == 'cmd' else True) if dispatch: monkeypatch.setattr(supervisor, dispatch, mock.Mock(return_value=True)) assert supervisor.main() == 0 supervisor.signal.signal.assert_not_called() supervisor.ClusterAuthorityLock.assert_not_called() supervisor.SupervisorInstanceLock.assert_not_called() @pytest.mark.parametrize('stage', ['load', 'authority', 'publish', 'event']) def test_pending_term_cannot_admit_startup(runtime, stage): h = runtime if stage == 'load': supervisor.load_instance_metadata.side_effect = lambda path: h.handler(15, None) or dict(h.metadata) elif stage == 'authority': supervisor.runtime_authority_error.side_effect = lambda context: h.handler(15, None) or '' elif stage == 'publish': publish = supervisor.update_instance_activation.side_effect def signal_during_publish(*args): publish(*args) if args[-1] == supervisor.PHASE_ACTIVE: h.handler(15, None) supervisor.update_instance_activation.side_effect = signal_during_publish else: def install_event_hook(): h.context['activation_event'].set.side_effect = lambda: h.handler(15, None) h.on_install = install_event_hook assert supervisor.main() == 0 for source in h.children: source.start.assert_not_called() h.dashboard.start.assert_not_called() h.loop.assert_not_called() assert h.shutdown.called is (stage in ('publish', 'event')) assert h.context['lifecycle_phase'] == supervisor.PHASE_STOPPING def test_background_activation_wait_consumes_term(runtime): h = runtime h.args.background_child = True supervisor.threading.Thread.return_value.start.side_effect = None h.clock.sleep.side_effect = lambda seconds: h.handler(15, None) assert supervisor.main() == 0 assert supervisor.PHASE_ACTIVE not in [call.args[-1] for call in supervisor.update_instance_activation.call_args_list] h.shutdown.assert_not_called() h.controller.request_stop.assert_not_called() assert h.context['shutdown_event'].is_set() def test_admission_rejects_pending_term_before_checkpoint(): context = lifecycle_context(shutdown_requested=True) assert not supervisor.lifecycle_start_allowed(context) assert context['lifecycle_phase'] == supervisor.PHASE_ACTIVE context['shutdown_event'].set.assert_not_called() supervisor.update_instance_activation.assert_not_called() @pytest.mark.parametrize('during_callback', [False, True]) def test_control_activation_cannot_reopen_pending_shutdown(during_callback): context = lifecycle_context( lifecycle_phase=supervisor.PHASE_ACTIVATING, activation_state=supervisor.PHASE_ACTIVATING, start_gate_open=False, shutdown_requested=not during_callback, ) callback = mock.Mock(side_effect=lambda: context.__setitem__('shutdown_requested', True)) context['activation_callback'] = callback server = object.__new__(supervisor.SupervisorControlServer) server.context, server.lock = context, context['control_lock'] server.instance_id, server.token = 'exact', 'credential' result = server.run_request({ 'schema': supervisor.CONTROL_SCHEMA, 'instance_id': 'exact', 'token': 'credential', 'action': 'activate', }) assert not result['ok'] assert callback.called is during_callback assert context['lifecycle_phase'] == supervisor.PHASE_STOPPING assert not supervisor.lifecycle_start_allowed(context) def test_background_activation_callback_failure_is_sticky(runtime): h = runtime h.args.background_child = True publish = supervisor.update_instance_activation.side_effect def partial_publication(*args): publish(*args) if args[-1] == supervisor.PHASE_ACTIVE: raise SystemExit(None) def request_activation(): server = object.__new__(supervisor.SupervisorControlServer) server.context, server.lock = h.context, h.guard server.instance_id, server.token = h.metadata['instance_id'], h.metadata['token'] h.activation_result = server.run_request({ 'schema': supervisor.CONTROL_SCHEMA, 'instance_id': server.instance_id, 'token': server.token, 'action': 'activate', }) supervisor.update_instance_activation.side_effect = partial_publication supervisor.threading.Thread.return_value.start.side_effect = request_activation assert supervisor.main() != 0 assert not h.activation_result['ok'] h.shutdown.assert_called_once() h.source.start.assert_not_called() assert supervisor.write_shutdown_receipt.call_args.args[-1] != 0 @pytest.mark.parametrize('autostart', [True, False]) def test_term_between_pipeline_starts_admits_no_more_children(runtime, monkeypatch, autostart): h = runtime h.pipeline_enabled = True h.args.autostart = autostart if not autostart: config = supervisor.load_supervisor_runtime.return_value[0] config['keychecks'].update(enabled=True, autostart=True) constructor = supervisor.ManagedPipelineWorker.__init__ def initialize(source, name, *args): constructor(source, name, *args) if name == 'result-ingester': source.start.side_effect = lambda **kw: h.handler(15, None) or True monkeypatch.setattr(supervisor.ManagedPipelineWorker, '__init__', initialize) assert supervisor.main() == 0 for source in h.children: assert source.start.call_count == int(source.source == 'result-ingester') h.dashboard.start.assert_not_called() h.shutdown.assert_called_once() supervisor.write_shutdown_receipt.assert_called_once_with(h.instance_file, h.metadata['instance_id'], 0) @pytest.mark.parametrize('loop_name', ['non_interactive_loop', 'interactive_loop_blocking', 'watch_local', 'tick_supervisor_runtime']) @pytest.mark.parametrize('pending_signal', [False, True]) def test_checkpoints_exit_without_polling_or_starting(loop_name, pending_signal): context = lifecycle_context(shutdown_requested=pending_signal, instance_file='mock', instance_id='exact') if not pending_signal: context['shutdown_event'].set() source = mock.Mock(status='running', manual_only=False) context['postgres_controller'] = mock.Mock() def publish(*args): assert context['control_lock'].depth > 0 supervisor.update_instance_activation.side_effect = publish function = getattr(supervisor, loop_name) if loop_name == 'tick_supervisor_runtime': with context['control_lock']: function(context) elif loop_name == 'interactive_loop_blocking': function([source], autostart=True, context=context) else: function([source], context=context) source.start.assert_not_called() source.poll.assert_not_called() context['postgres_controller'].tick.assert_not_called() supervisor.read_single_key.assert_not_called() assert context['shutdown_event'].is_set() if pending_signal: assert context['lifecycle_phase'] == supervisor.PHASE_STOPPING def test_term_between_source_polls_rejects_next_poll(): context = lifecycle_context() first, second = mock.Mock(status='running'), mock.Mock(status='running') first.poll.side_effect = lambda: context.__setitem__('shutdown_requested', True) supervisor.non_interactive_loop([first, second], context=context) first.poll.assert_called_once() second.poll.assert_not_called() assert context['lifecycle_phase'] == supervisor.PHASE_STOPPING @pytest.mark.parametrize('watch', [False, True]) def test_foreground_waits_consume_term_without_input_or_next_refresh(monkeypatch, watch): context = lifecycle_context() source = mock.Mock(status='running') def pending(*args, **kwargs): context['shutdown_requested'] = True return None if watch else 'start all' if watch: supervisor.read_single_key.side_effect = pending supervisor.watch_local([source], poll_sec=100, context=context) else: commands = mock.Mock() commands.get.side_effect = pending monkeypatch.setattr(supervisor.queue, 'Queue', mock.Mock(return_value=commands)) supervisor.interactive_loop_blocking([source], context=context) assert context['lifecycle_phase'] == supervisor.PHASE_STOPPING source.start.assert_not_called() source.poll.assert_called_once() @pytest.mark.parametrize('transition,phase', [ ('begin_stopping', supervisor.PHASE_STOPPING), ('enter_failed_hold', supervisor.PHASE_FAILED_HOLD), ]) @pytest.mark.parametrize('error', [OSError('publication'), KeyboardInterrupt(), SystemExit(0)]) def test_metadata_publication_failure_leaves_closed_memory_gates(transition, phase, error): context = lifecycle_context(instance_file='mock', instance_id='exact') def fail_publication(*args): assert context['lifecycle_phase'] == context['activation_state'] == phase assert not supervisor.lifecycle_start_allowed(context) assert context['authority_release_safe'] is False raise error supervisor.update_instance_activation.side_effect = fail_publication with context['control_lock'], pytest.raises(type(error)): getattr(supervisor, transition)(context) assert context['runtime_failed'] is True assert not supervisor.lifecycle_start_allowed(context) def test_repeated_term_checkpoint_does_not_republish_or_rearm_retry(): context = lifecycle_context(shutdown_requested=True, instance_file='mock', instance_id='exact') with context['control_lock']: assert supervisor.shutdown_checkpoint(context) assert supervisor.shutdown_checkpoint(context) supervisor.update_instance_activation.assert_called_once_with('mock', 'exact', supervisor.PHASE_STOPPING) context['shutdown_retry_event'].set.assert_not_called() supervisor.enter_failed_hold(context) supervisor.begin_stopping(context) assert context['lifecycle_phase'] == supervisor.PHASE_FAILED_HOLD context['shutdown_retry_event'].set.assert_called_once() def test_repeated_shutdown_does_not_discard_confirmed_stop_proof(): context = lifecycle_context( lifecycle_phase=supervisor.PHASE_STOPPING, activation_state=supervisor.PHASE_STOPPING, start_gate_open=False, authority_release_safe=True, ) with context['control_lock']: supervisor.begin_stopping(context) assert context['authority_release_safe'] is True assert not supervisor.lifecycle_start_allowed(context) @pytest.mark.parametrize('error', [KeyboardInterrupt(), SystemExit(0), RuntimeError('cleanup')]) @pytest.mark.parametrize('publication_fails', [False, True]) def test_initial_cleanup_interruption_retains_authority_and_failure(runtime, error, publication_fails): h = runtime h.shutdown.side_effect = error publish = supervisor.update_instance_activation.side_effect def update(*args): if publication_fails and args[-1] == supervisor.PHASE_FAILED_HOLD: raise OSError('hold publication') publish(*args) supervisor.update_instance_activation.side_effect = update def retain(sources, context): assert context['lifecycle_phase'] == supervisor.PHASE_FAILED_HOLD assert context['authority_release_safe'] is False assert not supervisor.lifecycle_start_allowed(context) h.instance_lock.release.assert_not_called() h.cluster_lock.release.assert_not_called() h.server.shutdown.assert_not_called() supervisor.write_shutdown_receipt.assert_not_called() assert supervisor.signal.signal.call_count == 1 h.handler(15, None) context['authority_release_safe'] = True return True h.retain.side_effect = retain assert supervisor.main() != 0 h.retain.assert_called_once() assert supervisor.write_shutdown_receipt.call_args.args[-1] != 0 def test_interrupted_cleanup_lock_acquisition_retains_authority(runtime, monkeypatch): h = runtime interrupt = [False] enter = Guard.__enter__ def acquire(guard): if guard is h.guard and interrupt[0]: interrupt[0] = False raise KeyboardInterrupt() return enter(guard) def retain(sources, context): assert not supervisor.lifecycle_start_allowed(context) assert context['authority_release_safe'] is False h.instance_lock.release.assert_not_called() h.cluster_lock.release.assert_not_called() h.server.shutdown.assert_not_called() supervisor.write_shutdown_receipt.assert_not_called() context['authority_release_safe'] = True monkeypatch.setattr(Guard, '__enter__', acquire) h.loop.side_effect = lambda *a, **kw: interrupt.__setitem__(0, True) h.retain.side_effect = retain assert supervisor.main() != 0 h.retain.assert_called_once() @pytest.mark.parametrize('error', [RuntimeError('ACTIVE publication'), SystemExit(7)]) def test_partially_published_activation_uses_full_cleanup(runtime, error): h = runtime publish = supervisor.update_instance_activation.side_effect def partial_publication(*args): publish(*args) if args[-1] == supervisor.PHASE_ACTIVE: raise error supervisor.update_instance_activation.side_effect = partial_publication assert supervisor.main() != 0 h.shutdown.assert_called_once() h.source.stop.assert_called_once_with(final=True) assert h.trace.index('fixture.stop') < h.trace.index('postgres.stop') h.source.start.assert_not_called() @pytest.mark.parametrize('error', [KeyboardInterrupt(), SystemExit(None)]) def test_coordinated_shutdown_interruption_closes_gate_before_propagating(error): context = lifecycle_context() source = mock.Mock(source='fixture') source.stop.side_effect = error with pytest.raises(type(error)): supervisor.coordinated_shutdown([source], context) assert context['authority_release_safe'] is False assert context['lifecycle_phase'] == supervisor.PHASE_FAILED_HOLD assert context['runtime_failed'] is True def test_failed_hold_survives_interruption_and_logging_failure(monkeypatch): context = lifecycle_context(authority_release_safe=False) source = mock.Mock(source='fixture') source.stop.side_effect = [KeyboardInterrupt(), False, True] source.is_running.return_value = False monkeypatch.setattr(supervisor, 'print', mock.Mock(side_effect=OSError('log unavailable')), raising=False) assert supervisor.retain_unsafe_authority([source], context, max_attempts=3) assert context['authority_release_safe'] is True assert context['runtime_failed'] is True assert context['start_gate_open'] is False def test_failed_hold_does_not_release_with_unconfirmed_children(): context = lifecycle_context(authority_release_safe=False, postgres_controller=mock.Mock()) source = mock.Mock(source='fixture') source.stop.return_value = False source.is_running.return_value = True assert not supervisor.retain_unsafe_authority([source], context, max_attempts=2) assert source.stop.call_count == 2 assert context['authority_release_safe'] is False context['postgres_controller'].request_stop.assert_not_called() context['postgres_controller'].close.assert_not_called() @pytest.mark.parametrize('error', [OSError('receipt'), KeyboardInterrupt()]) def test_receipt_write_failure_is_nonzero(runtime, error): supervisor.write_shutdown_receipt.side_effect = error assert supervisor.main() != 0 runtime.instance_lock.release.assert_called_once() runtime.cluster_lock.release.assert_called_once() @pytest.mark.parametrize('operation', ['control.shutdown', 'control.close', 'stdout.flush', 'stderr.flush', 'log.close']) def test_late_cleanup_failures_are_in_receipt(runtime, operation): h = runtime h.args.background_child = True target = { 'control.shutdown': h.server.shutdown, 'control.close': h.server.server_close, 'stdout.flush': supervisor.sys.stdout.flush, 'stderr.flush': supervisor.sys.stderr.flush, 'log.close': h.writer.close, }[operation] target.side_effect = KeyboardInterrupt() assert supervisor.main() != 0 assert supervisor.write_shutdown_receipt.call_args.args[-1] != 0 h.server.server_close.assert_called_once() h.writer.close.assert_called_once() def test_failure_latched_by_draining_control_worker_is_in_receipt(runtime): runtime.server.server_close.side_effect = lambda: runtime.context.__setitem__('runtime_failed', True) assert supervisor.main() != 0 assert supervisor.write_shutdown_receipt.call_args.args[-1] != 0 @pytest.mark.parametrize('status,background,failed', [ ('failed', False, True), ('waiting', False, False), ('done', False, False), ('failed', True, False), ]) def test_terminal_foreground_failure_is_sticky_without_changing_restart_policy(status, background, failed): context = lifecycle_context(background_child=background) source = mock.Mock(status=status) supervisor.poll_managed_sources([source], context) source.status = 'stopped' supervisor.poll_managed_sources([source], context) assert bool(context.get('runtime_failed')) is failed source.start.assert_not_called() def test_foreground_source_failure_survives_cleanup_status_reset(runtime): runtime.loop.side_effect = lambda *a, **kw: setattr(runtime.source, 'status', 'failed') assert supervisor.main() != 0 assert runtime.source.status == 'stopped' assert supervisor.write_shutdown_receipt.call_args.args[-1] != 0 def test_foreground_missing_autostart_is_nonzero(runtime): runtime.args.autostart = False assert supervisor.main() != 0 runtime.loop.assert_not_called() assert supervisor.write_shutdown_receipt.call_args.args[-1] != 0 @pytest.mark.parametrize('code,expected', [(None, 0), (0, 0), ('', 1), ('refused', 1), (9, 9)]) def test_system_exit_status_is_preserved(runtime, code, expected): runtime.loop.side_effect = SystemExit(code) assert supervisor.main() == expected assert supervisor.write_shutdown_receipt.call_args.args[-1] == expected def test_interrupt_message_failure_cannot_publish_success(runtime): runtime.loop.side_effect = KeyboardInterrupt() log = supervisor.print.side_effect def print_message(*args, **kwargs): if args[0] == '\nStopping child processes...': raise OSError('log unavailable') return log(*args, **kwargs) supervisor.print.side_effect = print_message assert supervisor.main() != 0 assert supervisor.write_shutdown_receipt.call_args.args[-1] != 0 def test_authority_drift_failure_remains_sticky(runtime): def drift(*args, context, **kwargs): supervisor.inhibit_for_authority_drift(context, 'mock drift') context['authority_drift'] = '' runtime.loop.side_effect = drift assert supervisor.main() != 0 assert supervisor.write_shutdown_receipt.call_args.args[-1] != 0 @pytest.mark.parametrize('failure', [False, KeyboardInterrupt()]) def test_preactivation_close_failure_is_sticky(runtime, failure): h = runtime h.on_install = lambda: h.handler(15, None) h.controller.lifecycle_action_required = False h.controller.close.side_effect = [failure, True] assert supervisor.main() != 0 h.retain.assert_called_once() h.controller.request_stop.assert_not_called() assert supervisor.write_shutdown_receipt.call_args.args[-1] != 0 @pytest.mark.parametrize('stay_alive', [False, True]) def test_noninteractive_keeps_existing_completion_policy(isolated_runtime, stay_alive): context = lifecycle_context(background_child=stay_alive) source = mock.Mock(status='done') isolated_runtime.sleep.side_effect = lambda seconds: context.__setitem__('shutdown_requested', True) supervisor.non_interactive_loop([source], context=context, stay_alive=stay_alive) assert isolated_runtime.sleep.called is stay_alive assert not context.get('runtime_failed') @pytest.mark.parametrize('exit_code,receipt,removed,expected', [ (None, 0, True, True), (None, None, True, False), (None, 1, True, False), (None, ValueError('wrong instance'), True, False), (None, 0, False, False), (0, 0, True, True), (0, 1, True, False), (1, 0, True, False), (1, 1, True, False), (0, None, True, True), ]) def test_stopper_requires_posix_receipt_and_preserves_windows_crosscheck(monkeypatch, tmp_path, exit_code, receipt, removed, expected): metadata = {'instance_id': 'exact-instance', 'manages_postgres': True} process = mock.Mock() process.wait.return_value = True process.exit_code.return_value = exit_code instance_path = str(tmp_path / 'instance.json') monkeypatch.setattr(supervisor, 'background_paths', mock.Mock(return_value=(instance_path, '', '', ''))) monkeypatch.setattr(supervisor, 'background_lock_path', mock.Mock(return_value='mock-lock')) monkeypatch.setattr(supervisor, 'load_instance_metadata', mock.Mock(return_value=metadata)) monkeypatch.setattr(supervisor, 'verify_instance_process', mock.Mock(return_value=process)) monkeypatch.setattr(supervisor, 'send_control_request', mock.Mock(return_value='shutdown requested')) load_receipt = mock.Mock(return_value=receipt) if isinstance(receipt, BaseException): load_receipt.side_effect = receipt monkeypatch.setattr(supervisor, 'load_shutdown_receipt', load_receipt) supervisor.remove_instance_if_matches.return_value = removed assert supervisor.stop_background('mock-config', 'mock-results', {}) is expected load_receipt.assert_called_once_with(instance_path, 'exact-instance') assert supervisor.remove_shutdown_receipt.called is expected if expected: supervisor.remove_instance_if_matches.assert_called_once_with(instance_path, 'exact-instance', lock_path='mock-lock') process.close.assert_called_once() @pytest.mark.parametrize('observer', [False, True]) @pytest.mark.parametrize('running', [True, None]) def test_unconfirmed_child_defers_all_postgres_actions(observer, running): controller = mock.Mock(lifecycle_action_required=not observer) source = object.__new__(supervisor.ManagedPipelineWorker) source.source = 'unexpected-pipeline-worker' source.stop = mock.Mock(return_value=True) source.is_running = mock.Mock(return_value=running) context = lifecycle_context(postgres_controller=controller) assert not supervisor.coordinated_shutdown([source], context) controller.request_stop.assert_not_called() controller.close.assert_not_called() assert context['authority_release_safe'] is False @pytest.mark.parametrize('observer,elapsed,remaining', [(False, 3, 7), (False, 15, 0), (True, 0, 10)]) def test_children_precede_postgres_and_stop_close_share_budget(isolated_runtime, observer, elapsed, remaining): clock = isolated_runtime trace = [] controller = mock.Mock( lifecycle_action_required=not observer, terminal=True, stop_succeeded=True, authority_release_safe=True, detail='', ) def request_stop(): trace.append('postgres.stop') clock.now += elapsed controller.request_stop.side_effect = request_stop controller.close.side_effect = lambda **kw: trace.append('postgres.close') or True sources = [] for name in ( 'janitor', 'result-ingester', 'keychecks', 'jsonl-projector', 'scanner', 'worker-api', ): source = mock.Mock(source=name) source.stop.side_effect = lambda name=name, **kw: trace.append(name + '.stop') or True source.is_running.side_effect = lambda name=name: (trace.append(name + '.confirmed'), False)[1] sources.append(source) dashboard = mock.Mock(process=None) dashboard.stop.side_effect = lambda **kw: trace.append('dashboard.stop') or True context = lifecycle_context(postgres_controller=controller, dashboard_manager=dashboard) context['supervisor_config']['postgres_shutdown_timeout_sec'] = 10 assert supervisor.coordinated_shutdown(sources, context) assert trace[:7] == [ 'scanner.stop', 'worker-api.stop', 'keychecks.stop', 'jsonl-projector.stop', 'result-ingester.stop', 'janitor.stop', 'dashboard.stop', ] assert all(trace.index(source.source + '.confirmed') < trace.index('postgres.close') for source in sources) assert controller.request_stop.called is not observer controller.close.assert_called_once_with(wait=False, timeout_sec=remaining) @pytest.mark.parametrize('terminal_after_tick,remaining', [(True, 5.95), (False, 0)]) def test_postgres_wait_consumes_close_budget(isolated_runtime, terminal_after_tick, remaining): clock = isolated_runtime controller = mock.Mock( lifecycle_action_required=True, terminal=False, stop_succeeded=True, authority_release_safe=True, detail='', ) def tick(): clock.now += 3 controller.terminal = terminal_after_tick def close(**kwargs): controller.terminal = True return True controller.request_stop.side_effect = lambda: setattr(clock, 'now', clock.now + 1) controller.tick.side_effect = tick controller.close.side_effect = close context = lifecycle_context(postgres_controller=controller) context['supervisor_config']['postgres_shutdown_timeout_sec'] = 10 assert supervisor.coordinated_shutdown([], context) assert controller.tick.called assert controller.close.call_args.kwargs['timeout_sec'] == pytest.approx(remaining)