test_journal_gaps.py #1

"""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()