#!/usr/bin/env python3 """Deterministic utility behavior against temporary files; no real P4 calls.""" import bz2 import contextlib import gzip import importlib.util import io import lzma import os from pathlib import Path import shutil import subprocess import sys import tempfile import unittest from unittest.mock import patch import test_maintenance_review as maintenance SDP = Path(__file__).resolve().parents[1] BIN = SDP / "Server/Unix/p4/common/bin" BASH = os.environ.get("SDP_TEST_BASH", shutil.which("bash")) def load(name): spec = importlib.util.spec_from_file_location("utility_" + name, BIN / (name + ".py")) module = importlib.util.module_from_spec(spec) sys.modules[spec.name] = module spec.loader.exec_module(module) return module class LocalFilters(unittest.TestCase): def setUp(self): temporary = tempfile.TemporaryDirectory(prefix="sdp-utilities-") self.addCleanup(temporary.cleanup) self.root = Path(temporary.name) def run_python(self, name, *args): return subprocess.run([sys.executable, SDP / "Maintenance" / name, *args], cwd=self.root, capture_output=True, text=True, timeout=10) def test_revision_report_counts_file_heads_and_usage_is_nonzero(self): source = self.root / "files.txt" source.write_text("//depot/a#3 - edit change 3 (text)\n//depot/sp ace#12 - add change 4 (binary)\n") result = self.run_python("countrevs.py", str(source)) self.assertEqual(result.returncode, 0, result.stderr) self.assertIn("files = 2", result.stdout) self.assertIn("revisions = 15", result.stdout) self.assertNotEqual(self.run_python("countrevs.py").returncode, 0) def test_proxy_search_deduplicates_and_reports_missing_input(self): source = self.root / "server.log" source.write_text("pid 1 user@ws 10.1.2.3/192.168.1.1 'sync'\n" "pid 2 user@ws 10.1.2.3/192.168.1.2 'sync'\n" "pid 3 user@ws 10.2.3.4/192.168.1.3 'sync'\n" "pid 4 user@ws 192.168.1.4 'sync'\n") result = self.run_python("proxysearch.py", str(source)) self.assertEqual(result.returncode, 0, result.stderr) self.assertEqual(result.stdout.split(), ["10.1.2.3", "10.2.3.4"]) self.assertEqual(self.run_python("proxysearch.py", str(self.root / "absent")).returncode, 2) def test_lowercase_copy_preserves_existing_destination_on_collision(self): source = self.root / "Source" source.mkdir() (source / "Mixed.TXT").write_text("new content") target = self.root / "target" target.mkdir() with maintenance.module("lowercp") as module, patch.object(module, "TARGETDEPOTPATH", str(target) + "/"): previous = Path.cwd() try: os.chdir(self.root) module.rename_dir("Source") copied = target / "source/mixed.txt" self.assertEqual(copied.read_text(), "new content") copied.write_text("preserved") with contextlib.redirect_stderr(io.StringIO()), self.assertRaises(FileExistsError): module.rename_dir("Source") self.assertEqual(copied.read_text(), "preserved") self.assertEqual((source / "Mixed.TXT").read_text(), "new content") finally: os.chdir(previous) def test_protection_group_report_matches_case_and_preserves_query_failure(self): source = self.root / "groups.txt" source.write_text("Developers\nMISSING\n") with maintenance.module("protect_groups", [str(source)]) as module: with patch.object(module.sdputils, "run_checked", return_value=subprocess.CompletedProcess([], 0, "developers\n")): output = io.StringIO() with contextlib.redirect_stdout(output): module.main() self.assertEqual(output.getvalue(), "missing\n") with patch.object(module.sdputils, "run_checked", side_effect=subprocess.CalledProcessError(7, ["fake-p4"])): with self.assertRaises(subprocess.CalledProcessError): module.main() def test_clean_log_removes_virtual_lock_blocks_and_keeps_database_metrics(self): source = "--- clients/ws(W)\n--- total lock 1\n--- db.rev\n--- total lock 2\n--- peek count 1\ncommand completed\n" result = subprocess.run([BASH, BIN / "clean_log.sh"], input=source, text=True, capture_output=True, timeout=10) self.assertEqual(result.returncode, 0, result.stderr) self.assertEqual(result.stdout, "--- db.rev\n--- total lock 2\ncommand completed\n") def test_progress_report_counts_completed_batches_and_handles_empty_input(self): source = self.root / "files_to_fstat.txt" source.write_text("a\nb\nc\nd\n") (self.root / "batch_verify_001.log").write_text("2026 date batch 1 completed successfully (2 files)\n") command = [BASH, BIN / "check_pull_status.sh"] result = subprocess.run(command, cwd=self.root, capture_output=True, text=True, timeout=10) self.assertEqual(result.returncode, 0, result.stderr) self.assertIn("Processed: 2 files", result.stdout) self.assertIn("Progress: 50.00%", result.stdout) source.write_text("") result = subprocess.run(command, cwd=self.root, capture_output=True, text=True, timeout=10) self.assertEqual(result.returncode, 0, result.stderr) self.assertIn("Progress: 0.00%", result.stdout) def test_modern_edge_table_lists_include_storage_tables(self): for version in ["2024.2.12345", "2026.1.12345"]: with self.subTest(version=version): env = dict(os.environ, P4D_VERSION=version, SCRIPT=str(BIN / "edge_vars")) result = subprocess.run([BASH, "-c", 'source "$SCRIPT"; printf "%s\\n%s\\n" "$ExcludedTables" "$CheckpointTables"'], env=env, capture_output=True, text=True, timeout=10) self.assertEqual(result.returncode, 0, result.stderr) excluded, checkpoint = result.stdout.splitlines() self.assertIn("db.storagesx", excluded.split(",")) self.assertIn("db.storage", checkpoint.split(",")) self.assertNotIn("db.storage", excluded.split(",")) class TransferCancellation(unittest.TestCase): def setUp(self): self.module = load("cancel_pull_transfers") self.record = "//depot/literal $(file) name 1.2.3 text failed add " + "a" * 32 + "\n" def invoke(self, responses, *args): with patch.object(self.module, "run_p4", side_effect=responses) as command, \ contextlib.redirect_stdout(io.StringIO()), contextlib.redirect_stderr(io.StringIO()): rc = self.module.main(list(args)) return rc, command def test_snapshot_validation_precedes_any_cancellation(self): listing = subprocess.CompletedProcess([], 0, self.record + "malformed later record\n", "") rc, command = self.invoke([listing]) self.assertEqual(rc, 1) self.assertEqual(command.call_count, 1) def test_duplicate_literal_path_is_cancelled_once_with_full_archive_revision(self): listing = subprocess.CompletedProcess([], 0, self.record * 2, "") rc, command = self.invoke([listing, subprocess.CompletedProcess([], 0, "", "")], "-p", "loopback:19999") self.assertEqual(rc, 0) self.assertEqual(command.call_count, 2) self.assertEqual(command.call_args.args[0], ["p4", "-p", "loopback:19999", "pull", "-d", "-f", "//depot/literal $(file) name", "-r", "1.2.3"]) def test_dry_run_and_existing_snapshot_never_cancel(self): listing = subprocess.CompletedProcess([], 0, self.record, "") rc, command = self.invoke([listing], "--dry-run") self.assertEqual(rc, 0) self.assertEqual(command.call_count, 1) with tempfile.TemporaryDirectory(prefix="sdp-cancel-") as temporary: path = Path(temporary) / "snapshot" path.write_text("original") rc, command = self.invoke([listing], "--save-listing", str(path)) self.assertEqual(rc, 1) self.assertEqual(command.call_count, 1) self.assertEqual(path.read_text(), "original") class GroupAudit(unittest.TestCase): def test_only_group_owners_are_notified_and_ownerless_groups_are_reported(self): with maintenance.module("group_audit") as module: replies = [subprocess.CompletedProcess([], 0, "owned\nownerless\n"), subprocess.CompletedProcess([], 0, "... Owners0 alice\n... Users0 member\n"), subprocess.CompletedProcess([], 0, "... Users0 member\n")] output = io.StringIO() with patch.object(module.sdputils, "run_checked", side_effect=replies), \ patch.object(module, "email_owner") as notify, contextlib.redirect_stdout(output): module.main() notify.assert_called_once_with("owned", "alice") self.assertEqual(output.getvalue(), "ownerless,No owner.\n") def test_failed_group_query_never_sends_notifications(self): with maintenance.module("group_audit") as module: with patch.object(module.sdputils, "run_checked", side_effect=subprocess.CalledProcessError(7, ["fake-p4"])), \ patch.object(module, "email_owner") as notify: with self.assertRaises(subprocess.CalledProcessError): module.main() notify.assert_not_called() class TemplateEntrypoints(unittest.TestCase): def test_broker_proxy_templates_preserve_instance_arguments_and_child_status(self): with tempfile.TemporaryDirectory(prefix="sdp-init-wrapper-") as temporary: root = Path(temporary) for daemon in ["p4broker", "p4p"]: with self.subTest(daemon=daemon): base = root / (daemon + "_base") base.write_text('#!/bin/bash\nprintf "%s\\0" "$SDP_INSTANCE" "$@" > "$TEST_RECORD"\nexit 7\n') base.chmod(0o755) source = SDP / "Server/Unix/p4/common/etc/init.d" / (daemon + "_instance_init.template") script = root / (daemon + "_23_init") script.write_text(source.read_text().replace("REPL_SDP_INSTANCE", "23").replace("/p4/common/bin/", str(root) + "/")) record = root / "arguments" result = subprocess.run([BASH, script, "restart", "space argument", "$(literal)"], env=dict(os.environ, TEST_RECORD=str(record)), capture_output=True, text=True) self.assertEqual(result.returncode, 7, result.stderr) self.assertEqual(record.read_bytes().split(b"\0")[:-1], [b"23", b"23", b"restart", b"space argument", b"$(literal)"]) def test_sample_p4d_wrapper_adds_case_flag_and_preserves_child_status(self): with tempfile.TemporaryDirectory(prefix="sdp-p4d-wrapper-") as temporary: root = Path(temporary) binary = root / "p4d_1_bin" binary.write_text('#!/bin/bash\nprintf "%s\\0" "$@" > "$TEST_RECORD"\nexit 9\n') binary.chmod(0o755) script = root / "p4d_1" script.write_text((SDP / "Server/Unix/p4/1/bin/p4d_1").read_text().replace("/p4/common/bin/", str(root) + "/")) record = root / "arguments" result = subprocess.run([BASH, script, "-r", "path with spaces", "-V"], env=dict(os.environ, TEST_RECORD=str(record)), capture_output=True, text=True) self.assertEqual(result.returncode, 9, result.stderr) self.assertEqual(record.read_bytes().split(b"\0")[:-1], [b"-C1", b"-vtrack=0", b"-r", b"path with spaces", b"-V"]) class RoleWrappers(unittest.TestCase): def setUp(self): temporary = tempfile.TemporaryDirectory(prefix="sdp-roles-") self.addCleanup(temporary.cleanup) self.root = Path(temporary.name) self.common = self.root / "common/bin" self.common.mkdir(parents=True) (self.root / "1").mkdir() (self.common / "p4_vars").write_text('export SERVER_TYPE="$TEST_ROLE" P4REPLICA=FALSE EDGESERVER=FALSE\n') (self.common / "backup_functions.sh").write_text('set_vars () { :; }\n') self.command = self.root / "record" self.command.write_text('#!/bin/bash\nprintf "%s\\0" "$@" > "$TEST_CALLS"\nexit 7\n') self.command.chmod(0o755) self.calls = self.root / "calls" def run_wrapper(self, name, role): script = self.root / name script.write_text((BIN / name).read_text().replace("/p4/", str(self.root) + "/")) env = dict(os.environ, TEST_ROLE=role, TEST_CALLS=str(self.calls), P4BrokerInitScript=str(self.command), P4ProxyInitScript=str(self.command)) return subprocess.run([BASH, script, "1", self.command, "space argument", "$(literal)"], env=env, text=True, capture_output=True, timeout=10) def test_server_role_routing_preserves_arguments_and_child_exit(self): mapping = {"run_if_master.sh": "p4d_master", "run_if_edge.sh": "p4d_edge", "run_if_replica.sh": "p4d_standby"} for script, accepted in mapping.items(): with self.subTest(script=script): result = self.run_wrapper(script, accepted) self.assertEqual(result.returncode, 7, result.stderr) self.assertEqual(self.calls.read_bytes().split(b"\0")[:-1], [b"space argument", b"$(literal)"]) self.calls.unlink() self.assertEqual(self.run_wrapper(script, "p4proxy").returncode, 0) self.assertFalse(self.calls.exists()) for other in (set(mapping.values()) | {"p4d_replica", "p4broker"}) - {accepted}: self.assertEqual(self.run_wrapper(script, other).returncode, 0) self.assertFalse(self.calls.exists()) self.assertEqual(self.run_wrapper(script, "unrecognized_role").returncode, 1) self.assertFalse(self.calls.exists()) def test_broker_proxy_route_only_when_init_script_is_executable(self): for script in ["run_if_broker.sh", "run_if_proxy.sh"]: with self.subTest(script=script): self.command.chmod(0o755) self.assertEqual(self.run_wrapper(script, "unused").returncode, 7) self.calls.unlink() self.command.chmod(0o644) self.assertEqual(self.run_wrapper(script, "unused").returncode, 0) self.assertFalse(self.calls.exists()) def test_process_helper_excludes_init_and_search_process(self): ps = self.root / "ps" ps.write_text("#!/bin/bash\ncat <<'EOF'\nuser 123 1 0 x x /p4/1/bin/p4d_1 -p 19999\nuser 456 1 0 x x /p4/1/bin/p4d_1_init start\nuser 789 1 0 x x grep /p4/1/bin/p4d_1\nEOF\n") ps.chmod(0o755) env = dict(os.environ, PS=str(ps), GREP="grep", AWK="awk", SCRIPT=str(BIN / "ps_functions.sh")) result = subprocess.run([BASH, "-c", 'source "$SCRIPT"; get_pids /p4/1/bin/p4d_1'], env=env, capture_output=True, text=True) self.assertEqual(result.returncode, 0, result.stderr) self.assertEqual(result.stdout.strip(), "123") class LogAnalysis(unittest.TestCase): def test_compression_magic_and_suffix_inputs_are_read_without_modification(self): text = "2026/10/16 10:00:00 pid 123 user@ws 127.0.0.1 'user-sync //depot/...'\n--- lapse 2s\n" with tempfile.TemporaryDirectory(prefix="sdp-log-read-") as temporary: root = Path(temporary) for module_name in ["find_blocker", "log_analyzer2"]: module = load(module_name) for suffix, payload in [(".log", text.encode()), (".gz", gzip.compress(text.encode())), (".bz2", bz2.compress(text.encode())), (".xz", lzma.compress(text.encode())), (".unknown", gzip.compress(text.encode()))]: with self.subTest(module=module_name, suffix=suffix): path = root / (module_name + suffix) path.write_bytes(payload) with module.open_log(str(path)) as source: self.assertEqual(source.read(), text) self.assertEqual(path.read_bytes(), payload) def test_parser_extracts_command_and_waits_and_filters_users(self): module = load("log_analyzer2") text = "2026/10/16 10:00:00 pid 123 alice@ws 127.0.0.1 'user-sync //depot/...'\n--- lapse 2s\n--- db.rev\n--- total lock wait+held read/write 1500ms+200ms/0ms+0ms\n" with tempfile.TemporaryDirectory(prefix="sdp-log-parse-") as temporary: path = Path(temporary) / "server.log" path.write_text(text) entries = module.P4LogParser().parse_file(str(path)) self.assertEqual(len(entries), 1) self.assertEqual(entries[0].user, "alice") self.assertEqual(entries[0].lapse_seconds, 2) self.assertEqual(entries[0].total_read_lock_wait_ms, 1500) analyzer = module.P4LogAnalyzer(entries) self.assertEqual(len(analyzer.filter_entries(user="alice")), 1) self.assertEqual(analyzer.filter_entries(user="bob"), []) class OwnershipDiagnostics(unittest.TestCase): def setUp(self): temporary = tempfile.TemporaryDirectory(prefix="sdp-ownership-test-") self.addCleanup(temporary.cleanup) self.root = Path(temporary.name) self.mockbin = self.root / "mockbin" self.mockbin.mkdir() self.directory = self.root / "depot with spaces" self.directory.mkdir() self.env = dict(os.environ, PATH=str(self.mockbin) + os.pathsep + os.environ["PATH"], OWNERSHIP_PATH=str(self.directory / "literal $(command) file"), SCAN_FAIL="0", HOSTNAME="fixture") find = self.mockbin / "find" find.write_text('#!/bin/bash\nprintf "actual\\0group\\0%s\\0" "$OWNERSHIP_PATH"\nexit "$SCAN_FAIL"\n') find.chmod(0o755) def run_check(self, owner="actual", group="group", depth=None): command = [BASH, BIN / "check_dir_ownership.sh", self.directory, owner, group] if depth is not None: command.append(depth) return subprocess.run(command, env=self.env, text=True, capture_output=True, timeout=10) def test_partial_failed_scan_never_reports_success(self): self.env["SCAN_FAIL"] = "1" result = self.run_check() self.assertNotEqual(result.returncode, 0) self.assertIn("Ownership scan failed", result.stdout) self.assertNotIn("No ownership anomalies", result.stdout) def test_anomaly_preserves_complete_literal_filename_and_group_optional(self): result = self.run_check("expected") self.assertEqual(result.returncode, 1) self.assertIn(self.env["OWNERSHIP_PATH"] + " is owned by actual:group", result.stdout) self.assertEqual(self.run_check(group="unset").returncode, 0) self.assertNotEqual(self.run_check(depth="1 -delete").returncode, 0) @unittest.skipUnless(sys.platform.startswith("linux"), "GNU stat/find integration runs in Linux containers") def test_real_local_scan_detects_wrong_owner_and_preserves_names(self): self.env["PATH"] = os.environ["PATH"] path = Path(self.env["OWNERSHIP_PATH"]) path.write_text("unchanged") result = self.run_check("nonexistent-sdp-owner", "unset") self.assertEqual(result.returncode, 1, result.stderr) self.assertIn(str(path) + " is owned by", result.stdout) self.assertEqual(path.read_text(), "unchanged") class RuntimeCommandFixture: def setUp(self): temporary = tempfile.TemporaryDirectory(prefix="sdp-runtime-commands-") self.addCleanup(temporary.cleanup) self.root = Path(temporary.name) self.common = self.root / "common/bin" self.logs = self.root / "logs" self.db = self.root / "root" for path in [self.common, self.logs, self.db, self.root / "1/bin"]: path.mkdir(parents=True) self.actions = self.root / "actions" self.env = dict(os.environ, TEST_ROOT=str(self.root), TEST_ACTIONS=str(self.actions), LOGS=str(self.logs), P4ROOT=str(self.db), P4CBIN=str(self.common), P4BIN=str(self.common / "p4"), P4DBIN=str(self.root / "1/bin/p4d_1"), P4PORT="local port", P4MASTERPORT="master port", P4USER="fixture", P4SERVER="p4_1", SERVERID="replica", P4MASTER_ID="master", P4D_VERSION="2026.1", HOSTNAME="fixture", LOGIN_RC="0", JOURNAL_RC="0", STATUS_MODE="healthy", CONFIG_RC="0", ARCHIVE_RC="0", SUMMARY_RC="0", PATH=str(self.common) + os.pathsep + os.environ["PATH"]) (self.common / "p4_vars").write_text(":\n") (self.common / "backup_functions.sh").write_text('''log () { printf '%s\\n' "$*" >> "$LOGFILE"; } die () { log "$*"; exit 97; } log_init () { :; }; sdp_log_init () { :; }; sdp_add_field () { :; } sdp_set_status () { :; }; sdp_set_error () { log "$*"; } check_vars () { :; }; set_vars () { :; }; check_uid () { :; }; check_dirs () { :; } rotate_last_run_logs () { :; }; rotate_log_file () { :; }; get_journalnum () { :; } check_disk_space () { :; }; mail_log_file () { :; } ckp_running () { echo lock >> "$TEST_ACTIONS"; } ckp_complete () { echo unlock >> "$TEST_ACTIONS"; } get_offline_journal_num () { echo offline_floor >> "$TEST_ACTIONS"; } remove_old_checkpoints_and_journals () { echo retention >> "$TEST_ACTIONS"; } remove_old_logs () { echo log_cleanup >> "$TEST_ACTIONS"; } request_replica_checkpoint () { echo checkpoint_request >> "$TEST_ACTIONS"; } copy_jd_table () { JDTmpDir="$TEST_ROOT"; } remove_jd_tables () { echo remove_jd >> "$TEST_ACTIONS"; } is_edge () { echo NO; } ''') self.executable(self.common / "p4login", '#!/bin/bash\necho login >> "$TEST_ACTIONS"\necho login-diagnostic\nexit "$LOGIN_RC"\n') self.executable(self.common / "p4login_master", '#!/bin/bash\necho master_login >> "$TEST_ACTIONS"\nexit 0\n') self.executable(self.common / "timeout", '#!/bin/bash\nshift\nexec "$@"\n') self.executable(self.common / "p4", '''#!/bin/bash printf 'p4 %s\\n' "$*" >> "$TEST_ACTIONS" case "$*" in *'admin journal'*) echo journal-diagnostic; exit "$JOURNAL_RC";; *'pull -ljv'*) if [[ "$STATUS_MODE" == malformed ]]; then echo 'info: no usable counters'; exit 0; fi if [[ "$STATUS_MODE" == failed ]]; then echo pull-diagnostic >&2; exit 7; fi printf '... replicaJournalCounter 08\\n... masterJournalNumber 08\\n';; *'pull -ls'*) echo summary-diagnostic; exit "$SUMMARY_RC";; *'pull -l'*) echo archive-diagnostic; exit "$ARCHIVE_RC";; esac ''') self.executable(self.root / "1/bin/p4d_1", '''#!/bin/bash if [[ "$*" == *-cshow* ]]; then echo 'replica: lbr.replication = readonly'; exit "$CONFIG_RC"; fi printf '%s\\0' "$@" > "$TEST_ROOT/dump-argv" ''') @staticmethod def executable(path, source): path.write_text(source) path.chmod(0o755) def script(self, name, *args): path = self.root / name path.write_text((BIN / name).read_text().replace("/p4/", str(self.root) + "/")) return subprocess.run([BASH, path, "1", *args], env=self.env, text=True, capture_output=True, timeout=10) class RuntimeCommands(RuntimeCommandFixture, unittest.TestCase): def test_sizes_uses_tagged_ids_and_preserves_query_failure(self): self.executable(self.common / "p4", '''#!/bin/bash printf '%s\\n' "$*" >> "$TEST_ACTIONS" if [[ "$*" == *changes* ]]; then [[ "$*" == *-ztag* ]] || exit 88 printf '%s\\n' "${CHANGE_IDS:-12}" exit "${LIST_RC:-0}" fi echo '//...@12,@12 2 files 3.5G bytes' exit "${SIZE_RC:-0}" ''') result = self.script("sizes.sh") self.assertEqual(result.returncode, 0, result.stderr) self.assertIn("3.5G", result.stdout) self.assertIn("//...@12,@12", self.actions.read_text()) for variable in ["LIST_RC", "SIZE_RC"]: self.env[variable] = "7" self.assertEqual(self.script("sizes.sh").returncode, 7) self.env[variable] = "0" self.actions.write_text("") self.env["CHANGE_IDS"] = "12\n$(touch forbidden)" self.assertEqual(self.script("sizes.sh").returncode, 1) self.assertNotIn("sizes", self.actions.read_text()) def test_cachepurge_passes_scope_and_reports_failed_command(self): self.env["KEEPLOGS"] = "7" with (self.common / "backup_functions.sh").open("a") as output: output.write('remove_log_files () { :; }\n') self.executable(self.common / "p4", '#!/bin/bash\nprintf "%s\\0" "$@" > "$TEST_ROOT/purge-argv"\necho purge-diagnostic\nexit "$JOURNAL_RC"\n') self.env["JOURNAL_RC"] = "7" result = self.script("p4cachepurge.sh") self.assertEqual(result.returncode, 1, result.stderr) self.assertIn("purge-diagnostic", (self.logs / "p4cachepurge.log").read_text()) arguments = (self.root / "purge-argv").read_bytes().split(b"\0") self.assertIn(b"cachepurge", arguments) self.assertIn(b"//depot/release/...", arguments) self.env["JOURNAL_RC"] = "0" self.assertEqual(self.script("p4cachepurge.sh").returncode, 0) def test_broker_proxy_login_failure_stops_cleanup_and_keeps_diagnostic(self): for name, logfile in [("broker_rotate.sh", "broker_rotate.log"), ("proxy_rotate.sh", "rotate_proxy.log")]: with self.subTest(script=name): self.env["LOGIN_RC"] = "7" self.actions.write_text("") result = self.script(name) self.assertEqual(result.returncode, 97, result.stderr) self.assertNotIn("log_cleanup", self.actions.read_text()) self.assertIn("login-diagnostic", (self.logs / logfile).read_text()) self.env["LOGIN_RC"] = "0" self.assertEqual(self.script(name).returncode, 0) self.assertIn("log_cleanup", self.actions.read_text()) def test_replica_request_reports_failed_master_rotation(self): self.env["JOURNAL_RC"] = "7" result = self.script("request_replica_checkpoint.sh", "-now") self.assertEqual(result.returncode, 97, result.stderr) self.assertIn("checkpoint_request", self.actions.read_text()) self.assertIn("master_login", self.actions.read_text()) self.assertIn("journal-diagnostic", (self.logs / "request_checkpoint.log").read_text()) def test_replica_cleanup_reads_offline_floor_before_retention_and_releases_lock(self): result = self.script("replica_cleanup.sh") self.assertEqual(result.returncode, 0, result.stderr) actions = self.actions.read_text().splitlines() self.assertEqual(actions[0], "lock") self.assertLess(actions.index("offline_floor"), actions.index("retention")) self.assertEqual(actions[-1], "unlock") def test_edge_dump_keeps_checkpoint_filename_literal_and_releases_lock(self): target = str(self.root / "checkpoint $(literal) name.gz") result = self.script("edge_dump.sh", "edge-one", target) self.assertEqual(result.returncode, 0, result.stderr) self.assertEqual((self.root / "dump-argv").read_bytes().split(b"\0")[-2], target.encode()) self.assertEqual(self.actions.read_text().splitlines(), ["lock", "unlock"]) def test_replica_status_rejects_missing_counters_and_keeps_failed_query_diagnostic(self): self.env["STATUS_MODE"] = "malformed" self.assertEqual(self.script("replica_status.sh").returncode, 1) self.assertIn("valid journal counters", (self.logs / "replica_status.log").read_text()) self.env["STATUS_MODE"] = "failed" self.assertEqual(self.script("replica_status.sh").returncode, 97) self.assertIn("pull-diagnostic", (self.logs / "replica_status.log").read_text()) def test_replica_status_archive_query_failures_are_nonzero(self): (self.db / "db.config").touch() for variable in ["ARCHIVE_RC", "SUMMARY_RC", "CONFIG_RC"]: with self.subTest(variable=variable): self.env[variable] = "7" self.assertNotEqual(self.script("replica_status.sh").returncode, 0) self.env[variable] = "0" self.assertEqual(self.script("replica_status.sh").returncode, 0) class UpgradeFailures(RuntimeCommandFixture, unittest.TestCase): def setUp(self): super().setUp() self.env.update(P4REPLICA="TRUE", EDGESERVER="0", STOP_RC="0", COPY_RC="0", VERSION_BAD="0") with (self.common / "backup_functions.sh").open("a") as output: output.write('stop_p4d () { echo stop >> "$TEST_ACTIONS"; return "$STOP_RC"; }\n' 'start_p4d () { echo start >> "$TEST_ACTIONS"; }\n') (self.db / "license").write_text("Support: 2099/01/01\n") self.executable(self.root / "1/bin/p4d_1_init", "#!/bin/bash\nexit 0\n") self.executable(self.common / "p4", '''#!/bin/bash if [[ "$VERSION_BAD" == 1 ]]; then echo invalid-version; exit 0; fi echo 'Rev. P4/LINUX26X86_64/2026.1/1234567 (2026/07/01).' ''') self.executable(self.common / "p4d", '''#!/bin/bash echo 'Rev. P4D/LINUX26X86_64/2026.1/1234567 (2026/07/01).' ''') self.executable(self.common / "cp", '#!/bin/bash\n[[ "$COPY_RC" == 0 ]] || exit "$COPY_RC"\nexec /bin/cp "$@"\n') # GNU-date behavior for the two license-validation cases, portable to # the macOS authoring host; other date uses execute the real program. self.executable(self.common / "date", "#!" + sys.executable + ''' import datetime, os, sys if '-d' in sys.argv: value = sys.argv[sys.argv.index('-d') + 1] print(datetime.datetime.fromisoformat(value.replace('/', '-')).strftime('%Y/%m/%d')) else: os.execv('/bin/date', ['date'] + sys.argv[1:]) ''') self.old_binary = self.root / "1/bin/p4_1" self.old_binary.write_text("old binary entry point") def test_upgrade_dry_run_does_not_take_lock_or_replace_binaries(self): result = self.script("upgrade.sh") self.assertEqual(result.returncode, 0, result.stderr) self.assertEqual(self.old_binary.read_text(), "old binary entry point") self.assertFalse(self.actions.exists()) self.assertIn("DRY RUN complete", result.stdout) def test_upgrade_failed_stop_preserves_binaries_and_ownership(self): self.env["STOP_RC"] = "7" result = self.script("upgrade.sh", "-y") self.assertEqual(result.returncode, 97, result.stderr) self.assertEqual(self.old_binary.read_text(), "old binary entry point") actions = self.actions.read_text().splitlines() self.assertIn("lock", actions) self.assertIn("stop", actions) self.assertNotIn("start", actions) self.assertNotIn("unlock", actions) def test_upgrade_bad_staged_version_fails_before_stop(self): self.env["VERSION_BAD"] = "1" result = self.script("upgrade.sh", "-y") self.assertEqual(result.returncode, 97, result.stderr) self.assertNotIn("stop", self.actions.read_text().splitlines()) self.assertEqual(self.old_binary.read_text(), "old binary entry point") def test_upgrade_failed_copy_stops_before_database_upgrade_and_restart(self): self.env["COPY_RC"] = "7" result = self.script("upgrade.sh", "-y") self.assertEqual(result.returncode, 97, result.stderr) self.assertFalse((self.root / "dump-argv").exists()) self.assertNotIn("start", self.actions.read_text().splitlines()) self.assertNotIn("unlock", self.actions.read_text().splitlines()) if __name__ == "__main__": unittest.main()