mirror of
https://github.com/MacRimi/ProxMenux.git
synced 2026-11-01 10:22:08 +00:00
156 lines
7.0 KiB
Python
156 lines
7.0 KiB
Python
"""The record of an installation travels inside its container, so a backup
|
|
restored on another host still carries it."""
|
|
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
sys.path.insert(0, str(ROOT / "remote"))
|
|
|
|
import oci_carried_record as carried
|
|
|
|
|
|
def bundle(**extra):
|
|
return {"schema_version": 1, "kind": carried.KIND, "node": "pve", "vmid": 120,
|
|
"record": {"vmid": 120, "installation_id": "488ed3cb-1145-477a-b90f-c46777d69fb2"}, **extra}
|
|
|
|
|
|
class CarriedRecordTests(unittest.TestCase):
|
|
def setUp(self):
|
|
tmp = tempfile.TemporaryDirectory()
|
|
self.addCleanup(tmp.cleanup)
|
|
self.rootfs = Path(tmp.name) / "rootfs"
|
|
self.rootfs.mkdir()
|
|
self.outside = Path(tmp.name) / "outside"
|
|
self.outside.mkdir()
|
|
self.owner = (os.getuid(), os.getgid())
|
|
|
|
def test_the_copy_is_private_and_read_back_as_written(self):
|
|
carried.write_copy(self.rootfs, bundle(), self.owner)
|
|
path = self.rootfs / carried.DIRECTORY / carried.NAME
|
|
self.assertEqual(path.stat().st_mode & 0o777, 0o600)
|
|
self.assertEqual(path.parent.stat().st_mode & 0o777, 0o700)
|
|
self.assertEqual(carried.read_copy(self.rootfs), bundle())
|
|
self.assertEqual(os.listdir(path.parent), [carried.NAME])
|
|
|
|
def test_a_newer_record_replaces_the_copy(self):
|
|
carried.write_copy(self.rootfs, bundle(), self.owner)
|
|
carried.write_copy(self.rootfs, bundle(node="other"), self.owner)
|
|
self.assertEqual(carried.read_copy(self.rootfs)["node"], "other")
|
|
|
|
def test_a_link_left_in_place_of_the_folder_is_not_followed(self):
|
|
(self.rootfs / carried.DIRECTORY).symlink_to(self.outside)
|
|
with self.assertRaises(OSError):
|
|
carried.write_copy(self.rootfs, bundle(), self.owner)
|
|
self.assertEqual(os.listdir(self.outside), [])
|
|
(self.outside / carried.NAME).write_text(json.dumps(bundle()))
|
|
self.assertIsNone(carried.read_copy(self.rootfs))
|
|
|
|
def test_a_link_left_in_place_of_the_copy_is_not_followed(self):
|
|
(self.outside / "secret").write_text(json.dumps(bundle()))
|
|
(self.rootfs / carried.DIRECTORY).mkdir()
|
|
(self.rootfs / carried.DIRECTORY / carried.NAME).symlink_to(self.outside / "secret")
|
|
self.assertIsNone(carried.read_copy(self.rootfs))
|
|
|
|
def test_what_is_not_a_copy_is_not_read(self):
|
|
self.assertIsNone(carried.read_copy(self.rootfs))
|
|
folder = self.rootfs / carried.DIRECTORY
|
|
folder.mkdir()
|
|
for content in ("not json", json.dumps([1]), json.dumps(bundle(kind="other")),
|
|
json.dumps(bundle(schema_version=2)), json.dumps({**bundle(), "record": "text"})):
|
|
(folder / carried.NAME).write_text(content)
|
|
self.assertIsNone(carried.read_copy(self.rootfs), content)
|
|
|
|
def test_the_time_it_was_saved_does_not_make_a_copy_different(self):
|
|
self.assertEqual(carried.digest(bundle(saved_at="2026-10-03")), carried.digest(bundle(saved_at="2026-10-04")))
|
|
self.assertEqual(carried.digest(bundle(generation="a")), carried.digest(bundle(generation="b")))
|
|
self.assertNotEqual(carried.digest(bundle()), carried.digest(bundle(node="other")))
|
|
|
|
def test_the_owner_is_root_of_the_container(self):
|
|
self.assertEqual(carried.mapped_root("arch: amd64\nunprivileged: 1\n"), (100000, 100000))
|
|
self.assertEqual(carried.mapped_root("arch: amd64\n"), (0, 0))
|
|
custom = "unprivileged: 1\nlxc.idmap: u 0 200000 65536\nlxc.idmap: g 0 300000 65536\nlxc.idmap: u 1000 1000 1\n"
|
|
self.assertEqual(carried.mapped_root(custom), (200000, 300000))
|
|
|
|
|
|
|
|
|
|
class StoppedContainerTests(unittest.TestCase):
|
|
"""A stopped container is mounted to reach its copy of the record, unless
|
|
a directory of the host it mounts is missing: Proxmox would fail to mount
|
|
it and leave it locked."""
|
|
|
|
def reach(self, config):
|
|
from types import SimpleNamespace
|
|
from unittest.mock import patch
|
|
calls = []
|
|
|
|
def run(*args):
|
|
calls.append(args[:2])
|
|
if args[0] == "lxc-info":
|
|
return SimpleNamespace(returncode=1, stdout="")
|
|
if args[1] == "config":
|
|
return SimpleNamespace(returncode=0, stdout=config)
|
|
return SimpleNamespace(returncode=0, stdout="")
|
|
with patch.object(carried, "_run", side_effect=run):
|
|
with carried.container_root(120) as rootfs:
|
|
pass
|
|
return rootfs, calls
|
|
|
|
def test_it_is_mounted_when_its_host_directories_are_there(self):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
rootfs, calls = self.reach(f"rootfs: local:120/disk.raw\nmp0: {directory},mp=/media,backup=0\n")
|
|
self.assertEqual(rootfs, Path("/var/lib/lxc/120/rootfs"))
|
|
self.assertEqual(calls, [("lxc-info", "-n"), ("pct", "config"), ("pct", "mount"), ("pct", "unmount")])
|
|
|
|
def test_it_is_left_alone_when_one_is_missing(self):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
rootfs, calls = self.reach(f"rootfs: local:120/disk.raw\nmp0: {directory}/gone,mp=/media,backup=0\n")
|
|
self.assertIsNone(rootfs)
|
|
self.assertNotIn(("pct", "mount"), calls)
|
|
|
|
|
|
class ShownSyncTests(unittest.TestCase):
|
|
"""Every installed application is looked at after an operation, which
|
|
takes a while with many of them: the screen says what is being done."""
|
|
|
|
def setUp(self):
|
|
sys.path.insert(0, str(ROOT / "src"))
|
|
from proxmenux_oci import console, management
|
|
self.console, self.management = console, management
|
|
|
|
def run_sync(self, outcome, **options):
|
|
from unittest.mock import patch
|
|
events = []
|
|
with patch.object(self.console, "msg_info", side_effect=lambda text: events.append(("info", text))), \
|
|
patch.object(self.console, "stop_spinner", side_effect=lambda *a, **k: events.append(("stop",))), \
|
|
patch.object(carried, "sync", side_effect=outcome) as sync:
|
|
result = self.management.carry_records(ROOT, **options)
|
|
return result, events, sync
|
|
|
|
def test_the_sync_is_announced_and_the_notice_is_cleared(self):
|
|
result, events, sync = self.run_sync(lambda *args: {120: "current"}, shown=True)
|
|
self.assertEqual(result, {120: "current"})
|
|
self.assertEqual([event[0] for event in events], ["info", "stop"])
|
|
self.assertEqual(events[0][1], "Updating the copy of the record...")
|
|
sync.assert_called_once()
|
|
|
|
def test_the_notice_is_cleared_when_the_sync_fails(self):
|
|
result, events, _sync = self.run_sync(OSError("registry"), shown=True)
|
|
self.assertEqual(result, {})
|
|
self.assertEqual([event[0] for event in events], ["info", "stop"])
|
|
|
|
def test_nothing_is_shown_unless_asked(self):
|
|
_result, events, sync = self.run_sync(lambda *args: {}, vmids=[120], verify=True)
|
|
self.assertEqual(events, [])
|
|
self.assertEqual(sync.call_args.args[1:], ([120], True, True))
|
|
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|