136 lines
5.8 KiB
Python
136 lines
5.8 KiB
Python
import io
|
|
import re
|
|
import subprocess
|
|
import tarfile
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
|
|
|
|
class RecoveryScriptTests(unittest.TestCase):
|
|
def test_latest_alias_points_to_production_backup(self) -> None:
|
|
script = (ROOT / "restore.sh").read_text(encoding="utf-8")
|
|
self.assertIn("if [[ $ARCHIVE == latest ]]", script)
|
|
self.assertIn("ARCHIVE=/data/docker-backups/athena-latest.tar.gz", script)
|
|
|
|
def _archive(self, members: list[str]) -> Path:
|
|
tmp = tempfile.NamedTemporaryFile(suffix=".tar.gz", delete=False)
|
|
tmp.close()
|
|
archive = Path(tmp.name)
|
|
with tarfile.open(archive, "w:gz") as handle:
|
|
for name in members:
|
|
payload = b"test\n"
|
|
info = tarfile.TarInfo(name)
|
|
info.size = len(payload)
|
|
handle.addfile(info, io.BytesIO(payload))
|
|
self.addCleanup(archive.unlink, missing_ok=True)
|
|
return archive
|
|
|
|
def test_current_backup_layout_is_accepted(self) -> None:
|
|
archive = self._archive([
|
|
"/backup/etc-mike-ai/stack.env",
|
|
"/backup/volumes/router-state/state.json",
|
|
])
|
|
result = subprocess.run(
|
|
[str(ROOT / "restore.sh"), "--check", str(archive)],
|
|
check=False,
|
|
capture_output=True,
|
|
text=True,
|
|
)
|
|
self.assertEqual(result.returncode, 0, result.stderr)
|
|
self.assertIn("ATHENA_BACKUP_CHECK_OK", result.stdout)
|
|
|
|
def test_unexpected_tree_is_rejected(self) -> None:
|
|
archive = self._archive([
|
|
"/backup/etc-mike-ai/stack.env",
|
|
"/backup/volumes/router-state/state.json",
|
|
"/etc/shadow",
|
|
])
|
|
result = subprocess.run(
|
|
[str(ROOT / "restore.sh"), "--check", str(archive)],
|
|
check=False,
|
|
capture_output=True,
|
|
text=True,
|
|
)
|
|
self.assertNotEqual(result.returncode, 0)
|
|
self.assertIn("Unerwarteter Pfad", result.stderr)
|
|
|
|
def test_backup_prefix_extracts_to_restore_layout(self) -> None:
|
|
archive = self._archive([
|
|
"/backup/etc-mike-ai/stack.env",
|
|
"/backup/volumes/router-state/state.json",
|
|
])
|
|
with tempfile.TemporaryDirectory() as target:
|
|
result = subprocess.run(
|
|
[
|
|
"tar", "-xzf", str(archive), "-C", target,
|
|
],
|
|
check=False,
|
|
capture_output=True,
|
|
text=True,
|
|
)
|
|
self.assertEqual(result.returncode, 0, result.stderr)
|
|
root = Path(target, "backup")
|
|
self.assertTrue(Path(root, "etc-mike-ai", "stack.env").is_file())
|
|
self.assertTrue(Path(root, "volumes", "router-state", "state.json").is_file())
|
|
|
|
def test_installer_uses_current_profiles(self) -> None:
|
|
installer = (ROOT / "install.sh").read_text(encoding="utf-8")
|
|
self.assertNotIn("llama-experimental", installer)
|
|
for profile in ("fast", "medium", "large", "ultra", "uncensored"):
|
|
self.assertIn(f"llama-{profile}", installer)
|
|
|
|
def test_gateway_consumers_do_not_require_rebinding(self) -> None:
|
|
manager = (ROOT / "manage.sh").read_text(encoding="utf-8")
|
|
installer = (ROOT / "install.sh").read_text(encoding="utf-8")
|
|
self.assertNotIn('stop llama-dashboard portainer', manager)
|
|
self.assertNotIn('stop llama-dashboard portainer', installer)
|
|
self.assertNotIn('force-recreate llama-dashboard portainer', manager)
|
|
|
|
def test_gateway_proxies_stable_ui_services_and_portainer_backup(self) -> None:
|
|
compose = (ROOT / "compose.yaml").read_text(encoding="utf-8")
|
|
gateway = (ROOT / "platform/docker/wireguard-gateway/entrypoint.sh").read_text(
|
|
encoding="utf-8"
|
|
)
|
|
# WebRTC deliberately shares the gateway for private ICE candidates.
|
|
# Dashboard and Portainer must retain their independent namespaces.
|
|
for service in ("llama-dashboard", "portainer"):
|
|
block = re.search(
|
|
rf"(?ms)^ {service}:\n(.*?)(?=^ [a-zA-Z0-9_-]+:|\Z)", compose)
|
|
self.assertIsNotNone(block)
|
|
self.assertNotIn('network_mode: "service:wireguard-gateway"', block.group(1))
|
|
self.assertNotIn('"8099:8099"', compose)
|
|
self.assertNotIn('"9443:9443"', compose)
|
|
self.assertIn('start_proxy 8099 llama-dashboard:8099', gateway)
|
|
self.assertIn('start_proxy 9443 portainer:9443', gateway)
|
|
self.assertIn('portainer-data:/backup/volumes/portainer-data:ro', compose)
|
|
|
|
def test_disaster_recovery_covers_all_three_scenarios_without_formatting(self) -> None:
|
|
script = (ROOT / "disaster-recovery.sh").read_text(encoding="utf-8")
|
|
for scenario in ("system", "data", "all"):
|
|
self.assertIn(scenario, script)
|
|
self.assertIn("mountpoint -q /data", script)
|
|
self.assertIn("--portable", script)
|
|
for destructive in ("mkfs", "fdisk", "parted", "reboot", "shutdown"):
|
|
self.assertNotIn(f"{destructive} ", script)
|
|
|
|
def test_backup_layers_include_code_and_irreplaceable_data(self) -> None:
|
|
compose = (ROOT / "compose.yaml").read_text(encoding="utf-8")
|
|
export = (ROOT / "platform/backup/athena-export-backup").read_text(
|
|
encoding="utf-8"
|
|
)
|
|
self.assertIn("/opt/mike-ai:/backup/opt-mike-ai:ro", compose)
|
|
self.assertIn("/data/voice/applio/logs", export)
|
|
self.assertIn("/data/voice/applio/datasets", export)
|
|
self.assertIn("/data/voice/applio/mikes-applio-ui", export)
|
|
self.assertIn("/data/voice/applio/models/pretraineds/custom", export)
|
|
self.assertIn("ATHENA_EXPORT_KEEP:-5", export)
|
|
self.assertNotIn("add_path /data/models", export)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|