749 lines
32 KiB
Python
749 lines
32 KiB
Python
import asyncio
|
|
import os
|
|
import re
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
|
|
REPO_ROOT = Path(__file__).resolve().parents[2]
|
|
BACKUP_SCRIPT = REPO_ROOT / "app" / "sovran_systemsos_web" / "scripts" / "sovran-hub-backup.sh"
|
|
SERVER_FILE = REPO_ROOT / "app" / "sovran_systemsos_web" / "server.py"
|
|
SUPPORT_JS = REPO_ROOT / "app" / "sovran_systemsos_web" / "static" / "js" / "support.js"
|
|
NIX_HUB_FILE = REPO_ROOT / "modules" / "core" / "sovran-hub.nix"
|
|
|
|
|
|
class ManualBackupWorkflowTests(unittest.TestCase):
|
|
# ── Core design: rsync, no tar, no DB, no LND ─────────────────────────────
|
|
|
|
def test_backup_script_uses_rsync_not_tar(self):
|
|
"""New design: backup must use rsync, not tar archives."""
|
|
source = BACKUP_SCRIPT.read_text()
|
|
self.assertIn("rsync", source, "backup script must use rsync")
|
|
self.assertIn("--archive", source, "rsync must be called with --archive flag")
|
|
self.assertNotIn("tar --create", source, "backup script must not create tar archives")
|
|
self.assertNotIn("--file ", source, "backup script must not write tar --file output")
|
|
|
|
def test_backup_script_has_no_database_dump_functions(self):
|
|
"""Removed: PostgreSQL and MariaDB dump functions must not exist."""
|
|
source = BACKUP_SCRIPT.read_text()
|
|
self.assertNotIn("pg_dump", source, "backup script must not contain pg_dump")
|
|
self.assertNotIn("pg_dumpall", source, "backup script must not contain pg_dumpall")
|
|
self.assertNotIn("mariadb-dump", source, "backup script must not contain mariadb-dump")
|
|
self.assertNotIn("mysqldump", source, "backup script must not contain mysqldump")
|
|
self.assertNotIn("export_postgresql_dumps", source)
|
|
self.assertNotIn("export_mariadb_dumps", source)
|
|
|
|
def test_backup_script_has_no_lnd_service_orchestration(self):
|
|
"""Removed: LND stop/restart and SCB export must not exist."""
|
|
source = BACKUP_SCRIPT.read_text()
|
|
self.assertNotIn("export_lnd_scb_if_possible", source)
|
|
self.assertNotIn("stop_lnd_stack_if_needed", source)
|
|
self.assertNotIn("lncli", source, "backup script must not call lncli")
|
|
self.assertNotIn("LND_STOPPED", source, "backup script must not track LND_STOPPED state")
|
|
self.assertNotIn("LND_UNITS_TO_RESTART", source)
|
|
|
|
def test_backup_script_has_no_tar_dependencies(self):
|
|
"""Removed: sha256sum checksums for tar artifacts must not exist."""
|
|
source = BACKUP_SCRIPT.read_text()
|
|
self.assertNotIn("sha256sum", source, "backup script must not generate tar checksums")
|
|
self.assertNotIn("SHA256SUMS", source, "backup script must not write SHA256SUMS.txt")
|
|
|
|
# ── Rsync options ──────────────────────────────────────────────────────────
|
|
|
|
def test_backup_script_rsync_uses_metadata_preserving_options(self):
|
|
"""Rsync must be called with Linux metadata-preserving options."""
|
|
source = BACKUP_SCRIPT.read_text()
|
|
self.assertIn("--archive", source)
|
|
self.assertIn("--acls", source)
|
|
self.assertIn("--xattrs", source)
|
|
self.assertIn("--hard-links", source)
|
|
self.assertIn("--numeric-ids", source)
|
|
self.assertIn("--one-file-system", source)
|
|
|
|
def test_backup_script_rsync_no_delete(self):
|
|
"""Rsync must NOT use --delete or --delete-delay.
|
|
Accidental source deletion must not silently wipe the backup copy."""
|
|
source = BACKUP_SCRIPT.read_text()
|
|
self.assertNotIn("--delete", source,
|
|
"rsync must not use --delete; accidental source deletion must not erase backup")
|
|
|
|
def test_backup_script_uses_stable_current_mirror_path(self):
|
|
"""Backup must write to a stable 'current/' path, not timestamped directories.
|
|
Later runs should update the same mirror."""
|
|
source = BACKUP_SCRIPT.read_text()
|
|
self.assertIn("Sovran_SystemsOS_Backup/current", source,
|
|
"backup must use a stable 'current' mirror path")
|
|
self.assertIn("BACKUP_SUBPATH", source,
|
|
"stable sub-path must be defined in BACKUP_SUBPATH variable")
|
|
# Must not create new timestamped directories per run
|
|
self.assertNotIn(
|
|
"date '+%Y%m%d_%H%M%S'",
|
|
source,
|
|
"backup must not create timestamped per-run directories",
|
|
)
|
|
|
|
def test_backup_script_source_destination_mapping(self):
|
|
"""Each source tree must be synced to a matching destination path."""
|
|
source = BACKUP_SCRIPT.read_text()
|
|
self.assertIn("/etc/nixos/", source)
|
|
self.assertIn('"$BACKUP_DIR/etc/nixos/"', source)
|
|
self.assertIn("/home/", source)
|
|
self.assertIn('"$BACKUP_DIR/home/"', source)
|
|
self.assertIn("/var/lib/", source)
|
|
self.assertIn('"$BACKUP_DIR/var/lib/"', source)
|
|
|
|
# ── ext4 filesystem validation ─────────────────────────────────────────────
|
|
|
|
def test_backup_script_requires_ext4_filesystem(self):
|
|
"""Backup script must reject non-ext4 filesystems and require ext4."""
|
|
source = BACKUP_SCRIPT.read_text()
|
|
self.assertIn(
|
|
'fstype" != "ext4"',
|
|
source,
|
|
"backup script must check for ext4 filesystem",
|
|
)
|
|
self.assertIn("ext4", source, "backup script must reference ext4")
|
|
self.assertNotIn(
|
|
'"exfat"',
|
|
source,
|
|
"backup script must not accept exFAT (old requirement)",
|
|
)
|
|
self.assertNotIn(
|
|
'"fuseblk"',
|
|
source,
|
|
"backup script must not accept fuseblk/NTFS",
|
|
)
|
|
|
|
def test_backup_script_rejects_unsupported_filesystems_in_error_message(self):
|
|
"""The ext4 rejection message must mention the unsupported filesystem types."""
|
|
source = BACKUP_SCRIPT.read_text()
|
|
self.assertIn(
|
|
"exFAT, FAT32, and NTFS are not supported",
|
|
source,
|
|
"error message must clearly list unsupported filesystem types",
|
|
)
|
|
|
|
def test_backend_accepts_ext4_rejects_exfat(self):
|
|
"""Backend _is_supported_backup_fstype must accept ext4 and reject exFAT."""
|
|
server_source = SERVER_FILE.read_text()
|
|
# Must accept ext4
|
|
self.assertIn(
|
|
'return fstype == "ext4"',
|
|
server_source,
|
|
"backend must accept only ext4",
|
|
)
|
|
# Must not accept exFAT
|
|
self.assertNotIn(
|
|
'fstype == "exfat"',
|
|
server_source,
|
|
"backend must not accept exFAT",
|
|
)
|
|
self.assertNotIn(
|
|
'fuseblk',
|
|
server_source.split("_is_supported_backup_fstype")[1][:500],
|
|
"backend must not accept fuseblk",
|
|
)
|
|
|
|
def test_frontend_requires_ext4_not_exfat(self):
|
|
"""Frontend UI copy must require ext4 and not mention exFAT as the requirement."""
|
|
support_source = SUPPORT_JS.read_text()
|
|
self.assertIn(
|
|
"ext4",
|
|
support_source,
|
|
"support.js must mention ext4 as the required filesystem",
|
|
)
|
|
# The word exFAT must not appear as a requirement (it may appear in a
|
|
# note about unsupported formats, but the requirement itself must say ext4)
|
|
requirements_section = re.search(
|
|
r"Requirements.*?What gets backed up",
|
|
support_source,
|
|
re.DOTALL,
|
|
)
|
|
if requirements_section:
|
|
req_text = requirements_section.group(0)
|
|
self.assertNotIn(
|
|
"Drive must be formatted as <strong>exFAT</strong>",
|
|
req_text,
|
|
"Requirements section must not say 'formatted as exFAT'",
|
|
)
|
|
self.assertIn(
|
|
"ext4",
|
|
req_text,
|
|
"Requirements section must say ext4",
|
|
)
|
|
|
|
def test_frontend_failure_message_says_ext4(self):
|
|
"""Failure message in renderBackupDone must say ext4, not exFAT."""
|
|
support_source = SUPPORT_JS.read_text()
|
|
self.assertNotIn(
|
|
"formatted as exFAT",
|
|
support_source,
|
|
"failure message must not say 'formatted as exFAT'",
|
|
)
|
|
self.assertIn(
|
|
"formatted as ext4",
|
|
support_source,
|
|
"failure message must say 'formatted as ext4'",
|
|
)
|
|
|
|
# ── Database exclusions ────────────────────────────────────────────────────
|
|
|
|
def test_backup_script_excludes_postgresql_and_mariadb(self):
|
|
"""PostgreSQL and MariaDB raw database directories must be excluded."""
|
|
source = BACKUP_SCRIPT.read_text()
|
|
self.assertIn("--exclude='postgresql/'", source,
|
|
"rsync must exclude postgresql directory")
|
|
self.assertIn("--exclude='mysql/'", source,
|
|
"rsync must exclude mysql directory")
|
|
self.assertIn("--exclude='mariadb/'", source,
|
|
"rsync must exclude mariadb directory")
|
|
|
|
def test_backup_script_excludes_bitcoin_and_electrs(self):
|
|
"""Bitcoin blockchain and Electrs index data must be excluded."""
|
|
source = BACKUP_SCRIPT.read_text()
|
|
self.assertIn("--exclude='bitcoind/'", source,
|
|
"rsync must exclude bitcoind directory")
|
|
self.assertIn("--exclude='electrs/'", source,
|
|
"rsync must exclude electrs directory")
|
|
|
|
def test_manifest_states_database_exclusion(self):
|
|
"""The manifest must explicitly state that PostgreSQL and MariaDB are excluded."""
|
|
source = BACKUP_SCRIPT.read_text()
|
|
self.assertIn(
|
|
"PostgreSQL and MariaDB/MySQL application databases are NOT included",
|
|
source,
|
|
"manifest must state that databases are not included",
|
|
)
|
|
self.assertIn(
|
|
"Bitcoin blockchain data and Electrs indexes are NOT included",
|
|
source,
|
|
"manifest must state that blockchain data is not included",
|
|
)
|
|
|
|
def test_manifest_has_no_tar_restore_instructions(self):
|
|
"""Manifest must not mention tar extraction, pg_restore, or LND SCB."""
|
|
source = BACKUP_SCRIPT.read_text()
|
|
self.assertNotIn(
|
|
"tar --acls --xattrs",
|
|
source,
|
|
"manifest must not give tar restore instructions",
|
|
)
|
|
self.assertNotIn(
|
|
"pg_restore",
|
|
source,
|
|
"manifest must not reference pg_restore",
|
|
)
|
|
self.assertNotIn(
|
|
"lnd-static-channel-backup.scb",
|
|
source,
|
|
"manifest must not reference LND SCB restore procedures",
|
|
)
|
|
|
|
def test_manifest_has_rsync_restore_guidance(self):
|
|
"""Manifest must provide rsync-based restore guidance."""
|
|
source = BACKUP_SCRIPT.read_text()
|
|
self.assertIn(
|
|
"rsync -aAXH --numeric-ids",
|
|
source,
|
|
"manifest must show rsync restore command",
|
|
)
|
|
|
|
# ── Exit code 24 (vanished files) for /home only ──────────────────────────
|
|
|
|
def test_backup_script_exit_code_24_nonfatal_for_home(self):
|
|
"""Rsync exit code 24 (vanished files) must be nonfatal for /home only."""
|
|
source = BACKUP_SCRIPT.read_text()
|
|
self.assertIn(
|
|
'"$rc" -eq 24',
|
|
source,
|
|
"backup script must handle rsync exit code 24",
|
|
)
|
|
self.assertIn(
|
|
'allow_vanished" == "yes"',
|
|
source,
|
|
"exit 24 acceptance must be gated on allow_vanished flag",
|
|
)
|
|
|
|
def test_backup_script_home_stage_allows_vanished(self):
|
|
"""Stage 3 (/home) must call run_rsync with allow_vanished=yes."""
|
|
source = BACKUP_SCRIPT.read_text()
|
|
# The /home call must pass "yes" as the allow_vanished argument
|
|
self.assertIn(
|
|
'run_rsync "/home" yes',
|
|
source,
|
|
"/home stage must pass allow_vanished=yes to run_rsync",
|
|
)
|
|
|
|
def test_backup_script_other_stages_disallow_vanished(self):
|
|
"""Stages other than /home must call run_rsync with allow_vanished=no."""
|
|
source = BACKUP_SCRIPT.read_text()
|
|
self.assertIn(
|
|
'run_rsync "/etc/nixos" no',
|
|
source,
|
|
"/etc/nixos stage must use allow_vanished=no",
|
|
)
|
|
self.assertIn(
|
|
'run_rsync "/var/lib" no',
|
|
source,
|
|
"/var/lib stage must use allow_vanished=no",
|
|
)
|
|
|
|
# ── INCOMPLETE / BACKUP_COMPLETE markers ──────────────────────────────────
|
|
|
|
def test_backup_script_writes_incomplete_marker(self):
|
|
"""A backup directory must be marked INCOMPLETE immediately after
|
|
creation, so interrupted or failed runs are identifiable."""
|
|
source = BACKUP_SCRIPT.read_text()
|
|
self.assertIn(
|
|
'touch "$BACKUP_DIR/INCOMPLETE"',
|
|
source,
|
|
"backup script must create INCOMPLETE marker after mkdir",
|
|
)
|
|
|
|
def test_backup_script_writes_backup_complete_marker(self):
|
|
"""A BACKUP_COMPLETE file must be written only after all work succeeds."""
|
|
source = BACKUP_SCRIPT.read_text()
|
|
self.assertIn(
|
|
'BACKUP_COMPLETE"',
|
|
source,
|
|
"backup script must write BACKUP_COMPLETE file on success",
|
|
)
|
|
self.assertIn(
|
|
'rm -f "$BACKUP_DIR/INCOMPLETE"',
|
|
source,
|
|
"backup script must remove INCOMPLETE marker on success",
|
|
)
|
|
|
|
# ── Concurrency lock ──────────────────────────────────────────────────────
|
|
|
|
def test_backup_script_uses_flock_concurrency_lock(self):
|
|
"""The script must acquire an exclusive flock before starting work."""
|
|
source = BACKUP_SCRIPT.read_text()
|
|
self.assertIn("flock", source, "backup script must use flock")
|
|
self.assertIn("LOCK_FILE", source, "backup script must define LOCK_FILE")
|
|
|
|
# ── Status states ─────────────────────────────────────────────────────────
|
|
|
|
def test_backup_script_uses_running_success_failed_states(self):
|
|
"""Status values must be RUNNING, SUCCESS, and FAILED."""
|
|
source = BACKUP_SCRIPT.read_text()
|
|
self.assertIn('set_status "RUNNING"', source)
|
|
self.assertIn('set_status "SUCCESS"', source)
|
|
self.assertIn('set_status "FAILED"', source)
|
|
|
|
def test_backend_and_frontend_use_explicit_backup_terminal_states(self):
|
|
"""Backend and frontend must use consistent terminal state handling."""
|
|
server_source = SERVER_FILE.read_text()
|
|
support_source = SUPPORT_JS.read_text()
|
|
|
|
self.assertIn("_write_backup_status, \"RUNNING\"", server_source)
|
|
self.assertIn("_monitor_backup_subprocess", server_source)
|
|
self.assertIn("asyncio.create_task(_monitor_backup_subprocess(proc))", server_source)
|
|
self.assertIn("result === \"success\" || result === \"failed\"", support_source)
|
|
|
|
# ── Nix service PATH ──────────────────────────────────────────────────────
|
|
|
|
def test_nix_service_path_includes_rsync_and_acl(self):
|
|
"""pkgs.rsync and pkgs.acl must appear in the sovran-hub-web service path
|
|
so that rsync with ACL/xattr support is available at runtime."""
|
|
nix_source = NIX_HUB_FILE.read_text()
|
|
self.assertIn(
|
|
"pkgs.rsync",
|
|
nix_source,
|
|
"pkgs.rsync must be declared in the sovran-hub-web service path",
|
|
)
|
|
self.assertIn(
|
|
"pkgs.acl",
|
|
nix_source,
|
|
"pkgs.acl must be declared in the sovran-hub-web service path",
|
|
)
|
|
|
|
def test_nix_service_path_includes_bash_and_gawk(self):
|
|
"""Regression: pkgs.bash and pkgs.gawk must appear in the service path."""
|
|
nix_source = NIX_HUB_FILE.read_text()
|
|
self.assertIn("pkgs.bash", nix_source,
|
|
"pkgs.bash must be declared in the sovran-hub-web service path")
|
|
self.assertIn("pkgs.gawk", nix_source,
|
|
"pkgs.gawk must be declared in the sovran-hub-web service path")
|
|
|
|
def test_nix_service_path_has_no_gnutar(self):
|
|
"""pkgs.gnutar must be removed from the service path (no longer needed)."""
|
|
nix_source = NIX_HUB_FILE.read_text()
|
|
self.assertNotIn(
|
|
"pkgs.gnutar",
|
|
nix_source,
|
|
"pkgs.gnutar must be removed from the service path (tar is no longer used)",
|
|
)
|
|
|
|
# ── Regression tests for exit-code-127 / missing-interpreter bug ──────────
|
|
|
|
def test_server_resolves_bash_via_shutil_which(self):
|
|
"""Regression: server must locate bash with shutil.which()."""
|
|
source = SERVER_FILE.read_text()
|
|
self.assertIn('shutil.which("bash")', source)
|
|
self.assertNotIn('"/usr/bin/env", "bash"', source)
|
|
|
|
def test_server_logs_actionable_message_when_bash_missing(self):
|
|
"""Regression: when bash is absent the server must write an actionable log."""
|
|
source = SERVER_FILE.read_text()
|
|
self.assertIn("bash_path is None", source)
|
|
self.assertIn("interpreter", source)
|
|
|
|
def test_server_captures_stderr_from_backup_subprocess(self):
|
|
"""Regression: backup subprocess must use stderr=PIPE."""
|
|
source = SERVER_FILE.read_text()
|
|
self.assertIn("stderr=asyncio.subprocess.PIPE", source)
|
|
self.assertIn("stderr_text", source)
|
|
|
|
def test_monitor_includes_stderr_detail_in_failed_message(self):
|
|
"""Regression: _monitor_backup_subprocess must append captured stderr."""
|
|
source = SERVER_FILE.read_text()
|
|
self.assertIn("stderr_chunks", source)
|
|
self.assertIn("stderr_text", source)
|
|
self.assertIn('detail = f" — stderr: {stderr_text}"', source)
|
|
|
|
def test_exit_code_127_subprocess_stderr_drain(self):
|
|
"""Behavioral regression: a subprocess that exits 127 drains stderr without deadlock."""
|
|
async def _run():
|
|
proc = await asyncio.create_subprocess_exec(
|
|
"bash", "-c", "echo 'bash: command not found' >&2; exit 127",
|
|
stdout=asyncio.subprocess.DEVNULL,
|
|
stderr=asyncio.subprocess.PIPE,
|
|
)
|
|
chunks: list[bytes] = []
|
|
|
|
async def _drain():
|
|
async for line in proc.stderr:
|
|
chunks.append(line)
|
|
|
|
drain_task = asyncio.create_task(_drain())
|
|
rc = await proc.wait()
|
|
await drain_task
|
|
return rc, b"".join(chunks).decode()
|
|
|
|
rc, stderr_out = asyncio.run(_run())
|
|
self.assertEqual(rc, 127)
|
|
self.assertIn("bash: command not found", stderr_out)
|
|
|
|
# ── Rsync exit code 24 behavioral tests ──────────────────────────────────
|
|
|
|
def test_run_rsync_exit_24_nonfatal_for_home(self):
|
|
"""Behavioral: run_rsync with allow_vanished=yes treats exit 24 as a
|
|
nonfatal warning (recorded in RSYNC_WARNINGS) and does not fail."""
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
dest = os.path.join(tmpdir, "backup")
|
|
os.makedirs(dest, exist_ok=True)
|
|
|
|
script = f"""#!/usr/bin/env bash
|
|
set -euo pipefail
|
|
RSYNC_WARNINGS=()
|
|
|
|
log() {{ echo "$*"; }}
|
|
fail() {{ echo "FAILED: $*" >&2; exit 1; }}
|
|
|
|
run_rsync() {{
|
|
local label="$1"
|
|
local allow_vanished="$2"
|
|
shift 2
|
|
|
|
local rc=24 # simulate rsync exit 24
|
|
|
|
if [[ "$rc" -eq 0 ]]; then
|
|
return 0
|
|
elif [[ "$allow_vanished" == "yes" && "$rc" -eq 24 ]]; then
|
|
log "NOTE: $label — some files vanished during sync (normal on an active desktop)."
|
|
RSYNC_WARNINGS+=("$label: some files vanished during sync")
|
|
return 0
|
|
else
|
|
fail "rsync failed for $label (exit code $rc)"
|
|
fi
|
|
}}
|
|
|
|
run_rsync "/home" yes /home/ "{dest}/home/"
|
|
echo "WARNINGS: ${{#RSYNC_WARNINGS[@]}}"
|
|
echo "SUCCESS"
|
|
"""
|
|
script_file = os.path.join(tmpdir, "test.sh")
|
|
with open(script_file, "w") as sf:
|
|
sf.write(script)
|
|
result = subprocess.run(["bash", script_file], capture_output=True, text=True)
|
|
self.assertEqual(result.returncode, 0, f"stderr: {result.stderr}")
|
|
self.assertIn("SUCCESS", result.stdout)
|
|
self.assertIn("WARNINGS: 1", result.stdout)
|
|
|
|
def test_run_rsync_exit_24_fatal_for_non_home(self):
|
|
"""Behavioral: run_rsync with allow_vanished=no treats exit 24 as fatal."""
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
dest = os.path.join(tmpdir, "backup")
|
|
os.makedirs(dest, exist_ok=True)
|
|
|
|
script = f"""#!/usr/bin/env bash
|
|
RSYNC_WARNINGS=()
|
|
|
|
log() {{ echo "$*"; }}
|
|
fail() {{ echo "CORRECTLY_FAILED: $*"; exit 1; }}
|
|
|
|
run_rsync() {{
|
|
local label="$1"
|
|
local allow_vanished="$2"
|
|
shift 2
|
|
|
|
local rc=24 # simulate rsync exit 24
|
|
|
|
if [[ "$rc" -eq 0 ]]; then
|
|
return 0
|
|
elif [[ "$allow_vanished" == "yes" && "$rc" -eq 24 ]]; then
|
|
RSYNC_WARNINGS+=("$label: vanished")
|
|
return 0
|
|
else
|
|
fail "rsync failed for $label (exit code $rc)"
|
|
fi
|
|
}}
|
|
|
|
run_rsync "/etc/nixos" no /etc/nixos/ "{dest}/etc/nixos/"
|
|
echo "SHOULD_NOT_REACH_HERE"
|
|
"""
|
|
script_file = os.path.join(tmpdir, "test.sh")
|
|
with open(script_file, "w") as sf:
|
|
sf.write(script)
|
|
result = subprocess.run(["bash", script_file], capture_output=True, text=True)
|
|
self.assertEqual(result.returncode, 1)
|
|
self.assertIn("CORRECTLY_FAILED", result.stdout)
|
|
self.assertNotIn("SHOULD_NOT_REACH_HERE", result.stdout)
|
|
|
|
def test_run_rsync_other_nonzero_codes_always_fatal(self):
|
|
"""Behavioral: rsync exit codes other than 0 and 24 (for home) are fatal."""
|
|
for rc in [1, 10, 11, 12, 23, 25]:
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
dest = os.path.join(tmpdir, "backup")
|
|
os.makedirs(dest, exist_ok=True)
|
|
|
|
script = f"""#!/usr/bin/env bash
|
|
RSYNC_WARNINGS=()
|
|
|
|
fail() {{ echo "CORRECTLY_FAILED: $*"; exit 1; }}
|
|
log() {{ echo "$*"; }}
|
|
|
|
run_rsync() {{
|
|
local label="$1"
|
|
local allow_vanished="$2"
|
|
shift 2
|
|
|
|
local rc={rc} # simulated exit code
|
|
|
|
if [[ "$rc" -eq 0 ]]; then
|
|
return 0
|
|
elif [[ "$allow_vanished" == "yes" && "$rc" -eq 24 ]]; then
|
|
RSYNC_WARNINGS+=("$label: vanished")
|
|
return 0
|
|
else
|
|
fail "rsync failed for $label (exit code $rc)"
|
|
fi
|
|
}}
|
|
|
|
run_rsync "/home" yes /home/ "{dest}/home/"
|
|
echo "SHOULD_NOT_REACH_HERE"
|
|
"""
|
|
script_file = os.path.join(tmpdir, "test.sh")
|
|
with open(script_file, "w") as sf:
|
|
sf.write(script)
|
|
result = subprocess.run(["bash", script_file], capture_output=True, text=True)
|
|
self.assertEqual(result.returncode, 1,
|
|
f"Exit code {rc} should fail: {result.stdout}")
|
|
self.assertIn("CORRECTLY_FAILED", result.stdout)
|
|
|
|
# ── Behavioral: repeated runs target same 'current' mirror ──────────────
|
|
|
|
def test_repeated_runs_use_same_current_directory(self):
|
|
"""Behavioral: a second call to the backup logic must sync to the same
|
|
BACKUP_DIR (not create a new timestamped directory) so only changed
|
|
files are transferred."""
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
target = os.path.join(tmpdir, "drive")
|
|
os.makedirs(target, exist_ok=True)
|
|
|
|
script = f"""#!/usr/bin/env bash
|
|
set -euo pipefail
|
|
TARGET="{target}"
|
|
BACKUP_SUBPATH="Sovran_SystemsOS_Backup/current"
|
|
BACKUP_DIR="${{TARGET}}/${{BACKUP_SUBPATH}}"
|
|
mkdir -p "$BACKUP_DIR"
|
|
echo "$BACKUP_DIR"
|
|
"""
|
|
for _ in range(2):
|
|
script_file = os.path.join(tmpdir, "test.sh")
|
|
with open(script_file, "w") as sf:
|
|
sf.write(script)
|
|
result = subprocess.run(["bash", script_file], capture_output=True, text=True)
|
|
self.assertEqual(result.returncode, 0)
|
|
backup_dir = result.stdout.strip()
|
|
self.assertEqual(
|
|
backup_dir,
|
|
os.path.join(target, "Sovran_SystemsOS_Backup", "current"),
|
|
)
|
|
|
|
# ── Behavioral: INCOMPLETE/BACKUP_COMPLETE marker lifecycle ─────────────
|
|
|
|
def test_incomplete_marker_written_before_work_starts(self):
|
|
"""Behavioral: INCOMPLETE marker must exist during backup and be removed on success."""
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
backup_dir = os.path.join(tmpdir, "current")
|
|
os.makedirs(backup_dir, exist_ok=True)
|
|
|
|
script = f"""#!/usr/bin/env bash
|
|
set -euo pipefail
|
|
BACKUP_DIR="{backup_dir}"
|
|
BACKUP_COMPLETE=0
|
|
|
|
touch "$BACKUP_DIR/INCOMPLETE"
|
|
[[ -f "$BACKUP_DIR/INCOMPLETE" ]] && echo "INCOMPLETE_EXISTS"
|
|
|
|
# Simulate successful completion
|
|
rm -f "$BACKUP_DIR/INCOMPLETE"
|
|
echo "done" > "$BACKUP_DIR/BACKUP_COMPLETE"
|
|
BACKUP_COMPLETE=1
|
|
|
|
[[ ! -f "$BACKUP_DIR/INCOMPLETE" ]] && echo "INCOMPLETE_REMOVED"
|
|
[[ -f "$BACKUP_DIR/BACKUP_COMPLETE" ]] && echo "COMPLETE_EXISTS"
|
|
"""
|
|
script_file = os.path.join(tmpdir, "test.sh")
|
|
with open(script_file, "w") as sf:
|
|
sf.write(script)
|
|
result = subprocess.run(["bash", script_file], capture_output=True, text=True)
|
|
self.assertEqual(result.returncode, 0, f"stderr: {result.stderr}")
|
|
self.assertIn("INCOMPLETE_EXISTS", result.stdout)
|
|
self.assertIn("INCOMPLETE_REMOVED", result.stdout)
|
|
self.assertIn("COMPLETE_EXISTS", result.stdout)
|
|
|
|
# ── Behavioral: actual rsync with real source/dest trees ─────────────────
|
|
|
|
def test_rsync_mirrors_source_to_dest(self):
|
|
"""Behavioral: rsync correctly mirrors a source tree to a destination."""
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
src = os.path.join(tmpdir, "etc", "nixos")
|
|
dest = os.path.join(tmpdir, "backup", "etc", "nixos")
|
|
os.makedirs(src, exist_ok=True)
|
|
os.makedirs(os.path.join(tmpdir, "backup", "etc"), exist_ok=True)
|
|
|
|
# Write test files
|
|
with open(os.path.join(src, "configuration.nix"), "w") as f:
|
|
f.write("{ ... }: {}\n")
|
|
with open(os.path.join(src, "custom.nix"), "w") as f:
|
|
f.write("{ ... }: {}\n")
|
|
|
|
result = subprocess.run(
|
|
["rsync", "--archive", "--one-file-system",
|
|
src + "/", dest + "/"],
|
|
capture_output=True, text=True,
|
|
)
|
|
self.assertEqual(result.returncode, 0, f"rsync failed: {result.stderr}")
|
|
self.assertTrue(os.path.exists(os.path.join(dest, "configuration.nix")))
|
|
self.assertTrue(os.path.exists(os.path.join(dest, "custom.nix")))
|
|
|
|
def test_rsync_second_run_only_transfers_changes(self):
|
|
"""Behavioral: a second rsync run to the same dest only copies changed files."""
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
src = os.path.join(tmpdir, "source")
|
|
dest = os.path.join(tmpdir, "backup")
|
|
os.makedirs(src)
|
|
os.makedirs(dest)
|
|
|
|
with open(os.path.join(src, "file.txt"), "w") as f:
|
|
f.write("initial content\n")
|
|
|
|
# First run
|
|
r1 = subprocess.run(
|
|
["rsync", "--archive", "--one-file-system", src + "/", dest + "/"],
|
|
capture_output=True, text=True,
|
|
)
|
|
self.assertEqual(r1.returncode, 0)
|
|
|
|
# Modify one file and add another; advance mtime so rsync detects the change
|
|
with open(os.path.join(src, "file.txt"), "w") as f:
|
|
f.write("updated content\n")
|
|
import time as _time
|
|
future_mtime = _time.time() + 2
|
|
os.utime(os.path.join(src, "file.txt"), (future_mtime, future_mtime))
|
|
|
|
with open(os.path.join(src, "new.txt"), "w") as f:
|
|
f.write("new file\n")
|
|
|
|
# Second run
|
|
r2 = subprocess.run(
|
|
["rsync", "--archive", "--one-file-system", src + "/", dest + "/"],
|
|
capture_output=True, text=True,
|
|
)
|
|
self.assertEqual(r2.returncode, 0)
|
|
|
|
# Both files exist in dest
|
|
with open(os.path.join(dest, "file.txt")) as fh:
|
|
self.assertEqual(fh.read(), "updated content\n")
|
|
self.assertTrue(os.path.exists(os.path.join(dest, "new.txt")))
|
|
|
|
# ── Script syntax check ──────────────────────────────────────────────────
|
|
|
|
def test_backup_script_passes_bash_syntax_check(self):
|
|
"""bash -n must report no syntax errors in the backup script."""
|
|
result = subprocess.run(
|
|
["bash", "-n", str(BACKUP_SCRIPT)],
|
|
capture_output=True, text=True,
|
|
)
|
|
self.assertEqual(
|
|
result.returncode, 0,
|
|
f"bash -n failed:\n{result.stderr}",
|
|
)
|
|
|
|
# ── Home exclusions ──────────────────────────────────────────────────────
|
|
|
|
def test_backup_script_home_excludes_browser_caches(self):
|
|
"""Home sync must exclude browser disk caches."""
|
|
source = BACKUP_SCRIPT.read_text()
|
|
self.assertIn("--exclude='.mozilla/firefox/*/cache2/'", source)
|
|
self.assertIn("--exclude='.config/google-chrome/*/Cache/'", source)
|
|
self.assertIn("--exclude='.config/chromium/*/Cache/'", source)
|
|
self.assertIn("--exclude='.config/BraveSoftware/Brave-Browser/*/Cache/'", source)
|
|
|
|
def test_backup_script_home_excludes_other_volatile_caches(self):
|
|
"""Home sync must exclude baloo, thumbnails, and X session error logs."""
|
|
source = BACKUP_SCRIPT.read_text()
|
|
self.assertIn("--exclude='.local/share/baloo/'", source)
|
|
self.assertIn("--exclude='.thumbnails/'", source)
|
|
self.assertIn("--exclude='.xsession-errors'", source)
|
|
|
|
# ── require_cmd ──────────────────────────────────────────────────────────
|
|
|
|
def test_backup_script_requires_rsync_and_flock(self):
|
|
"""Script must require rsync and flock via require_cmd."""
|
|
source = BACKUP_SCRIPT.read_text()
|
|
self.assertIn("require_cmd rsync", source)
|
|
self.assertIn("require_cmd flock", source)
|
|
|
|
def test_backup_script_does_not_require_tar_or_sha256sum(self):
|
|
"""Script must not require tar or sha256sum (no longer used)."""
|
|
source = BACKUP_SCRIPT.read_text()
|
|
self.assertNotIn("require_cmd tar", source)
|
|
self.assertNotIn("require_cmd sha256sum", source)
|
|
|
|
# ── Frontend database limitation notice ──────────────────────────────────
|
|
|
|
def test_frontend_mentions_database_limitation(self):
|
|
"""Frontend must explain that PostgreSQL/MariaDB databases are excluded."""
|
|
support_source = SUPPORT_JS.read_text()
|
|
self.assertIn(
|
|
"PostgreSQL",
|
|
support_source,
|
|
"UI must mention PostgreSQL exclusion",
|
|
)
|
|
self.assertIn(
|
|
"not included",
|
|
support_source,
|
|
"UI must explain that databases are not included",
|
|
)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|