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})()) @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()