Require ownership evidence before rotating configuration snapshots

A literal backup prefix can also name another live instance. Publish private source/path/inode receipts without changing copied payloads, and require matching records before rotation or discard. Preserve unmarked historical copies and fail closed on receipt collisions; register filesystem controls and explain retention limits.

Change-source: pulse-maintainer
This commit is contained in:
pulse-triage[bot]
2026-10-31 07:52:43 +00:00
parent b3fb7d2e4e
commit 4c8e2eb075
7 changed files with 348 additions and 16 deletions
+9 -2
View File
@@ -298,8 +298,15 @@ sessions and monitoring data, even if their names say only “configuration”.
The installer normally retains five completed copies for that configuration
name, ordered by their timestamped names, not by verified restore readiness.
An update that aborts before replacing files can remove the copy it just made.
Later runs can rotate older copies away, so these are not a durable archive.
New installers rotate only copies with a matching `.pulse-snapshot` ownership
sidecar beside the directory. Older copies without a sidecar, unrecognised
records and replaced directories are left unchanged and reported; they do not
count towards that limit and can still occupy disk space. Do not create or
copy a sidecar to make a directory eligible for deletion. The record identifies
an installer copy; it does not establish that the copy can be restored.
An update that aborts before replacing files can remove the verified copy it
just made. Later runs can rotate older verified copies away, so these are not
a durable archive.
An absent copy does not prove that the update completed or that data was lost.
Preserve the current state and original evidence; do not delete another
instance's snapshots, change retention or reset keys to make recovery fit.
@@ -36,6 +36,41 @@ all operations with private text files and inert ownership/lifecycle doubles,
including every failure boundary and all five callers. It is source-bound
failure/preservation proof, not native signed upgrade or original recovery.
### Configuration snapshot deletion requires ownership evidence
A directory name, even an exact timestamp-shaped snapshot name, is not deletion
ownership. Before rotation or pre-mutation discard can remove a completed copy,
a bounded regular non-symlink `.pulse-snapshot` sidecar must match the format
version, exact configured source pathname, exact snapshot pathname and current
directory device/inode. NUL-delimited fields preserve all pathname bytes. A
missing, malformed, oversized, foreign, symlinked or stale record never admits
deletion. This is accidental-deletion protection, not defence against a hostile
privileged writer or a power-loss/concurrent-filesystem transaction guarantee.
Publication prepares the mode-0600 sidecar in the private outer stage, without
altering or adding files inside the configuration payload. Both final renames
are non-overwriting and check that their source disappeared. Receipt-preparation
failure leaves no published copy and never rotates. If receipt publication fails
after the copy moved, the new unmarked private copy remains, the update fails
with its location and no rotation, and prior copies/foreign receipts stay intact.
Only recognised copies count toward default, positive and zero retention. Old
unmarked snapshots are deliberately preserved, not automatically adopted from
their names; one warning reports the skipped count. This changes the earlier
legacy-name eligibility choice to prevent deletion of a different live directory
sharing that prefix. Operators can still inspect/recover older copies privately;
the ownership record is not restore-readiness or application-consistency proof.
Successful pruning/discard removes that copy's sidecar too. A replaced directory
must survive even if its old receipt remains. Existing fallback, privacy, literal
pathname, clock reversal and explicit backup-failure admission remain in force.
`TestRootInstallScriptConfigBackupPublicationPreservesPriorSnapshots` registers
real-file controls for each receipt boundary, unknown/legacy preservation at
0/1/5 retention, repeated real publications, unchanged copied hidden files,
late receipt collisions and replaced-directory rotation/discard. The in-app
installer guide uses the real Markdown renderer and explains disk-space and
recovery limits without instructing operators to fabricate ownership records.
### Configuration snapshot rotation stays in its literal instance namespace
The installer's configuration snapshot rotation treats the configured basename
@@ -47,7 +82,8 @@ the C locale. Only immediate real directories qualify: regular files, directory
symlinks, broken links and absent inventories are not removal candidates.
Default/fixed retention counts, explicit zero retention and preservation of the
current completed snapshot after a backwards clock step remain unchanged. The
current completed snapshot after a backwards clock step remain unchanged for
copies admitted by the ownership receipt above. Unrecognised copies stay put. The
existing registered `TestRootInstallScriptConfigBackupPublicationPreservesPriorSnapshots`
executes the real backup/rotation with literal-name, embedded-newline, foreign
state, file/link and empty-inventory controls. These private text-file fixtures
@@ -9064,8 +9100,9 @@ Incomplete hidden stages are not retention candidates. After publication,
rotation still keeps the configured number of snapshots (five by default),
including the current invocation's completed copy for positive retention counts
after same-second retries or a backwards clock step. Zero still disables
completed-snapshot retention. Legacy timestamp-only snapshots remain eligible for
ordinary rotation, not rewritten or treated as incomplete. The existing
completed-snapshot retention for recognised copies. Legacy timestamp-only copies
without ownership receipts stay unchanged and outside the retention count; names
alone do not establish deletion ownership. The existing
read-only-parent fallback and disk-headroom admission remain. Discard after a
pre-mutation staging failure names only this invocation's completed snapshot.
All three existing-install update/reinstall entries explicitly stop if the
+9 -2
View File
@@ -298,8 +298,15 @@ sessions and monitoring data, even if their names say only “configuration”.
The installer normally retains five completed copies for that configuration
name, ordered by their timestamped names, not by verified restore readiness.
An update that aborts before replacing files can remove the copy it just made.
Later runs can rotate older copies away, so these are not a durable archive.
New installers rotate only copies with a matching `.pulse-snapshot` ownership
sidecar beside the directory. Older copies without a sidecar, unrecognised
records and replaced directories are left unchanged and reported; they do not
count towards that limit and can still occupy disk space. Do not create or
copy a sidecar to make a directory eligible for deletion. The record identifies
an installer copy; it does not establish that the copy can be restored.
An update that aborts before replacing files can remove the verified copy it
just made. Later runs can rotate older verified copies away, so these are not
a durable archive.
An absent copy does not prove that the update completed or that data was lost.
Preserve the current state and original evidence; do not delete another
instance's snapshots, change retention or reset keys to make recovery fit.
@@ -79,6 +79,21 @@ describe('installer configuration snapshot guidance', () => {
expect(rendered.querySelector('pre')).toBeNull();
});
it('explains the ownership limit without adopting unmarked or historical directories', () => {
const copy = section().textContent!.replace(/\s+/g, ' ');
for (const distinction of [
'matching .pulse-snapshot ownership sidecar',
'Older copies without a sidecar',
'left unchanged and reported',
'do not count towards that limit',
'can still occupy disk space',
'Do not create or copy a sidecar',
'does not establish that the copy can be restored',
]) {
expect(copy).toContain(distinction);
}
});
it('reaches real shipped recovery sections from the existing installation guide', () => {
expect(
article('INSTALL').querySelector(
+56 -3
View File
@@ -3052,6 +3052,29 @@ create_user() {
fi
}
config_backup_receipt_matches() {
local backup_dir="$1" receipt="$1.pulse-snapshot"
local size identity version source recorded_path recorded_identity remainder=""
# A name is not evidence that this installer created the directory. Bind
# the private sidecar to the exact source, destination and directory inode.
# Replaced directories, copied receipts, links and malformed records stay put.
[[ -d "$backup_dir" && ! -L "$backup_dir" && -f "$receipt" && ! -L "$receipt" ]] || return 1
size=$(stat -c '%s' -- "$receipt") || return 1
[[ "$size" =~ ^[0-9]+$ ]] && (( size <= 16384 )) || return 1
identity=$(stat -c '%d:%i' -- "$backup_dir") || return 1
{
IFS= read -r -d '' version &&
IFS= read -r -d '' source &&
IFS= read -r -d '' recorded_path &&
IFS= read -r -d '' recorded_identity || return 1
if IFS= read -r -d '' remainder || [[ -n "$remainder" ]]; then
return 1
fi
[[ "$version" == pulse-config-snapshot-v1 && "$source" == "$CONFIG_DIR" &&
"$recorded_path" == "$backup_dir" && "$recorded_identity" == "$identity" ]]
} < "$receipt"
}
prune_config_backups() {
# Unattended updates create one snapshot per run; without rotation they
# fill small root filesystems (issue #1646). Keep the newest snapshots
@@ -3064,7 +3087,7 @@ prune_config_backups() {
local current_backup="${3:-}" current_count=0
local LC_ALL=C
local backups=()
local entry
local entry unrecognised=0
# Expand only the final wildcard, never the instance's literal prefix.
# A find -name pattern borrows []?* from a custom config name; line-based
@@ -3073,12 +3096,19 @@ prune_config_backups() {
# Match find's non-following directory scope, including an empty inventory.
for entry in "$backup_parent/$snapshot_prefix"*; do
[[ -d "$entry" && ! -L "$entry" ]] || continue
if ! config_backup_receipt_matches "$entry"; then
(( unrecognised += 1 ))
continue
fi
if [[ "$entry" == "$current_backup" ]] && (( keep > 0 )); then
current_count=1
continue
fi
backups+=("$entry")
done
if (( unrecognised > 0 )); then
print_warn "Left $unrecognised unrecognised configuration backup directories unchanged; retention applies only to verified installer snapshots"
fi
local excess=$(( ${#backups[@]} + current_count - keep ))
if (( excess <= 0 )); then
@@ -3089,6 +3119,8 @@ prune_config_backups() {
for (( i = 0; i < excess; i++ )); do
if ! rm -rf -- "${backups[$i]}"; then
print_warn "Could not remove old configuration backup ${backups[$i]}"
elif ! rm -f -- "${backups[$i]}.pulse-snapshot"; then
print_warn "Could not remove configuration backup receipt ${backups[$i]}.pulse-snapshot"
fi
done
}
@@ -3097,7 +3129,7 @@ backup_existing() {
LAST_CONFIG_BACKUP_DIR=""
if [[ -d "$CONFIG_DIR" ]]; then
print_info "Backing up existing configuration..."
local backup_parent backup_dir backup_stage backup_payload stage_name config_name
local backup_parent backup_dir backup_stage backup_payload stage_name config_name backup_identity
# Command substitution strips every trailing newline, including bytes
# in the configured name or its parent. Keep a sentinel until after
# capture, then remove only the utility's one output newline and marker.
@@ -3147,6 +3179,17 @@ backup_existing() {
print_error "Could not protect configuration backup; existing snapshots are unchanged"
return 1
fi
# The sidecar is outside the copied tree: never overwrite a source
# file or add metadata to the recovery payload itself. NUL delimiters
# retain every pathname byte, including embedded/trailing newlines.
if ! backup_identity=$(stat -c '%d:%i' -- "$backup_payload") ||
! (umask 077; printf 'pulse-config-snapshot-v1\0%s\0%s\0%s\0' \
"$CONFIG_DIR" "$backup_dir" "$backup_identity" > "$backup_stage/receipt") ||
[[ -e "$backup_dir.pulse-snapshot" || -L "$backup_dir.pulse-snapshot" ]]; then
rm -rf -- "$backup_stage"
print_error "Could not prepare configuration backup ownership receipt; existing snapshots are unchanged"
return 1
fi
# GNU mv -n can return zero without moving anything. Admit publication
# only when our stage actually disappeared; never merge or overwrite a
# file, directory or link that appeared at the final path.
@@ -3155,6 +3198,12 @@ backup_existing() {
print_error "Could not publish configuration backup; existing snapshots are unchanged"
return 1
fi
if ! mv -T -n -- "$backup_stage/receipt" "$backup_dir.pulse-snapshot" ||
[[ -e "$backup_stage/receipt" || -L "$backup_stage/receipt" ]]; then
rm -rf -- "$backup_stage"
print_error "Configuration copy retained at $backup_dir, but its ownership receipt could not be published; no snapshots were rotated"
return 1
fi
rmdir -- "$backup_stage" || print_warn "Could not remove empty configuration backup staging directory"
LAST_CONFIG_BACKUP_DIR="$backup_dir"
prune_config_backups "$backup_parent" "$config_name.backup." "$backup_dir"
@@ -3168,8 +3217,12 @@ backup_existing() {
# automatic retry until the filesystem fills (#2127).
discard_last_config_backup() {
if [[ -n "${LAST_CONFIG_BACKUP_DIR:-}" && -e "$LAST_CONFIG_BACKUP_DIR" ]]; then
if ! rm -rf -- "$LAST_CONFIG_BACKUP_DIR"; then
if ! config_backup_receipt_matches "$LAST_CONFIG_BACKUP_DIR"; then
print_warn "Configuration backup ownership is unverified; leaving $LAST_CONFIG_BACKUP_DIR unchanged"
elif ! rm -rf -- "$LAST_CONFIG_BACKUP_DIR"; then
print_warn "Could not remove unneeded configuration backup ${LAST_CONFIG_BACKUP_DIR}"
elif ! rm -f -- "$LAST_CONFIG_BACKUP_DIR.pulse-snapshot"; then
print_warn "Could not remove configuration backup receipt $LAST_CONFIG_BACKUP_DIR.pulse-snapshot"
fi
fi
LAST_CONFIG_BACKUP_DIR=""
+20 -4
View File
@@ -604,8 +604,13 @@ func TestRootInstallScriptConfigBackupRotatesOldSnapshots(t *testing.T) {
` + extractRootInstallShellFunction(t, "get_available_bytes_for_path") + `
` + extractRootInstallShellFunction(t, "get_directory_size_bytes") + `
` + extractRootInstallShellFunction(t, "ensure_config_backup_headroom") + `
` + extractRootInstallShellFunction(t, "prune_config_backups") + `
` + extractRootInstallShellFunction(t, "config_backup_receipt_matches") + "\n" +
extractRootInstallShellFunction(t, "prune_config_backups") + `
` + extractRootInstallShellFunction(t, "backup_existing") + `
for entry in "$INSTALL_DIR/config-backups"/*; do
(umask 077; printf 'pulse-config-snapshot-v1\0%s\0%s\0%s\0' \
"$CONFIG_DIR" "$entry" "$(stat -c '%d:%i' -- "$entry")" > "$entry.pulse-snapshot")
done
date() { printf '20260706-020000\n'; }
chmod a-w "$(dirname "$CONFIG_DIR_UNDER_TEST")" 2>/dev/null || true
trap 'chmod u+w "$(dirname "$CONFIG_DIR_UNDER_TEST")" 2>/dev/null || true' EXIT
@@ -628,7 +633,9 @@ func TestRootInstallScriptConfigBackupRotatesOldSnapshots(t *testing.T) {
}
names := make([]string, 0, len(entries))
for _, entry := range entries {
names = append(names, entry.Name())
if entry.IsDir() {
names = append(names, entry.Name())
}
}
if len(names) != 5 {
t.Fatalf("expected 5 snapshots after rotation, got %d: %v", len(names), names)
@@ -675,7 +682,8 @@ func TestRootInstallScriptDiscardsUnneededConfigBackup(t *testing.T) {
` + extractRootInstallShellFunction(t, "get_available_bytes_for_path") + `
` + extractRootInstallShellFunction(t, "get_directory_size_bytes") + `
` + extractRootInstallShellFunction(t, "ensure_config_backup_headroom") + `
` + extractRootInstallShellFunction(t, "prune_config_backups") + `
` + extractRootInstallShellFunction(t, "config_backup_receipt_matches") + "\n" +
extractRootInstallShellFunction(t, "prune_config_backups") + `
` + extractRootInstallShellFunction(t, "backup_existing") + `
` + extractRootInstallShellFunction(t, "discard_last_config_backup") + `
date() { printf '20260920-160000\n'; }
@@ -2960,7 +2968,8 @@ func TestRootInstallScriptConfigBackupZeroRetention(t *testing.T) {
print_info() { :; }; print_warn() { :; }; print_error() { :; }
ensure_config_backup_headroom() { :; }
CONFIG_BACKUP_KEEP_COUNT=0
` + extractRootInstallShellFunction(t, "prune_config_backups") + "\n" +
` + extractRootInstallShellFunction(t, "config_backup_receipt_matches") + "\n" +
extractRootInstallShellFunction(t, "prune_config_backups") + "\n" +
extractRootInstallShellFunction(t, "backup_existing") + `
backup_existing
test -f "$CONFIG_DIR/settings.json"
@@ -3000,6 +3009,13 @@ func TestRootInstallScriptConfigBackupPublicationPreservesPriorSnapshots(t *test
"test_discard_removes_only_this_invocations_new_snapshot",
"test_missing_config_clears_stale_ownership_without_deleting_prior_snapshot",
"test_read_only_config_parent_uses_the_existing_hardened_unit_fallback",
"test_unmarked_prefix_directories_and_legacy_copies_are_never_deleted",
"test_only_exact_bounded_regular_ownership_receipts_admit_rotation",
"test_receipt_is_private_and_does_not_change_the_copied_configuration",
"test_receipt_preparation_failure_stops_without_publication_or_rotation",
"test_receipt_publication_failures_retain_the_copy_without_rotation",
"test_repeated_real_publications_rotate_receipts_and_preserve_legacy_copies",
"test_replaced_snapshot_cannot_be_deleted_by_rotation_or_discard",
"test_every_update_entry_stops_on_failed_backup_even_in_an_or_list",
}, 30*time.Second, "PULSE_INSTALLER_UNDER_TEST="+repoFile("install.sh"))
}
+199 -2
View File
@@ -43,12 +43,24 @@ class ConfigBackupPublication(unittest.TestCase):
self.install = self.root / "install"
self.install.mkdir()
def old_snapshot(self, stamp=STAMP):
def old_snapshot(self, stamp=STAMP, owned=True):
path = self.config.parent / (self.config.name + ".backup." + stamp)
path.mkdir()
(path / "settings.json").write_text("prior recovery state\n")
if owned:
self.write_receipt(path)
return path
def write_receipt(self, path, source=None, destination=None, identity=None):
observed = path.stat()
fields = (b"pulse-config-snapshot-v1", os.fsencode(source or self.config),
os.fsencode(destination or path),
(identity or f"{observed.st_dev}:{observed.st_ino}").encode())
receipt = Path(str(path) + ".pulse-snapshot")
receipt.write_bytes(b"\0".join(fields) + b"\0")
receipt.chmod(0o600)
return receipt
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,
@@ -62,7 +74,9 @@ class ConfigBackupPublication(unittest.TestCase):
def completed(self):
# The configured namespace is data, not a glob in the oracle either.
return sorted(p for p in self.config.parent.iterdir()
if p.name.startswith(self.config.name + ".backup."))
if p.is_dir() and not p.is_symlink()
and p.name.startswith(self.config.name + ".backup.")
and not p.name.endswith(".pulse-snapshot"))
def assert_no_stages(self):
self.assertEqual([p for p in self.config.parent.iterdir()
@@ -433,6 +447,189 @@ discard_last_config_backup
finally:
self.config.parent.chmod(0o700)
def test_unmarked_prefix_directories_and_legacy_copies_are_never_deleted(self):
for keep in (0, 1, 5):
with self.subTest(keep=keep):
self.config = self.root / f"unmarked-{keep}" / "config"
self.config.mkdir(parents=True)
(self.config / "settings.json").write_text("selected live state\n")
peers = [self.old_snapshot("000-live", owned=False),
self.old_snapshot("19000101-000000.Abc123", owned=False)]
legacy = [self.old_snapshot(f"202610{day:02d}-000000", owned=False)
for day in range(1, 7)]
for peer in peers:
(peer / ".encryption.key").write_text("synthetic peer key\n")
result = self.shell('backup_existing\n', CONFIG_BACKUP_KEEP_COUNT=str(keep))
self.assertIn("unrecognised", result.stderr)
for path in peers + legacy:
self.assertEqual((path / "settings.json").read_text(), "prior recovery state\n")
for peer in peers:
self.assertEqual((peer / ".encryption.key").read_text(), "synthetic peer key\n")
self.assertEqual(len(self.completed()), len(peers + legacy) + min(keep, 1))
self.assertEqual((self.config / "settings.json").read_text(), "selected live state\n")
self.assert_no_stages()
def test_only_exact_bounded_regular_ownership_receipts_admit_rotation(self):
for kind in ("source", "destination", "identity", "version", "truncated",
"trailing", "extra-null", "oversized", "link", "directory"):
with self.subTest(kind=kind):
old = self.old_snapshot("invalid-" + kind)
receipt = Path(str(old) + ".pulse-snapshot")
if kind == "source":
self.write_receipt(old, source=self.root / "other-config")
elif kind == "destination":
self.write_receipt(old, destination=self.root / "other-copy")
elif kind == "identity":
self.write_receipt(old, identity="0:0")
elif kind == "version":
receipt.write_bytes(receipt.read_bytes().replace(b"-v1", b"-v2", 1))
elif kind == "truncated":
receipt.write_bytes(receipt.read_bytes()[:-1])
elif kind == "trailing":
receipt.write_bytes(receipt.read_bytes() + b"unexpected")
elif kind == "extra-null":
receipt.write_bytes(receipt.read_bytes() + b"\0")
elif kind == "oversized":
receipt.write_bytes(b"x" * 16385)
else:
saved = self.root / ("receipt-" + kind)
receipt.rename(saved)
if kind == "link":
receipt.symlink_to(saved)
else:
receipt.mkdir()
self.shell('backup_existing\n', CONFIG_BACKUP_KEEP_COUNT="0")
self.assert_prior(old)
self.assertTrue(receipt.exists())
self.assert_no_stages()
def test_receipt_is_private_and_does_not_change_the_copied_configuration(self):
(self.config / ".pulse-snapshot").write_text("existing private source file\n")
self.shell('backup_existing\nprintf "%s" "$LAST_CONFIG_BACKUP_DIR" > "$FIXTURE/latest"\n')
latest = Path((self.root / "latest").read_text())
receipt = Path(str(latest) + ".pulse-snapshot")
observed = latest.stat()
self.assertEqual(receipt.read_bytes(), b"\0".join((b"pulse-config-snapshot-v1",
os.fsencode(self.config), os.fsencode(latest),
f"{observed.st_dev}:{observed.st_ino}".encode())) + b"\0")
self.assertEqual(stat.S_IMODE(receipt.stat().st_mode), 0o600)
self.assertEqual((latest / ".pulse-snapshot").read_bytes(),
(self.config / ".pulse-snapshot").read_bytes())
self.assertEqual((latest / ".env").read_bytes(), (self.config / ".env").read_bytes())
self.assert_no_stages()
def test_receipt_preparation_failure_stops_without_publication_or_rotation(self):
for kind in ("stat", "write", "collision"):
with self.subTest(kind=kind):
old = self.old_snapshot("prepare-" + kind)
result = self.shell(r'''
case "$KIND" in
stat) stat() { return 23; } ;;
write) printf() { [[ "$1" != pulse-config-snapshot-v1* ]] || return 23; builtin printf "$@"; } ;;
collision) mktemp() {
local stage
stage=$(command mktemp "$@") || return
local name="${stage##*/}"
local final="${stage%/*}/${name#.}"
printf 'foreign receipt\n' > "$final.pulse-snapshot"
printf '%s' "$final.pulse-snapshot" > "$FIXTURE/collision-receipt"
printf '%s\n' "$stage"
} ;;
esac
if backup_existing; then exit 90; fi
test -z "$LAST_CONFIG_BACKUP_DIR"
''', KIND=kind)
self.assertIn("Could not prepare configuration backup ownership receipt", result.stderr)
self.assert_prior(old)
self.assert_no_stages()
if kind == "collision":
receipt = Path((self.root / "collision-receipt").read_text())
self.assertEqual(receipt.read_text(), "foreign receipt\n")
self.assertFalse(Path(str(receipt).removesuffix(".pulse-snapshot")).exists())
def test_receipt_publication_failures_retain_the_copy_without_rotation(self):
for kind in ("failure", "skipped", "file", "directory", "link"):
with self.subTest(kind=kind):
old = self.old_snapshot("publish-" + kind)
result = self.shell(r'''
mv() {
local destination="${@: -1}"
if [[ "$destination" == *.pulse-snapshot ]]; then
printf '%s' "${destination%.pulse-snapshot}" > "$FIXTURE/retained-copy"
case "$KIND" in
failure) return 23 ;;
skipped) return 0 ;;
file) printf 'foreign receipt\n' > "$destination" ;;
directory) mkdir "$destination"; printf 'foreign receipt\n' > "$destination/retained" ;;
link) ln -s "$FIXTURE/missing-receipt" "$destination" ;;
esac
fi
command mv "$@"
}
if backup_existing; then exit 90; fi
test -z "$LAST_CONFIG_BACKUP_DIR"
''', KIND=kind)
self.assertIn("no snapshots were rotated", result.stderr)
copy = Path((self.root / "retained-copy").read_text())
self.assertEqual((copy / "settings.json").read_text(), "current state\n")
self.assertEqual((copy / ".env").read_bytes(), (self.config / ".env").read_bytes())
self.assert_prior(old)
self.assert_no_stages()
receipt = Path(str(copy) + ".pulse-snapshot")
if kind == "file":
self.assertEqual(receipt.read_text(), "foreign receipt\n")
elif kind == "directory":
self.assertEqual((receipt / "retained").read_text(), "foreign receipt\n")
elif kind == "link":
self.assertEqual(os.readlink(receipt), str(self.root / "missing-receipt"))
else:
self.assertFalse(receipt.exists())
def test_repeated_real_publications_rotate_receipts_and_preserve_legacy_copies(self):
for keep in (0, 1, 5):
with self.subTest(keep=keep):
self.config = self.root / f"real-{keep}" / "config"
self.config.mkdir(parents=True)
(self.config / "settings.json").write_text("selected live state\n")
legacy = self.old_snapshot("20260101-000000", owned=False)
self.shell(r'''
for day in 01 02 03 04 05 06 07 08; do
date() { printf '202610%s-000000\n' "$day"; }
backup_existing
done
''', CONFIG_BACKUP_KEEP_COUNT=str(keep))
self.assert_prior(legacy)
copies = [p for p in self.completed() if p != legacy]
self.assertEqual(len(copies), keep)
receipts = list(self.config.parent.glob("*.pulse-snapshot"))
self.assertEqual(len(receipts), keep)
self.assertEqual({Path(str(p) + ".pulse-snapshot") for p in copies}, set(receipts))
for copy in copies:
self.assertEqual((copy / "settings.json").read_text(), "selected live state\n")
self.assert_no_stages()
def test_replaced_snapshot_cannot_be_deleted_by_rotation_or_discard(self):
for action in ("rotate", "discard"):
with self.subTest(action=action):
result = self.shell(r'''
backup_existing
current="$LAST_CONFIG_BACKUP_DIR"
command mv -- "$current" "$FIXTURE/owned-original-$ACTION"
mkdir -- "$current"
printf 'replacement live state\n' > "$current/settings.json"
if [[ "$ACTION" == rotate ]]; then
CONFIG_BACKUP_KEEP_COUNT=0 prune_config_backups "${current%/*}" "${current##*/}"
else
discard_last_config_backup
test -z "$LAST_CONFIG_BACKUP_DIR"
fi
test "$(cat "$current/settings.json")" = 'replacement live state'
test "$(cat "$FIXTURE/owned-original-$ACTION/settings.json")" = 'current state'
''', ACTION=action)
self.assertIn("unverified" if action == "discard" else "unrecognised", result.stderr)
self.assert_no_stages()
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):