"""Journal archival policy and checkpoint defaults, using disposable files only.""" import gzip import os from pathlib import Path import shutil import subprocess import tempfile import unittest BIN = Path(__file__).resolve().parents[1] / "Server/Unix/p4/common/bin" BASH = os.environ.get("SDP_TEST_BASH", shutil.which("bash")) PREFIX = "p4.site.with.dots" class JournalGaps(unittest.TestCase): def setUp(self): temp = tempfile.TemporaryDirectory(prefix="sdp-journal-gaps-") self.addCleanup(temp.cleanup) self.root = Path(temp.name) self.journals = self.root / "rotated journals" self.checkpoints = self.root / "checkpoints with spaces" self.journals.mkdir() self.checkpoints.mkdir() self.log = self.root / "archive.log" self.env = dict(os.environ, BACKUP_FUNCTIONS=str(BIN / "backup_functions.sh"), P4CBIN=str(self.root), JOURNALS=str(self.journals), CHECKPOINTS=str(self.checkpoints), P4CKP_PREFIX=PREFIX, LOGFILE=str(self.log), TEST_ROOT=str(self.root), REAL_GZIP=shutil.which("gzip")) def raw(self, number): return self.journals / (PREFIX + ".jnl." + str(number)) def archived(self, number): return self.checkpoints / (PREFIX + ".jnl." + str(number) + ".gz") @staticmethod def payload(number): return ("journal %s\n" % number).encode() + bytes(range(256)) def populate(self, numbers): for number in numbers: self.raw(number).write_bytes(self.payload(number)) (self.journals / "journal").write_bytes(b"live journal stays untouched\n") (self.journals / "other.instance.jnl.1").write_bytes(b"foreign instance\n") def run_shell(self, commands, extra=None): return subprocess.run([BASH, "-c", '''source "$BACKUP_FUNCTIONS" log () { printf '%s\\n' "$*" >> "$LOGFILE"; } die () { log "$*"; printf '%s\\n' "$*" >&2; exit 97; } ''' + commands], env=dict(self.env, **(extra or {})), cwd=self.root, capture_output=True, text=True, timeout=15) def assert_contents(self, remaining, moved): actual_raw = {file.name for file in self.journals.iterdir() if file.name.startswith(PREFIX + ".jnl.")} self.assertEqual(actual_raw, {self.raw(number).name for number in remaining}) self.assertEqual({file.name for file in self.checkpoints.iterdir()}, {self.archived(number).name for number in moved}) for number in remaining: self.assertEqual(self.raw(number).read_bytes(), self.payload(number)) for number in moved: self.assertEqual(gzip.decompress(self.archived(number).read_bytes()), self.payload(number)) self.assertEqual((self.journals / "journal").read_bytes(), b"live journal stays untouched\n") self.assertEqual((self.journals / "other.instance.jnl.1").read_bytes(), b"foreign instance\n") def test_two_or_fewer_raw_journals_remain_untouched_even_with_gaps(self): self.populate((1, 100)) result = self.run_shell("gzip_mv_journals") self.assertEqual(result.returncode, 0, result.stderr) self.assert_contents((1, 100), ()) def test_three_raw_journals_archive_only_the_oldest(self): self.populate((2, 30, 400)) result = self.run_shell("gzip_mv_journals") self.assertEqual(result.returncode, 0, result.stderr) self.assert_contents((30, 400), (2,)) def test_four_raw_journals_archive_only_the_oldest(self): self.populate((2, 30, 400, 5000)) result = self.run_shell("gzip_mv_journals") self.assertEqual(result.returncode, 0, result.stderr) self.assert_contents((30, 400, 5000), (2,)) def test_missing_unarchived_number_does_not_stop_later_selected_journal(self): self.populate((1, 3, 4, 5, 6, 7)) result = self.run_shell("gzip_mv_journals") self.assertEqual(result.returncode, 0, result.stderr) self.assert_contents((4, 5, 6, 7), (1, 3)) def test_wide_gaps_do_not_change_fifth_newest_cutoff(self): self.populate((1, 1000000, 1000001, 1000002, 1000003, 1000004)) result = self.run_shell("gzip_mv_journals") self.assertEqual(result.returncode, 0, result.stderr) self.assert_contents((1000001, 1000002, 1000003, 1000004), (1, 1000000)) def test_existing_archive_in_gap_is_preserved(self): self.populate((1, 3, 4, 5, 6, 7)) archive = gzip.compress(self.payload(2)) self.archived(2).write_bytes(archive) result = self.run_shell("gzip_mv_journals") self.assertEqual(result.returncode, 0, result.stderr) self.assert_contents((4, 5, 6, 7), (1, 2, 3)) self.assertEqual(self.archived(2).read_bytes(), archive) def test_selected_file_removed_after_enumeration_is_skipped(self): self.populate((1, 2, 3, 4, 5, 6)) result = self.run_shell('''gzip () { "$REAL_GZIP" "$@" || return $? rm -f "$JOURNALS/$P4CKP_PREFIX.jnl.2" } gzip_mv_journals''') self.assertEqual(result.returncode, 0, result.stderr) self.assert_contents((3, 4, 5, 6), (1,)) self.assertIn(str(self.raw(2)), self.log.read_text()) self.assertIn("skipping", self.log.read_text().lower()) def test_enumeration_failure_stops_before_any_journal_is_changed(self): self.populate((1, 2, 3, 4, 5, 6)) for failed_command in ("journal_numbers_in_cwd", "sort"): with self.subTest(failed_command=failed_command): result = self.run_shell(failed_command + ''' () { printf '1\\n' printf 'injected journal enumeration failure\\n' >&2 return 23 } gzip_mv_journals''') self.assertEqual(result.returncode, 97) self.assert_contents((1, 2, 3, 4, 5, 6), ()) self.assertIn("injected journal enumeration failure", result.stderr) self.assertIn("Cannot enumerate", self.log.read_text()) def test_gzip_command_error_is_not_hidden_by_existing_archive(self): self.populate((1, 2, 3, 4, 5, 6)) old_archive = gzip.compress(b"previous archive must not mask a new failure") self.archived(1).write_bytes(old_archive) result = self.run_shell('''gzip () { printf 'injected compression failure\\n' >&2; return 23; } gzip_mv_journals''') self.assertEqual(result.returncode, 97) for number in (1, 2, 3, 4, 5, 6): self.assertEqual(self.raw(number).read_bytes(), self.payload(number)) self.assertEqual(self.archived(1).read_bytes(), old_archive) self.assertFalse(self.archived(2).exists()) self.assertIn("injected compression failure", self.log.read_text()) self.assertIn("gzip", result.stderr) def test_concurrent_archival_success_remains_tolerated(self): for gzip_status in (0, 23): with self.subTest(gzip_status=gzip_status): self.populate((1, 2, 3)) result = self.run_shell('''gzip () { "$REAL_GZIP" "$@" || return $? command mv "$1.gz" "$CHECKPOINTS/" || return $? return "$SIMULATED_GZIP_STATUS" } gzip_mv_journals''', {"SIMULATED_GZIP_STATUS": str(gzip_status)}) self.assertEqual(result.returncode, 0, result.stderr) self.assert_contents((2, 3), (1,)) self.archived(1).unlink() def test_real_gzip_collision_still_fails_without_losing_inputs(self): self.populate((1, 2, 3)) compressed = Path(str(self.raw(1)) + ".gz") original = gzip.compress(b"existing compressed recovery evidence") compressed.write_bytes(original) result = self.run_shell("gzip_mv_journals") self.assertEqual(result.returncode, 97) self.assertEqual(self.raw(1).read_bytes(), self.payload(1)) self.assertEqual(compressed.read_bytes(), original) self.assertFalse(list(self.checkpoints.iterdir())) self.assertIn("already exists", self.log.read_text()) def test_success_without_compressed_output_is_an_error(self): self.populate((1, 2, 3)) result = self.run_shell('''gzip () { return 0; } gzip_mv_journals''') self.assertEqual(result.returncode, 97) self.assert_contents((1, 2, 3), ()) self.assertIn("gzip did not create", result.stderr) def test_real_move_failure_preserves_compressed_source_and_stops(self): self.populate((1, 2, 3, 4, 5, 6)) destination = self.root / "not a directory" destination.write_bytes(b"preserve unrelated destination") result = self.run_shell("gzip_mv_journals", {"CHECKPOINTS": str(destination)}) self.assertEqual(result.returncode, 97) compressed = Path(str(self.raw(1)) + ".gz") self.assertEqual(gzip.decompress(compressed.read_bytes()), self.payload(1)) self.assertFalse(self.raw(1).exists()) self.assertEqual(self.raw(2).read_bytes(), self.payload(2)) self.assertEqual(destination.read_bytes(), b"preserve unrelated destination") self.assertIn("move of", result.stderr) def test_set_vars_defaults_parallel_checkpoint_threads_to_sixteen(self): result = self.run_shell('''unset NumCheckPointThreads set_vars printf '%s|%s' "$DoParallelCheckpoints" "$NumCheckPointThreads" ''', {"P4HOME": str(self.root), "SDP_INSTANCE": "1", "P4D_VERSION": "2026.1", "SERVERID": "", "P4CCFG": str(self.root), "P4SERVER": "p4_test"}) self.assertEqual(result.returncode, 0, result.stderr) self.assertEqual(result.stdout, "1|16") def test_parallel_verification_and_restore_default_to_sixteen_and_honor_override(self): source = self.checkpoints / "parallel checkpoint" source.mkdir() Path(str(source) + ".md5").write_text("fixture checksum\n") for command in ('verify_checkpoint "$SOURCE"', 'restore_checkpoint "$SOURCE" "$TEST_ROOT/target"'): for override, expected in ((None, "16"), ("6", "6")): with self.subTest(command=command, override=override): result = self.run_shell('''P4DBIN=fake_p4d fake_p4d () { if [[ "$1" == -h ]]; then printf '%s\\n' '-jvp'; return 0; fi printf '%s\\n' "$@" > "$TEST_ROOT/p4d.args" } ''' + ("unset NumCheckPointThreads\n" if override is None else "NumCheckPointThreads=" + override + "\n") + command, {"SOURCE": str(source)}) self.assertEqual(result.returncode, 0, result.stderr) args = (self.root / "p4d.args").read_text().splitlines() self.assertEqual(args[args.index("-N") + 1], expected) self.assertEqual(args[-1], str(source)) if __name__ == "__main__": unittest.main()