from concurrent.futures import Future import os from pathlib import Path import sys import tempfile import unittest ROOT = Path(__file__).resolve().parents[1] APP_DIR = ROOT / 'app' sys.path.insert(0, str(APP_DIR)) import postgres_runtime from runtime_security import ensure_private_directory, private_file_ready class ImmediateExecutor: def submit(self, function): future = Future() try: future.set_result(function()) except BaseException as exc: future.set_exception(exc) return future class ControlledStartExecutor: def __init__(self): self.submissions = [] def submit(self, function): future = Future() self.submissions.append((function.__name__, future)) if function.__name__ == 'stop': try: future.set_result(function()) except BaseException as exc: future.set_exception(exc) return future class ScriptedBackend: def __init__(self, probes, starts=()): self.probes = list(probes) self.starts = list(starts) self.start_calls = 0 self.stop_calls = 0 self.closed = False def probe(self): return self.probes.pop(0) def start(self): self.start_calls += 1 return self.starts.pop(0) def stop(self): self.stop_calls += 1 return postgres_runtime.StopResult(True, True, 'stopped') def close(self): self.closed = True @unittest.skipUnless(os.name == 'nt', 'Windows ACL inheritance semantics required') class PostgresCollectorAclTests(unittest.TestCase): def make_logging_backend(self, postgres_dir): backend = postgres_runtime.PostgresBackend.__new__(postgres_runtime.PostgresBackend) backend.paths = { 'log_dir': os.path.join(postgres_dir, 'logs'), 'log_path': os.path.join(postgres_dir, 'postgres.log'), } backend.log_max_mb = 1 backend.log_keep = 2 return backend def test_prepare_hardens_recreated_startup_log_with_inherited_acl(self): with tempfile.TemporaryDirectory() as temp_dir: postgres_dir = os.path.join(temp_dir, 'postgres') ensure_private_directory(postgres_dir, reject_reparse=True) backend = self.make_logging_backend(postgres_dir) backend._prepare_logging() path = backend.paths['log_path'] os.remove(path) Path(path).write_text('pg_ctl startup output', encoding='ascii') self.assertFalse(private_file_ready(path)) backend._prepare_logging() self.assertTrue(private_file_ready(path)) def test_prepare_rejects_startup_log_hard_link_without_hardening_target(self): with tempfile.TemporaryDirectory() as temp_dir: postgres_dir = os.path.join(temp_dir, 'postgres') ensure_private_directory(postgres_dir, reject_reparse=True) backend = self.make_logging_backend(postgres_dir) target = os.path.join(temp_dir, 'external.log') Path(target).write_text('external', encoding='ascii') os.link(target, backend.paths['log_path']) with self.assertRaisesRegex(postgres_runtime.ClusterIdentityError, 'unsafe'): backend._prepare_logging() self.assertFalse(private_file_ready(target)) self.assertEqual(Path(target).read_text(encoding='ascii'), 'external') def test_prune_hardens_normally_created_collector_file(self): with tempfile.TemporaryDirectory() as temp_dir: log_dir = os.path.join(temp_dir, 'collector') ensure_private_directory(log_dir, reject_reparse=True) path = os.path.join(log_dir, 'postgresql-20260719-123456.log') Path(path).write_text('active log', encoding='ascii') self.assertFalse(private_file_ready(path)) self.assertEqual(postgres_runtime.prune_postgres_collector_logs(log_dir, 2), 0) self.assertTrue(private_file_ready(path)) def test_prune_rejects_collector_hard_link_without_hardening_target(self): with tempfile.TemporaryDirectory() as temp_dir: log_dir = os.path.join(temp_dir, 'collector') ensure_private_directory(log_dir, reject_reparse=True) target = os.path.join(temp_dir, 'external.log') Path(target).write_text('external', encoding='ascii') link = os.path.join(log_dir, 'postgresql-20260719-123456.log') os.link(target, link) with self.assertRaisesRegex(postgres_runtime.ClusterIdentityError, 'unsafe'): postgres_runtime.prune_postgres_collector_logs(log_dir, 2) self.assertFalse(private_file_ready(target)) self.assertEqual(Path(target).read_text(encoding='ascii'), 'external') class PostgresOwnedLifecycleTests(unittest.TestCase): def make_controller(self, backend): return postgres_runtime.PostgresController( backend, executor=ImmediateExecutor(), health_interval_sec=1, stable_ready_interval_sec=0, ) def pending_start(self): backend = ScriptedBackend([]) executor = ControlledStartExecutor() controller = postgres_runtime.PostgresController(backend, executor=executor) controller.tick(0) executor.submissions[0][1].set_result( postgres_runtime.ProbeResult(postgres_runtime.ProbeKind.STOPPED, 'initially offline') ) controller.tick(0) self.assertEqual([name for name, _ in executor.submissions], ['probe', 'start']) controller.request_stop() return controller, backend, executor def test_shutdown_after_rejected_stale_start_closes_without_stop(self): results = ( postgres_runtime.StartResult(False, 'start refused while cluster state is READY'), postgres_runtime.StartResult( False, 'PostgreSQL logging setup failed while cluster remained offline', foreign_or_config_error=True, ), ) for result in results: with self.subTest(detail=result.detail): controller, backend, executor = self.pending_start() executor.submissions[1][1].set_result(result) controller.tick(0) self.assertEqual([name for name, _ in executor.submissions], ['probe', 'start']) self.assertEqual(backend.stop_calls, 0) self.assertTrue(controller.terminal) self.assertTrue(controller.authority_release_safe) self.assertFalse(controller.lifecycle_action_required) self.assertIn('no controller-owned side effect', controller.detail) self.assertTrue(controller.close(timeout_sec=0.1)) self.assertTrue(backend.closed) def test_shutdown_after_accepted_or_uncertain_stale_start_compensates_once(self): results = ( postgres_runtime.StartResult(True, 'accepted'), postgres_runtime.StartResult(True, 'outcome uncertain', uncertain=True), ) for result in results: with self.subTest(detail=result.detail): controller, backend, executor = self.pending_start() executor.submissions[1][1].set_result(result) controller.tick(0) self.assertEqual([name for name, _ in executor.submissions], ['probe', 'start', 'stop']) self.assertEqual(backend.stop_calls, 1) self.assertFalse(controller.authority_release_safe) controller.tick(0) self.assertEqual(controller.state, postgres_runtime.PostgresState.STOPPED) self.assertTrue(controller.authority_release_safe) def test_shutdown_after_stale_start_exception_fails_closed_with_stop(self): controller, backend, executor = self.pending_start() executor.submissions[1][1].set_exception(RuntimeError('uncertain worker failure')) controller.tick(0) self.assertEqual([name for name, _ in executor.submissions], ['probe', 'start', 'stop']) self.assertEqual(backend.stop_calls, 1) self.assertFalse(controller.authority_release_safe) controller.tick(0) self.assertTrue(controller.authority_release_safe) def test_owned_ready_recovery_death_backs_off_and_starts_again(self): backend = ScriptedBackend( probes=[ postgres_runtime.ProbeResult(postgres_runtime.ProbeKind.STOPPED, 'initially stopped'), postgres_runtime.ProbeResult(postgres_runtime.ProbeKind.READY, 'owned ready', 'epoch-1'), postgres_runtime.ProbeResult(postgres_runtime.ProbeKind.RECOVERING, 'owned recovery', 'epoch-1'), postgres_runtime.ProbeResult(postgres_runtime.ProbeKind.STOPPED, 'owned cluster stopped'), postgres_runtime.ProbeResult(postgres_runtime.ProbeKind.STOPPED, 'backoff verification'), postgres_runtime.ProbeResult(postgres_runtime.ProbeKind.READY, 'restarted', 'epoch-2'), ], starts=[ postgres_runtime.StartResult(True, 'first start accepted'), postgres_runtime.StartResult(True, 'second start accepted'), ], ) controller = self.make_controller(backend) for now in (0, 0, 0, 0): controller.tick(now) self.assertEqual(controller.state, postgres_runtime.PostgresState.READY) self.assertTrue(controller._owned_start) for now in (1, 1, 2, 2): controller.tick(now) self.assertEqual(controller.state, postgres_runtime.PostgresState.BACKOFF) self.assertFalse(controller._owned_start) for now in (32, 32, 32, 32): controller.tick(now) self.assertEqual(controller.state, postgres_runtime.PostgresState.READY) self.assertEqual(backend.start_calls, 2) self.assertEqual(backend.stop_calls, 0) def test_preexisting_cluster_remains_inert_after_ready_and_recovery(self): backend = ScriptedBackend(probes=[ postgres_runtime.ProbeResult(postgres_runtime.ProbeKind.READY, 'preexisting ready', 'epoch-1'), postgres_runtime.ProbeResult(postgres_runtime.ProbeKind.RECOVERING, 'preexisting recovery', 'epoch-1'), postgres_runtime.ProbeResult(postgres_runtime.ProbeKind.STOPPED, 'preexisting cluster stopped'), postgres_runtime.ProbeResult(postgres_runtime.ProbeKind.STOPPED, 'still stopped'), ]) controller = self.make_controller(backend) controller.tick(0) controller.tick(0) self.assertEqual(controller.state, postgres_runtime.PostgresState.READY) self.assertTrue(controller.snapshot()['lifecycle_inert']) self.assertFalse(controller._owned_start) for now in (1, 1, 2, 2, 3, 3): controller.tick(now) self.assertEqual(controller.state, postgres_runtime.PostgresState.FOREIGN_OR_CONFIG_ERROR) self.assertEqual(backend.start_calls, 0) self.assertEqual(backend.stop_calls, 0) if __name__ == '__main__': unittest.main()