Merge remote-tracking branch 'origin/develop' into HEAD

tests/test_apprise_service_help.cjs

@@ -0,0 +1,30 @@
+// Render the real help component; only its translation provider is replaced.
+const assert = require('node:assert/strict');
+const fs = require('node:fs');
+const vm = require('node:vm');
+const ts = require('typescript');
+const React = require('react');
+const { renderToStaticMarkup } = require('react-dom/server');
+const source = fs.readFileSync('AppImage/components/apprise-service-help.tsx', 'utf8');
+const js = ts.transpileModule(source, { compilerOptions: { module: ts.ModuleKind.CommonJS, jsx: ts.JsxEmit.ReactJSX } }).outputText;
+const english = JSON.parse(fs.readFileSync('AppImage/messages/en/common.json'));
+function resolve(catalog, key) { return key.split('.').reduce((value, part) => value?.[part], catalog); }
+for (const locale of ['en', 'es', 'de', 'fr', 'it', 'pt', 'sk', 'sv', 'missing']) {
+  const catalog = locale === 'missing' ? {} : JSON.parse(fs.readFileSync(`AppImage/messages/${locale}/common.json`));
+  const context = { exports: {}, require: name => name === '../lib/i18n/provider' ? {
+    useT: () => key => resolve(catalog, key) ?? resolve(english, key),
+  } : require(name) };
+  vm.runInNewContext(js, context);
+  const html = renderToStaticMarkup(React.createElement(context.exports.AppriseServiceHelp));
+  assert.ok(html.includes('https://hooks.slack.com/services/TOKEN_A/TOKEN_B/TOKEN_C'));
+  assert.ok(html.includes('matrixs://USER:PASSWORD@matrix.example.com/!ROOM_ID:matrix.example.com'));
+  assert.ok(html.includes('https://appriseit.com/services/slack/'));
+  assert.ok(html.includes('https://appriseit.com/services/matrix/'));
+  assert.ok(!html.includes('settings.notifications.'), `${locale}: no unresolved labels`);
+}
+for (const [, key] of source.matchAll(/t\("([^"]+)"\)/g)) assert.ok(resolve(english, key), key);
+assert.ok(!source.includes('onChange') && !source.includes('fetch(') && !source.includes('updateChannel'));
+const settings = fs.readFileSync('AppImage/components/notification-settings.tsx', 'utf8');
+const section = settings.slice(settings.indexOf('<TabsContent value="apprise"'), settings.indexOf('</TabsContent>', settings.indexOf('<TabsContent value="apprise"')));
+for (const part of ['<AppriseServiceHelp />', 'renderChannelCategories("apprise")', 'handleTest("apprise")']) assert.ok(section.includes(part));
+console.log('PASS: real provider-help JSX, shipped locales and English fallback, no secret mutation, native category/test controls retained');

tests/test_apprise_services.py

@@ -0,0 +1,137 @@
+"""Real Apprise Slack/Matrix plugins with the HTTP boundary replaced.
+
+No provider account or network is used. These contracts run against both
+the minimum bundled Apprise version and the current tested version.
+"""
+import gc
+import json
+from pathlib import Path
+import sys
+import unittest
+from unittest.mock import patch
+from urllib.parse import urlparse
+
+import apprise
+import requests
+
+sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "AppImage/scripts"))
+from notification_channels import create_channel
+
+SLACK = "https://hooks.slack.com/services/T00000000/B00000000/fixturesecret1234567890"
+MATRIX = "matrixs://notify:fixture-password@matrix.example.test/!room:matrix.example.test"
+
+
+class AppriseServiceTests(unittest.TestCase):
+    def setUp(self):
+        self.calls = []
+        self.status = 200
+        self.transport = patch.object(requests.Session, "request", side_effect=self.http)
+        self.transport.start()
+        self.addCleanup(self.transport.stop)
+        self.sleep = patch("notification_channels.time.sleep")
+        self.sleep.start()
+        self.addCleanup(self.sleep.stop)
+        self.addCleanup(gc.collect)
+
+    def http(self, method, url, **kwargs):
+        self.calls.append((method, url, kwargs))
+        response = requests.Response()
+        response.status_code = self.status
+        response.url = url
+        response.reason = "OK" if self.status == 200 else "Rejected"
+        response._content_consumed = True
+        host, path = urlparse(url).hostname, urlparse(url).path
+        if host == "hooks.slack.com":
+            self.assertTrue(path.startswith("/services/"))
+            response._content = b"ok" if self.status == 200 else b"invalid_auth"
+        else:
+            self.assertEqual(host, "matrix.example.test", "no unexpected destinations")
+            if self.status != 200:
+                payload = {"errcode": "M_FORBIDDEN", "error": "Invalid credentials"}
+            elif path == "/.well-known/matrix/client":
+                payload = {"m.homeserver": {"base_url": "https://" + host}}
+            elif path.endswith("/login"):
+                payload = {"access_token": "fixture-token", "home_server": host, "user_id": "@notify:" + host}
+            elif path.endswith("/joined_rooms"):
+                payload = {"joined_rooms": ["!room:" + host]}
+            elif "/join/" in path:
+                payload = {"room_id": "!room:" + host}
+            elif "/send/" in path:
+                payload = {"event_id": "$fixture"}
+            elif path.endswith("/logout"):
+                payload = {}
+            elif path.endswith("/versions"):
+                payload = {"versions": ["v1.1"]}
+            else:
+                raise AssertionError("Unhandled Matrix request: " + path)
+            response._content = json.dumps(payload).encode()
+        return response
+
+    def send(self, url):
+        channel = create_channel("apprise", {"url": url})
+        self.assertIsNotNone(channel)
+        return channel.send("ProxMenux probe", "Controlled message", "WARNING")
+
+    def test_slack_accepts_the_native_webhook_without_conversion(self):
+        result = self.send(SLACK)
+        self.assertTrue(result["success"], result)
+        self.assertTrue(any(url == SLACK for _, url, _ in self.calls))
+        self.assertIn("Controlled message", str(self.calls))
+
+    def test_slack_apprise_scheme_uses_the_same_transport(self):
+        result = self.send(SLACK.replace("https://hooks.slack.com/services/", "slack://"))
+        self.assertTrue(result["success"], result)
+        self.assertTrue(any(url == SLACK for _, url, _ in self.calls))
+
+    def test_matrix_authenticates_and_delivers_to_the_full_room_id_over_https(self):
+        result = self.send(MATRIX)
+        self.assertTrue(result["success"], result)
+        sends = [(url, kwargs) for _, url, kwargs in self.calls if "/send/" in url]
+        self.assertTrue(sends, self.calls)
+        self.assertTrue(all(url.startswith("https://matrix.example.test/") for url, _ in sends))
+        self.assertTrue(all(kwargs.get("verify", True) for _, kwargs in sends))
+        self.assertIn("Controlled message", str(sends))
+        self.assertTrue(any(url.endswith("/login") for _, url, _ in self.calls))
+
+    def test_test_button_contract_uses_the_same_provider(self):
+        ok, error = create_channel("apprise", {"url": SLACK}).test()
+        self.assertTrue(ok, error)
+        self.assertTrue(self.calls)
+
+    def test_rejected_slack_webhook_is_not_reported_as_delivered(self):
+        self.status = 403
+        result = self.send(SLACK)
+        self.assertFalse(result["success"])
+        self.assertTrue(result["error"])
+
+    def test_invalid_matrix_credentials_are_not_reported_as_delivered(self):
+        self.status = 403
+        result = self.send(MATRIX)
+        self.assertFalse(result["success"])
+        self.assertFalse(any("/send/" in url for _, url, _ in self.calls))
+
+    def test_invalid_urls_and_missing_library_fail_before_any_request(self):
+        channel = create_channel("apprise", {"url": "not-a-url"})
+        self.assertFalse(channel.send("Probe", "Body")["success"])
+        channel = create_channel("apprise", {"url": SLACK})
+        with patch.object(channel, "_load_apprise", return_value=None):
+            self.assertFalse(channel.send("Probe", "Body")["success"])
+        self.assertEqual(self.calls, [])
+
+    def test_connection_failure_never_claims_delivery(self):
+        for url in (SLACK, MATRIX):
+            with patch.object(requests.Session, "request", side_effect=requests.ConnectionError("fixture connection failure")):
+                result = self.send(url)
+            self.assertFalse(result["success"])
+            self.assertTrue(result["error"])
+
+    def test_severity_mapping_keeps_native_notification_semantics(self):
+        channel = create_channel("apprise", {"url": SLACK})
+        for severity, expected in (("CRITICAL", apprise.NotifyType.FAILURE), ("WARNING", apprise.NotifyType.WARNING),
+                                   ("INFO", apprise.NotifyType.INFO), ("OK", apprise.NotifyType.SUCCESS),
+                                   ("SUCCESS", apprise.NotifyType.SUCCESS)):
+            self.assertEqual(channel._severity_to_notify_type(apprise, severity), expected)
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_backup_storage_retention.cjs

@@ -0,0 +1,27 @@
+const assert = require('node:assert/strict');
+const fs = require('node:fs');
+const vm = require('node:vm');
+const ts = require('typescript');
+const source = fs.readFileSync('AppImage/lib/backup-retention.ts', 'utf8');
+const context = { exports: {} };
+vm.runInNewContext(ts.transpileModule(source, { compilerOptions: { module: ts.ModuleKind.CommonJS } }).outputText, context);
+const { parseRetention, buildRetention } = context.exports;
+for (const policy of ['keep-all=1', 'keep-last=3', 'keep-last=3,keep-weekly=2', 'keep-hourly=24,keep-daily=7,keep-monthly=12,keep-yearly=2']) {
+  const { fields, keepAll } = parseRetention(policy);
+  assert.equal(buildRetention(fields, keepAll), policy);
+}
+assert.equal(buildRetention({}, false), '', 'empty policy remains explicit');
+assert.equal(buildRetention({}, true), 'keep-all=1');
+assert.equal(buildRetention({ daily: '3', hourly: '0', last: '' }, false), 'keep-daily=3');
+assert.equal(buildRetention({ last: '3' }, true), 'keep-last=3', 'existing job editor precedence preserved');
+const component = fs.readFileSync('AppImage/components/backup-storage-retention.tsx', 'utf8');
+const english = JSON.parse(fs.readFileSync('AppImage/messages/en/common.json')).backup.storageRetention;
+for (const [, key] of component.matchAll(/t\("backup\.storageRetention\.([A-Za-z]+)"/g)) {
+  assert.ok(typeof english[key] === 'string' && english[key], `translation: ${key}`);
+}
+assert.ok(component.includes('digest: editing.digest'));
+assert.ok(component.includes('method: "PUT"'));
+assert.ok(!component.includes('/prune'));
+assert.ok(fs.readFileSync('AppImage/components/vm-backup-jobs.tsx', 'utf8').includes('from "../lib/backup-retention"'));
+assert.ok(fs.readFileSync('AppImage/components/backup-tab.tsx', 'utf8').includes('<BackupStorageRetention />'));
+console.log('PASS: shared native retention properties, keep-all/count round trips, explicit empty policy, translations and digest wiring');

tests/test_figurine_installation.py

@@ -0,0 +1,204 @@
+"""Exercise the real Figurine installer with isolated files and no network/root access."""
+
+import ast
+import importlib.util
+import io
+from pathlib import Path
+import re
+import subprocess
+import tarfile
+import tempfile
+import unittest
+
+
+ROOT = Path(__file__).resolve().parents[1]
+SCRIPT = ROOT / "scripts/post_install/customizable_post_install.sh"
+FUNCTION = re.search(
+    r"^configure_figurine\(\) \{\n.*?^\}", SCRIPT.read_text(), re.M | re.S
+).group(0)
+VERSION = re.search(r'FIGURINE_VERSION:-([^}]+)', FUNCTION).group(1)
+FUNCTION_VERSION = re.search(r'local FUNC_VERSION="([^"]+)"', FUNCTION).group(1)
+
+
+class FigurineInstallation(unittest.TestCase):
+    def run_installer(self, arch="amd64", *, kernel="x86_64", dpkg_exit=0,
+                      release=None, download_exit=0, package="valid", existing=True):
+        with tempfile.TemporaryDirectory(prefix="figurine-test-") as directory:
+            root = Path(directory)
+            binary_dir = root / "bin"
+            profile_dir = root / "profile.d"
+            binary_dir.mkdir()
+            profile_dir.mkdir()
+            (root / "tmp").mkdir()
+            binary = binary_dir / "figurine"
+            profile = profile_dir / "figurine.sh"
+            bashrc = root / "bashrc"
+            if existing:
+                binary.write_text("previous binary\n")
+                binary.chmod(0o755)
+                profile.write_text("previous profile\n")
+                bashrc.write_text("# previous bashrc\n")
+            archive = root / "fixture.tar.gz"
+            payload = b"#!/bin/sh\nprintf 'fixture Figurine\\n'\n"
+            if package == "corrupt":
+                archive.write_bytes(b"invalid tarball")
+            else:
+                with tarfile.open(archive, "w:gz") as tar:
+                    name = "deploy/figurine" if package == "valid" else "README"
+                    member = tarfile.TarInfo(name)
+                    member.size = len(payload)
+                    tar.addfile(member, io.BytesIO(payload))
+
+            # Relocate every installer filesystem boundary, including its login script.
+            function = (FUNCTION.replace("/usr/local/bin", str(binary_dir))
+                        .replace("/etc/profile.d", str(profile_dir))
+                        .replace("/root/.bashrc", str(bashrc)))
+            harness = r'''
+translate() { printf '%s' "$1"; }
+msg_info2() { :; }
+msg_info() { :; }
+msg_ok() { :; }
+msg_success() { :; }
+msg_error() { printf '%s\n' "$*" >&2; }
+pmx_journal_context() { :; }
+pmx_write_file() { cat > "$1"; }
+pmx_append_file() { cat >> "$1"; }
+pmx_edit_file() { :; }
+chown() { :; }
+dpkg() {
+    [[ "$TEST_DPKG_EXIT" == 0 ]] || return "$TEST_DPKG_EXIT"
+    printf '%s\n' "$TEST_ARCH"
+}
+uname() { printf '%s\n' "$TEST_KERNEL"; }
+wget() {
+    [[ "$1" == -qO ]] || return 90
+    printf '%s\n' "$3" >> "$TEST_ROOT/downloads"
+    [[ "$TEST_DOWNLOAD_EXIT" == 0 ]] || return "$TEST_DOWNLOAD_EXIT"
+    cp "$TEST_ROOT/fixture.tar.gz" "$2"
+}
+register_tool() { printf '%s\n' "$*" > "$TEST_ROOT/registered"; }
+'''
+            harness += function + "\nconfigure_figurine\nexit \"$?\"\n"
+            environment = {
+                "PATH": f"{binary_dir}:/usr/bin:/bin",
+                "HOME": str(root),
+                "TMPDIR": str(root / "tmp"),
+                "LC_ALL": "C",
+                "TEST_ROOT": str(root),
+                "TEST_ARCH": arch,
+                "TEST_DPKG_EXIT": str(dpkg_exit),
+                "TEST_KERNEL": kernel,
+                "TEST_DOWNLOAD_EXIT": str(download_exit),
+            }
+            if release is not None:
+                environment["FIGURINE_VERSION"] = release
+            result = subprocess.run(
+                ["/bin/bash", "--noprofile", "--norc"], input=harness,
+                text=True, capture_output=True, env=environment, timeout=10,
+            )
+            return {
+                "result": result,
+                "binary": binary.read_bytes(),
+                "mode": binary.stat().st_mode & 0o777,
+                "profile": profile.read_text(),
+                "bashrc": bashrc.read_text(),
+                "downloads": (root / "downloads").read_text() if (root / "downloads").exists() else "",
+                "registered": (root / "registered").read_text() if (root / "registered").exists() else "",
+                "payload": payload,
+                "binary_path": str(binary),
+            }
+
+    def test_installs_release_matching_debian_userspace(self):
+        for arch, asset in (("amd64", "amd64"), ("arm64", "arm64"), ("armhf", "arm")):
+            with self.subTest(arch=arch):
+                state = self.run_installer(arch)
+                self.assertEqual(state["result"].returncode, 0, state["result"].stderr)
+                self.assertEqual(state["downloads"],
+                                 f"https://github.com/arsham/figurine/releases/download/v{VERSION}/"
+                                 f"figurine_linux_{asset}_v{VERSION}.tar.gz\n")
+                self.assertEqual(state["binary"], state["payload"])
+                self.assertEqual(state["mode"], 0o755)
+                self.assertIn(f'{state["binary_path"]} -f "3d.flf" $(hostname)', state["profile"])
+                self.assertEqual(state["registered"], f"figurine true {FUNCTION_VERSION}\n")
+
+    def test_falls_back_to_kernel_when_dpkg_is_unavailable(self):
+        for kernel, asset in (("x86_64", "amd64"), ("aarch64", "arm64"),
+                              ("armv7l", "arm"), ("armv8l", "arm")):
+            with self.subTest(kernel=kernel):
+                state = self.run_installer("", kernel=kernel, dpkg_exit=127)
+                self.assertEqual(state["result"].returncode, 0, state["result"].stderr)
+                self.assertIn(f"figurine_linux_{asset}_v{VERSION}.tar.gz", state["downloads"])
+
+    def test_userspace_takes_precedence_over_kernel(self):
+        state = self.run_installer("armhf", kernel="aarch64")
+        self.assertEqual(state["result"].returncode, 0, state["result"].stderr)
+        self.assertIn(f"figurine_linux_arm_v{VERSION}.tar.gz", state["downloads"])
+
+    def test_fresh_install_creates_binary_banner_and_aliases(self):
+        state = self.run_installer(existing=False)
+        self.assertEqual(state["result"].returncode, 0, state["result"].stderr)
+        self.assertEqual(state["binary"], state["payload"])
+        self.assertIn(' -f "3d.flf" $(hostname)', state["profile"])
+        self.assertIn("shopt -s expand_aliases", state["bashrc"])
+        self.assertIn("alias aptup='apt update && apt dist-upgrade'", state["bashrc"])
+        self.assertEqual(state["registered"], f"figurine true {FUNCTION_VERSION}\n")
+
+    def test_unsupported_architecture_fails_before_changes_or_download(self):
+        for arch in ("i386", "riscv64", "armel", "armv6l", "ppc64el", ""):
+            with self.subTest(arch=arch):
+                state = self.run_installer(arch)
+                self.assertNotEqual(state["result"].returncode, 0)
+                self.assertIn("Unsupported architecture for Figurine:", state["result"].stderr)
+                self.assertEqual(state["downloads"], "")
+                self.assert_preserved(state)
+                self.assertEqual(state["bashrc"], "# previous bashrc\n")
+
+    def assert_preserved(self, state):
+        self.assertEqual(state["binary"], b"previous binary\n")
+        self.assertEqual(state["profile"], "previous profile\n")
+        self.assertEqual(state["registered"], "")
+
+    def test_download_failure_preserves_previous_installation(self):
+        state = self.run_installer(download_exit=8)
+        self.assertNotEqual(state["result"].returncode, 0)
+        self.assertIn("Failed to download Figurine", state["result"].stderr)
+        self.assert_preserved(state)
+
+    def test_invalid_packages_preserve_previous_installation(self):
+        for package, error in (("corrupt", "Failed to extract package"),
+                               ("missing", "Binary not found in extracted content.")):
+            with self.subTest(package=package):
+                state = self.run_installer(package=package)
+                self.assertNotEqual(state["result"].returncode, 0)
+                self.assertIn(error, state["result"].stderr)
+                self.assert_preserved(state)
+
+    def test_release_override_is_preserved(self):
+        state = self.run_installer("arm64", release="2.1.0")
+        self.assertEqual(state["result"].returncode, 0, state["result"].stderr)
+        self.assertIn("/v2.1.0/figurine_linux_arm64_v2.1.0.tar.gz", state["downloads"])
+
+    def test_existing_installation_is_offered_the_update_once(self):
+        spec = importlib.util.spec_from_file_location(
+            "post_install_versions", ROOT / "AppImage/scripts/post_install_versions.py")
+        versions = importlib.util.module_from_spec(spec)
+        spec.loader.exec_module(versions)
+        metadata = versions.parse_post_install_script(SCRIPT)
+        installed = {"figurine": {"installed": True, "version": "1.1", "source": "custom"}}
+        updates = versions._detect_updates({}, metadata, installed)
+        self.assertEqual(len(updates), 1)
+        self.assertEqual(updates[0]["available_version"], FUNCTION_VERSION)
+        self.assertEqual(updates[0]["function"], "configure_figurine")
+        installed["figurine"]["version"] = FUNCTION_VERSION
+        self.assertEqual(versions._detect_updates({}, metadata, installed), [])
+
+        tree = ast.parse((ROOT / "AppImage/scripts/flask_proxmenux_routes.py").read_text())
+        fallback = next(ast.literal_eval(node.value) for node in tree.body
+                        if isinstance(node, ast.Assign)
+                        and any(isinstance(t, ast.Name) and t.id == "TOOL_METADATA"
+                                for t in node.targets))
+        self.assertEqual(fallback["figurine"]["version"], FUNCTION_VERSION)
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_vm_backup_jobs_api.py

@@ -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,63 @@ 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_an_empty_policy_returns_the_storage_to_the_proxmox_default(self):
+        self.assertEqual(self.storage_policy("").status_code, 200)
+        self.assertEqual(self.changes(), [["set", "/storage/local", "--digest", "a" * 40, "--delete", "prune-backups"]])
+
+    def test_invalid_storage_retention_is_rejected_before_commands(self):
+        invalid = [None, [], {}, {"digest": "a" * 40, "prune_backups": []}]
+        for prune in ("keep-last=3,keep-daily=0", "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",