mirror of
https://github.com/rcourtman/Pulse.git
synced 2026-10-25 13:10:03 +00:00
pulse-triage[bot]
b79496219f
Reconcile the retained snapshot publication repair with current main without importing unrelated containing history. Unique private staging, no-clobber publication and failure-before-mutation protect earlier recovery copies; retain zero-retention and hardened-unit fallback semantics. Retained-source: 854ea3c81a94ffe88145d663d25d5040d4637a47, 91bd7393fc4413b6629ad851f90c2c7f6c2c0991 Change-source: pulse-maintainer
275 lines
11 KiB
Python
275 lines
11 KiB
Python
#!/usr/bin/env python3
|
|
"""Run real configuration snapshot copy/publication against private fixtures.
|
|
|
|
Only the clock, disk-capacity observation and injected command failures are
|
|
controlled. No service, database, production config or network is touched.
|
|
These checks do not establish application-consistent backup or native recovery.
|
|
"""
|
|
|
|
import os
|
|
from pathlib import Path
|
|
import stat
|
|
import subprocess
|
|
import tempfile
|
|
import unittest
|
|
|
|
|
|
ROOT = Path(__file__).resolve().parents[2]
|
|
INSTALLER = Path(os.environ.get("PULSE_INSTALLER_UNDER_TEST", ROOT / "install.sh")).resolve()
|
|
STAMP = "20261023-154500"
|
|
|
|
PRELUDE = r'''
|
|
source "$INSTALLER"
|
|
print_info() { :; }
|
|
print_warn() { echo "$*" >&2; }
|
|
print_error() { echo "$*" >&2; }
|
|
ensure_config_backup_headroom() { return "${HEADROOM_EXIT:-0}"; }
|
|
date() { printf '%s\n' "$STAMP"; }
|
|
'''
|
|
|
|
|
|
class ConfigBackupPublication(unittest.TestCase):
|
|
def setUp(self):
|
|
self.temporary = tempfile.TemporaryDirectory(prefix="pulse-config-backup-")
|
|
self.addCleanup(self.temporary.cleanup)
|
|
self.root = Path(self.temporary.name)
|
|
self.config = self.root / "source parent" / "config with spaces"
|
|
self.config.mkdir(parents=True, mode=0o700)
|
|
(self.config / "settings.json").write_text("current state\n")
|
|
(self.config / ".env").write_text("private synthetic setting\n")
|
|
(self.config / ".env").chmod(0o600)
|
|
(self.config / "empty").mkdir()
|
|
(self.config / "setting-link").symlink_to("settings.json")
|
|
self.install = self.root / "install"
|
|
self.install.mkdir()
|
|
|
|
def old_snapshot(self, stamp=STAMP):
|
|
path = self.config.parent / (self.config.name + ".backup." + stamp)
|
|
path.mkdir()
|
|
(path / "settings.json").write_text("prior recovery state\n")
|
|
return path
|
|
|
|
def shell(self, body, **extra):
|
|
env = {k: v for k, v in os.environ.items() if not k.startswith("PULSE_")}
|
|
env.update(INSTALLER=str(INSTALLER), FIXTURE=str(self.root), STAMP=STAMP,
|
|
PULSE_INSTALL_DIR=str(self.install), PULSE_CONFIG_DIR=str(self.config))
|
|
env.update(extra)
|
|
result = subprocess.run(["bash", "-c", PRELUDE + body], cwd=self.root,
|
|
env=env, capture_output=True, text=True, timeout=10)
|
|
self.assertEqual(result.returncode, 0, result.stdout + result.stderr)
|
|
return result
|
|
|
|
def completed(self):
|
|
return sorted(self.config.parent.glob(self.config.name + ".backup.*"))
|
|
|
|
def assert_no_stages(self):
|
|
self.assertEqual(list(self.config.parent.glob("." + self.config.name + ".backup.*")), [])
|
|
|
|
def assert_prior(self, old):
|
|
self.assertEqual((old / "settings.json").read_text(), "prior recovery state\n")
|
|
self.assertEqual(sorted(p.name for p in old.iterdir()), ["settings.json"])
|
|
|
|
def test_two_successful_same_second_copies_are_separate_complete_snapshots(self):
|
|
old = self.old_snapshot()
|
|
self.shell(r'''
|
|
backup_existing
|
|
printf '%s\n' "$LAST_CONFIG_BACKUP_DIR" > "$FIXTURE/first"
|
|
printf 'second state\n' > "$CONFIG_DIR/settings.json"
|
|
backup_existing
|
|
printf '%s\n' "$LAST_CONFIG_BACKUP_DIR" > "$FIXTURE/second"
|
|
''')
|
|
first = Path((self.root / "first").read_text().strip())
|
|
second = Path((self.root / "second").read_text().strip())
|
|
self.assertNotEqual(first, second)
|
|
self.assertEqual((first / "settings.json").read_text(), "current state\n")
|
|
self.assertEqual((second / "settings.json").read_text(), "second state\n")
|
|
self.assertEqual(len(self.completed()), 3)
|
|
for snapshot in (first, second):
|
|
self.assertEqual((snapshot / ".env").read_bytes(), (self.config / ".env").read_bytes())
|
|
self.assertEqual(stat.S_IMODE((snapshot / ".env").stat().st_mode), 0o600)
|
|
self.assertEqual(stat.S_IMODE(snapshot.stat().st_mode), 0o700)
|
|
self.assertTrue((snapshot / "empty").is_dir())
|
|
self.assertTrue((snapshot / "setting-link").is_symlink())
|
|
self.assertEqual(os.readlink(snapshot / "setting-link"), "settings.json")
|
|
self.assert_prior(old)
|
|
self.assert_no_stages()
|
|
|
|
def test_failed_partial_copy_does_not_mutate_or_delete_a_previous_snapshot(self):
|
|
old = self.old_snapshot()
|
|
result = self.shell(r'''
|
|
LAST_CONFIG_BACKUP_DIR="$CONFIG_DIR.backup.$STAMP"
|
|
cp() { command cp "$@"; return 23; }
|
|
if backup_existing; then exit 90; fi
|
|
test -z "$LAST_CONFIG_BACKUP_DIR"
|
|
''')
|
|
self.assertIn("existing snapshots are unchanged", result.stderr)
|
|
self.assert_prior(old)
|
|
self.assertEqual(self.completed(), [old])
|
|
self.assert_no_stages()
|
|
|
|
def test_allocation_failure_stops_before_copy_or_rotation(self):
|
|
old = self.old_snapshot()
|
|
self.shell(r'''
|
|
mktemp() { return 23; }
|
|
cp() { echo unexpected-copy > "$FIXTURE/unexpected"; return 97; }
|
|
prune_config_backups() { echo unexpected-prune > "$FIXTURE/unexpected"; }
|
|
if backup_existing; then exit 90; fi
|
|
test -z "$LAST_CONFIG_BACKUP_DIR"
|
|
''')
|
|
self.assertFalse((self.root / "unexpected").exists())
|
|
self.assert_prior(old)
|
|
self.assert_no_stages()
|
|
|
|
def test_headroom_failure_creates_no_stage_and_preserves_prior_snapshot(self):
|
|
old = self.old_snapshot()
|
|
self.shell('if backup_existing; then exit 90; fi\ntest -z "$LAST_CONFIG_BACKUP_DIR"\n', HEADROOM_EXIT="23")
|
|
self.assert_prior(old)
|
|
self.assert_no_stages()
|
|
|
|
def test_failed_publication_does_not_rotate_or_publish_a_snapshot(self):
|
|
old = self.old_snapshot()
|
|
self.shell(r'''
|
|
mv() { return 23; }
|
|
prune_config_backups() { echo unexpected-prune > "$FIXTURE/unexpected"; }
|
|
if backup_existing; then exit 90; fi
|
|
test -z "$LAST_CONFIG_BACKUP_DIR"
|
|
''')
|
|
self.assertFalse((self.root / "unexpected").exists())
|
|
self.assertEqual(self.completed(), [old])
|
|
self.assert_prior(old)
|
|
self.assert_no_stages()
|
|
|
|
def test_exit_zero_skipped_publication_preserves_file_directory_and_link(self):
|
|
for kind in ("file", "directory", "link"):
|
|
with self.subTest(kind=kind):
|
|
old = self.old_snapshot("collision-" + kind)
|
|
result = self.shell(r'''
|
|
mv() {
|
|
local destination="${@: -1}"
|
|
printf '%s\n' "$destination" > "$FIXTURE/collision"
|
|
case "$KIND" in
|
|
file) printf 'do not replace\n' > "$destination" ;;
|
|
directory) mkdir "$destination"; printf 'do not replace\n' > "$destination/marker" ;;
|
|
link) ln -s "$FIXTURE/absent-target" "$destination" ;;
|
|
esac
|
|
command mv "$@"
|
|
}
|
|
if backup_existing; then exit 90; fi
|
|
test -z "$LAST_CONFIG_BACKUP_DIR"
|
|
''', KIND=kind)
|
|
collision = Path((self.root / "collision").read_text().strip())
|
|
self.assertIn("Could not publish configuration backup", result.stderr)
|
|
if kind == "file":
|
|
self.assertEqual(collision.read_text(), "do not replace\n")
|
|
collision.unlink()
|
|
elif kind == "directory":
|
|
self.assertEqual([p.name for p in collision.iterdir()], ["marker"])
|
|
self.assertEqual((collision / "marker").read_text(), "do not replace\n")
|
|
(collision / "marker").unlink()
|
|
collision.rmdir()
|
|
else:
|
|
self.assertTrue(collision.is_symlink())
|
|
self.assertEqual(os.readlink(collision), str(self.root / "absent-target"))
|
|
collision.unlink()
|
|
self.assert_prior(old)
|
|
self.assert_no_stages()
|
|
|
|
def test_current_snapshot_survives_rotation_even_after_clock_moves_backwards(self):
|
|
for day in range(24, 29):
|
|
self.old_snapshot(f"202610{day}-154500")
|
|
self.shell('backup_existing\nprintf "%s\\n" "$LAST_CONFIG_BACKUP_DIR" > "$FIXTURE/latest"\n')
|
|
latest = Path((self.root / "latest").read_text().strip())
|
|
self.assertTrue(latest.is_dir())
|
|
self.assertEqual((latest / "settings.json").read_text(), "current state\n")
|
|
self.assertEqual(len(self.completed()), 5)
|
|
self.assertFalse((self.config.parent / (self.config.name + ".backup.20261024-154500")).exists())
|
|
self.assert_no_stages()
|
|
|
|
def test_incomplete_copy_is_not_another_invocations_rotation_candidate(self):
|
|
self.shell(r'''
|
|
cp() {
|
|
local stage="${@: -1}" peer="$CONFIG_DIR.backup.20261024-154500.peer"
|
|
test -d "$stage" || return 91
|
|
mkdir "$peer"
|
|
CONFIG_BACKUP_KEEP_COUNT=1 prune_config_backups "$(dirname "$CONFIG_DIR")" "$(basename "$CONFIG_DIR").backup." "$peer"
|
|
test -d "$stage" || return 92
|
|
command cp "$@"
|
|
}
|
|
backup_existing
|
|
test -f "$LAST_CONFIG_BACKUP_DIR/settings.json"
|
|
''')
|
|
self.assert_no_stages()
|
|
|
|
def test_discard_removes_only_this_invocations_new_snapshot(self):
|
|
old = self.old_snapshot()
|
|
self.shell(r'''
|
|
backup_existing
|
|
current="$LAST_CONFIG_BACKUP_DIR"
|
|
discard_last_config_backup
|
|
test -z "$LAST_CONFIG_BACKUP_DIR"
|
|
test ! -e "$current"
|
|
''')
|
|
self.assert_prior(old)
|
|
self.assertEqual(self.completed(), [old])
|
|
self.assert_no_stages()
|
|
|
|
def test_missing_config_clears_stale_ownership_without_deleting_prior_snapshot(self):
|
|
old = self.old_snapshot()
|
|
self.shell(r'''
|
|
LAST_CONFIG_BACKUP_DIR="$CONFIG_DIR.backup.$STAMP"
|
|
CONFIG_DIR="$FIXTURE/absent-config"
|
|
backup_existing
|
|
test -z "$LAST_CONFIG_BACKUP_DIR"
|
|
discard_last_config_backup
|
|
''')
|
|
self.assert_prior(old)
|
|
self.assert_no_stages()
|
|
|
|
def test_read_only_config_parent_uses_the_existing_hardened_unit_fallback(self):
|
|
if os.geteuid() == 0:
|
|
self.skipTest("root bypasses ordinary write permissions")
|
|
self.config.parent.chmod(0o500)
|
|
try:
|
|
self.shell('backup_existing\nprintf "%s\\n" "$LAST_CONFIG_BACKUP_DIR" > "$FIXTURE/latest"\n')
|
|
latest = Path((self.root / "latest").read_text().strip())
|
|
self.assertEqual(latest.parent, self.install / "config-backups")
|
|
self.assertEqual((latest / "settings.json").read_text(), "current state\n")
|
|
self.assertEqual(list(latest.parent.glob(".*.backup.*")), [])
|
|
finally:
|
|
self.config.parent.chmod(0o700)
|
|
|
|
def test_every_update_entry_stops_on_failed_backup_even_in_an_or_list(self):
|
|
for flow in ("pinned", "menu-update", "menu-reinstall"):
|
|
with self.subTest(flow=flow):
|
|
result = self.shell(r'''
|
|
CURRENT_VERSION=v6.5.0
|
|
FORCE_VERSION=""
|
|
FORCE_CHANNEL=stable
|
|
IN_CONTAINER=true
|
|
IN_DOCKER=false
|
|
[[ "$FLOW" != pinned ]] || FORCE_VERSION=v6.6.0
|
|
print_header() { :; }
|
|
check_root() { :; }
|
|
detect_os() { :; }
|
|
check_docker_environment() { :; }
|
|
check_existing_installation() { return 0; }
|
|
check_proxmox_host() { return 1; }
|
|
resolve_latest_release_tag_for_channel() { printf 'v6.6.0\n'; }
|
|
detect_service_name() { printf 'pulse\n'; }
|
|
safe_read() { local selection=1; [[ "$FLOW" != menu-reinstall ]] || selection=2; printf -v "$2" '%s' "$selection"; }
|
|
offer_existing_auto_updates() { :; }
|
|
run_upgrade_readiness_preflight() { :; }
|
|
backup_existing() { printf 'backup refused\n' >&2; return 23; }
|
|
create_user() { echo user > "$FIXTURE/unexpected"; }
|
|
download_pulse() { echo download > "$FIXTURE/unexpected"; exit 0; }
|
|
status=0
|
|
main || status=$?
|
|
test "$status" = 1
|
|
''', FLOW=flow)
|
|
self.assertIn("backup refused", result.stderr)
|
|
self.assertFalse((self.root / "unexpected").exists())
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|