Mirrors/ProxMenux
1
0
Fork 0
mirror of https://github.com/MacRimi/ProxMenux.git synced 2026-11-02 19:02:31 +00:00
Code Issues Packages Projects Releases Wiki Activity

Expose native storage retention for VM and CT backups

Browse Source
This commit is contained in:
Jonatan
2026-10-16 20:32:05 +02:00
parent 3000c1cd66
commit 113659c0e3
11 changed files with 353 additions and 22 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_vm_backup_jobs_api.py
+81 -2
View File
@@ -1,5 +1,6 @@
"""VM/CT backup jobs API: what reaches pvesh, with pvesh itself replaced."""
import importlib
import importlib.util
import json
import sys
import types
@@ -23,6 +24,34 @@ GUESTS = [{"vmid": 100, "name": "web", "type": "qemu", "node": "pve1"},
{"vmid": 200, "name": "other", "type": "qemu", "node": "pve2"}]
@unittest.skipIf(Flask is None, "Flask runtime required")
class StorageRetentionAuthorizationTests(unittest.TestCase):
def test_real_authentication_guard_blocks_read_only_writes_before_pvesh(self):
scripts = Path(__file__).resolve().parents[1] / "AppImage/scripts"
auth = types.ModuleType("auth_manager")
auth.load_auth_config = lambda: {"enabled": True}
auth.verify_token = lambda token: "tester" if token in ("admin", "reader") else None
auth.verify_token_full = lambda token: ("tester", "full_admin" if token == "admin" else "read_only")
spec = importlib.util.spec_from_file_location("storage_retention_middleware", scripts / "jwt_middleware.py")
middleware = importlib.util.module_from_spec(spec)
with patch.dict(sys.modules, auth_manager=auth):
spec.loader.exec_module(middleware)
spec = importlib.util.spec_from_file_location("storage_retention_auth_routes", scripts / "flask_vm_backup_routes.py")
routes = importlib.util.module_from_spec(spec)
with patch.dict(sys.modules, jwt_middleware=middleware):
spec.loader.exec_module(routes)
app = Flask(__name__)
app.register_blueprint(routes.vm_backup_bp)
client = app.test_client()
data = {"digest": "a" * 40, "prune_backups": "keep-last=3"}
with patch.object(routes, "_pvesh") as run:
self.assertEqual(client.get("/api/vm-backup-storages").status_code, 401)
self.assertEqual(client.put("/api/vm-backup-storages/local", json=data).status_code, 401)
self.assertEqual(client.put("/api/vm-backup-storages/local", json=data,
headers={"Authorization": "Bearer reader"}).status_code, 403)
run.assert_not_called()
@unittest.skipIf(Flask is None, "Flask runtime required")
class VmBackupJobsApiTests(unittest.TestCase):
def setUp(self):
@@ -73,9 +102,59 @@ class VmBackupJobsApiTests(unittest.TestCase):
def test_reading_needs_a_session_and_changing_needs_an_administrator(self):
guards = dict((name, kind) for kind, name in self.guards)
self.assertEqual({name for name, kind in guards.items() if kind == "auth"}, {"list_jobs", "options", "get_job"})
self.assertEqual({name for name, kind in guards.items() if kind == "auth"}, {"list_jobs", "options", "get_job", "backup_storage_retention"})
self.assertEqual({name for name, kind in guards.items() if kind == "admin"},
{"create_job", "update_job", "delete_job", "toggle_job", "run_job"})
{"create_job", "update_job", "delete_job", "toggle_job", "run_job", "update_backup_storage_retention"})
def test_storage_retention_lists_only_backup_storages_and_no_secrets(self):
with patch.object(self.routes, "_pvesh_json", return_value=(True, [
{"storage": "local", "type": "dir", "content": "backup,iso", "digest": "a" * 40,
"prune-backups": {"keep-last": 3}, "password": "private"},
{"storage": "images", "type": "dir", "content": "images", "digest": "a" * 40},
])):
answer = self.client.get("/api/vm-backup-storages").get_json()
self.assertEqual(answer, {"storages": [{"id": "local", "type": "dir", "digest": "a" * 40, "prune_backups": "keep-last=3"}]})
def storage_policy(self, prune="keep-last=3,keep-weekly=2", digest="a" * 40, current=None):
if current is None:
current = {"storage": "local", "content": "backup,iso", "digest": "a" * 40}
with patch.object(self.routes, "_pvesh_json", return_value=(True, current)), self.pvesh():
return self.client.put("/api/vm-backup-storages/local", json={"digest": digest, "prune_backups": prune})
def test_saving_native_storage_retention_never_runs_pruning_or_changes_other_properties(self):
response = self.storage_policy()
self.assertEqual(response.status_code, 200)
self.assertEqual(self.changes(), [["set", "/storage/local", "--digest", "a" * 40,
"--prune-backups", "keep-last=3,keep-weekly=2"]])
self.calls.clear()
self.assertEqual(self.storage_policy("keep-all=1").status_code, 200)
self.assertEqual(self.changes()[0][-1], "keep-all=1")
def test_invalid_storage_retention_is_rejected_before_commands(self):
invalid = [None, [], {}, {"digest": "a" * 40, "prune_backups": []}]
for prune in ("", "keep-last=-1", "keep-all=0", "keep-all=2", "keep-last=0", "keep-last=1,keep-last=2",
"keep-all=1,keep-last=2", "keep-last=2 --delete path", "keep-last=1\n", "keep-other=1"):
invalid.append({"digest": "a" * 40, "prune_backups": prune})
invalid += [{"digest": "invalid", "prune_backups": "keep-last=1"}]
with patch.object(self.routes, "_pvesh") as run:
for data in invalid:
with self.subTest(data=data):
self.assertEqual(self.client.put("/api/vm-backup-storages/local", json=data).status_code, 400)
self.assertEqual(self.client.put("/api/vm-backup-storages/-bad", json=invalid[-1]).status_code, 400)
run.assert_not_called()
def test_stale_storage_configuration_or_non_backup_storage_is_not_overwritten(self):
for current in ({"content": "backup", "digest": "b" * 40},
{"content": "images", "digest": "a" * 40}):
self.assertEqual(self.storage_policy(current=current).status_code, 409)
self.assertEqual(self.changes(), [])
def test_native_storage_update_failure_is_visible(self):
with patch.object(self.routes, "_pvesh_json", return_value=(True, {"content": "backup", "digest": "a" * 40})), \
patch.object(self.routes, "_pvesh", return_value=(False, "configuration changed under lock")):
response = self.client.put("/api/vm-backup-storages/local", json={"digest": "a" * 40, "prune_backups": "keep-last=3"})
self.assertEqual(response.status_code, 409)
self.assertIn("changed", response.get_json()["error"])
def test_the_list_gives_text_properties_and_the_host_backups_that_hang_from_a_job(self):
with self.pvesh(), patch.object(self.routes, "_host_backups_attached",