import ast import contextlib import json import ntpath import os from pathlib import Path import posixpath import signal import subprocess import sys import tempfile from types import SimpleNamespace import unittest from unittest import mock from urllib.parse import quote ROOT = Path(__file__).resolve().parents[1] sys.path.insert(0, str(ROOT / 'app')) import paths import postgres_runtime as runtime class NativePostgresPathTests(unittest.TestCase): def setUp(self): self.environment = {} self.os = SimpleNamespace( name='posix', path=posixpath, environ=self.environment, getenv=self.environment.get, fspath=os.fspath, getcwd=lambda: '/unrelated', ) self.enterContext(mock.patch.object(paths, 'os', self.os)) self.enterContext(mock.patch.object(runtime, 'os', self.os)) self.enterContext(mock.patch.object(runtime, 'IS_WINDOWS', False)) self.enterContext(mock.patch.object(runtime, 'canonical_path', side_effect=posixpath.normpath)) self.reject = self.enterContext(mock.patch.object(runtime, 'reject_reparse_components', side_effect=lambda path: path)) self.trust = self.enterContext(mock.patch.object(runtime, 'require_trusted_native_executable', side_effect=lambda path: path)) def test_optional_bin_dir_expands_and_preserves_old_fallback(self): config = paths.apply_path_config({'global': { 'root_dir': '/opt/truf', 'project_dir': '/opt/truf/app', 'runtime_dir': '/data/runtime-linux', 'postgres_data_dir': '/data/postgres-linux', }}) self.assertNotIn('postgres_bin_dir', config['global']) self.assertEqual(paths.resolve_postgres_bin_dir(config['global']), '/data/runtime-linux/postgres/pgsql/bin') config['global']['postgres_bin_dir'] = '{root_dir}/native/../native/bin' paths.apply_path_config(config) self.assertEqual(config['global']['postgres_bin_dir'], '/opt/truf/native/bin') config['global']['postgres_bin_dir'] = '/usr/lib/postgresql/16/bin' resolved = runtime.postgres_runtime_paths(config) for name in ('postgres', 'pg_ctl', 'pg_isready', 'pg_controldata', 'initdb', 'psql'): self.assertEqual(resolved[name], '/usr/lib/postgresql/16/bin/' + name) self.assertEqual(resolved['data_dir'], '/data/postgres-linux') self.assertEqual(resolved['identity_path'], '/data/runtime-linux/postgres/cluster_identity.json') def test_windows_fallback_and_executable_suffix_are_unchanged(self): self.os.name = 'nt' self.os.path = ntpath config = {'global': {'runtime_dir': r'E:\fixture\runtime'}} resolved = runtime.postgres_runtime_paths(config) self.assertEqual(resolved['pg_ctl'], r'E:\fixture\runtime\postgres\pgsql\bin\pg_ctl.exe') self.trust.assert_not_called() self.reject.assert_not_called() def test_executable_link_is_rejected_before_canonicalization(self): def reject(path): if path.endswith('/postgres') and '/bin/' in path: raise OSError('linked executable') return path self.reject.side_effect = reject with self.assertRaisesRegex(OSError, 'linked executable'): runtime.postgres_runtime_paths({'global': { 'runtime_dir': '/data/runtime-linux', 'postgres_bin_dir': '/usr/lib/postgresql/16/bin', }}) def test_every_native_utility_is_verified_before_execution_and_pg_environment_is_scrubbed(self): self.environment.update({'PGHOSTADDR': 'foreign', 'PGPASSWORD': 'secret', 'PATH': '/usr/bin'}) with mock.patch.object(runtime.subprocess, 'run') as run: for name in ('postgres', 'pg_ctl', 'pg_isready', 'pg_controldata', 'initdb', 'psql'): executable = '/usr/lib/postgresql/16/bin/' + name runtime._run_bounded([executable, '--version'], 1) self.trust.assert_called_with(executable) self.assertEqual(run.call_args.args[0][0], executable) self.assertEqual(run.call_args.kwargs['env'], {'PATH': '/usr/bin', 'LC_ALL': 'C', 'LANG': 'C'}) runtime._run_bounded(['/usr/lib/postgresql/16/bin/psql'], 1, environment_overrides={'PGPASSFILE': '/private/pass'}) self.assertEqual(run.call_args.kwargs['env']['PGPASSFILE'], '/private/pass') def test_untrusted_utility_never_executes(self): self.trust.side_effect = OSError('native file is writable') with mock.patch.object(runtime.subprocess, 'run') as run: with self.assertRaisesRegex(OSError, 'writable'): runtime._run_bounded(['/usr/lib/postgresql/16/bin/postgres', '--version'], 1) run.assert_not_called() def test_identity_verification_rejects_untrusted_native_code_even_with_matching_hashes(self): config = {'global': { 'runtime_dir': '/data/runtime-linux', 'postgres_data_dir': '/data/postgres-linux', 'postgres_bin_dir': '/usr/lib/postgresql/16/bin', }} resolved = runtime.postgres_runtime_paths(config) executable_keys = ('postgres', 'pg_ctl', 'pg_isready', 'pg_controldata') identity = { 'schema': 1, 'private_file_ready': True, 'data_directory': resolved['data_dir'], 'database': 'truf', 'user': 'truf', 'port': 5432, 'pg_major': 16, 'system_identifier': '12345', 'created_at': 'fixture', 'executables': {key: resolved[key] for key in executable_keys}, 'executable_sha256': {key: 'a' * 64 for key in executable_keys}, } self.trust.side_effect = OSError('native executable is writable') with mock.patch.object(runtime, 'read_private_json', return_value=identity), \ mock.patch.object(self.os.path, 'isfile', return_value=True), \ mock.patch.object(runtime, 'sha256_file', return_value='a' * 64), \ mock.patch.object(runtime, '_run_bounded') as run: with self.assertRaisesRegex(OSError, 'writable'): runtime.verify_cluster_identity(config, verify_offline_system_identifier=False) run.assert_not_called() def test_windows_utility_does_not_use_posix_trust_policy(self): self.os.name = 'nt' self.environment['PGHOST'] = 'legacy' with mock.patch.object(runtime.subprocess, 'run') as run: runtime._run_bounded(['pg_ctl.exe', 'status'], 1) self.trust.assert_not_called() self.assertEqual(run.call_args.kwargs['env']['PGHOST'], 'legacy') def test_normal_posix_start_explicitly_disables_unix_socket_defaults(self): backend = runtime.PostgresBackend.__new__(runtime.PostgresBackend) backend.paths = {'log_path': '/data/runtime-linux/postgres/postgres.log'} backend._verified_identity = { 'data_directory': '/data/postgres-linux', 'port': 5432, 'executables': {'pg_ctl': '/usr/lib/postgresql/16/bin/pg_ctl'}, } with mock.patch.object(backend, 'probe', return_value=runtime.ProbeResult(runtime.ProbeKind.STOPPED)), \ mock.patch.object(runtime, 'IS_WINDOWS', False), \ mock.patch.object(runtime, '_run_bounded', return_value=SimpleNamespace(returncode=0)) as run: self.assertTrue(backend.start().accepted) command = run.call_args.args[0] self.assertIn('unix_socket_directories=', command[command.index('-o') + 1]) class OwnedChild: def __init__(self, case, name, pid): self.case = case self.name = name self.pid = pid self.running = True self.wait_errors = [] self.returncode = 0 self.signals = [] def poll(self): return None if self.running else self.returncode def wait(self, timeout): self.case.assertTrue(self.case.lock_held) self.case.events.append(self.name + '-wait') if self.wait_errors: raise self.wait_errors.pop(0) self.running = False return self.returncode def send_signal(self, value): self.case.assertTrue(self.case.lock_held) self.case.assertEqual(value, signal.SIGINT) self.signals.append(value) class InitializeEmptyTests(unittest.TestCase): def setUp(self): self.temp = Path(self.enterContext(tempfile.TemporaryDirectory())) self.config = {'global': { 'root_dir': str(self.temp / 'application'), 'project_dir': str(self.temp / 'application' / 'app'), 'runtime_dir': str(self.temp / 'runtime'), 'postgres_data_dir': str(self.temp / 'postgres-linux'), 'postgres_bin_dir': str(self.temp / 'native' / 'bin'), }} self.paths = runtime.postgres_runtime_paths(self.config) for key in ('postgres_dir', 'data_dir'): Path(self.paths[key]).mkdir(parents=True) self.os = SimpleNamespace(**vars(os)) self.os.name = 'posix' self.os.geteuid = lambda: 10001 self.os.environ = {'PGHOSTADDR': 'must-not-inherit', 'PGPASSWORD': 'must-not-inherit'} self.enterContext(mock.patch.object(runtime, 'os', self.os)) self.enterContext(mock.patch.object(runtime, 'postgres_runtime_paths', return_value=self.paths)) self.private = self.enterContext(mock.patch.object(runtime, 'require_private_directory', side_effect=lambda path, **_: path)) self.enterContext(mock.patch.object(runtime, 'private_file_ready', return_value=True)) self.trust = self.enterContext(mock.patch.object(runtime, 'require_trusted_native_executable', side_effect=lambda path: path)) self.values = {'database': 'truf', 'user': 'truf', 'port': 55432} self.password = "fixture:'\\password" self.enterContext(mock.patch.object(runtime, 'configured_cluster_values', return_value=self.values)) self.dsn = f'postgresql://truf:{quote(self.password, safe="")}@127.0.0.1:55432/truf' self.enterContext(mock.patch.object(runtime, 'canonical_database_url', return_value=self.dsn)) self.listener = self.enterContext(mock.patch.object(runtime, '_listener_present', return_value=False)) self.port_socket = self.enterContext(mock.patch.object(runtime.socket, 'socket')) self.major = self.enterContext(mock.patch.object(runtime, '_bundled_postgres_major', return_value=16)) self.enterContext(mock.patch.object(runtime, '_system_identifier', return_value='12345')) self.lock_held = False self.events = [] @contextlib.contextmanager def authority(*_, **kwargs): self.assertEqual(kwargs['endpoint_dsn'], self.dsn) self.assertFalse(self.lock_held) self.lock_held = True self.events.append('lock') try: yield finally: self.lock_held = False self.events.append('unlock') self.lock = self.enterContext(mock.patch.object(runtime, 'ClusterAuthorityLock', side_effect=authority)) self.initdb = OwnedChild(self, 'initdb', 101) self.postgres = OwnedChild(self, 'postgres', 202) self.backend = mock.Mock() self.backend._prepare_logging.side_effect = lambda: Path(self.paths['log_path']).touch() self.backend._open_postmaster.return_value = mock.Mock(pid=self.postgres.pid) self.enterContext(mock.patch.object(runtime, 'PostgresBackend', return_value=self.backend)) self.commands = [] self.private_paths = [] def spawn(command, **kwargs): self.assertTrue(self.lock_held) self.assertFalse(any(key.startswith('PG') for key in kwargs['env'])) self.assertNotIn(self.password, repr(command)) self.commands.append(command) self.assertTrue(kwargs['start_new_session']) if command[0] == self.paths['initdb']: password_path = next(value.split('=', 1)[1] for value in command if value.startswith('--pwfile=')) self.assertEqual(Path(password_path).read_text(encoding='utf-8'), self.password + '\n') if os.name != 'nt': self.assertEqual(os.stat(password_path).st_mode & 0o777, 0o600) self.private_paths.append(password_path) Path(self.paths['data_dir'], 'PG_VERSION').write_text('16\n', encoding='ascii') for name in ('pg_hba.conf', 'pg_ident.conf'): auth_path = Path(self.paths['data_dir'], name) auth_path.write_text('# initdb fixture\n', encoding='ascii') os.chmod(auth_path, 0o600) if os.name != 'nt' and os.geteuid() == 0: os.chown(auth_path, 10001, 10001) return self.initdb self.assertEqual(command[0], self.paths['postgres']) self.assertFalse(self.initdb.running) return self.postgres self.spawn = self.enterContext(mock.patch.object(runtime.subprocess, 'Popen', side_effect=spawn)) self.sql = [] self.sql_failure = False self.control_state = 'shut down' def utility(command, **kwargs): self.assertTrue(self.lock_held) if command[0] == self.paths['pg_isready']: return SimpleNamespace(returncode=0, stdout='accepting') if command[0] == self.paths['pg_controldata']: self.assertFalse(self.postgres.running) return SimpleNamespace(returncode=0, stdout='Database cluster state: ' + self.control_state + '\n') self.assertEqual(command[0], self.paths['psql']) self.assertIn('-X', command) self.assertIn('-w', command) self.assertIn('--set=ON_ERROR_STOP=1', command) passfile = kwargs['environment_overrides']['PGPASSFILE'] self.assertTrue(Path(passfile).is_file()) escaped = self.password.replace('\\', '\\\\').replace(':', '\\:') self.assertTrue(Path(passfile).read_text(encoding='utf-8').endswith(':' + escaped + '\n')) self.private_paths.append(passfile) statement = command[command.index('-c') + 1] self.sql.append(statement) if 'json_build_object' in statement: return SimpleNamespace(returncode=0, stdout=json.dumps({ 'data_directory': self.paths['data_dir'], 'system_identifier': '12345', 'user': self.values['user'], 'port': self.values['port'], 'listen_addresses': '', 'unix_socket': True, })) return SimpleNamespace(returncode=1 if self.sql_failure else 0, stdout='fixture SQL failure' if self.sql_failure else '') self.run = self.enterContext(mock.patch.object(runtime, '_run_bounded', side_effect=utility)) self.identity = {'pg_major': 16, 'system_identifier': '12345'} def bind(config): self.assertTrue(self.lock_held) self.assertFalse(self.postgres.running) self.events.append('bind') Path(self.paths['identity_path']).write_text(json.dumps(self.identity), encoding='ascii') return self.identity self.bind = self.enterContext(mock.patch.object(runtime, 'bootstrap_cluster_identity', side_effect=bind)) self.enterContext(mock.patch.object(runtime.time, 'sleep')) def test_initialization_is_locked_private_socket_only_stopped_and_bound_once(self): self.assertEqual(runtime.initialize_empty(self.config), self.identity) self.assertEqual(self.events[-3:], ['postgres-wait', 'bind', 'unlock']) self.lock.assert_called_once_with(self.config, endpoint_dsn=self.dsn) self.assertEqual(self.postgres.signals, [signal.SIGINT]) self.assertEqual(self.initdb.signals, []) initdb, postgres = self.commands self.assertIn('--encoding=UTF8', initdb) self.assertIn('--locale=C', initdb) self.assertIn('--auth-local=scram-sha-256', initdb) self.assertIn('--auth-host=scram-sha-256', initdb) self.assertIn('--username=truf', initdb) self.assertNotIn('--no-sync', initdb) self.assertIn('listen_addresses=', postgres) self.assertIn('unix_socket_permissions=0700', postgres) self.assertTrue(any(item.startswith('unix_socket_directories="') for item in postgres)) self.assertTrue(any('CREATE DATABASE "truf" OWNER "truf"' in sql for sql in self.sql)) self.assertTrue(any('REVOKE CREATE ON SCHEMA public FROM PUBLIC' in sql for sql in self.sql)) if not runtime.IS_WINDOWS: hba = Path(self.paths['data_dir'], 'pg_hba.conf').read_text(encoding='ascii') ident = Path(self.paths['data_dir'], 'pg_ident.conf').read_text(encoding='ascii') self.assertTrue(hba.startswith( '# BEGIN TRUF HOST AGENT AUTHORITY\n' 'local truf truf peer map=truf_host_agent\n' '# END TRUF HOST AGENT AUTHORITY\n' )) self.assertIn('truf_host_agent root truf\n', ident) self.assertIn('truf_host_agent truf truf\n', ident) self.assertTrue(all(not Path(path).exists() for path in self.private_paths)) self.assertTrue(all(call.kwargs == {'create': False} for call in self.private.call_args_list)) with self.assertRaisesRegex(runtime.ClusterIdentityError, 'existing cluster identity'): runtime.initialize_empty(self.config) self.assertEqual(self.spawn.call_count, 2) self.assertEqual(self.bind.call_count, 1) @unittest.skipIf(os.name == 'nt', 'POSIX ownership checks required') def test_managed_peer_authority_is_exact_idempotent_and_conflict_closed(self): self.os.geteuid = os.geteuid hba = Path(self.paths['data_dir'], 'pg_hba.conf') ident = Path(self.paths['data_dir'], 'pg_ident.conf') for path in (hba, ident): path.write_text('# original\n', encoding='ascii') os.chmod(path, 0o600) runtime._configure_host_agent_peer_authority(self.paths, self.values) first = (hba.read_bytes(), ident.read_bytes()) runtime._configure_host_agent_peer_authority(self.paths, self.values) self.assertEqual((hba.read_bytes(), ident.read_bytes()), first) hba.write_text( '# BEGIN TRUF HOST AGENT AUTHORITY\nconflict\n' '# END TRUF HOST AGENT AUTHORITY\n', encoding='ascii', ) with self.assertRaisesRegex(runtime.ClusterIdentityError, 'conflicts'): runtime._configure_host_agent_peer_authority(self.paths, self.values) def test_sql_identifiers_are_quoted_not_interpolated_as_commands(self): self.values.update(user='owner"quoted', database='database"quoted') runtime.initialize_empty(self.config) self.assertTrue(any('CREATE DATABASE "database""quoted" OWNER "owner""quoted"' in sql for sql in self.sql)) self.assertTrue(any('ALTER ROLE "owner""quoted"' in sql for sql in self.sql)) def test_existing_identity_refuses_before_any_pg_tool(self): Path(self.paths['identity_path']).write_text('untouched', encoding='ascii') with self.assertRaisesRegex(runtime.ClusterIdentityError, 'existing cluster identity'): runtime.initialize_empty(self.config) self.assertEqual(Path(self.paths['identity_path']).read_text(), 'untouched') self.spawn.assert_not_called() self.run.assert_not_called() self.bind.assert_not_called() def test_nonempty_pgdata_is_not_adopted_or_repaired(self): Path(self.paths['data_dir'], 'PG_VERSION').write_text('16\n', encoding='ascii') with self.assertRaisesRegex(runtime.ClusterIdentityError, 'nonempty PGDATA'): runtime.initialize_empty(self.config) self.spawn.assert_not_called() self.run.assert_not_called() self.bind.assert_not_called() def test_postmaster_and_all_supervisor_metadata_refuse_initialization(self): runtime_dir = Path(self.config['global']['runtime_dir']) for path in ( Path(self.paths['data_dir'], 'postmaster.pid'), runtime_dir / 'control' / 'supervisor.instance.json', runtime_dir / 'logs' / 'supervisor.instance.json', runtime_dir / 'logs' / 'supervisor.pid', ): with self.subTest(path=path): path.parent.mkdir(parents=True, exist_ok=True) path.write_text('untouched', encoding='ascii') with self.assertRaisesRegex(runtime.ClusterIdentityError, 'refusing bootstrap'): runtime.initialize_empty(self.config) self.assertEqual(path.read_text(), 'untouched') path.unlink() self.spawn.assert_not_called() self.bind.assert_not_called() def test_occupied_listener_or_bound_tcp_port_refuses_initialization(self): self.listener.return_value = True with self.assertRaisesRegex(runtime.ClusterIdentityError, 'listener'): runtime.initialize_empty(self.config) self.listener.return_value = False self.port_socket.return_value.__enter__.return_value.bind.side_effect = OSError('address in use') with self.assertRaisesRegex(runtime.ClusterIdentityError, 'occupied or unavailable'): runtime.initialize_empty(self.config) self.spawn.assert_not_called() def test_root_and_windows_are_rejected_before_lock_or_tool(self): self.os.geteuid = lambda: 0 with self.assertRaisesRegex(runtime.ClusterIdentityError, 'non-root POSIX'): runtime.initialize_empty(self.config) self.os.geteuid = lambda: 10001 self.os.name = 'nt' with self.assertRaisesRegex(runtime.ClusterIdentityError, 'non-root POSIX'): runtime.initialize_empty(self.config) self.lock.assert_not_called() self.spawn.assert_not_called() def test_explicit_independent_prepared_data_directory_is_required(self): del self.config['global']['postgres_data_dir'] with self.assertRaisesRegex(runtime.ClusterIdentityError, 'explicit resolved absolute'): runtime.initialize_empty(self.config) self.config['global']['postgres_data_dir'] = self.config['global']['runtime_dir'] self.paths['data_dir'] = runtime.canonical_path(self.config['global']['runtime_dir']) with self.assertRaisesRegex(runtime.ClusterIdentityError, 'independent'): runtime.initialize_empty(self.config) self.lock.assert_not_called() self.spawn.assert_not_called() def test_private_directory_failure_never_repairs_or_spawns(self): self.private.side_effect = OSError('private directory absent or unsafe') with self.assertRaisesRegex(OSError, 'absent or unsafe'): runtime.initialize_empty(self.config) self.lock.assert_not_called() self.spawn.assert_not_called() def test_lock_conflict_or_untrusted_tool_precedes_initialization(self): self.lock.side_effect = BlockingIOError('owned by another maintenance process') with self.assertRaises(BlockingIOError): runtime.initialize_empty(self.config) self.spawn.assert_not_called() def test_untrusted_tools_and_wrong_major_are_rejected(self): self.trust.side_effect = OSError('native executable untrusted') with self.assertRaisesRegex(OSError, 'untrusted'): runtime.initialize_empty(self.config) self.trust.side_effect = lambda path: path self.major.return_value = 17 with self.assertRaisesRegex(runtime.ClusterIdentityError, 'PostgreSQL 16'): runtime.initialize_empty(self.config) self.spawn.assert_not_called() def test_initdb_timeout_retains_owner_without_killing_or_proceeding(self): self.initdb.wait_errors = [subprocess.TimeoutExpired('initdb', 120)] with self.assertRaises(subprocess.TimeoutExpired): runtime.initialize_empty(self.config) self.assertEqual(self.events, ['lock', 'initdb-wait', 'initdb-wait', 'unlock']) self.assertEqual(self.initdb.signals, []) self.assertEqual(self.spawn.call_count, 1) self.bind.assert_not_called() def test_sql_failure_compensates_and_retains_lock_and_socket_until_exit_confirmed(self): self.sql_failure = True self.postgres.wait_errors = [subprocess.TimeoutExpired('postgres', 60)] def held_sleep(_): self.assertTrue(self.lock_held) self.assertTrue(self.postgres.running) self.assertTrue(all(Path(path).exists() for path in self.private_paths)) with mock.patch.object(runtime.time, 'sleep', side_effect=held_sleep), \ mock.patch('builtins.print') as output: with self.assertRaisesRegex(runtime.ClusterIdentityError, 'SQL setup failed'): runtime.initialize_empty(self.config) self.assertEqual(self.events[-3:], ['postgres-wait', 'postgres-wait', 'unlock']) self.assertTrue(any('FAILED_HOLD' in call.args[0] for call in output.call_args_list)) self.assertFalse(self.postgres.running) self.bind.assert_not_called() self.assertFalse(Path(self.paths['identity_path']).exists()) def test_mismatched_postmaster_never_receives_a_guessed_pid_stop(self): self.backend._open_postmaster.return_value.pid = 999 with self.assertRaisesRegex(runtime.ClusterIdentityError, 'directly owned child'): runtime.initialize_empty(self.config) self.assertEqual(self.postgres.signals, [signal.SIGINT]) self.assertFalse(self.sql) self.bind.assert_not_called() def test_unclean_control_state_does_not_bind_identity(self): self.control_state = 'in production' with self.assertRaisesRegex(runtime.ClusterIdentityError, 'cleanly stopped'): runtime.initialize_empty(self.config) self.bind.assert_not_called() class InitializationEntrypointTests(unittest.TestCase): def test_cli_dispatches_to_lock_owning_initializer_after_preflight(self): config = {'global': {}} with mock.patch.object(sys, 'argv', ['postgres_runtime.py', 'initialize-empty', '--config', 'fixture.yaml']), \ mock.patch.object(runtime, '_load_config', return_value=config), \ mock.patch.object(runtime, 'preflight_lifecycle_paths') as preflight, \ mock.patch.object(runtime, 'load_postgres_environment'), \ mock.patch.object(runtime, 'canonical_database_url', return_value='fixture-dsn'), \ mock.patch.object(runtime, 'initialize_empty', return_value={'pg_major': 16, 'system_identifier': '12345'}) as initialize, \ mock.patch.object(runtime, 'postgres_runtime_paths', return_value={'identity_path': 'fixture-identity'}), \ mock.patch.object(runtime, 'ClusterAuthorityLock') as outer_lock, \ mock.patch('builtins.print'): runtime.main() preflight.assert_called_once() initialize.assert_called_once_with(config) outer_lock.assert_not_called() def test_container_refusal_and_full_layout_gate_precede_application_imports(self): tree = ast.parse((ROOT / 'app' / 'postgres_runtime.py').read_text(encoding='utf-8')) expected = ast.parse( "import sys\nimport os\nif __name__ == '__main__':\n" " if sys.platform != 'linux' or not os.path.isfile('/.dockerenv') " "or os.path.abspath(__file__) != '/opt/truf/app/postgres_runtime.py':\n" " raise SystemExit('Docker development copy: runtime control is disabled outside the prepared container. See DOCKER_MIGRATION.md.')\n" ' import runpy\n' " runpy.run_path('/opt/truf/app/container_runtime.py')['require_container']()\n" ).body self.assertEqual([ast.dump(node) for node in tree.body[:3]], [ast.dump(node) for node in expected]) def test_bounded_maintenance_stop_requires_proof_and_never_closes_a_supplied_backend(self): for completed, stopped in ((False, False), (True, False), (False, True), (1, 1), (True, True)): backend = mock.Mock() backend.probe.return_value = runtime.ProbeResult(runtime.ProbeKind.READY) backend.stop.return_value = runtime.StopResult(completed, stopped) with self.subTest(completed=completed, stopped=stopped): if completed is True and stopped is True: self.assertEqual(runtime.maintenance_stop({}, backend), backend.stop.return_value) else: with self.assertRaisesRegex(runtime.ClusterIdentityError, 'stop did not complete'): runtime.maintenance_stop({}, backend) backend.stop.assert_called_once_with() backend.probe.assert_not_called() backend.close.assert_not_called() def test_bootstrap_cannot_replace_an_existing_identity(self): with tempfile.TemporaryDirectory() as directory: identity_path = Path(directory, 'identity.json') identity_path.write_text('original', encoding='ascii') with mock.patch.object(runtime, 'postgres_runtime_paths', return_value={'identity_path': str(identity_path)}), \ mock.patch.object(runtime, '_run_bounded') as run, \ mock.patch.object(runtime, 'write_private_json_exclusive') as publish: with self.assertRaisesRegex(runtime.ClusterIdentityError, 'existing cluster identity'): runtime.bootstrap_cluster_identity({}) self.assertEqual(identity_path.read_text(), 'original') run.assert_not_called() publish.assert_not_called() class MaintenanceEntrypointTests(unittest.TestCase): def setUp(self): self.config = {'global': {'runtime_dir': 'fixture-runtime'}} self.dsn = 'postgresql://fixture:fixture@127.0.0.1:55432/fixture' self.events = [] self.lock_held = False self.closed = False self.stopped = False self.ready = False self.pg_live = True self.launch_owned = True self.start_error = None self.start_result = runtime.StartResult(True) self.probes = [] self.stops = [] self.sleep_errors = [] self.close_errors = [] self.print_errors = [] self.cancel_when_locked = False self.backend = mock.Mock(spec=runtime.PostgresBackend) self.backend.owns_start = False self.backend.start_settle_timeout_sec = 0.1 self.retained = object() self.backend._expected_process = self.retained def start(): self.events.append(('start', self.lock_held, self.closed)) if self.launch_owned: self.backend.owns_start = True self.pg_live = True if self.start_error is not None: raise self.start_error return self.start_result def probe(): value = self.probes.pop(0) if self.probes else runtime.ProbeResult(runtime.ProbeKind.READY) if callable(value): value = value() if isinstance(value, BaseException): raise value if value.kind == runtime.ProbeKind.READY: self.ready = True self.events.append(('ready', self.lock_held, self.closed)) return value def stop(): self.events.append(('stop', self.lock_held, self.closed, self.backend._expected_process is self.retained)) value = self.stops.pop(0) if self.stops else runtime.StopResult(True, True, 'verified stopped') if callable(value): value = value() if isinstance(value, BaseException): raise value if isinstance(value, runtime.StopResult) and value.completed is True and value.stopped is True: self.stopped = True self.pg_live = False self.backend.owns_start = False self.events.append(('proof', self.lock_held, self.closed)) return value def close(): self.events.append(('close', self.lock_held, self.stopped, self.ready)) if self.close_errors: raise self.close_errors.pop(0) self.closed = True self.backend.start.side_effect = start self.backend.probe.side_effect = probe self.backend.stop.side_effect = stop self.backend.close.side_effect = close self.construct = self.enterContext(mock.patch.object(runtime, 'PostgresBackend', return_value=self.backend)) self.enterContext(mock.patch.object(runtime, '_load_config', return_value=self.config)) self.enterContext(mock.patch.object(runtime, 'preflight_lifecycle_paths')) self.enterContext(mock.patch.object(runtime, 'load_postgres_environment')) self.enterContext(mock.patch.object(runtime, 'canonical_database_url', return_value=self.dsn)) self.verify = self.enterContext(mock.patch.object(runtime, 'verify_cluster_identity', side_effect=AssertionError('verification belongs inside the owned backend'))) self.lookup = self.enterContext(mock.patch.object(runtime, 'open_process')) self.native = self.enterContext(mock.patch.object(runtime, '_run_bounded')) self.kill = self.enterContext(mock.patch.object(runtime.os, 'kill')) self.original_handlers = {signal.SIGINT: object(), signal.SIGTERM: object()} self.handlers = dict(self.original_handlers) def set_handler(signum, handler): old = self.handlers[signum] self.handlers[signum] = handler phase = 'restore-signal' if handler is self.original_handlers[signum] else 'install-signal' self.events.append((phase, signum, self.lock_held)) return old self.enterContext(mock.patch.object(runtime.signal, 'signal', side_effect=set_handler)) @contextlib.contextmanager def authority(*args, **kwargs): self.events.append(('lock',)) self.lock_held = True if self.cancel_when_locked: self.handlers[signal.SIGTERM](signal.SIGTERM, None) try: yield finally: self.events.append(('unlock', self.stopped, self.closed, self.ready)) self.lock_held = False self.authority = authority self.lock = self.enterContext(mock.patch.object(runtime, 'ClusterAuthorityLock', side_effect=authority)) def sleep(_): self.events.append(('sleep', self.lock_held, self.closed)) if self.sleep_errors: raise self.sleep_errors.pop(0) self.enterContext(mock.patch.object(runtime.time, 'sleep', side_effect=sleep)) def output(message, **kwargs): self.events.append(('output', message)) if 'FAILED_HOLD' in message and self.print_errors: raise self.print_errors.pop(0) self.output = self.enterContext(mock.patch('builtins.print', side_effect=output)) def run_cli(self, action): if action == 'maintenance-start': self.pg_live = False with mock.patch.object(sys, 'argv', ['postgres_runtime.py', action, '--config', 'fixture.yaml']): return runtime.main() def assert_stopped_before_release(self): self.assertTrue(self.stopped) self.assertFalse(self.pg_live) self.assertTrue(self.closed) self.assertFalse(self.lock_held) self.construct.assert_called_once_with(self.config) self.lock.assert_called_once_with(self.config, endpoint_dsn=self.dsn) self.backend.close.assert_called_once_with() self.assertTrue(all(event[1:3] == (True, True) for event in self.events if event[0] == 'close')) self.assertTrue(all(event[1:] == (True, False, True) for event in self.events if event[0] == 'stop')) phases = [event[0] for event in self.events] self.assertLess(phases.index('proof'), phases.index('close')) self.assertLess(phases.index('close'), phases.index('unlock')) self.assertLess(phases.index('unlock'), phases.index('restore-signal')) self.assertEqual(self.handlers, self.original_handlers) self.verify.assert_not_called() self.lookup.assert_not_called() self.native.assert_not_called() self.kill.assert_not_called() def test_stop_retains_same_backend_and_locks_through_repeated_failures_and_interrupts(self): self.stops = [ runtime.StopResult(False, False, 'first stop failure'), KeyboardInterrupt(), SystemExit(0), runtime.StopResult(True, False, 'still live'), runtime.StopResult(True, True, 'verified stopped'), ] self.sleep_errors = [KeyboardInterrupt(), SystemExit(0)] self.print_errors = [KeyboardInterrupt(), SystemExit(0)] with self.assertRaisesRegex(SystemExit, 'first stop failure') as caught: self.run_cli('maintenance-stop') self.assertNotEqual(caught.exception.code, 0) self.assertEqual(self.backend.stop.call_count, 5) self.assertTrue(any(event[0] == 'output' and 'FAILED_HOLD' in event[1] for event in self.events)) self.assert_stopped_before_release() def test_system_exit_zero_is_sticky_failure_not_a_successful_cleanup_exit(self): self.stops = [SystemExit(0), KeyboardInterrupt(), SystemExit(0)] with self.assertRaisesRegex(SystemExit, 'SystemExit: 0') as caught: self.run_cli('maintenance-stop') self.assertNotEqual(caught.exception.code, 0) self.assertEqual(self.backend.stop.call_count, 4) self.assert_stopped_before_release() def test_truthy_or_incomplete_stop_results_are_not_stop_proof(self): self.stops = [ SimpleNamespace(completed=True, stopped=True), runtime.StopResult(False, True), runtime.StopResult(True, False), runtime.StopResult(1, 1), ] with self.assertRaisesRegex(SystemExit, 'stop did not complete'): self.run_cli('maintenance-stop') self.assertEqual(self.backend.stop.call_count, 5) self.assert_stopped_before_release() def test_stop_identity_failure_remains_inert_and_retained_until_proof(self): self.stops = [ runtime.StopResult(False, False, 'foreign identity; stop refused'), OSError('identity temporarily unavailable'), ] with self.assertRaisesRegex(SystemExit, 'foreign identity'): self.run_cli('maintenance-stop') self.assert_stopped_before_release() def test_uncertain_start_compensates_before_sticky_failure_can_release_locks(self): self.start_result = runtime.StartResult(True, 'launch outcome uncertain', uncertain=True) self.probes = [ runtime.ProbeResult(runtime.ProbeKind.RECOVERING), runtime.ProbeResult(runtime.ProbeKind.OWNED_START_UNCERTAIN, 'first start failure'), ] self.stops = [runtime.StopResult(False, False, 'not stopped'), KeyboardInterrupt(), SystemExit(0)] with self.assertRaisesRegex(SystemExit, 'first start failure'): self.run_cli('maintenance-start') self.backend.start.assert_called_once_with() self.assertEqual(self.backend.stop.call_count, 4) self.assert_stopped_before_release() def test_interrupt_during_start_is_compensated_on_the_exact_owned_backend(self): self.start_error = KeyboardInterrupt() self.stops = [SystemExit(0), runtime.StopResult(False, False, 'uncertain')] with self.assertRaisesRegex(SystemExit, 'KeyboardInterrupt'): self.run_cli('maintenance-start') self.backend.start.assert_called_once_with() self.assert_stopped_before_release() def test_rejected_but_possibly_launched_start_still_requires_compensation(self): self.start_result = runtime.StartResult(False, 'start result was lost') self.stops = [runtime.StopResult(False, False)] with self.assertRaisesRegex(SystemExit, 'start result was lost'): self.run_cli('maintenance-start') self.assert_stopped_before_release() def test_lost_stop_result_cannot_clear_the_latched_compensation_requirement(self): self.probes = [runtime.ProbeResult(runtime.ProbeKind.OWNED_START_UNCERTAIN, 'start failed')] def lost_stop_result(): self.backend.owns_start = False raise KeyboardInterrupt() self.stops = [lost_stop_result, runtime.StopResult(False, False, 'still unconfirmed')] with self.assertRaisesRegex(SystemExit, 'start failed'): self.run_cli('maintenance-start') self.assertEqual(self.backend.stop.call_count, 3) self.assert_stopped_before_release() def test_prelaunch_foreign_rejection_never_adopts_or_stops_that_cluster(self): self.launch_owned = False self.start_result = runtime.StartResult(False, 'foreign before launch', foreign_or_config_error=True) with self.assertRaisesRegex(SystemExit, 'foreign before launch'): self.run_cli('maintenance-start') self.backend.stop.assert_not_called() self.backend.close.assert_called_once_with() self.assertFalse(self.backend.owns_start) self.assertFalse(self.lock_held) self.lookup.assert_not_called() self.native.assert_not_called() self.kill.assert_not_called() def test_successful_start_releases_for_migration_without_stopping_ready_postgres(self): self.probes = [runtime.ProbeResult(runtime.ProbeKind.RECOVERING), runtime.ProbeResult(runtime.ProbeKind.READY)] self.assertIsNone(self.run_cli('maintenance-start')) self.assertTrue(self.ready) self.assertTrue(self.pg_live) self.assertFalse(self.stopped) self.assertFalse(self.lock_held) self.construct.assert_called_once_with(self.config) self.backend.start.assert_called_once_with() self.backend.stop.assert_not_called() self.backend.close.assert_called_once_with() self.assertEqual(self.handlers, self.original_handlers) def test_sigterm_after_ready_is_flag_only_and_compensates_before_release(self): def ready_then_cancel(): self.assertIsNone(self.handlers[signal.SIGTERM](signal.SIGTERM, None)) self.assertIsNone(self.handlers[signal.SIGTERM](signal.SIGTERM, None)) return runtime.ProbeResult(runtime.ProbeKind.READY) self.probes = [ready_then_cancel] self.stops = [runtime.StopResult(False, False), KeyboardInterrupt()] with self.assertRaisesRegex(SystemExit, 'shutdown requested'): self.run_cli('maintenance-start') self.assert_stopped_before_release() def test_sigterm_during_stop_cannot_interrupt_verified_cleanup(self): def cancelled_stop(): self.assertIsNone(self.handlers[signal.SIGTERM](signal.SIGTERM, None)) self.assertIsNone(self.handlers[signal.SIGINT](signal.SIGINT, None)) return runtime.StopResult(True, True, 'stopped after cancellation') self.stops = [cancelled_stop] with self.assertRaisesRegex(SystemExit, 'shutdown requested'): self.run_cli('maintenance-stop') self.backend.stop.assert_called_once_with() self.assert_stopped_before_release() def test_cancel_before_launch_does_not_start_or_adopt_a_postmaster(self): self.cancel_when_locked = True with self.assertRaisesRegex(SystemExit, 'shutdown requested'): self.run_cli('maintenance-start') self.backend.start.assert_not_called() self.backend.stop.assert_not_called() self.backend.close.assert_called_once_with() self.assertFalse(self.lock_held) def test_status_interrupt_after_readiness_still_compensates(self): def output(message, **kwargs): if 'authenticated-ready' in message: raise SystemExit(0) self.output.side_effect = output self.stops = [KeyboardInterrupt(), runtime.StopResult(False, False)] with self.assertRaisesRegex(SystemExit, 'SystemExit: 0'): self.run_cli('maintenance-start') self.assert_stopped_before_release() def test_failed_ready_handoff_close_requires_compensation_before_failure_return(self): self.close_errors = [SystemExit(0)] self.stops = [runtime.StopResult(False, False), KeyboardInterrupt()] with self.assertRaisesRegex(SystemExit, 'SystemExit: 0'): self.run_cli('maintenance-start') self.assertTrue(self.stopped) self.assertFalse(self.pg_live) self.assertEqual(self.backend.close.call_count, 2) close_events = [event for event in self.events if event[0] == 'close'] self.assertEqual(close_events, [('close', True, False, True), ('close', True, True, True)]) self.construct.assert_called_once_with(self.config) self.lock.assert_called_once_with(self.config, endpoint_dsn=self.dsn) def test_default_helper_owns_backend_until_failed_stop_is_confirmed(self): self.stops = [runtime.StopResult(False, False, 'initial stop failed'), KeyboardInterrupt()] with self.authority(), self.assertRaisesRegex(runtime.ClusterIdentityError, 'initial stop failed'): runtime.maintenance_stop(self.config) self.assertTrue(self.stopped) self.assertFalse(self.pg_live) self.construct.assert_called_once_with(self.config) self.backend.close.assert_called_once_with() self.assertTrue(all(event[1:3] == (True, True) for event in self.events if event[0] == 'close')) def test_default_start_helper_compensates_failure_before_closing(self): self.probes = [runtime.ProbeResult(runtime.ProbeKind.OWNED_START_UNCERTAIN, 'uncertain launch')] self.stops = [KeyboardInterrupt(), SystemExit(0)] with self.authority(), self.assertRaisesRegex(runtime.ClusterIdentityError, 'uncertain launch'): runtime.maintenance_start(self.config) self.assertTrue(self.stopped) self.assertFalse(self.pg_live) self.construct.assert_called_once_with(self.config) self.backend.close.assert_called_once_with() self.assertTrue(all(event[1:3] == (True, True) for event in self.events if event[0] == 'close')) if __name__ == '__main__': unittest.main()