"""Opt-in sendq cleanup around real temporary database/symlink switches. The cleanup helper and service commands are stubs; helper path safety and live P4 configuration are exercised by their separate suites. Filesystem switching, metadata preservation, rollback, and retained recovery copies are real here. """ import unittest import test_runtime_improvements as runtime_fixtures class SendqSwitch(unittest.TestCase): def setUp(self): self.fixture = runtime_fixtures.RuntimeImprovements(methodName="runTest") self.fixture.setUp() self.addCleanup(self.fixture.doCleanups) f = self.fixture f.prepare_switch() self.old_queue = f.old / "old sendq" self.new_queue = f.new / "candidate sendq" for root, queue in ((f.old, self.old_queue), (f.new, self.new_queue)): queue.mkdir() (queue / "queue.tmp").write_text("temporary queue at " + str(root)) (root / ".fixture-sendq-config").write_text(queue.name) self.state = f.root / "service-state" self.state.write_text("stopped") self.trace_file = f.root / "switch-trace" f.prefix += ''' compare_journal_numbers () { printf 'compare|%s\\n' "$(cd "$P4ROOT" && pwd -P)" >> "$TEST_ROOT/switch-trace" } copy_readonly_clients_dir () { printf 'copy|%s|%s\\n' "$(cd "$P4ROOT" && pwd -P)" "$(cd "$OFFLINE_DB" && pwd -P)" >> "$TEST_ROOT/switch-trace" } clean_client_sendq_dir () { local physical= configured= state= physical=$(cd "$P4ROOT" && pwd -P) || return 41 configured=$(cat "$P4ROOT/.fixture-sendq-config") || return 42 state=$(cat "$TEST_ROOT/service-state") || return 43 printf 'clean|%s|%s|%s\\n' "$physical" "$configured" "$state" >> "$TEST_ROOT/switch-trace" [[ "$state" == stopped ]] || { log 'ERROR: attempted cleanup with service running'; return 44; } if [[ "$physical" == "${FAIL_CLEAN_ROOT:-}" ]]; then log "Injected sendq cleanup failure for $physical" return 45 fi rm -- "$P4ROOT/$configured/queue.tmp" || return 46 } start_p4d () { printf 'start|%s\\n' "$(cd "$P4ROOT" && pwd -P)" >> "$TEST_ROOT/switch-trace" printf started > "$TEST_ROOT/service-state" } ''' def trace(self): return [line.split("|") for line in self.trace_file.read_text().splitlines()] def assert_original_database(self): f = self.fixture self.assertEqual((f.root / "root").resolve(), f.old) self.assertEqual((f.root / "offline_db").resolve(), f.new) self.assertEqual((f.old / "db.counters").read_text(), "old counters") self.assertEqual((f.old / "db.rev").read_text(), "old revisions") self.assertEqual((f.old / "server.id").read_text(), "original server") self.assertEqual((f.new / "server.id").read_text(), "offline server") self.assertEqual((f.old / "save/db.counters").read_text(), "prior saved counters") def test_each_physical_root_uses_its_local_config_while_stopped(self): f = self.fixture result = f.run_bash("switch_db_files clean_sendq") self.assertEqual(result.returncode, 0, result.stderr) self.assertEqual(self.trace(), [ ["compare", str(f.old)], ["clean", str(f.old), self.old_queue.name, "stopped"], ["copy", str(f.old), str(f.new)], ["clean", str(f.new), self.new_queue.name, "stopped"], ]) self.assertEqual(self.state.read_text(), "stopped") self.assertFalse((self.old_queue / "queue.tmp").exists()) self.assertFalse((self.new_queue / "queue.tmp").exists()) self.assertEqual((f.root / "root").resolve(), f.new) self.assertEqual((f.root / "offline_db").resolve(), f.old) def test_first_cleanup_failure_preserves_database_links_and_stays_down(self): f = self.fixture result = f.run_bash("switch_db_files clean_sendq", {"FAIL_CLEAN_ROOT": str(f.old)}) self.assertEqual(result.returncode, 97) self.assert_original_database() self.assertEqual(self.trace(), [ ["compare", str(f.old)], ["clean", str(f.old), self.old_queue.name, "stopped"], ]) self.assertEqual(self.state.read_text(), "stopped") self.assertTrue((self.old_queue / "queue.tmp").exists()) self.assertTrue((self.new_queue / "queue.tmp").exists()) self.assertFalse(list(f.old.glob(".sdp-switch.*"))) self.assertIn("Injected sendq cleanup failure", (f.logs / "checkpoint.log").read_text()) def test_second_cleanup_failure_rolls_back_and_restarts_original(self): f = self.fixture result = f.run_bash("switch_db_files clean_sendq", {"FAIL_CLEAN_ROOT": str(f.new)}) self.assertEqual(result.returncode, 97) self.assert_original_database() self.assertEqual(self.trace(), [ ["compare", str(f.old)], ["clean", str(f.old), self.old_queue.name, "stopped"], ["copy", str(f.old), str(f.new)], ["clean", str(f.new), self.new_queue.name, "stopped"], ["start", str(f.old)], ]) self.assertEqual(self.state.read_text(), "started") self.assertFalse((self.old_queue / "queue.tmp").exists()) self.assertTrue((self.new_queue / "queue.tmp").exists()) self.assertFalse((f.old / "P4ROOT_not_usable.txt").exists()) self.assertFalse((f.new / "P4ROOT_not_usable.txt").exists()) self.assertIn("rolled back", result.stderr.lower()) def test_success_retains_previous_database_until_completion(self): f = self.fixture result = f.run_bash("switch_db_files clean_sendq") self.assertEqual(result.returncode, 0, result.stderr) self.assertEqual((f.old / "save/db.counters").read_text(), "old counters") self.assertEqual((f.old / "save/db.rev").read_text(), "old revisions") self.assertEqual((f.new / "db.counters").read_text(), "new counters") stages = list(f.old.glob(".sdp-switch.*")) self.assertEqual(len(stages), 1) self.assertEqual((stages[0] / "previous-save/db.counters").read_text(), "prior saved counters") self.assertEqual(self.state.read_text(), "stopped") def test_recovery_copy_is_removed_only_after_replacement_start_and_completion(self): f = self.fixture result = f.run_bash('''switch_db_files clean_sendq [[ -f "$SWITCH_DB_OLDROOT/save/db.counters" ]] || exit 70 start_p4d complete_db_switch ''') self.assertEqual(result.returncode, 0, result.stderr) self.assertEqual(self.trace()[-1], ["start", str(f.new)]) self.assertFalse((f.old / "save/db.counters").exists()) self.assertFalse(list(f.old.glob(".sdp-switch.*"))) self.assertEqual((f.new / "db.counters").read_text(), "new counters") def test_callers_without_opt_in_do_not_clean_either_root(self): f = self.fixture result = f.run_bash("switch_db_files") self.assertEqual(result.returncode, 0, result.stderr) self.assertEqual(self.trace(), [["compare", str(f.old)], ["copy", str(f.old), str(f.new)]]) self.assertTrue((self.old_queue / "queue.tmp").exists()) self.assertTrue((self.new_queue / "queue.tmp").exists()) self.assertEqual((f.old / "save/db.counters").read_text(), "old counters") if __name__ == "__main__": unittest.main()