Files
AI-Profile-Router/dev/test_recovery.py
T

101 lines
3.8 KiB
Python

import io
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_are_stopped_before_gateway_recreation(self) -> None:
manager = (ROOT / "manage.sh").read_text(encoding="utf-8")
stop = 'stop llama-dashboard portainer'
deploy = 'up -d --build'
self.assertIn(stop, manager)
self.assertLess(manager.index(stop), manager.index(deploy))
def test_gateway_owns_shared_ui_ports_and_portainer_backup(self) -> None:
compose = (ROOT / "compose.yaml").read_text(encoding="utf-8")
self.assertIn('"8099:8099"', compose)
self.assertIn('"9443:9443"', compose)
self.assertIn('portainer-data:/backup/volumes/portainer-data:ro', compose)
if __name__ == "__main__":
unittest.main()