from contextlib import ExitStack import os from pathlib import Path import sys from types import SimpleNamespace import unittest from unittest import mock ROOT = Path(__file__).resolve().parents[1] APP_DIR = ROOT / 'app' sys.path.insert(0, str(APP_DIR)) import supervisor def supervisor_args(**overrides): values = { 'config': 'config.yaml', 'sources': None, 'once': False, 'dry_run': False, 'status_interval': None, 'no_clear': True, 'autostart': False, 'non_interactive': True, 'background_child': False, 'launch_nonce': None, 'expected_config_sha256': None, 'expected_supervisor_sha256': None, 'expected_code_manifest_sha256': None, 'background': False, 'stop_background': False, 'background_status': False, 'attach': False, 'cmd': None, 'instance_file': None, 'pid_file': None, 'dashboard': False, 'no_dashboard': False, 'with_postgres': False, } values.update(overrides) return SimpleNamespace(**values) def runtime_fixture(): config = { 'global': {'queue_dir': 'queues'}, 'sources': {'fixture': {'enabled': True}}, 'keychecks': {'enabled': False}, } return config, str(APP_DIR), 'results', {'dashboard': {'enabled': False}}, ['fixture'] class SupervisorManagedPostgresGateTests(unittest.TestCase): def main_setup(self, stack, args): runtime = runtime_fixture() stack.enter_context(mock.patch.dict(os.environ, { supervisor.RUNTIME_BOOTSTRAP_ENV: supervisor.RUNTIME_BOOTSTRAP_VALUE, }, clear=False)) stack.enter_context(mock.patch.object(supervisor, 'parse_args', return_value=args)) stack.enter_context(mock.patch.object(supervisor, 'sha256_file', return_value='a' * 64)) stack.enter_context(mock.patch.object(supervisor, 'load_supervisor_runtime', return_value=runtime)) stack.enter_context(mock.patch.object(supervisor, 'preflight_lifecycle_paths')) validator = stack.enter_context(mock.patch.object( supervisor, 'validate_managed_runtime_startup', return_value=SimpleNamespace(config=runtime[0], config_sha256='a' * 64), )) stack.enter_context(mock.patch.object( supervisor, 'supervisor_runtime_from_config', return_value=runtime, )) stack.enter_context(mock.patch.object(supervisor, 'runtime_authority', return_value={ 'config_sha256': 'a' * 64, 'supervisor_sha256': 'b' * 64, 'code_manifest': {}, 'code_manifest_sha256': 'c' * 64, })) return validator def test_runtime_without_flag_stops_before_locks_children_or_database_setup(self): args = supervisor_args() with ExitStack() as stack: self.main_setup(stack, args) load_env = stack.enter_context(mock.patch.object(supervisor, 'load_postgres_env')) database_url = stack.enter_context(mock.patch.object(supervisor, 'canonical_database_url')) cluster_lock = stack.enter_context(mock.patch.object(supervisor, 'ClusterAuthorityLock')) instance_lock = stack.enter_context(mock.patch.object(supervisor, 'SupervisorInstanceLock')) source = stack.enter_context(mock.patch.object(supervisor, 'ManagedSource')) keychecks = stack.enter_context(mock.patch.object(supervisor, 'ManagedKeychecks')) dashboard = stack.enter_context(mock.patch.object(supervisor, 'ManagedDashboard')) controller = stack.enter_context(mock.patch.object(supervisor, 'controller_from_config')) control_server = stack.enter_context(mock.patch.object(supervisor, 'start_control_server')) with self.assertRaisesRegex(SystemExit, 'unmanaged PostgreSQL mutation is retired.*--with-postgres'): supervisor.main() for blocked in ( load_env, database_url, cluster_lock, instance_lock, source, keychecks, dashboard, controller, control_server, ): blocked.assert_not_called() def test_background_without_flag_stops_before_spawning_runtime(self): args = supervisor_args(background=True) with ExitStack() as stack: self.main_setup(stack, args) start = stack.enter_context(mock.patch.object(supervisor, 'start_background')) with self.assertRaisesRegex(SystemExit, 'unmanaged PostgreSQL mutation is retired.*--with-postgres'): supervisor.main() start.assert_not_called() def test_production_background_with_flag_reaches_existing_launcher(self): args = supervisor_args(background=True, with_postgres=True) with ExitStack() as stack: validator = self.main_setup(stack, args) legacy_loader = supervisor.load_supervisor_runtime start = stack.enter_context(mock.patch.object(supervisor, 'start_background')) self.assertEqual(supervisor.main(), 0) validator.assert_called_once_with(str(ROOT / 'config.yaml')) legacy_loader.assert_not_called() start.assert_called_once() def test_managed_runtime_with_flag_reaches_existing_lifecycle_controller(self): args = supervisor_args(with_postgres=True) held_cluster_lock = mock.Mock() held_instance_lock = mock.Mock() managed_source = SimpleNamespace(enabled=True) with ExitStack() as stack: validator = self.main_setup(stack, args) load_env = stack.enter_context(mock.patch.object(supervisor, 'load_postgres_env')) stack.enter_context(mock.patch.object( supervisor, 'canonical_database_url', return_value='postgresql://truf:test@127.0.0.1:5432/truf', )) stack.enter_context(mock.patch.object( supervisor, 'background_paths', return_value=('instance', 'log', 'status', 'pid'), )) stack.enter_context(mock.patch.object(supervisor, 'background_lock_path', return_value='lock')) cluster_lock = stack.enter_context(mock.patch.object(supervisor, 'ClusterAuthorityLock')) cluster_lock.return_value.acquire.return_value = held_cluster_lock instance_lock = stack.enter_context(mock.patch.object(supervisor, 'SupervisorInstanceLock')) instance_lock.return_value.acquire.return_value = held_instance_lock stack.enter_context(mock.patch.object(supervisor.os.path, 'exists', return_value=False)) source = stack.enter_context(mock.patch.object(supervisor, 'ManagedSource', return_value=managed_source)) controller = stack.enter_context(mock.patch.object( supervisor, 'controller_from_config', side_effect=RuntimeError('managed controller sentinel'), )) self.assertEqual(supervisor.main(), 1) load_env.assert_called_once_with(str(ROOT / 'config.yaml'), {'queue_dir': 'queues'}, enforce_canonical=True) validator.assert_called_once_with(str(ROOT / 'config.yaml')) cluster_lock.assert_called_once_with( runtime_fixture()[0], endpoint_dsn='postgresql://truf:test@127.0.0.1:5432/truf', ) instance_lock.assert_called_once_with('instance', lock_path='lock') controller.assert_called_once_with(runtime_fixture()[0], runtime_fixture()[3]) source.assert_not_called() held_instance_lock.release.assert_called_once() held_cluster_lock.release.assert_called_once() def test_dry_run_without_flag_does_not_construct_lifecycle(self): args = supervisor_args(dry_run=True) dry_source = SimpleNamespace( enabled=True, source='fixture', build_command=lambda: ['python', 'scanner.py'], log_path='fixture.log', state_path='fixture.state', use_per_source_state=True, once=False, repeat=True, restart=True, interval=30, ) with ExitStack() as stack: validator = self.main_setup(stack, args) stack.enter_context(mock.patch.object(supervisor, 'ManagedSource', return_value=dry_source)) load_env = stack.enter_context(mock.patch.object(supervisor, 'load_postgres_env')) cluster_lock = stack.enter_context(mock.patch.object(supervisor, 'ClusterAuthorityLock')) instance_lock = stack.enter_context(mock.patch.object(supervisor, 'SupervisorInstanceLock')) controller = stack.enter_context(mock.patch.object(supervisor, 'controller_from_config')) dashboard = stack.enter_context(mock.patch.object(supervisor, 'ManagedDashboard')) self.assertEqual(supervisor.main(), 0) for lifecycle in (load_env, cluster_lock, instance_lock, controller, dashboard): lifecycle.assert_not_called() validator.assert_not_called() def test_remote_control_actions_without_flag_do_not_construct_lifecycle(self): cases = ( ({'stop_background': True}, 'stop_background'), ({'background_status': True}, 'background_status'), ({'attach': True}, 'attach_background'), ({'cmd': 'status'}, 'send_background_command'), ) for overrides, expected_call in cases: with self.subTest(action=expected_call), ExitStack() as stack: validator = self.main_setup(stack, supervisor_args(**overrides)) actions = { name: stack.enter_context(mock.patch.object(supervisor, name, return_value=True)) for name in ('stop_background', 'background_status', 'attach_background', 'send_background_command') } load_env = stack.enter_context(mock.patch.object(supervisor, 'load_postgres_env')) cluster_lock = stack.enter_context(mock.patch.object(supervisor, 'ClusterAuthorityLock')) instance_lock = stack.enter_context(mock.patch.object(supervisor, 'SupervisorInstanceLock')) controller = stack.enter_context(mock.patch.object(supervisor, 'controller_from_config')) self.assertEqual(supervisor.main(), 0) actions[expected_call].assert_called_once() for name, action in actions.items(): if name != expected_call: action.assert_not_called() for lifecycle in (load_env, cluster_lock, instance_lock, controller): lifecycle.assert_not_called() validator.assert_not_called() def test_validation_precedes_preflight_and_config_drift_fails_before_launch(self): args = supervisor_args(background=True, with_postgres=True) trace = [] with ExitStack() as stack: validator = self.main_setup(stack, args) supervisor.preflight_lifecycle_paths.side_effect = lambda *_args: trace.append('preflight') validator.side_effect = lambda *_args: ( trace.append('validate') or SimpleNamespace(config=runtime_fixture()[0], config_sha256='a' * 64) ) supervisor.sha256_file.side_effect = ['a' * 64, 'b' * 64] start = stack.enter_context(mock.patch.object(supervisor, 'start_background')) with self.assertRaisesRegex(SystemExit, 'changed while it was being validated'): supervisor.main() self.assertEqual(trace, ['validate']) start.assert_not_called() if __name__ == '__main__': unittest.main()