98 lines
4.8 KiB
Python
98 lines
4.8 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_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()
|
|
|