fix(ops): close Gitea backup restore verification
This commit is contained in:
ogt
@@ -19,3 +19,17 @@ def test_backup_status_keeps_aggregate_failure_out_of_core_blockers() -> None:
|
||||
assert "component_failed_188" in core_line
|
||||
assert "failed_total_110" not in core_line
|
||||
assert "failed_total_188" not in core_line
|
||||
|
||||
|
||||
def test_backup_status_uses_dynamic_healthy_over_monitored_ratio() -> None:
|
||||
text = BACKUP_STATUS.read_text(encoding="utf-8")
|
||||
core_line = next(line for line in text.splitlines() if line.startswith("core_blockers="))
|
||||
|
||||
assert 'fresh_ok_110="$(metric_count "${TEXTFILE_110}" "awoooi_backup_job_fresh" 1)"' in text
|
||||
assert 'fresh_ok_188="$(metric_count "${TEXTFILE_188_TMP}" "awoooi_backup_job_fresh" 1)"' in text
|
||||
assert "110備份=${fresh_ok_110}/${monitored_total_110}" in text
|
||||
assert "188備份=${fresh_ok_188}/${monitored_total_188}" in text
|
||||
assert "freshness_metrics_missing_110" in core_line
|
||||
assert "freshness_metrics_missing_188" in core_line
|
||||
assert "/13 fresh" not in text
|
||||
assert "/2 fresh" not in text
|
||||
|
||||
309
scripts/backup/tests/test_gitea_backup_automation.py
Normal file
309
scripts/backup/tests/test_gitea_backup_automation.py
Normal file
@@ -0,0 +1,309 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
import shutil
|
||||
import subprocess
|
||||
import textwrap
|
||||
import zipfile
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
|
||||
ROOT = Path(__file__).resolve().parents[3]
|
||||
BUNDLE_SCRIPT = ROOT / "scripts" / "backup" / "gitea-repo-bundle-backup.sh"
|
||||
RESTORE_DRILL_SCRIPT = ROOT / "scripts" / "backup" / "verify-gitea-backup.sh"
|
||||
|
||||
|
||||
def _fake_git(tmp_path: Path) -> Path:
|
||||
fake_bin = tmp_path / "bin"
|
||||
fake_bin.mkdir()
|
||||
git = fake_bin / "git"
|
||||
git.write_text(
|
||||
textwrap.dedent(
|
||||
"""\
|
||||
#!/usr/bin/env bash
|
||||
set -euo pipefail
|
||||
if [ -n "${FAKE_GIT_FAIL_PATTERN:-}" ] && [[ "$*" == *"${FAKE_GIT_FAIL_PATTERN}"* ]]; then
|
||||
exit 42
|
||||
fi
|
||||
if [ "$1" = "ls-remote" ]; then
|
||||
printf '0000000000000000000000000000000000000001\trefs/heads/main\n'
|
||||
exit 0
|
||||
fi
|
||||
if [ "$1" = "clone" ]; then
|
||||
mkdir -p "${!#}"
|
||||
exit 0
|
||||
fi
|
||||
if [ "$1" = "-C" ] && [ "$3" = "bundle" ] && [ "$4" = "create" ]; then
|
||||
printf 'synthetic bundle\n' > "$5"
|
||||
exit 0
|
||||
fi
|
||||
if [ "$1" = "-C" ] && [ "$3" = "bundle" ] && [ "$4" = "verify" ]; then
|
||||
exit 0
|
||||
fi
|
||||
printf 'unexpected fake git invocation: %s\n' "$*" >&2
|
||||
exit 64
|
||||
"""
|
||||
),
|
||||
encoding="utf-8",
|
||||
)
|
||||
git.chmod(0o755)
|
||||
return fake_bin
|
||||
|
||||
|
||||
def _fake_docker(fake_bin: Path, tmp_path: Path) -> None:
|
||||
docker = fake_bin / "docker"
|
||||
docker.write_text(
|
||||
textwrap.dedent(
|
||||
"""\
|
||||
#!/usr/bin/env bash
|
||||
set -euo pipefail
|
||||
root="${FAKE_DOCKER_ROOT:?}"
|
||||
if [ "$1" = "exec" ]; then
|
||||
shift 4
|
||||
if [ "$1" = "test" ]; then
|
||||
exit 0
|
||||
fi
|
||||
if [ "$1" = "git" ] && [ "$4" = "for-each-ref" ]; then
|
||||
printf '0000000000000000000000000000000000000001\trefs/heads/main\n'
|
||||
exit 0
|
||||
fi
|
||||
if [ "$1" = "git" ] && [ "$4" = "bundle" ] && [ "$5" = "create" ]; then
|
||||
mkdir -p "${root}$(dirname "$6")"
|
||||
printf 'synthetic container bundle\n' > "${root}$6"
|
||||
exit 0
|
||||
fi
|
||||
if [ "$1" = "git" ] && [ "$4" = "bundle" ] && [ "$5" = "verify" ]; then
|
||||
exit 0
|
||||
fi
|
||||
if [ "$1" = "rm" ]; then
|
||||
rm -f "${root}${3}"
|
||||
exit 0
|
||||
fi
|
||||
fi
|
||||
if [ "$1" = "cp" ]; then
|
||||
source_path="${2#*:}"
|
||||
cp "${root}${source_path}" "$3"
|
||||
exit 0
|
||||
fi
|
||||
printf 'unexpected fake docker invocation: %s\n' "$*" >&2
|
||||
exit 64
|
||||
"""
|
||||
),
|
||||
encoding="utf-8",
|
||||
)
|
||||
docker.chmod(0o755)
|
||||
|
||||
|
||||
def test_bundle_backup_promotes_only_a_complete_repo_set(tmp_path: Path) -> None:
|
||||
fake_bin = _fake_git(tmp_path)
|
||||
output_root = tmp_path / "bundles"
|
||||
env = os.environ | {"PATH": f"{fake_bin}:{os.environ['PATH']}"}
|
||||
|
||||
complete = subprocess.run(
|
||||
[
|
||||
"bash",
|
||||
str(BUNDLE_SCRIPT),
|
||||
"--output-root",
|
||||
str(output_root),
|
||||
"--git-remote-base",
|
||||
"ssh://git@example.invalid:2222",
|
||||
"--stamp",
|
||||
"complete",
|
||||
"--repo",
|
||||
"owner/one",
|
||||
"--repo",
|
||||
"owner/two",
|
||||
],
|
||||
env=env,
|
||||
capture_output=True,
|
||||
text=True,
|
||||
check=False,
|
||||
)
|
||||
assert complete.returncode == 0, complete.stderr
|
||||
assert (output_root / "latest-private-complete").resolve() == output_root / "complete"
|
||||
assert (output_root / "latest").resolve() == output_root / "complete"
|
||||
assert "BUNDLE_BACKUP_COMPLETE=1 success=2 expected=2" in complete.stdout
|
||||
|
||||
failed = subprocess.run(
|
||||
[
|
||||
"bash",
|
||||
str(BUNDLE_SCRIPT),
|
||||
"--output-root",
|
||||
str(output_root),
|
||||
"--git-remote-base",
|
||||
"ssh://git@example.invalid:2222",
|
||||
"--stamp",
|
||||
"partial",
|
||||
"--repo",
|
||||
"owner/one",
|
||||
"--repo",
|
||||
"owner/fail",
|
||||
],
|
||||
env=env | {"FAKE_GIT_FAIL_PATTERN": "owner/fail"},
|
||||
capture_output=True,
|
||||
text=True,
|
||||
check=False,
|
||||
)
|
||||
assert failed.returncode != 0
|
||||
assert "complete links unchanged" in failed.stderr
|
||||
assert (output_root / "latest-private-complete").resolve() == output_root / "complete"
|
||||
assert "owner/fail\tls_remote_failed" in (output_root / "partial" / "manifest.tsv").read_text(
|
||||
encoding="utf-8"
|
||||
)
|
||||
|
||||
|
||||
def test_bundle_backup_supports_local_gitea_container_without_credentials(tmp_path: Path) -> None:
|
||||
fake_bin = _fake_git(tmp_path)
|
||||
_fake_docker(fake_bin, tmp_path)
|
||||
output_root = tmp_path / "local-bundles"
|
||||
docker_root = tmp_path / "fake-docker"
|
||||
docker_root.mkdir()
|
||||
env = os.environ | {
|
||||
"PATH": f"{fake_bin}:{os.environ['PATH']}",
|
||||
"FAKE_DOCKER_ROOT": str(docker_root),
|
||||
}
|
||||
|
||||
result = subprocess.run(
|
||||
[
|
||||
"bash",
|
||||
str(BUNDLE_SCRIPT),
|
||||
"--output-root",
|
||||
str(output_root),
|
||||
"--local-container",
|
||||
"gitea",
|
||||
"--stamp",
|
||||
"local-complete",
|
||||
"--repo",
|
||||
"owner/one",
|
||||
"--repo",
|
||||
"owner/two",
|
||||
],
|
||||
env=env,
|
||||
capture_output=True,
|
||||
text=True,
|
||||
check=False,
|
||||
)
|
||||
assert result.returncode == 0, result.stderr
|
||||
assert (output_root / "latest-private-complete").resolve() == output_root / "local-complete"
|
||||
assert "BUNDLE_BACKUP_COMPLETE=1 success=2 expected=2" in result.stdout
|
||||
|
||||
|
||||
def test_bundle_backup_preflights_both_promotion_links(tmp_path: Path) -> None:
|
||||
fake_bin = _fake_git(tmp_path)
|
||||
output_root = tmp_path / "blocked-promotion"
|
||||
output_root.mkdir()
|
||||
(output_root / "latest-private-complete").mkdir()
|
||||
env = os.environ | {"PATH": f"{fake_bin}:{os.environ['PATH']}"}
|
||||
|
||||
result = subprocess.run(
|
||||
[
|
||||
"bash",
|
||||
str(BUNDLE_SCRIPT),
|
||||
"--output-root",
|
||||
str(output_root),
|
||||
"--git-remote-base",
|
||||
"ssh://git@example.invalid:2222",
|
||||
"--stamp",
|
||||
"candidate",
|
||||
"--repo",
|
||||
"owner/one",
|
||||
],
|
||||
env=env,
|
||||
capture_output=True,
|
||||
text=True,
|
||||
check=False,
|
||||
)
|
||||
assert result.returncode != 0
|
||||
assert "promotion path exists and is not a symlink" in result.stderr
|
||||
assert not (output_root / "latest").exists()
|
||||
assert (output_root / "latest-private-complete").is_dir()
|
||||
|
||||
|
||||
@pytest.mark.skipif(shutil.which("unzip") is None, reason="unzip is required by the production verifier")
|
||||
def test_gitea_restore_drill_blocks_same_count_inventory_mismatch(tmp_path: Path) -> None:
|
||||
archive = tmp_path / "gitea-dump.zip"
|
||||
with zipfile.ZipFile(archive, "w") as payload:
|
||||
payload.writestr("gitea-repo/owner/one.git/HEAD", "ref: refs/heads/main\n")
|
||||
payload.writestr("gitea-repo/owner/two.git/HEAD", "ref: refs/heads/main\n")
|
||||
payload.writestr("gitea-db.sql", "synthetic database placeholder\n")
|
||||
payload.writestr("config/app.ini", "synthetic config placeholder\n")
|
||||
|
||||
state_dir = tmp_path / "state"
|
||||
work_dir = tmp_path / "work"
|
||||
state_dir.mkdir()
|
||||
work_dir.mkdir()
|
||||
matching_inventory = tmp_path / "matching.txt"
|
||||
matching_inventory.write_text("owner/one.git\nowner/two.git\n", encoding="utf-8")
|
||||
mismatch_inventory = tmp_path / "mismatch.txt"
|
||||
mismatch_inventory.write_text("owner/one.git\nowner/different.git\n", encoding="utf-8")
|
||||
base_env = os.environ | {
|
||||
"GITEA_RESTORE_DRILL_TEST_MODE": "1",
|
||||
"GITEA_RESTORE_DRILL_SOURCE_ZIP": str(archive),
|
||||
"GITEA_RESTORE_DRILL_STATE_DIR": str(state_dir),
|
||||
"GITEA_RESTORE_DRILL_WORK_ROOT": str(work_dir),
|
||||
"BACKUP_BASE": str(tmp_path),
|
||||
}
|
||||
|
||||
success = subprocess.run(
|
||||
["bash", str(RESTORE_DRILL_SCRIPT), "--no-notify"],
|
||||
env=base_env | {"GITEA_RESTORE_DRILL_LIVE_INVENTORY_FILE": str(matching_inventory)},
|
||||
capture_output=True,
|
||||
text=True,
|
||||
check=False,
|
||||
)
|
||||
assert success.returncode == 0, success.stderr
|
||||
status = (state_dir / "gitea-restore-drill.status").read_text(encoding="utf-8")
|
||||
assert "success=1" in status
|
||||
assert "inventory_digest_match=1" in status
|
||||
|
||||
mismatch = subprocess.run(
|
||||
["bash", str(RESTORE_DRILL_SCRIPT), "--no-notify"],
|
||||
env=base_env | {"GITEA_RESTORE_DRILL_LIVE_INVENTORY_FILE": str(mismatch_inventory)},
|
||||
capture_output=True,
|
||||
text=True,
|
||||
check=False,
|
||||
)
|
||||
assert mismatch.returncode != 0
|
||||
status = (state_dir / "gitea-restore-drill.status").read_text(encoding="utf-8")
|
||||
assert "success=0" in status
|
||||
assert "reason=repository_inventory_mismatch" in status
|
||||
|
||||
|
||||
@pytest.mark.skipif(shutil.which("unzip") is None, reason="unzip is required by the production verifier")
|
||||
def test_gitea_restore_drill_records_live_inventory_read_failure(tmp_path: Path) -> None:
|
||||
archive = tmp_path / "gitea-dump.zip"
|
||||
with zipfile.ZipFile(archive, "w") as payload:
|
||||
payload.writestr("gitea-repo/owner/one.git/HEAD", "ref: refs/heads/main\n")
|
||||
payload.writestr("gitea-db.sql", "synthetic database placeholder\n")
|
||||
payload.writestr("config/app.ini", "synthetic config placeholder\n")
|
||||
|
||||
fake_bin = tmp_path / "bin"
|
||||
fake_bin.mkdir()
|
||||
docker = fake_bin / "docker"
|
||||
docker.write_text("#!/usr/bin/env bash\nexit 42\n", encoding="utf-8")
|
||||
docker.chmod(0o755)
|
||||
state_dir = tmp_path / "state"
|
||||
work_dir = tmp_path / "work"
|
||||
state_dir.mkdir()
|
||||
work_dir.mkdir()
|
||||
result = subprocess.run(
|
||||
["bash", str(RESTORE_DRILL_SCRIPT), "--no-notify"],
|
||||
env=os.environ
|
||||
| {
|
||||
"PATH": f"{fake_bin}:{os.environ['PATH']}",
|
||||
"GITEA_RESTORE_DRILL_TEST_MODE": "1",
|
||||
"GITEA_RESTORE_DRILL_SOURCE_ZIP": str(archive),
|
||||
"GITEA_RESTORE_DRILL_STATE_DIR": str(state_dir),
|
||||
"GITEA_RESTORE_DRILL_WORK_ROOT": str(work_dir),
|
||||
"BACKUP_BASE": str(tmp_path),
|
||||
},
|
||||
capture_output=True,
|
||||
text=True,
|
||||
check=False,
|
||||
)
|
||||
assert result.returncode != 0
|
||||
status = (state_dir / "gitea-restore-drill.status").read_text(encoding="utf-8")
|
||||
assert "success=0" in status
|
||||
assert "reason=live_inventory_read_failed" in status
|
||||
Reference in New Issue
Block a user