310 lines
10 KiB
Python
310 lines
10 KiB
Python
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
|