187 lines
10 KiB
Python
187 lines
10 KiB
Python
import contextlib
|
|
import hashlib
|
|
import importlib.util
|
|
import io
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import tarfile
|
|
import tempfile
|
|
import unittest
|
|
from unittest.mock import patch
|
|
|
|
spec = importlib.util.spec_from_file_location("ops", Path(__file__).resolve().parents[1] / "deploy" / "ops.py")
|
|
ops = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(ops)
|
|
|
|
|
|
class ArchiveTests(unittest.TestCase):
|
|
def test_persistence_layout_resolves_shared_keys_and_database(self):
|
|
config = {
|
|
"services": {
|
|
"api": {"volumes": [{"type": "volume", "source": "app-keys", "target": "/var/lib/guestops/keys"}]},
|
|
"worker": {"volumes": [{"type": "volume", "source": "app-keys", "target": "/var/lib/guestops/keys"}]},
|
|
"mongo": {"volumes": [{"type": "volume", "source": "mongo-data", "target": "/data/db"}]},
|
|
},
|
|
"volumes": {
|
|
"app-keys": {"name": "guestops_app-keys"},
|
|
"mongo-data": {"name": "guestops_mongo-data"},
|
|
},
|
|
}
|
|
self.assertEqual(ops.persistence_layout(config), {"keys": "guestops_app-keys", "database": "guestops_mongo-data"})
|
|
|
|
def test_persistence_layout_rejects_anonymous_or_split_keys(self):
|
|
config = {
|
|
"services": {
|
|
"api": {"volumes": [{"type": "volume", "source": "api-keys", "target": "/var/lib/guestops/keys"}]},
|
|
"worker": {"volumes": [{"type": "volume", "source": "worker-keys", "target": "/var/lib/guestops/keys"}]},
|
|
"mongo": {"volumes": [{"type": "volume", "source": "mongo-data", "target": "/data/db"}]},
|
|
},
|
|
"volumes": {"api-keys": {}, "worker-keys": {}, "mongo-data": {}},
|
|
}
|
|
with self.assertRaisesRegex(RuntimeError, "key volumes differ"):
|
|
ops.persistence_layout(config)
|
|
|
|
config["services"]["worker"]["volumes"][0] = {"type": "volume", "target": "/var/lib/guestops/keys"}
|
|
with self.assertRaisesRegex(RuntimeError, "named persistent volume"):
|
|
ops.persistence_layout(config)
|
|
|
|
def test_persistence_drill_requires_explicit_restart_confirmation(self):
|
|
with self.assertRaisesRegex(RuntimeError, "confirm-restart"):
|
|
ops.persistence_drill(type("Args", (), {"confirm_restart": False})())
|
|
|
|
def test_persistence_drill_restarts_then_recreates_stateless_services(self):
|
|
compose_calls = []
|
|
inventory = json.dumps({"bytes": 1, "collections": {"hotels": {"count": 1, "indexes": []}}}).encode()
|
|
|
|
def compose(*args, **kwargs):
|
|
compose_calls.append(args)
|
|
return b"a" * 30 if args[-1] == "--backup-probe" else b""
|
|
|
|
def mongo(script):
|
|
return inventory if "storageSize" in script else b""
|
|
|
|
with patch.object(ops, "configuration", return_value={}), \
|
|
patch.object(ops, "persistence_layout", return_value={"keys": "keys", "database": "data"}), \
|
|
patch.object(ops, "volume_identity", side_effect=lambda name: {"name": name, "driver": "local", "scope": "local"}), \
|
|
patch.object(ops, "compose", side_effect=compose), \
|
|
patch.object(ops, "mongo", side_effect=mongo), \
|
|
patch.object(ops, "wait_for", side_effect=lambda action, message: action()), \
|
|
patch.object(ops, "readiness"):
|
|
ops.persistence_drill(type("Args", (), {"confirm_restart": True})())
|
|
|
|
self.assertIn(("restart", "-t", "150", "mongo"), compose_calls)
|
|
self.assertIn(("restart", "-t", "150", "api", "worker"), compose_calls)
|
|
self.assertIn(("up", "-d", "--no-build", "--force-recreate", "api", "worker"), compose_calls)
|
|
self.assertTrue(any("--verify-backup-probe" in call for call in compose_calls))
|
|
|
|
def test_empty_provider_templates_are_not_secret_configuration(self):
|
|
for section, target in (("Pms", "/run/guestops/pms.json"), ("Payments", "/run/guestops/payments.json")):
|
|
self.assertFalse(ops.provider_configured({section: {"Hotels": {}}}, target))
|
|
self.assertTrue(ops.provider_configured({section: {"Hotels": {"fixture": {"Key": "fixture"}}}}, target))
|
|
self.assertTrue(ops.provider_configured({section: {"Unknown": "fixture"}}, target))
|
|
|
|
def bundle(self, root, change=None):
|
|
files = {name: b"fixture backup data" for name in ops.FILES - {"manifest.json"}}
|
|
files["manifest.json"] = json.dumps({"format": 1, "sha256": {name: hashlib.sha256(data).hexdigest() for name, data in files.items()}}).encode()
|
|
target = root / "bundle.tar"
|
|
with tarfile.open(target, "w") as archive:
|
|
for name, data in files.items():
|
|
member = tarfile.TarInfo(name); member.size = len(data)
|
|
if change:
|
|
change(member)
|
|
archive.addfile(member, io.BytesIO(data) if member.isfile() else None)
|
|
return target
|
|
|
|
def test_checked_archive_roundtrip(self):
|
|
with tempfile.TemporaryDirectory() as temp:
|
|
root = Path(temp); dest = root / "dest"; dest.mkdir()
|
|
self.assertEqual(ops.unpack(self.bundle(root), dest)["format"], 1)
|
|
|
|
def test_path_escape_rejected(self):
|
|
with tempfile.TemporaryDirectory() as temp:
|
|
root = Path(temp); dest = root / "dest"; dest.mkdir()
|
|
def corrupt(member):
|
|
if member.name == "configuration.json": member.name = "../outside"
|
|
with self.assertRaisesRegex(RuntimeError, "Unexpected backup members"):
|
|
ops.unpack(self.bundle(root, corrupt), dest)
|
|
self.assertFalse((root / "outside").exists())
|
|
|
|
def test_symlink_rejected(self):
|
|
with tempfile.TemporaryDirectory() as temp:
|
|
root = Path(temp); dest = root / "dest"; dest.mkdir()
|
|
def corrupt(member):
|
|
if member.name == "configuration.json": member.type = tarfile.SYMTYPE; member.linkname = "/etc/passwd"; member.size = 0
|
|
with self.assertRaisesRegex(RuntimeError, "Invalid or oversized"):
|
|
ops.unpack(self.bundle(root, corrupt), dest)
|
|
|
|
def test_checksum_mismatch_rejected(self):
|
|
with tempfile.TemporaryDirectory() as temp:
|
|
root = Path(temp); dest = root / "dest"; dest.mkdir()
|
|
with patch.object(ops, "digest", return_value="changed"):
|
|
with self.assertRaisesRegex(RuntimeError, "checksum"):
|
|
ops.unpack(self.bundle(root), dest)
|
|
|
|
def test_backup_requires_explicit_maintenance(self):
|
|
with self.assertRaisesRegex(RuntimeError, "confirm-maintenance"):
|
|
ops.backup(type("Args", (), {"confirm_maintenance": False})())
|
|
|
|
def test_scheduled_backup_requires_explicit_maintenance(self):
|
|
with self.assertRaisesRegex(RuntimeError, "confirm-maintenance"):
|
|
ops.scheduled_backup(type("Args", (), {"confirm_maintenance": False})())
|
|
|
|
@unittest.skipUnless(os.name == "posix", "Linux file permission semantics")
|
|
def test_scheduled_backup_uses_private_external_directory_and_environment(self):
|
|
with tempfile.TemporaryDirectory() as parent:
|
|
directory = Path(parent) / "backups"
|
|
directory.mkdir(mode=0o700)
|
|
captured = []
|
|
with patch.dict(os.environ, {"BACKUP_RECIPIENT": "A" * 40, "BACKUP_DIRECTORY": str(directory)}), \
|
|
patch.object(ops, "ROOT", Path(parent) / "checkout"), \
|
|
patch.object(ops, "backup", side_effect=lambda args: captured.append(args)), \
|
|
patch.object(ops.time, "strftime", return_value="20260929T020000Z"):
|
|
ops.scheduled_backup(type("Args", (), {"confirm_maintenance": True})())
|
|
self.assertEqual(captured[0].recipient, "A" * 40)
|
|
self.assertEqual(Path(captured[0].output), directory / "guestops-20260929T020000Z.tar.gpg")
|
|
self.assertTrue(captured[0].confirm_maintenance)
|
|
|
|
@unittest.skipUnless(os.name == "posix", "Linux file permission semantics")
|
|
def test_scheduled_backup_rejects_checkout_directory(self):
|
|
with tempfile.TemporaryDirectory() as parent:
|
|
checkout = Path(parent) / "checkout"
|
|
directory = checkout / "backups"
|
|
directory.mkdir(parents=True, mode=0o700)
|
|
with patch.dict(os.environ, {"BACKUP_RECIPIENT": "A" * 40, "BACKUP_DIRECTORY": str(directory)}), \
|
|
patch.object(ops, "ROOT", checkout):
|
|
with self.assertRaisesRegex(RuntimeError, "outside the application checkout"):
|
|
ops.scheduled_backup(type("Args", (), {"confirm_maintenance": True})())
|
|
|
|
@unittest.skipUnless(os.name == "posix", "Linux file permission semantics")
|
|
def test_dump_failure_restarts_services_and_ttl(self):
|
|
with tempfile.TemporaryDirectory() as temp:
|
|
calls = []
|
|
def compose(*args, **kwargs):
|
|
calls.append(args)
|
|
if args[0] == "ps": return b"a" * 64
|
|
if "sh" in args: raise RuntimeError("Simulated dump failure")
|
|
return b""
|
|
def mongo(script):
|
|
calls.append((script,))
|
|
if "getParameter" in script: return b"true"
|
|
if "storageSize" in script: return b'{"bytes":1,"collections":{}}'
|
|
return b""
|
|
def run(args, **kwargs):
|
|
return b'[{"Image":"sha256:fixture"}]' if "inspect" in args else b""
|
|
args = type("Args", (), {"confirm_maintenance": True, "recipient": "A" * 40, "output": str(Path(temp) / "backup.gpg")})()
|
|
with patch.object(ops, "configuration", return_value={}), patch.object(ops, "run", side_effect=run), patch.object(ops, "compose", side_effect=compose), patch.object(ops, "mongo", side_effect=mongo), patch.object(ops, "maintenance_lock", return_value=contextlib.nullcontext()):
|
|
with self.assertRaisesRegex(RuntimeError, "Simulated dump failure"):
|
|
ops.backup(args)
|
|
self.assertIn(("start", "api", "worker"), calls)
|
|
self.assertTrue(any("ttlMonitorEnabled:true" in call[0] for call in calls))
|
|
self.assertFalse(Path(args.output).exists())
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|
|
|