#!/usr/bin/env python3
"""Runtime review regressions using temporary artifacts and stubbed commands only."""
import os
from pathlib import Path
import shutil
import subprocess
import tempfile
import unittest

import test_maintenance_logging as maintenance_fixtures

BIN = Path(__file__).resolve().parents[1] / "Server/Unix/p4/common/bin"
BASH = os.environ.get("SDP_TEST_BASH", shutil.which("bash"))


class BackupRuntimeReview(unittest.TestCase):
    def setUp(self):
        self.temp = tempfile.TemporaryDirectory(prefix="sdp-runtime-")
        self.addCleanup(self.temp.cleanup)
        self.root = Path(self.temp.name)
        self.checkpoints = self.root / "checkpoints with spaces"
        self.offline = self.root / "offline"
        self.checkpoints.mkdir()
        self.offline.mkdir()
        self.env = dict(os.environ, P4CBIN=str(self.root), CHECKPOINTS=str(self.checkpoints),
                        P4CKP_PREFIX="p4.test", AWK="awk", TEST_OFFLINE=str(self.offline),
                        LOGFILE=str(self.root / "run.log"), TEST_ROOT=str(self.root),
                        BACKUP_FUNCTIONS=str(BIN / "backup_functions.sh"))

    def run_backup(self, commands, extra=None):
        env = dict(self.env, **(extra or {}))
        return subprocess.run([BASH, "-c", '''source "$BACKUP_FUNCTIONS"
function die () { printf '%s\\n' "$*" >&2; exit 97; }
OFFLINE_DB=$TEST_OFFLINE
''' + commands], env=env, cwd=self.root, text=True, capture_output=True, timeout=10)

    def artifact(self, kind, number, directory=False):
        path = self.checkpoints / f"p4.test.{kind}.{number}"
        if directory:
            path.mkdir()
            (path / "db.counters.gz").write_text("checkpoint")
        else:
            path = Path(str(path) + ".gz")
            path.write_text("artifact")
        Path(str(path) + ".md5").write_text("digest")
        return path

    def cleanup(self):
        return self.run_backup('KEEPCKPS=1; KEEPJNLS=2; OFFLINEJNLNUM=1000; remove_old_checkpoints_and_journals')

    def test_serial_cleanup_preserves_newer_numbers_with_same_prefix(self):
        old = self.artifact("ckp", 1)
        new = self.artifact("ckp", 100)
        result = self.cleanup()
        self.assertEqual(result.returncode, 0, result.stderr)
        self.assertFalse(old.exists())
        self.assertTrue(new.exists())
        self.assertTrue(Path(str(new) + ".md5").exists())

    def test_parallel_cleanup_preserves_newer_numbers_with_same_prefix(self):
        old = self.artifact("ckp", 1, directory=True)
        new = self.artifact("ckp", 100, directory=True)
        result = self.cleanup()
        self.assertEqual(result.returncode, 0, result.stderr)
        self.assertFalse(old.exists())
        self.assertTrue(new.exists())
        self.assertTrue(Path(str(new) + ".md5").exists())

    def test_journal_cleanup_preserves_newer_numbers_with_same_prefix(self):
        self.artifact("ckp", 100)
        old = self.artifact("jnl", 1)
        new = [self.artifact("jnl", n) for n in [10, 100]]
        result = self.cleanup()
        self.assertEqual(result.returncode, 0, result.stderr)
        self.assertFalse(old.exists())
        self.assertTrue(all(p.exists() for p in new))

    def test_cleanup_failure_is_reported(self):
        self.artifact("ckp", 3)
        self.artifact("jnl", 1)
        self.artifact("jnl", 2)
        self.artifact("jnl", 3)
        result = self.run_backup('''rm () { echo 'mock delete denied' >&2; return 13; }
KEEPCKPS=1; KEEPJNLS=2; OFFLINEJNLNUM=1000
remove_old_checkpoints_and_journals''')
        self.assertEqual(result.returncode, 97)
        self.assertIn("Failed to remove journal", result.stderr)
        self.assertIn("mock delete denied", (self.root / "run.log").read_text())

    def test_restore_ignores_foreign_or_journal_checksums(self):
        good = self.artifact("ckp", 5, directory=True)
        os.utime(Path(str(good) + ".md5"), (1, 1))
        self.artifact("jnl", 9)
        (self.checkpoints / "other.ckp.9").mkdir()
        (self.checkpoints / "other.ckp.9.md5").write_text("other")
        result = self.run_backup('''DoParallelCheckpoints=1; NumCheckPointThreads=2
P4DBIN=fake_p4d
fake_p4d () { if [[ ${1:-} == -h ]]; then echo -jvp; else printf '%s\\n' "$@" > "$TEST_ROOT/restore.args"; fi; }
recreate_offline_db_files''')
        self.assertEqual(result.returncode, 0, result.stderr)
        self.assertEqual((self.root / "restore.args").read_text().splitlines()[-1], str(good))

    def test_missing_parallel_checkpoint_preserves_offline_database(self):
        db = self.offline / "db.counters"
        db.write_text("preserve")
        (self.checkpoints / "p4.test.ckp.1.md5").write_text("orphan")
        result = self.run_backup('DoParallelCheckpoints=1; recreate_offline_db_files')
        self.assertEqual(result.returncode, 97)
        self.assertEqual(db.read_text(), "preserve")

    def test_missing_serial_checkpoint_preserves_offline_database(self):
        db = self.offline / "db.counters"
        db.write_text("preserve")
        (self.checkpoints / "p4.test.ckp.1.gz.md5").write_text("orphan")
        result = self.run_backup('DoParallelCheckpoints=0; recreate_offline_db_files')
        self.assertEqual(result.returncode, 97)
        self.assertEqual(db.read_text(), "preserve")

    def test_rsync_dry_run_failure_stops_copy_and_keeps_diagnostics(self):
        result = self.run_backup('''rsync () { echo 'mock dry run denied' >&2; return 23; }
rsync_with_preflight "$TEST_ROOT/source" "$TEST_ROOT/target" 0''')
        self.assertNotEqual(result.returncode, 0)
        self.assertIn("mock dry run denied", (self.root / "run.log").read_text())

    def test_rsync_missing_statistics_stops_copy(self):
        result = self.run_backup('''rsync () { echo 'unexpected rsync statistics'; }
rsync_with_preflight "$TEST_ROOT/source" "$TEST_ROOT/target" 0''')
        self.assertEqual(result.returncode, 1)
        self.assertIn("Cannot determine", (self.root / "run.log").read_text())

    def test_rsync_invalid_free_space_stops_copy(self):
        result = self.run_backup('''rsync () { echo 'Total transferred file size: 1,024 bytes'; }
df () { echo 'invalid output'; }
rsync_with_preflight "$TEST_ROOT/source" "$TEST_ROOT/target" 0''')
        self.assertEqual(result.returncode, 1)
        self.assertIn("Invalid available space", (self.root / "run.log").read_text())

    def test_rsync_keeps_paths_literal_and_converts_bytes(self):
        source = str(self.root / 'source $(touch injected)')
        result = self.run_backup('''rsync () {
  case " $* " in
    *' --dry-run '*) echo 'Total transferred file size: 1,024 bytes' ;;
    *) printf '%s\\n' "$@" > "$TEST_ROOT/rsync.args" ;;
  esac
}
df () { printf 'Filesystem 1024-blocks Used Available Capacity Mounted on\\nmock 100 98 2 98%% /\\n'; }
rsync_with_preflight "$SOURCE" "$TEST_ROOT/target with spaces" 1''', {"SOURCE": source})
        self.assertEqual(result.returncode, 0, result.stderr)
        self.assertEqual((self.root / "rsync.args").read_text().splitlines()[-2:], [source + "/", str(self.root / "target with spaces")])
        self.assertFalse((self.root / "injected").exists())

    def test_readonly_config_failure_prevents_swap_copy(self):
        (self.root / "db.config").touch()
        (self.root / "server.id").write_text("test\n")
        p4d = self.root / "fake_p4d"
        p4d.write_text("#!/bin/bash\necho 'mock configuration read denied' >&2\nexit 8\n")
        p4d.chmod(0o755)
        result = self.run_backup('''P4DBIN=$TEST_ROOT/fake_p4d; P4ROOT=$TEST_ROOT; SERVERID=test
copy_readonly_clients_dir''')
        self.assertEqual(result.returncode, 1)
        self.assertIn("mock configuration read denied", (self.root / "run.log").read_text())

    def test_missing_journals_directory_is_not_scanned_from_current_directory(self):
        result = self.run_backup('JOURNALS=$TEST_ROOT/missing; gzip_mv_journals')
        self.assertEqual(result.returncode, 97)
        self.assertIn("Cannot enter journals", result.stderr)

    def test_counter_dump_failure_is_not_an_empty_database(self):
        (self.root / "db.counters").write_text("not empty")
        result = self.run_backup('''P4ROOT=$TEST_ROOT; P4DBIN=fake_p4d
fake_p4d () { echo 'mock corrupt counters' >&2; return 8; }
get_journalnum''')
        self.assertEqual(result.returncode, 97)
        self.assertIn("Cannot read the journal counter", result.stderr)
        self.assertIn("mock corrupt counters", (self.root / "run.log").read_text())

    def test_valid_empty_counter_dump_stays_zero(self):
        (self.root / "db.counters").touch()
        result = self.run_backup('''P4ROOT=$TEST_ROOT; P4DBIN=fake_p4d
fake_p4d () { return 0; }
EDGESERVER=0; REPLICASERVER=0; STANDBYSERVER=0
get_journalnum
printf '%s %s' "$JOURNALNUM" "$CHECKPOINTNUM"''')
        self.assertEqual(result.returncode, 0, result.stderr)
        self.assertEqual(result.stdout, "0 1")

    def test_disk_threshold_supports_bytes_units_and_percent(self):
        for value, total, expected in [("0", "", "0"), ("250M", "", "262144000"),
                                       ("20%", "1000", "200"), ("08K", "", "8192")]:
            with self.subTest(value=value):
                result = self.run_backup('convert_to_bytes "$VALUE" "$TOTAL"',
                                         {"VALUE": value, "TOTAL": total})
                self.assertEqual(result.returncode, 0, result.stderr)
                self.assertEqual(result.stdout.strip(), expected)

    def test_invalid_disk_threshold_is_rejected(self):
        for value, total in [("Undefined", ""), ("1.5G", ""), ("101%", "1000"), ("20%", "bad")]:
            with self.subTest(value=value):
                result = self.run_backup('convert_to_bytes "$VALUE" "$TOTAL"',
                                         {"VALUE": value, "TOTAL": total})
                self.assertEqual(result.returncode, 1)
                self.assertIn("Invalid", result.stderr)

    def test_failed_disk_query_is_logged_and_rejected(self):
        result = self.run_backup('''P4BIN=fake_p4; AWK=awk; CUT=cut
fake_p4 () { echo 'mock disk query failed' >&2; return 8; }
get_journal_stats''')
        self.assertEqual(result.returncode, 97)
        self.assertIn("mock disk query failed", (self.root / "run.log").read_text())


class CacheCleanupReview(unittest.TestCase):
    def setUp(self):
        self.fixture = maintenance_fixtures.MaintenanceLogging(methodName="runTest")
        self.fixture.setUp()
        self.addCleanup(self.fixture.doCleanups)
        f = self.fixture
        (f.home / "depots/depot").mkdir(parents=True)
        f.write_executable(f.home / "bin/p4_1", "#!" + BASH + '''
case "$*" in
  *lbr.replication*) echo readonly ;;
  *dm.shelve.promote*)
    if [[ ${MOCK_PROMOTE_FAIL:-0} == 1 ]]; then echo 'mock config failure' >&2; exit 9; fi
    echo "${MOCK_PROMOTE:-1}" ;;
esac
''')
        f.write_executable(f.mockbin / "findmnt", "#!" + BASH + '\necho rw,relatime\n')
        f.write_executable(f.mockbin / "df", "#!" + BASH + "\nprintf 'Filesystem 1024-blocks Used Available Capacity Mounted on\\nmock 100 95 5 95%% /\\n'\n")
        f.write_executable(f.mockbin / "find", "#!" + BASH + '''
[[ $1 == "$SDP_TEST_ROOT/metrics" ]] && exit 0
echo invoked >> "$SDP_TEST_ROOT/deletions"
if [[ ${MOCK_FIND_FAIL:-0} == 1 ]]; then echo 'mock delete failure' >&2; exit 13; fi
printf '..'
''')

    def test_disabled_promotion_never_deletes_files(self):
        f = self.fixture
        result = f.run_script("cacheclean.sh", "1", extra={"MOCK_PROMOTE": "0"})
        self.assertEqual(result.returncode, 1)
        self.assertFalse((f.root / "deletions").exists())
        self.assertIn("dm.shelve.promote=1", f.log("cacheclean"))
        self.assertEqual(f.event("cacheclean")["status"], "failure")

    def test_promotion_query_failure_keeps_diagnostics(self):
        f = self.fixture
        result = f.run_script("cacheclean.sh", "1", extra={"MOCK_PROMOTE_FAIL": "1"})
        self.assertEqual(result.returncode, 1)
        self.assertFalse((f.root / "deletions").exists())
        self.assertIn("mock config failure", f.log("cacheclean"))

    def test_failed_deletion_worker_fails_run(self):
        f = self.fixture
        result = f.run_script("cacheclean.sh", "1", extra={"MOCK_FIND_FAIL": "1"})
        self.assertEqual(result.returncode, 1)
        self.assertIn("mock delete failure", f.log("cacheclean"))
        self.assertEqual(f.event("cacheclean")["status"], "failure")

    def test_successful_deletion_counts_are_logged(self):
        f = self.fixture
        result = f.run_script("cacheclean.sh", "1")
        self.assertEqual(result.returncode, 0, result.stderr)
        self.assertEqual(f.event("cacheclean")["files_removed"], 2)
        self.assertEqual(f.event("cacheclean")["status"], "success")


class RecoveryReview(unittest.TestCase):
    def setUp(self):
        self.fixture = maintenance_fixtures.MaintenanceLogging(methodName="runTest")
        self.fixture.setUp()
        self.addCleanup(self.fixture.doCleanups)
        f = self.fixture
        for name in ["offline_db", "checkpoints"]:
            (f.home / name).mkdir()
        with (f.common / "p4_vars").open("a") as out:
            out.write('''export P4DBIN="$P4HOME/bin/p4d_stub" P4JOURNAL="$LOGS/journal"
export CHECKPOINTS="$P4HOME/checkpoints" P4CKP_PREFIX=p4_1
''')
        f.write_executable(f.common / "backup_functions.sh", '''
OFFLINE_DB="$P4HOME/offline_db"
log () { echo "$*" >> "$LOGFILE"; }
die () { log "$*"; exit 97; }
check_vars () { :; }; set_vars () { :; }; check_uid () { :; }; check_dirs () { :; }
log_init () { :; }; rotate_log_file () { :; }; sdp_log_init () { :; }
ckp_running () { :; }; ckp_complete () { :; }; mail_log_file () { :; }
verify_checkpoint () { :; }
restore_checkpoint () { "$P4DBIN" -r "$2" -jr "$1" >> "$LOGFILE" 2>&1; }
switch_db_files () { local rebuilt=${2:-$OFFLINE_DB}; mv "$rebuilt"/db.* "$P4ROOT"/; }
complete_db_switch () { :; }
dump_checkpoint_from_root () { DoParallelCheckpoints=1; mkdir "$CHECKPOINTS/$P4CKP_PREFIX.ckp.$CHECKPOINTNUM"; }
DoParallelCheckpoints=1
stop_p4d () { echo stopped >> "$SDP_TEST_ROOT/actions"; }
start_p4d () { echo started >> "$SDP_TEST_ROOT/actions"; }
get_offline_journal_num () {
  [[ -f "$OFFLINE_DB/offline_db_usable.txt" ]] || die 'offline database marker missing'
  OFFLINECHECKPOINTNUM=1
}
''')
        f.write_executable(f.home / "bin/p4d_stub", "#!" + BASH + '''
count=0
[[ -f "$SDP_TEST_ROOT/p4d_count" ]] && count=$(cat "$SDP_TEST_ROOT/p4d_count")
count=$((count + 1)); echo "$count" > "$SDP_TEST_ROOT/p4d_count"
if [[ "$count" == "${MOCK_FAIL_STAGE:-0}" ]]; then echo 'mock p4d failure' >&2; exit 7; fi
# A checkpoint dump reads the source; it must not overwrite the live DB.
[[ " $* " != *" -jd "* ]] || exit 0
# Simulate restore into the target passed by the staging caller.
while [[ $# -gt 0 ]]; do
  if [[ $1 == -r ]]; then target=$2; shift 2; else shift; fi
done
printf restored > "${target:-$P4HOME/offline_db}/db.counters"
''')
        (f.home / "root/db.counters").write_text("live database")
        (f.home / "root/server.id").write_text("replica")
        (f.home / "root/sdp_server_type.txt").write_text("p4d_standby")
        (f.home / "root/statejcopy").write_text("3/1024")
        (f.logs / "journal.3").write_text("recovery journal")
        (f.root / "master.ckp").mkdir()
        for name in ["root", "offline_db"]:
            real = f.home / (name + "_physical")
            (f.home / name).rename(real)
            (f.home / name).symlink_to(real)
        f.write_executable(f.mockbin / "mv", "#!" + __import__("sys").executable + '\nimport os, subprocess, sys\nargs = sys.argv[1:]\nif "-Tf" in args: os.replace(args[-2], args[-1])\nelse: sys.exit(subprocess.call(["/bin/mv"] + args))\n')

    def test_failover_does_not_start_after_failed_journal_replay(self):
        f = self.fixture
        result = f.run_script("convert_standby_to_master.sh", "1", extra={"MOCK_FAIL_STAGE": "1"})
        self.assertEqual(result.returncode, 97, result.stderr)
        self.assertEqual((f.root / "actions").read_text(), "stopped\n")
        self.assertIn("mock p4d failure", f.log("convert_standby_to_master"))

    def test_failover_accepts_source_identity_after_position_record(self):
        f = self.fixture
        (f.home / "root/statejcopy").write_text("3/1024\nmaster-id\n")
        result = f.run_script("convert_standby_to_master.sh", "1")
        self.assertEqual(result.returncode, 0, result.stderr)
        self.assertEqual((f.root / "actions").read_text(), "stopped\nstarted\n")
        self.assertEqual((f.home / "root/server.id").read_text(), "master-id\n")
        self.assertEqual((f.logs / "journal").read_text(), "recovery journal")

    def test_failover_accepts_legacy_position_without_final_newline(self):
        f = self.fixture
        result = f.run_script("convert_standby_to_master.sh", "1")
        self.assertEqual(result.returncode, 0, result.stderr)
        self.assertEqual((f.root / "actions").read_text(), "stopped\nstarted\n")
        self.assertEqual((f.logs / "journal").read_text(), "recovery journal")

    def test_failover_rejects_malformed_position_before_identity_change(self):
        f = self.fixture
        for state in ("", "bad", "3\nmaster-id\n", "3/invalid\nmaster-id\n", "3/1024/extra\nmaster-id\n"):
            with self.subTest(state=state):
                (f.root / "actions").unlink(missing_ok=True)
                (f.home / "root/statejcopy").write_text(state)
                result = f.run_script("convert_standby_to_master.sh", "1")
                self.assertEqual(result.returncode, 97, result.stderr)
                self.assertEqual((f.root / "actions").read_text(), "stopped\n")
                self.assertEqual((f.home / "root/server.id").read_text(), "replica")
                self.assertTrue((f.logs / "journal.3").exists())
                self.assertFalse((f.root / "p4d_count").exists())

    def test_failover_validates_journal_before_changing_identity(self):
        f = self.fixture
        (f.logs / "journal.3").unlink()
        result = f.run_script("convert_standby_to_master.sh", "1")
        self.assertEqual(result.returncode, 97)
        self.assertEqual((f.root / "actions").read_text(), "stopped\n")
        self.assertEqual((f.home / "root/server.id").read_text(), "replica")

    def test_failover_reads_the_final_journal_after_stopping(self):
        f = self.fixture
        helper = f.common / "backup_functions.sh"
        helper.write_text(helper.read_text().replace(
            'stop_p4d () { echo stopped >> "$SDP_TEST_ROOT/actions"; }',
            '''stop_p4d () {
  echo stopped >> "$SDP_TEST_ROOT/actions"
  echo '4/2048' > "$P4ROOT/statejcopy"
  echo 'new final journal' > "$LOGS/journal.4"
}'''))
        result = f.run_script("convert_standby_to_master.sh", "1", extra={"MOCK_FAIL_STAGE": "1"})
        self.assertEqual(result.returncode, 97)
        self.assertEqual((f.logs / "journal").read_text(), "new final journal\n")
        self.assertTrue((f.logs / "journal.3").exists())

    def test_edge_recovery_failure_preserves_live_database(self):
        f = self.fixture
        for stage in [1, 2, 3]:
            with self.subTest(stage=stage):
                (f.root / "p4d_count").write_text("0")
                result = f.run_script("recover_edge.sh", "1", str(f.root / "master.ckp"),
                                      extra={"MOCK_FAIL_STAGE": str(stage)})
                self.assertEqual(result.returncode, 97, result.stderr)
                self.assertEqual((f.home / "root/db.counters").read_text(), "live database")
                if (f.root / "actions").exists():
                    self.assertNotIn("started", (f.root / "actions").read_text())
                self.assertIn("mock p4d failure", f.log("recover_edge"))

    def test_edge_success_marks_offline_db_and_preserves_number_prefix(self):
        f = self.fixture
        old = f.home / "checkpoints/p4_1.ckp.1"
        newer = f.home / "checkpoints/p4_1.ckp.10"
        old.mkdir(); newer.mkdir()
        result = f.run_script("recover_edge.sh", "1", str(f.root / "master.ckp"))
        self.assertEqual(result.returncode, 0, result.stderr + f.log("recover_edge"))
        self.assertTrue((f.home / "offline_db/offline_db_usable.txt").exists())
        self.assertTrue(newer.exists())
        self.assertTrue(old.exists())
        self.assertTrue(list((f.home / "checkpoints").glob(".sdp-edge-dump.*/previous-checkpoint/p4_1.ckp.1")))

    def test_latest_restore_ignores_journal_and_other_instance(self):
        f = self.fixture
        candidates = f.home / "checkpoints"
        good = candidates / "p4_1.ckp.2.gz.md5"
        good.touch(); os.utime(good, (1, 1))
        (candidates / "p4_1.jnl.3.gz.md5").touch()
        (candidates / "p4_10.ckp.4.gz.md5").touch()
        source = (BIN / "load_checkpoint.sh").read_text()
        start = source.index("function find_latest_checkpoint_md5 () {")
        end = source.index("\n#------------------------------------------------------------------------------", start)
        result = subprocess.run([BASH, "-c", source[start:end] + '\nfind_latest_checkpoint_md5 "$1"',
                                 "test", str(candidates / "p4_1")], capture_output=True, text=True)
        self.assertEqual(result.returncode, 0, result.stderr)
        self.assertEqual(result.stdout.strip(), str(good))


class PasswordFileReview(unittest.TestCase):
    def test_failed_service_password_setup_leaves_no_plaintext_file(self):
        for name in ["mkedge", "mkedge-standby", "mkstandby", "mkstandby_shared"]:
            with self.subTest(name=name):
                f = maintenance_fixtures.MaintenanceLogging(methodName="runTest")
                f.setUp()
                try:
                    f.write_executable(f.home / "bin/p4_1", "#!" + BASH + '''
case "$1 $2" in
  'server -o') printf 'Services:\tstandard\n' ;;
  'server -i'|'user -f') cat >/dev/null ;;
  'user -o') echo 'Type: standard' ;;
  '-ztag configure') printf '... Value 0\\n' ;;
  'passwd '*) cat > "$SDP_TEST_ROOT/password_received"; echo 'mock passwd failure' >&2; exit 7 ;;
esac
''')
                    # A preexisting unrelated file must not be overwritten.
                    sentinel = f.root / "passwd.txt"
                    sentinel.write_text("preserve unrelated file")
                    secret = "test password with * spaces"
                    result = f.run_script(name + ".sh", "1", "replica", secret, "master:1666", "upstream")
                    self.assertNotEqual(result.returncode, 0)
                    self.assertEqual(sentinel.read_text(), "preserve unrelated file")
                    self.assertEqual((f.root / "password_received").read_text(), secret + "\n" + secret + "\n")
                    self.assertNotIn(secret, f.log(name))
                finally:
                    f.doCleanups()


if __name__ == "__main__":
    unittest.main()