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

tests/lxc_updates/test_apply_updates.sh

@@ -162,6 +162,15 @@ run_case stopped_restore RUN_HELPER=0 UPDATE_COMMAND='update-custom' MOCK_INITIA
 [[ $(count 'pct start 101') -eq 1 ]] || fail "stopped CT was not started exactly once"
 [[ $(count 'pct shutdown 101 --timeout 60') -eq 1 ]] || fail "stopped CT state was not restored"
 
+run_case scheduled_stopped RUN_HELPER=0 UPDATE_COMMAND='update-custom' \
+  EXECUTION_MODE=scheduled SCHEDULED_STOPPED_MARKER='fixture-stopped-marker' MOCK_INITIAL_STATE=stopped
+[[ $CASE_RC -eq 9 ]] || fail "scheduled_stopped returned $CASE_RC"
+[[ $(count 'pct start 101') -eq 0 ]] || fail "scheduled update started a stopped CT"
+[[ $(count 'pct exec') -eq 0 ]] || fail "scheduled update executed inside a stopped CT"
+[[ $(count CUSTOM_EXEC) -eq 0 ]] || fail "scheduled update ran a custom updater"
+grep -qF 'fixture-stopped-marker' "$TEST_ROOT/scheduled_stopped.out" \
+  || fail "scheduled stopped result did not include its marker"
+
 run_case locked RUN_HELPER=0 UPDATE_COMMAND='update-custom' MOCK_FLOCK_FAIL=1
 [[ $CASE_RC -eq 7 ]] || fail "locked expected 7, got $CASE_RC"
 [[ $(count CUSTOM_EXEC) -eq 0 ]] || fail "locked run executed an updater"

tests/lxc_updates/test_observed_state_ui.cjs

@@ -78,7 +78,7 @@ function walk(node) {
   if (ts.isVariableDeclaration(node)) {
     const name = node.name.getText(tree)
     if (wanted.has(name)) definitions.push(`const ${node.getText(tree)};`)
-    if (["applySchedulePayload", "loadSchedule", "loadBulkUpdate"].includes(name))
+    if (["LEGACY_STOPPED_SCHEDULE_REASONS", "scheduleReasonCodeFromPersisted", "applySchedulePayload", "loadSchedule", "loadBulkUpdate"].includes(name))
       preferenceDefinitions.push(`const ${node.getText(tree)};`)
     if (["osUpdateStale", "osUpdateStatusKnown", "hasOsUpdates"].includes(name))
       status.push(`const ${node.getText(tree)};`)
@@ -325,11 +325,11 @@ context.selectedVMRef.current = { vmid: 100 }
   }
   for (const name of ["ScheduleEnabled", "ScheduleConfigured", "ScheduleCron", "SchedulePreset",
     "ScheduleTarget", "ScheduleTargets", "ScheduleReleaseDelayDays", "ApplyBackup", "ApplyBackupStorage",
-    "ApplyRestart", "ScheduleLastRunAt", "ScheduleLastRunStatus", "ScheduleLastRunReason", "ScheduleLastRunLog",
+    "ApplyRestart", "ScheduleLastRunAt", "ScheduleLastRunStatus", "ScheduleLastRunReason", "ScheduleLastRunReasonCode", "ScheduleLastRunLog",
     "ScheduleLastRunRebootRequired", "ScheduleLastRunRebootPackages", "ExternalCron", "ScheduleError",
     "ScheduleLoaded", "BulkError", "BulkLoaded", "BulkEditMode", "BulkConfigured", "BulkTargets", "BulkPersistedTargets"])
     preferenceContext[`set${name}`] = value => { preferenceState[name] = value }
-  assert.equal(preferenceDefinitions.length, 3)
+  assert.equal(preferenceDefinitions.length, 5)
   const preferences = compile(preferenceDefinitions.join("\n") + `
     exports.apply = applySchedulePayload; exports.schedule = loadSchedule; exports.bulk = loadBulkUpdate;
   `, preferenceContext)
@@ -340,6 +340,12 @@ context.selectedVMRef.current = { vmid: 100 }
   assert.equal(preferenceState.ApplyRestart, false, "missing restart cannot inherit a reboot")
   preferences.apply({ backup: false, backup_storage: "", restart: false })
   assert.equal(preferenceState.ApplyBackupStorage, "")
+  preferences.apply({ last_run_reason_code: "stopped_container", last_run_reason: "raw stopped reason" })
+  assert.equal(preferenceState.ScheduleLastRunReasonCode, "stopped_container",
+    "schedule payload publishes its structured stopped reason")
+  preferences.apply({})
+  assert.equal(preferenceState.ScheduleLastRunReasonCode, null,
+    "a missing reason code cannot leak from the previously selected guest")
   preferenceContext.vmModalCacheRef.current.schedule.set(213, { backup: true, backup_storage: "local", restart: true })
   const previousSchedule = preferences.schedule(213)
   assert.equal(preferenceState.ApplyBackup, true, "use only the selected guest's own cache")

tests/lxc_updates/test_scheduled_reason_i18n.cjs

@@ -0,0 +1,102 @@
+// Execute the real scheduled-run detail JSX: a known backend reason code is
+// localized, while an unknown/missing code keeps its raw diagnostic text.
+const assert = require('node:assert/strict')
+const fs = require('node:fs')
+const path = require('node:path')
+const vm = require('node:vm')
+const root = path.resolve(__dirname, '../../AppImage')
+const ts = require(path.join(root, 'node_modules/typescript'))
+const source = fs.readFileSync(path.join(root, 'components/virtual-machines.tsx'), 'utf8')
+const tree = ts.createSourceFile('vm.tsx', source, ts.ScriptTarget.Latest, true, ts.ScriptKind.TSX)
+let declaration
+let legacyReasons
+let reasonCodeHelper
+function walk(node) {
+  if (ts.isVariableDeclaration(node) && node.name.getText(tree) === 'renderScheduleRunDetails') declaration = node.getText(tree)
+  if (ts.isVariableDeclaration(node) && node.name.getText(tree) === 'LEGACY_STOPPED_SCHEDULE_REASONS') legacyReasons = node.getText(tree)
+  if (ts.isVariableDeclaration(node) && node.name.getText(tree) === 'scheduleReasonCodeFromPersisted') reasonCodeHelper = node.getText(tree)
+  ts.forEachChild(node, walk)
+}
+walk(tree)
+assert(declaration, 'scheduled run detail renderer must exist')
+assert(legacyReasons, 'legacy stopped-reason list must exist')
+assert(reasonCodeHelper, 'persisted schedule reason-code helper must exist')
+
+function compile(context) {
+  const result = ts.transpileModule(`const ${declaration}; exports.render = renderScheduleRunDetails;`, {
+    compilerOptions: { target: ts.ScriptTarget.ES2020, module: ts.ModuleKind.CommonJS, jsx: ts.JsxEmit.ReactJSX },
+    reportDiagnostics: true,
+  })
+  assert.equal(result.diagnostics.length, 0)
+  vm.runInNewContext(result.outputText, context)
+  return context.exports.render
+}
+
+function compileReasonCodeHelper() {
+  const result = ts.transpileModule(`const ${legacyReasons}; const ${reasonCodeHelper}; exports.reasonCode = scheduleReasonCodeFromPersisted;`, {
+    compilerOptions: { target: ts.ScriptTarget.ES2020, module: ts.ModuleKind.CommonJS },
+    reportDiagnostics: true,
+  })
+  assert.equal(result.diagnostics.length, 0)
+  const context = { exports: {}, Set }
+  vm.runInNewContext(result.outputText, context)
+  return context.exports.reasonCode
+}
+
+function flattenText(node) {
+  if (Array.isArray(node)) return node.map(flattenText).join(' ')
+  if (!node || typeof node !== 'object') return typeof node === 'string' ? node : ''
+  return flattenText(node.props?.children)
+}
+
+const persistedReasonCode = compileReasonCodeHelper()
+assert.equal(persistedReasonCode('stopped_container', 'anything'), 'stopped_container')
+assert.equal(persistedReasonCode(null, 'Container is stopped; scheduled update was skipped'), 'stopped_container')
+assert.equal(persistedReasonCode(null, 'container was stopped before scheduled update could begin'), 'stopped_container')
+assert.equal(persistedReasonCode('future_reason', 'raw diagnostic'), 'future_reason')
+assert.equal(persistedReasonCode(null, 'raw diagnostic'), null)
+
+for (const locale of ['en', 'sk']) {
+  const messages = JSON.parse(fs.readFileSync(path.join(root, `messages/${locale}/common.json`), 'utf8'))
+  const t = (key, variables = {}) => {
+    const message = key.split('.').reduce((value, part) => value?.[part], messages)
+    return String(message ?? key).replace(/\{([^}]+)\}/g, (_, name) => String(variables[name] ?? `{${name}}`))
+  }
+  const context = {
+    exports: {}, t, selectedVM: { vmid: 100, status: 'stopped' }, scheduleLastRunAt: '2026-10-16T05:10:02Z',
+    scheduleLastRunStatus: 'skipped', scheduleLastRunReasonCode: 'stopped_container',
+    scheduleLastRunReason: 'container is stopped; scheduled update was skipped',
+    scheduleLastRunRebootRequired: false, scheduleLastRunRebootPackages: [], scheduleLastRunLog: null,
+    openScheduleLog() {}, AlertTriangle: 'AlertTriangle', FileText: 'FileText',
+    require: () => ({ jsx: (type, props) => ({ type, props }), jsxs: (type, props) => ({ type, props }), Fragment: 'Fragment' }),
+  }
+  const render = compile(context)
+  const translated = flattenText(render(true))
+  const expectedCurrentReason = t('vmLxc.scheduled.reasonStoppedContainer')
+  const expectedHistoricalReason = t('vmLxc.scheduled.reasonStoppedContainerHistorical')
+  const skippedLabel = t('vmLxc.scheduled.runSkippedStatus')
+  const expectedCurrentText = locale === 'sk'
+    ? 'Kontajner je vypnutý; plánovaná aktualizácia sa preskočila.'
+    : 'Container is stopped; scheduled update was skipped.'
+  const expectedHistoricalText = locale === 'sk'
+    ? 'Kontajner bol v čase plánovanej aktualizácie vypnutý.'
+    : 'Container was stopped at the scheduled update time.'
+  assert.equal(expectedCurrentReason, expectedCurrentText,
+    `${locale}: stopped CT must describe its current stopped state`)
+  assert.equal(expectedHistoricalReason, expectedHistoricalText,
+    `${locale}: running CT must describe the stopped state at the prior scheduled run`)
+  assert(translated.includes(expectedCurrentReason), `${locale}: currently stopped CT must use the present-state catalog`)
+  assert(!translated.includes(expectedHistoricalReason), `${locale}: currently stopped CT must not use historical wording`)
+  assert(translated.includes(skippedLabel), `${locale}: skipped run must not claim that no updates are pending`)
+  assert(!translated.includes(context.scheduleLastRunReason), `${locale}: raw stopped reason must not leak into UI`)
+
+  context.selectedVM.status = 'running'
+  const historical = flattenText(render(true))
+  assert(historical.includes(expectedHistoricalReason), `${locale}: running CT must use historical wording for a past skipped run`)
+  assert(!historical.includes(expectedCurrentReason), `${locale}: running CT must not claim it is currently stopped`)
+
+  context.scheduleLastRunReasonCode = 'unknown_reason'
+  const fallback = flattenText(render(true))
+  assert(fallback.includes(context.scheduleLastRunReason), `${locale}: unknown code must retain the raw diagnostic`)
+}
+console.log('PASS scheduled reason i18n: EN/SK structured stopped reason and raw fallback')

tests/lxc_updates/test_stopped_update_observation.py

@@ -119,7 +119,8 @@ raise SystemExit(h.main())
         self.write("sleep", "#!/bin/sh\nexit 0\n")
         self.write("flock", "#!/bin/sh\nexit 0\n")
         names = {"_finalize_lxc_update", "_terminal_lxc_update_completed",
-                 "api_lxc_updates_applied", "_run_scheduled_update"}
+                 "api_lxc_updates_applied", "_run_scheduled_update",
+                 "_scheduled_runner_reported_stopped"}
         nodes = [n for n in ast.parse(observed.SERVER.read_text()).body
                  if isinstance(n, ast.FunctionDef) and n.name in names]
         self.scope = self.fixture.scope
@@ -134,6 +135,7 @@ raise SystemExit(h.main())
             "_json_list": lambda value: json.loads(value or '[]'),
             "_lxc_update_finalizations": {}, "_lxc_update_finalization_lock": threading.RLock(),
             "_LXC_UPDATE_FINALIZATION_TTL": 3600,
+            "_SCHEDULED_STOPPED_EXIT": 9,
             "_fast_guest_status": lambda *_: self.state.read_text(),
             "_lxc_update_snapshot": lambda *_: {"ct_name": "fixture"},
             "_lxc_update_target_labels": lambda *_: ["OS"],
@@ -315,6 +317,7 @@ raise SystemExit(h.main())
         self.scope['notification_manager'].emit_event.assert_called_once()
 
     def test_scheduled_path_executes_real_runner_and_real_finalization(self):
+        self.state.write_text("running")
         def run(command, **kwargs):
             kwargs['env'].update({key: value for key, value in self.env.items()
                                   if key != 'RUN_ID'})
@@ -322,7 +325,7 @@ raise SystemExit(h.main())
         self.scope['subprocess'] = SimpleNamespace(run=run, TimeoutExpired=subprocess.TimeoutExpired)
         result = self.scope['_run_scheduled_update'](100, {'targets': ['os'], 'target': 'os'})
         self.assertEqual(result['status'], 'success', result)
-        self.assertEqual(self.state.read_text(), 'stopped')
+        self.assertEqual(self.state.read_text(), 'running')
         self.assertEqual(self.fixture.row()['update_check']['count'], 0)
 
     def test_legacy_cli_without_run_id_preserves_update_and_original_state(self):

tests/lxc_updates/test_update_contract.py

@@ -1,4 +1,7 @@
 import ast
+import json
+import os
+import re
 import sys
 import subprocess
 import tempfile
@@ -312,6 +315,159 @@ class AppCacheWriteThroughContractTests(unittest.TestCase):
         self.assertIn("before_snapshot=before", scheduler_block)
 
 
+class ScheduledStoppedContainerTests(unittest.TestCase):
+    @staticmethod
+    def _worker(scope):
+        source = ast.parse((REPO_ROOT / "AppImage" / "scripts" / "flask_server.py").read_text())
+        node = next(item for item in source.body
+                    if isinstance(item, ast.FunctionDef) and item.name == "_run_scheduled_update")
+        exec(compile(ast.Module(body=[node], type_ignores=[]), "scheduled-update", "exec"), scope)
+        return scope["_run_scheduled_update"]
+
+    @staticmethod
+    def _stopped_marker_reader(scope):
+        source = ast.parse((REPO_ROOT / "AppImage" / "scripts" / "flask_server.py").read_text())
+        node = next(item for item in source.body
+                    if isinstance(item, ast.FunctionDef) and item.name == "_scheduled_runner_reported_stopped")
+        exec(compile(ast.Module(body=[node], type_ignores=[]), "scheduled-marker", "exec"), scope)
+        return scope["_scheduled_runner_reported_stopped"]
+
+    def test_stopped_container_skips_before_runner_or_any_guest_effect(self):
+        runner = MagicMock()
+        finalizer = MagicMock(return_value={"recorded": True})
+        append_log = MagicMock()
+        scope = {
+            "time": SimpleNamespace(monotonic=MagicMock(return_value=10.0)),
+            "uuid": SimpleNamespace(uuid4=lambda: SimpleNamespace(hex="a" * 32)),
+            "datetime": __import__("datetime").datetime,
+            "subprocess": SimpleNamespace(run=runner, STDOUT=object(), TimeoutExpired=subprocess.TimeoutExpired),
+            "_create_lxc_update_log": lambda _vmid, _run_id: ("fixture.log", "/tmp/fixture.log"),
+            "_fast_guest_status": MagicMock(return_value="stopped"),
+            "_normalise_schedule_targets": lambda _sched: ["os"],
+            "_lxc_update_snapshot": lambda _vmid: {"ct_name": "fixture"},
+            "_lxc_update_target_labels": lambda *_: ["OS"],
+            "_append_lxc_update_log": append_log,
+            "_prune_lxc_update_logs": MagicMock(),
+            "_finalize_lxc_update": finalizer,
+        }
+
+        outcome = self._worker(scope)(101, {"target": "os"})
+
+        self.assertEqual(outcome["status"], "skipped")
+        self.assertEqual(outcome["executed_targets"], [])
+        self.assertIn("container is stopped", outcome["reason"])
+        self.assertEqual(outcome["reason_code"], "stopped_container")
+        runner.assert_not_called()
+        self.assertEqual(finalizer.call_args.kwargs["status"], "skipped")
+        self.assertEqual(finalizer.call_args.kwargs["source"], "scheduled")
+
+    def test_stopped_marker_requires_the_scheduler_token_not_only_exit_code(self):
+        marker = "PROXMENUX_SCHEDULED_STOPPED:fixture"
+        reader = self._stopped_marker_reader({"_SCHEDULED_STOPPED_EXIT": 9, "os": os})
+        self.assertTrue(reader(SimpleNamespace(returncode=9, stdout=marker, stderr=""), None, marker))
+        self.assertFalse(reader(SimpleNamespace(returncode=9, stdout="", stderr=""), None, marker))
+        self.assertFalse(reader(SimpleNamespace(returncode=8, stdout=marker, stderr=""), None, marker))
+        self.assertFalse(reader(SimpleNamespace(returncode=4, stdout=marker, stderr=""), None, marker))
+        with tempfile.NamedTemporaryFile(mode="w", encoding="utf-8") as log:
+            log.write("prefix\\n" + marker + "\\n")
+            log.flush()
+            self.assertTrue(reader(SimpleNamespace(returncode=9, stdout="", stderr=""), log.name, marker))
+
+    def test_runner_race_to_stopped_is_reported_as_skipped_without_reboot_probe(self):
+        runner = MagicMock(return_value=SimpleNamespace(returncode=9, stdout="", stderr=""))
+        finalizer = MagicMock(return_value={"recorded": True})
+        reboot_probe = MagicMock()
+        stopped_marker = MagicMock(return_value=True)
+        lxc = SimpleNamespace(
+            _read_sidecar=lambda _vmid: {"apps": []},
+            partition_scheduled_release_targets=lambda targets, _apps: (set(), targets),
+        )
+        scope = {
+            "time": SimpleNamespace(monotonic=MagicMock(return_value=10.0)),
+            "uuid": SimpleNamespace(uuid4=lambda: SimpleNamespace(hex="b" * 32)),
+            "datetime": __import__("datetime").datetime,
+            "subprocess": SimpleNamespace(run=runner, STDOUT=object(), TimeoutExpired=subprocess.TimeoutExpired),
+            "os": os,
+            "json": json,
+            "_APPLY_UPDATES_SCRIPT": "/bin/true",
+            "_SCHEDULED_STOPPED_EXIT": 9,
+            "_DOCKER_ENGINE_INTEGRATED_COMMAND": "integrated-docker",
+            "_create_lxc_update_log": lambda _vmid, _run_id: (None, None),
+            "_fast_guest_status": MagicMock(return_value="running"),
+            "_normalise_schedule_targets": lambda _sched: ["os"],
+            "_lxc_update_snapshot": lambda _vmid: {"ct_name": "fixture"},
+            "_lxc_update_target_labels": lambda *_: ["OS"],
+            "_append_lxc_update_log": MagicMock(),
+            "_prune_lxc_update_logs": MagicMock(),
+            "_finalize_lxc_update": finalizer,
+            "_scheduled_helper_enabled": lambda *_: False,
+            "_compose_scheduled_update_command": lambda *_: "",
+            "_inspect_lxc_reboot_requirement": reboot_probe,
+            "_scheduled_runner_reported_stopped": stopped_marker,
+        }
+
+        with patch.dict(sys.modules, {"lxc_apps": lxc}):
+            outcome = self._worker(scope)(101, {"target": "os"})
+
+        self.assertEqual(outcome["status"], "skipped")
+        self.assertEqual(outcome["executed_targets"], [])
+        self.assertIn("before scheduled update could begin", outcome["reason"])
+        self.assertEqual(outcome["reason_code"], "stopped_container")
+        self.assertEqual(runner.call_args.args[0], ["bash", "/bin/true"])
+        self.assertEqual(runner.call_args.kwargs["env"]["EXECUTION_MODE"], "scheduled")
+        self.assertTrue(runner.call_args.kwargs["env"]["SCHEDULED_STOPPED_MARKER"].startswith(
+            "PROXMENUX_SCHEDULED_STOPPED:scheduled-"))
+        stopped_marker.assert_called_once()
+        reboot_probe.assert_not_called()
+
+        stopped_marker.reset_mock()
+        stopped_marker.return_value = False
+        runner.return_value = SimpleNamespace(returncode=8, stdout="", stderr="")
+        reboot_probe.return_value = (None, [], None)
+        outcome = self._worker(scope)(101, {"target": "os"})
+        self.assertEqual(outcome["status"], "failure")
+        self.assertIn("exited with code 8", outcome["reason"])
+        stopped_marker.assert_called_once()
+        reboot_probe.assert_called_once()
+
+    def test_oci_race_to_stopped_skips_before_the_oci_engine_runs(self):
+        runner = MagicMock(return_value=SimpleNamespace(returncode=0, stdout="", stderr=""))
+        finalizer = MagicMock(return_value={"recorded": True})
+        status = MagicMock(side_effect=["running", "stopped"])
+        scope = {
+            "time": SimpleNamespace(monotonic=MagicMock(return_value=10.0)),
+            "uuid": SimpleNamespace(uuid4=lambda: SimpleNamespace(hex="c" * 32)),
+            "datetime": __import__("datetime").datetime,
+            "subprocess": SimpleNamespace(
+                run=runner,
+                STDOUT=object(),
+                DEVNULL=object(),
+                TimeoutExpired=subprocess.TimeoutExpired,
+            ),
+            "os": os,
+            "re": re,
+            "_create_lxc_update_log": lambda _vmid, _run_id: (None, None),
+            "_fast_guest_status": status,
+            "_normalise_schedule_targets": lambda _sched: ["oci_image"],
+            "_lxc_update_snapshot": lambda _vmid: {"ct_name": "fixture"},
+            "_lxc_update_target_labels": lambda *_: ["OCI image"],
+            "_append_lxc_update_log": MagicMock(),
+            "_prune_lxc_update_logs": MagicMock(),
+            "_finalize_lxc_update": finalizer,
+        }
+
+        outcome = self._worker(scope)(101, {"targets": ["oci_image"]})
+
+        self.assertEqual(outcome["status"], "skipped")
+        self.assertEqual(outcome["target"], "oci_image")
+        self.assertEqual(outcome["executed_targets"], [])
+        self.assertEqual(outcome["reason_code"], "stopped_container")
+        self.assertIn("before scheduled update could begin", outcome["reason"])
+        self.assertEqual(status.call_count, 2)
+        runner.assert_not_called()
+        self.assertEqual(finalizer.call_args.kwargs["status"], "skipped")
+
+
 class DockerStackNotificationTests(unittest.TestCase):
     def _docker_app(self):
         return {

tests/lxc_updates/test_update_method_choice.py

@@ -37,6 +37,7 @@ def routes():
               _lxc_update_target_labels=lambda *_: [],
               _append_lxc_update_log=lambda *_: None,
               _prune_lxc_update_logs=lambda *_: None,
+              _scheduled_runner_reported_stopped=lambda *_: False,
               _finalize_lxc_update=Mock(return_value={}),
               _inspect_lxc_reboot_requirement=lambda *_, **__: (False, [], None))
     exec(compile(ast.Module(body=nodes, type_ignores=[]), 'update-routes', 'exec'), ns)

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_guest_backup_archives.cjs

@@ -0,0 +1,69 @@
+// Run the shipped deletion client with only its authenticated transport replaced.
+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/guest-backup-archives.ts', 'utf8');
+const js = ts.transpileModule(source, { compilerOptions: { module: ts.ModuleKind.CommonJS, target: ts.ScriptTarget.ES2020 } }).outputText;
+const backup = { volid: 'local:backup/vzdump-lxc-100-fixture.tar.zst', storage: 'local' };
+
+async function run(responses) {
+  const calls = [];
+  const context = { exports: {}, AbortSignal, Date, setTimeout: callback => callback(), require: name => {
+    assert.equal(name, './api-config');
+    return { fetchApi: async (path, options) => {
+      calls.push({ path, options });
+      assert.ok(responses.length, `unexpected call: ${path}`);
+      const response = responses.shift();
+      if (response instanceof Error) throw response;
+      return response;
+    } };
+  } };
+  vm.runInNewContext(js, context);
+  let results, error;
+  try { results = JSON.parse(JSON.stringify(await context.exports.deleteGuestBackups(100, [backup]))); }
+  catch (caught) { error = caught; }
+  return { calls, results, error };
+}
+
+(async () => {
+  const success = await run([{ results: [{ volid: backup.volid, status: 'deleted' }] }]);
+  assert.equal(success.results[0].status, 'deleted');
+  assert.deepEqual(JSON.parse(success.calls[0].options.body), { confirm: true, backups: [backup] });
+  assert.equal(success.calls[0].options.method, 'POST');
+  assert.equal(success.calls.length, 1);
+  const task = 'UPID:pve1:01:02:03:imgdel:100:root@pam:';
+  const pending = { results: [{ volid: backup.volid, status: 'pending', task }] };
+  const completed = await run([pending, { volid: backup.volid, status: 'pending' }, { volid: backup.volid, status: 'deleted' }]);
+  assert.equal(completed.results[0].status, 'deleted');
+  assert.equal(completed.calls.length, 3);
+  assert.equal(completed.calls[1].path, `/api/vms/100/backups/delete-status?task=${encodeURIComponent(task)}`);
+  assert.equal(completed.calls.filter(call => call.options.method === 'POST').length, 1);
+  const failed = await run([pending, { volid: backup.volid, status: 'unconfirmed', error: 'protected' }]);
+  assert.equal(failed.results[0].status, 'unconfirmed');
+  const lostPost = await run([new Error('network failure')]);
+  assert.equal(lostPost.error.message, 'network failure');
+  assert.equal(lostPost.calls.length, 1, 'never retry destructive POST');
+  const lostStatus = await run([pending, new Error('timeout'), new Error('timeout'), new Error('timeout')]);
+  assert.equal(lostStatus.results[0].status, 'unconfirmed');
+  assert.equal(lostStatus.calls.length, 4, 'bounded read failures');
+  for (const response of [{ results: [] }, { results: [{ volid: 'other', status: 'deleted' }] },
+                           { results: [{ volid: backup.volid, status: 'accepted' }] }]) {
+    assert.equal((await run([response])).error.message, 'deletionUnconfirmed');
+  }
+  const foreignStatus = await run([pending, { volid: 'other', status: 'deleted' }]);
+  assert.equal(foreignStatus.results[0].status, 'unconfirmed');
+
+  const component = fs.readFileSync('AppImage/components/guest-backup-list.tsx', 'utf8');
+  const english = JSON.parse(fs.readFileSync('AppImage/messages/en/common.json')).vmLxc.backups;
+  for (const [, key] of component.matchAll(/t\("vmLxc\.backups\.([A-Za-z]+)"/g)) {
+    assert.ok(typeof english[key] === 'string' && english[key].length, `English translation: ${key}`);
+  }
+  assert.ok(component.includes('disabled={disabled || !!backup.protected'));
+  assert.ok(component.includes('!backup.protected'));
+  assert.ok(component.includes('DialogDescription'));
+  const parent = fs.readFileSync('AppImage/components/virtual-machines.tsx', 'utf8');
+  assert.ok(parent.includes('<GuestBackupList'));
+  assert.ok(parent.includes('await fetchVmBackups(selectedVM.vmid, true)'));
+  console.log('PASS: native task completion, partial/unconfirmed outcomes, lost responses, no destructive retries, translated UI and forced refresh');
+})().catch(error => { console.error(error); process.exitCode = 1; });

tests/test_guest_backup_archives.py

@@ -0,0 +1,239 @@
+"""Guest archive deletion contract. No real Proxmox commands or backups."""
+
+import ast
+import copy
+import importlib.util
+import json
+from pathlib import Path
+import subprocess
+import sys
+import types
+import unittest
+from unittest.mock import Mock, patch
+
+from flask import Flask, jsonify, request
+
+ROOT = Path(__file__).resolve().parents[1]
+SCRIPTS = ROOT / "AppImage/scripts"
+sys.path.insert(0, str(SCRIPTS))
+import guest_backup_archives as archives
+
+VOLUME = "local:backup/vzdump-lxc-100-2026_10_16-12_00_00.tar.zst"
+OTHER = VOLUME.replace("12_00_00", "13_00_00")
+TASK = "UPID:pve1:00000001:00000002:00000003:imgdel:100:root@pam:"
+
+
+class DeletionTests(unittest.TestCase):
+    def setUp(self):
+        self.node = patch.object(archives, "_node", return_value="pve1")
+        self.node.start()
+        self.addCleanup(self.node.stop)
+        self.inventory = [{"volid": VOLUME, "vmid": 100, "content": "backup", "protected": False}]
+        self.calls = []
+        self.response = None
+        archives._tasks.clear()
+
+    def run_pvesh(self, args, **kwargs):
+        self.calls.append(args)
+        if args[0] == "get" and args[1].endswith("/content"):
+            return copy.deepcopy(self.inventory)
+        if args[0] == "delete":
+            if isinstance(self.response, Exception):
+                raise self.response
+            return self.response
+        return {"status": "stopped", "exitstatus": "OK"}
+
+    def delete(self, volumes=(VOLUME,), **kwargs):
+        data = {"confirm": True, "backups": [{"storage": v.split(":")[0], "volid": v} for v in volumes]}
+        data.update(kwargs)
+        with patch.object(archives, "_pvesh", side_effect=self.run_pvesh):
+            return archives.delete_guest_backups(100, data)
+
+    def test_success_uses_guest_filtered_native_inventory_and_delete(self):
+        self.assertEqual(self.delete()["results"], [{"storage": "local", "volid": VOLUME, "status": "deleted"}])
+        self.assertEqual(self.calls, [
+            ["get", "/nodes/pve1/storage/local/content", "--content", "backup", "--vmid", "100"],
+            ["delete", f"/nodes/pve1/storage/local/content/{VOLUME}", "--delay", "1"],
+        ])
+
+    def test_a_deletion_that_ends_within_the_delay_prints_its_log_and_counts_as_deleted(self):
+        # Real pvesh output on a directory storage: the task log, not JSON.
+        run = subprocess.CompletedProcess([], 0, stdout=f"Removed volume '{VOLUME}'\n", stderr="")
+        with patch.object(archives.subprocess, "run", return_value=run):
+            self.assertEqual(archives._pvesh(["delete", f"/nodes/pve1/storage/local/content/{VOLUME}", "--delay", "1"]),
+                             f"Removed volume '{VOLUME}'")
+        self.response = f"Removed volume '{VOLUME}'"
+        self.assertEqual(self.delete()["results"][0]["status"], "deleted")
+        self.response = "Removed volume 'local:backup/another-archive.tar.zst'"
+        self.assertEqual(self.delete()["results"][0]["status"], "unconfirmed")
+        with patch.object(archives.subprocess, "run", return_value=run), self.assertRaises(archives.BackupArchiveError):
+            archives._pvesh(["get", "/nodes/pve1/storage/local/content"])
+
+    def test_bad_payloads_and_paths_never_reach_proxmox(self):
+        payloads = [None, [], {}, {"confirm": 1}, {"confirm": True, "backups": []},
+                    {"confirm": True, "backups": [None]}, {"confirm": True, "backups": [VOLUME]},
+                    {"confirm": True, "backups": [{"storage": "local", "volid": VOLUME}] * 21}]
+        for volid in ("local:../etc/passwd", "local:backup/../x", "local:/etc/pve/storage.cfg", "local:backup//x",
+                      "local:backup/x%2fy", "local:backup/x\n--force", "local:backup/x;touch /tmp/x", "other:backup/x"):
+            payloads.append({"confirm": True, "backups": [{"storage": "local", "volid": volid}]})
+        for storage in ("../local", "-x", "local\n", 2, None):
+            payloads.append({"confirm": True, "backups": [{"storage": storage, "volid": VOLUME}]})
+        with patch.object(archives, "_pvesh") as run:
+            for data in payloads:
+                with self.subTest(data=data), self.assertRaises(archives.BackupArchiveError):
+                    archives.delete_guest_backups(100, data)
+            for vmid in (0, 99, 1000000000):
+                with self.assertRaises(archives.BackupArchiveError):
+                    archives.delete_guest_backups(vmid, {"confirm": True, "backups": [{"storage": "local", "volid": VOLUME}]})
+        run.assert_not_called()
+
+    def test_duplicate_selection_rejected_before_commands(self):
+        with self.assertRaises(archives.BackupArchiveError):
+            self.delete((VOLUME, VOLUME))
+        self.assertEqual(self.calls, [])
+
+    def test_newly_protected_archives_fail_the_whole_batch(self):
+        self.inventory.append({"volid": OTHER, "vmid": 100, "content": "backup", "protected": True})
+        for protection in (True, 1, "1", "true", "unexpected"):
+            self.inventory[-1]["protected"] = protection
+            with self.subTest(protection=protection), self.assertRaises(archives.BackupArchiveError) as error:
+                self.delete((VOLUME, OTHER))
+            self.assertEqual(error.exception.code, 409)
+        self.assertFalse(any(call[0] == "delete" for call in self.calls))
+
+    def test_missing_cross_guest_and_non_backup_volumes_are_never_deleted(self):
+        for entries in ([], [{"volid": VOLUME, "vmid": 101, "content": "backup"}],
+                        [{"volid": VOLUME, "content": "backup"}],
+                        [{"volid": VOLUME, "vmid": 100, "content": "images"}]):
+            self.inventory = entries
+            with self.subTest(entries=entries), self.assertRaises(archives.BackupArchiveError):
+                self.delete()
+        self.assertFalse(any(call[0] == "delete" for call in self.calls))
+
+    def test_storage_failure_prevents_every_delete(self):
+        def unavailable(args):
+            self.assertEqual(args[0], "get")
+            raise archives.BackupArchiveError("offline", 502)
+        with patch.object(archives, "_pvesh", side_effect=unavailable), self.assertRaises(archives.BackupArchiveError):
+            archives.delete_guest_backups(100, {"confirm": True, "backups": [
+                {"volid": VOLUME, "storage": "local"}, {"volid": "nas:backup/x", "storage": "nas"}]})
+
+    def test_partial_native_errors_remain_unconfirmed(self):
+        self.inventory.append({"volid": OTHER, "vmid": 100, "content": "backup"})
+        def native(args):
+            if args[0] == "delete" and OTHER in args[1]:
+                raise archives.BackupArchiveError("backup is protected", 502)
+            return self.run_pvesh(args)
+        with patch.object(archives, "_pvesh", side_effect=native):
+            result = archives.delete_guest_backups(100, {"confirm": True, "backups": [
+                {"volid": v, "storage": "local"} for v in (VOLUME, OTHER)]})
+        self.assertEqual([item["status"] for item in result["results"]], ["deleted", "unconfirmed"])
+        self.assertIn("protected", result["results"][1]["error"])
+
+    def test_native_task_is_not_a_success_until_exit_ok(self):
+        self.response = TASK
+        result = self.delete()["results"][0]
+        self.assertEqual(result["status"], "pending")
+        for native, expected in (({"status": "running"}, "pending"),
+                                 ({"status": "stopped", "exitstatus": "OK"}, "deleted"),
+                                 ({"status": "stopped", "exitstatus": "cannot delete protected backup"}, "unconfirmed"),
+                                 ({"status": "stopped"}, "unconfirmed")):
+            with patch.object(archives, "_pvesh", return_value=native) as run:
+                answer = archives.guest_backup_deletion_status(100, TASK)
+            self.assertEqual(answer["status"], expected)
+            run.assert_called_once_with(["get", f"/nodes/pve1/tasks/{TASK}/status"])
+
+    def test_poll_cannot_read_arbitrary_tasks_or_other_guests(self):
+        self.response = TASK
+        self.delete()
+        with patch.object(archives, "_pvesh") as run:
+            for vmid, task in ((101, TASK), (100, "../../status"), (100, None), (100, TASK + "x")):
+                with self.assertRaises(archives.BackupArchiveError):
+                    archives.guest_backup_deletion_status(vmid, task)
+            archives._tasks[TASK]["created"] -= 3601
+            with self.assertRaises(archives.BackupArchiveError):
+                archives.guest_backup_deletion_status(100, TASK)
+        run.assert_not_called()
+
+    def test_invalid_native_task_answers_never_claim_success(self):
+        for response in ("", {}, [TASK], TASK.replace("imgdel", "vzdump"), TASK.replace("pve1", "pve2")):
+            self.response = response
+            with self.subTest(response=response):
+                self.assertEqual(self.delete()["results"][0]["status"], "unconfirmed")
+
+    def test_pvesh_uses_argv_and_handles_failure_and_timeout(self):
+        with patch.object(archives.subprocess, "run", return_value=types.SimpleNamespace(returncode=0, stdout="null")) as run:
+            self.assertIsNone(archives._pvesh(["delete", "/native/path"]))
+        self.assertEqual(run.call_args.args[0], ["pvesh", "delete", "/native/path", "--output-format", "json"])
+        self.assertNotIn("shell", run.call_args.kwargs)
+        with patch.object(archives.subprocess, "run", return_value=types.SimpleNamespace(returncode=0, stdout="")):
+            self.assertIsNone(archives._pvesh(["delete", "/native/path"]))
+        for response in (types.SimpleNamespace(returncode=1, stderr="protected", stdout=""),
+                         types.SimpleNamespace(returncode=0, stdout="not JSON")):
+            with patch.object(archives.subprocess, "run", return_value=response), self.assertRaises(archives.BackupArchiveError):
+                archives._pvesh(["delete", "/native/path"])
+        for error in (FileNotFoundError(), subprocess.TimeoutExpired("pvesh", 20)):
+            with patch.object(archives.subprocess, "run", side_effect=error), self.assertRaises(archives.BackupArchiveError):
+                archives._pvesh(["delete", "/native/path"])
+
+
+class RouteTests(unittest.TestCase):
+    """Compile the real route bodies/decorators without starting host services."""
+    def setUp(self):
+        auth = types.ModuleType("auth_manager")
+        auth.load_auth_config = lambda: {"enabled": True}
+        auth.verify_token = lambda token: "tester" if token == "admin" else None
+        auth.verify_token_full = lambda token: ("tester", "full_admin" if token == "admin" else "read_only")
+        spec = importlib.util.spec_from_file_location("guest_backup_test_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)
+        self.app = Flask(__name__)
+        self.service = Mock(return_value={"results": []})
+        self.status = Mock(return_value={"volid": VOLUME, "status": "deleted"})
+        self.invalidate = Mock()
+        self.namespace = {"app": self.app, "require_admin_scope": middleware.require_admin_scope,
+                          "request": request, "jsonify": jsonify, "BackupArchiveError": archives.BackupArchiveError,
+                          "delete_guest_backups": self.service, "guest_backup_deletion_status": self.status,
+                          "_vm_cache_invalidate": self.invalidate, "_vm_backups_cache": {}}
+        source = ast.parse((SCRIPTS / "flask_server.py").read_text())
+        wanted = {"api_delete_vm_backups", "api_vm_backup_deletion_status"}
+        routes = [node for node in source.body if isinstance(node, ast.FunctionDef) and node.name in wanted]
+        self.assertEqual(len(routes), 2)
+        exec(compile(ast.Module(body=routes, type_ignores=[]), "flask_server.py", "exec"), self.namespace)
+        self.client = self.app.test_client()
+
+    def test_both_routes_reject_unauthenticated_and_read_only_requests(self):
+        for headers, status in (({}, 401), ({"Authorization": "Bearer reader"}, 403)):
+            self.assertEqual(self.client.post("/api/vms/100/backups/delete", json={}, headers=headers).status_code, status)
+            self.assertEqual(self.client.get("/api/vms/100/backups/delete-status?task=x", headers=headers).status_code, status)
+        self.service.assert_not_called()
+        self.status.assert_not_called()
+        self.invalidate.assert_not_called()
+
+    def test_admin_route_passes_payload_and_invalidates_even_after_errors(self):
+        data = {"confirm": True, "backups": [{"storage": "local", "volid": VOLUME}]}
+        headers = {"Authorization": "Bearer admin"}
+        self.assertEqual(self.client.post("/api/vms/100/backups/delete", json=data, headers=headers).status_code, 200)
+        self.service.assert_called_once_with(100, data)
+        self.invalidate.assert_called_once_with(100, {})
+        self.service.side_effect = archives.BackupArchiveError("protected", 409)
+        self.assertEqual(self.client.post("/api/vms/100/backups/delete", json=data, headers=headers).status_code, 409)
+        self.assertEqual(self.invalidate.call_count, 2)
+
+    def test_poll_invalidates_only_after_completion(self):
+        headers = {"Authorization": "Bearer admin"}
+        self.status.return_value = {"status": "pending"}
+        self.client.get("/api/vms/100/backups/delete-status?task=x", headers=headers)
+        self.invalidate.assert_not_called()
+        self.status.return_value = {"status": "deleted"}
+        self.client.get("/api/vms/100/backups/delete-status?task=x", headers=headers)
+        self.status.assert_called_with(100, "x")
+        self.invalidate.assert_called_once_with(100, {})
+
+    def test_runtime_module_is_shipped(self):
+        self.assertIn('cp "$SCRIPT_DIR/guest_backup_archives.py" "$APP_DIR/usr/bin/"', (SCRIPTS / "build_appimage.sh").read_text())
+
+
+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",