sashatrask
882 lines
46 KiB
Python
882 lines
46 KiB
Python
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()
|