Mirrors/ProxMenux
1
0
Fork 0
mirror of https://github.com/MacRimi/ProxMenux.git synced 2026-09-27 09:06:47 +00:00
Code Issues Packages Projects Releases Wiki Activity

Improve LXC updater selection, error handling and dashboard consistency

Browse Source
This commit is contained in:
MacRimi
2026-09-05 16:10:31 +02:00
parent 4f38d0e2e7
commit 6644b62f63
33 changed files with 5660 additions and 660 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/lxc_updates/test_apply_updates.sh
+164
View File
@@ -0,0 +1,164 @@
#!/bin/bash
set -euo pipefail
REPO_ROOT=$(cd "$(dirname "$0")/../.." && pwd)
RUNNER="$REPO_ROOT/scripts/lxc/apply_updates.sh"
TEST_ROOT=$(mktemp -d "${TMPDIR:-/tmp}/proxmenux-update-test.XXXXXX")
trap 'rm -rf "$TEST_ROOT"' EXIT
MOCK_BIN="$TEST_ROOT/bin"
mkdir -p "$MOCK_BIN" "$TEST_ROOT/locks"
cat >"$MOCK_BIN/pct" <<'EOF'
#!/bin/bash
echo "pct $*" >>"$MOCK_LOG"
case "$1" in
list)
printf 'VMID Status Name\n101 %s test\n' "${MOCK_INITIAL_STATE:-running}"
;;
status)
printf 'status: %s\n' "${MOCK_INITIAL_STATE:-running}"
;;
start|shutdown|reboot)
;;
exec)
shift 3
joined="$*"
case "$joined" in
*'grep -E "^ID="'*) echo 'ID=debian' ;;
'test -f /usr/bin/update') [[ "${MOCK_HAS_WRAPPER:-1}" == "1" ]] ;;
'cat /usr/bin/update')
case "${MOCK_WRAPPER_FORMAT:-legacy}" in
modern)
printf '%s\n' \
'#!/usr/bin/env bash' \
'# Regenerated on install and on every successful update.' \
"export SCRIPT_SLUG=\"${MOCK_HELPER_SLUG:-jellyfin}\"" \
"export UPDATE_SCRIPT_NAME=\"${MOCK_HELPER_SLUG:-jellyfin}\"" \
'bash -c "$(curl -fsSL "${COMMUNITY_SCRIPTS_URL}/ct/${UPDATE_SCRIPT_NAME}.sh")"'
;;
update-name)
printf "export UPDATE_SCRIPT_NAME='%s'\n" "${MOCK_HELPER_SLUG:-jellyfin}"
;;
unsafe)
printf '%s\n' 'export SCRIPT_SLUG="$(touch /tmp/proxmenux-unsafe)"'
;;
*)
echo "bash -c \"\$(curl -fsSL https://raw.githubusercontent.com/community-scripts/ProxmoxVE/main/ct/${MOCK_HELPER_SLUG:-jellyfin}.sh)\""
;;
esac
;;
*'command -v wget'*) ;;
bash\ -c*) echo HELPER_EXEC >>"$MOCK_LOG" ;;
sh\ -c*)
echo CUSTOM_EXEC >>"$MOCK_LOG"
[[ "${MOCK_CUSTOM_FAIL:-0}" != "1" ]]
;;
env\ DEBIAN_FRONTEND=noninteractive*) echo OS_EXEC >>"$MOCK_LOG" ;;
esac
;;
esac
EOF
cat >"$MOCK_BIN/flock" <<'EOF'
#!/bin/bash
[[ "${MOCK_FLOCK_FAIL:-0}" != "1" ]]
EOF
cat >"$MOCK_BIN/python3" <<'EOF'
#!/bin/bash
cat >/dev/null
if [[ "${2:-}" == 'protect-update-command' ]]; then
printf '%s' "$UPDATE_COMMAND"
exit 0
fi
[[ "${MOCK_HELPER_SELECTED:-1}" == "1" ]]
EOF
cat >"$MOCK_BIN/sleep" <<'EOF'
#!/bin/bash
exit 0
EOF
cat >"$MOCK_BIN/vzdump" <<'EOF'
#!/bin/bash
echo "vzdump $*" >>"$MOCK_LOG"
EOF
chmod +x "$MOCK_BIN"/*
fail() { echo "FAIL: $*" >&2; exit 1; }
count() { grep -c "$1" "$MOCK_LOG" 2>/dev/null || true; }
run_case() {
local name=$1
shift
export MOCK_LOG="$TEST_ROOT/$name.log"
: >"$MOCK_LOG"
set +e
env PATH="$MOCK_BIN:$PATH" PROXMENUX_LOCK_DIR="$TEST_ROOT/locks" \
VMID=101 TARGET=app BACKUP=0 RESTART=0 "$@" bash "$RUNNER" \
>"$TEST_ROOT/$name.out" 2>&1
CASE_RC=$?
set -e
}
run_case helper_only RUN_HELPER=1 UPDATE_COMMAND=
[[ $CASE_RC -eq 0 ]] || fail "helper_only returned $CASE_RC"
[[ $(count HELPER_EXEC) -eq 1 ]] || fail "helper_only did not run helper exactly once"
[[ $(count CUSTOM_EXEC) -eq 0 ]] || fail "helper_only unexpectedly ran custom"
run_case helper_not_selected RUN_HELPER=1 UPDATE_COMMAND= MOCK_HELPER_SELECTED=0
[[ $CASE_RC -eq 5 ]] || fail "helper_not_selected expected 5, got $CASE_RC"
[[ $(count HELPER_EXEC) -eq 0 ]] || fail "unselected helper was executed"
run_case modern_helper RUN_HELPER=1 UPDATE_COMMAND= MOCK_WRAPPER_FORMAT=modern MOCK_HELPER_SLUG=nginxproxymanager
[[ $CASE_RC -eq 0 ]] || fail "modern_helper returned $CASE_RC"
[[ $(count HELPER_EXEC) -eq 1 ]] || fail "modern_helper did not run helper exactly once"
grep -qF 'slug: nginxproxymanager' "$TEST_ROOT/modern_helper.out" \
|| fail "modern_helper did not resolve SCRIPT_SLUG"
run_case update_name_helper RUN_HELPER=1 UPDATE_COMMAND= MOCK_WRAPPER_FORMAT=update-name MOCK_HELPER_SLUG=qbittorrent
[[ $CASE_RC -eq 0 ]] || fail "update_name_helper returned $CASE_RC"
[[ $(count HELPER_EXEC) -eq 1 ]] || fail "update_name_helper did not run helper exactly once"
grep -qF 'slug: qbittorrent' "$TEST_ROOT/update_name_helper.out" \
|| fail "update_name_helper did not resolve UPDATE_SCRIPT_NAME"
run_case unsafe_wrapper RUN_HELPER=1 UPDATE_COMMAND= MOCK_WRAPPER_FORMAT=unsafe
[[ $CASE_RC -eq 5 ]] || fail "unsafe_wrapper expected 5, got $CASE_RC"
[[ $(count HELPER_EXEC) -eq 0 ]] || fail "unsafe_wrapper ran helper"
[[ ! -e /tmp/proxmenux-unsafe ]] || fail "unsafe_wrapper evaluated CT content"
run_case custom_override RUN_HELPER=0 UPDATE_COMMAND='update-custom'
[[ $CASE_RC -eq 0 ]] || fail "custom_override returned $CASE_RC"
[[ $(count HELPER_EXEC) -eq 0 ]] || fail "custom_override unexpectedly ran helper"
[[ $(count CUSTOM_EXEC) -eq 1 ]] || fail "custom_override did not run custom exactly once"
run_case custom_replaces_helper RUN_HELPER=1 UPDATE_COMMAND='replace-helper'
[[ $CASE_RC -eq 0 ]] || fail "custom_replaces_helper returned $CASE_RC"
[[ $(count HELPER_EXEC) -eq 0 ]] || fail "custom_replaces_helper unexpectedly ran helper"
[[ $(count CUSTOM_EXEC) -eq 1 ]] || fail "custom_replaces_helper did not run custom exactly once"
grep -qF 'skipping Proxmox VE Helper-Scripts updater' "$TEST_ROOT/custom_replaces_helper.out" \
|| fail "custom_replaces_helper did not report the replacement rule"
run_case explicit_multi_app RUN_HELPER=1 ALLOW_HELPER_WITH_CUSTOM=1 UPDATE_COMMAND='update-another-app'
[[ $CASE_RC -eq 0 ]] || fail "explicit_multi_app returned $CASE_RC"
[[ $(count HELPER_EXEC) -eq 1 ]] || fail "explicit_multi_app did not run helper exactly once"
[[ $(count CUSTOM_EXEC) -eq 1 ]] || fail "explicit_multi_app did not run custom exactly once"
run_case missing_wrapper RUN_HELPER=1 UPDATE_COMMAND= MOCK_HAS_WRAPPER=0
[[ $CASE_RC -eq 5 ]] || fail "missing_wrapper expected 5, got $CASE_RC"
[[ $(count HELPER_EXEC) -eq 0 ]] || fail "missing_wrapper ran helper"
run_case base_os_wrapper RUN_HELPER=1 UPDATE_COMMAND= MOCK_HELPER_SLUG=debian
[[ $CASE_RC -eq 5 ]] || fail "base_os_wrapper expected 5, got $CASE_RC"
[[ $(count HELPER_EXEC) -eq 0 ]] || fail "base_os_wrapper ran helper"
run_case stopped_restore RUN_HELPER=0 UPDATE_COMMAND='update-custom' MOCK_INITIAL_STATE=stopped
[[ $CASE_RC -eq 0 ]] || fail "stopped_restore returned $CASE_RC"
[[ $(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 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"
echo "apply_updates.sh: all deterministic tests passed"
tests/lxc_updates/test_download_update_guard.py
+163
View File
@@ -0,0 +1,163 @@
"""Run real shells and the real LXC runner against local download/guest fixtures."""
from __future__ import annotations
import ast
import os
from pathlib import Path
import shlex
import subprocess
import sys
import tempfile
import threading
import time
import unittest
from unittest.mock import Mock
ROOT = Path(__file__).resolve().parents[2]
sys.path.insert(0, str(ROOT / 'AppImage/scripts'))
import lxc_apps
URL = 'https://raw.githubusercontent.com/community-scripts/ProxmoxVE/main/ct/odoo.sh2'
LEGACY = f'''PHS_SILENT=1 bash -c "$(wget -qLO - '{URL}')"'''
class DownloadGuardTests(unittest.TestCase):
def setUp(self):
folder = tempfile.TemporaryDirectory(prefix='proxmenux-download-test-')
self.addCleanup(folder.cleanup)
self.folder = Path(folder.name)
self.bin = self.folder / 'bin'
self.bin.mkdir()
self.env = {**os.environ, 'PATH': f'{self.bin}:{os.environ["PATH"]}',
'PYTHONPATH': str(ROOT / 'AppImage/scripts'),
'FETCH_STATUS': '8', 'FETCH_BODY': '', 'VMID': '101',
'TARGET': 'app', 'BACKUP': '0', 'RESTART': '0',
'RUN_HELPER': '0', 'UPDATE_COMMAND': '',
'PROXMENUX_LOCK_DIR': str(self.folder)}
for tool in ('wget', 'curl'):
self.write(tool, '#!/bin/sh\nprintf "%s" "$FETCH_BODY"\nexit "$FETCH_STATUS"\n')
self.write('flock', '#!/bin/sh\nexit 0\n')
self.write('pct', '''#!/bin/bash
case "$1" in
list) printf 'VMID Status Name\\n101 running fixture\\n' ;;
status) echo 'status: running' ;;
exec)
shift 3
case "$*" in
*'/etc/os-release'*) echo 'ID=debian' ;;
'test -f /usr/bin/update') exit 0 ;;
'cat /usr/bin/update') echo 'SCRIPT_SLUG="odoo"' ;;
*) exec "$@" ;;
esac ;;
*) exit 90 ;;
esac
''')
# Only consent is stubbed. Preparation imports the real implementation.
self.write('python3', '#!/bin/bash\n'
'if [[ "$2" != "protect-update-command" ]]; then exit 0; fi\n'
# Preload the source under test, not a monitor already
# installed on the build host at the production path.
f'exec {shlex.quote(sys.executable)} -c '
"'import sys, lxc_apps; exec(sys.stdin.read())'\n")
def write(self, name, content):
path = self.bin / name
path.write_text(content)
path.chmod(0o755)
def shell(self, command, status=8, body=''):
return subprocess.run(['/bin/sh', '-c', command], text=True, capture_output=True,
env={**self.env, 'FETCH_STATUS': str(status), 'FETCH_BODY': body}, timeout=10)
def test_reproduces_original_false_success(self):
self.assertEqual(self.shell(LEGACY).returncode, 0)
fixed = self.shell(lxc_apps.protect_download_update_command(LEGACY))
self.assertNotEqual(fixed.returncode, 0)
self.assertIn('download failed', fixed.stderr)
def test_supported_literal_launchers_preserve_url_shell_and_environment(self):
for tool in ('wget -qLO -', 'wget -qO-', 'wget -qO -', 'curl -fsSL', 'curl -fSL'):
for shell in ('bash', 'sh', '/bin/bash', '/bin/sh'):
for prefix in ('', 'PHS_SILENT=0 ', 'PHS_SILENT=1 '):
original = f'''{prefix}{shell} -c "$({tool} '{URL}')"'''
with self.subTest(command=original):
prepared = lxc_apps.protect_download_update_command(original)
self.assertNotEqual(prepared, original)
self.assertEqual(lxc_apps.protect_download_update_command(prepared), prepared)
self.assertIn(URL, prepared)
result = self.shell(prepared, 0, 'echo "RAN:${PHS_SILENT:-unset}"')
self.assertEqual(result.returncode, 0, result.stderr)
self.assertEqual(result.stdout.strip(), 'RAN:' + (prefix.strip()[-1] if prefix else 'unset'))
def test_does_not_reinterpret_unrelated_or_dynamic_commands(self):
for command in ('/opt/odoo/update.sh', 'false; true', 'echo "$(date)"',
'bash -c "$(wget -qLO - \'$UPDATE_URL\')"',
'bash -c "$(curl -fsSL https://example.com/update.sh; echo injected)"',
'bash -c "$(curl -fsSL https://example.com/update.sh?x=1&y=2)"',
'PHS_SILENT=1 bash -c "$(wget -qLO - https://example.com/update.sh)"; true',
'curl -fsSL https://example.com/update.sh | bash'):
self.assertEqual(lxc_apps.protect_download_update_command(command), command)
def test_failed_partial_and_empty_downloads_never_execute(self):
prepared = lxc_apps.protect_download_update_command(LEGACY)
for status, body in ((8, ''), (8, 'echo BAD_EXECUTION'), (0, '')):
result = self.shell(prepared, status, body)
self.assertNotEqual(result.returncode, 0)
self.assertNotIn('BAD_EXECUTION', result.stdout)
result = self.shell(prepared, 0, 'echo REAL_UPDATER; exit 23')
self.assertEqual(result.returncode, 23)
self.assertIn('REAL_UPDATER', result.stdout)
def test_grouping_stops_later_apps_and_preserves_failures(self):
command = lxc_apps.protect_download_update_command(LEGACY) + ' && echo NEXT_APP'
self.assertNotIn('NEXT_APP', self.shell(command).stdout)
self.assertNotIn('NEXT_APP', self.shell(command, 0, 'exit 23').stdout)
self.assertIn('NEXT_APP', self.shell(command, 0, 'exit 0').stdout)
def test_real_runner_reports_failure_for_both_methods(self):
for method in ('helper', 'custom'):
for status, body, expected in ((8, '', 4), (8, 'echo BAD_EXECUTION', 4),
(0, '', 4), (0, 'exit 23', 4), (0, 'exit 0', 0)):
with self.subTest(method=method, status=status, body=body):
result = subprocess.run(['/bin/bash', str(ROOT / 'scripts/lxc/apply_updates.sh')],
capture_output=True, text=True, timeout=15, env={**self.env,
'RUN_HELPER': '1' if method == 'helper' else '0',
'UPDATE_COMMAND': LEGACY if method == 'custom' else '',
'FETCH_STATUS': str(status), 'FETCH_BODY': body})
self.assertEqual(result.returncode, expected, result.stdout + result.stderr)
self.assertEqual('=== Update complete' in result.stdout, expected == 0)
self.assertEqual('=== Update FAILED' in result.stdout, expected != 0)
self.assertNotIn('BAD_EXECUTION', result.stdout)
self.assert_terminal_notification(result.returncode)
def assert_terminal_notification(self, exit_code):
# Actual completion hook + finalizer; only IO/metadata are stubbed.
names = {'_terminal_lxc_update_completed', '_finalize_lxc_update'}
nodes = [n for n in ast.parse((ROOT / 'AppImage/scripts/flask_server.py').read_text()).body
if isinstance(n, ast.FunctionDef) and n.name in names]
notification = Mock()
ns = dict(os=os, time=time, re=__import__('re'), notification_manager=notification,
_LXC_APPLY_UPDATES_SCRIPT=str(ROOT / 'scripts/lxc/apply_updates.sh'),
_normalise_lxc_update_run_id=lambda value, **_: value,
_normalise_lxc_update_targets=lambda values, _: values,
_normalise_lxc_update_labels=lambda values: values,
_json_list=lambda _: [], _lxc_update_finalizations={},
_lxc_update_finalization_lock=threading.Lock(), _LXC_UPDATE_FINALIZATION_TTL=3600,
_fast_guest_status=lambda *_: 'stopped',
_lxc_update_snapshot=lambda *_: {'ct_name': 'fixture'},
_lxc_update_target_labels=lambda *_: ['Odoo'],
_lxc_update_details=lambda **kwargs: kwargs['status'],
get_proxmox_node_name=lambda: 'fixture-node')
exec(compile(ast.Module(body=nodes, type_ignores=[]), 'completion', 'exec'), ns)
params = {'RUN_ID': 'fixture-run', 'VMID': '101', 'TARGET': 'app'}
for _ in range(2):
ns['_terminal_lxc_update_completed'](script_path=ns['_LXC_APPLY_UPDATES_SCRIPT'],
params=params, exit_code=exit_code, duration_seconds=1)
notification.emit_event.assert_called_once()
event = notification.emit_event.call_args.kwargs
self.assertEqual(event['data']['result'], 'succeeded' if exit_code == 0 else 'failed')
self.assertEqual(event['severity'], 'INFO' if exit_code == 0 else 'WARNING')
if __name__ == '__main__':
unittest.main()
tests/lxc_updates/test_update_contract.py
+565
View File
@@ -0,0 +1,565 @@
import sys
import subprocess
import tempfile
import unittest
from types import SimpleNamespace
from pathlib import Path
from unittest.mock import MagicMock, patch
REPO_ROOT = Path(__file__).resolve().parents[2]
sys.path.insert(0, str(REPO_ROOT / "AppImage" / "scripts"))
import lxc_apps # noqa: E402
import managed_installs # noqa: E402
class UpdateStrategyValidationTests(unittest.TestCase):
def test_custom_command_defaults_to_override(self):
ok, config = lxc_apps.validate_config({
"name": "Jellyfin",
"update_command": "systemctl restart jellyfin",
})
self.assertTrue(ok, config)
self.assertEqual(config["update_strategy"], "custom_override")
def test_legacy_helper_then_custom_is_normalised_to_override(self):
ok, config = lxc_apps.validate_config({
"name": "Jellyfin",
"update_command": "systemctl restart jellyfin",
"update_strategy": "helper_then_custom",
})
self.assertTrue(ok, config)
self.assertEqual(config["update_strategy"], "custom_override")
def test_unknown_strategy_is_ignored_and_normalised(self):
ok, config = lxc_apps.validate_config({
"name": "Jellyfin",
"update_command": "true",
"update_strategy": "run-everything",
})
self.assertTrue(ok, config)
self.assertEqual(config["update_strategy"], "custom_override")
class HelperEvidenceTests(unittest.TestCase):
def test_legacy_literal_wrapper_slug_is_supported(self):
wrapper = (
'bash -c "$(curl -fsSL '
'https://raw.githubusercontent.com/community-scripts/ProxmoxVE/'
'main/ct/jellyfin.sh)"'
)
self.assertEqual(
managed_installs._extract_helper_slug_from_update_wrapper(wrapper),
"jellyfin",
)
def test_current_generated_wrapper_slug_is_supported(self):
wrapper = """#!/usr/bin/env bash
# Community-Scripts update entrypoint (generated - do not edit by hand).
# Regenerated on install and on every successful update.
export SCRIPT_SLUG="nginxproxymanager"
export UPDATE_SCRIPT_NAME="nginxproxymanager"
export COMMUNITY_SCRIPTS_URL="https://raw.githubusercontent.com/community-scripts/ProxmoxVE/main"
bash -c "$(curl -fsSL "${COMMUNITY_SCRIPTS_URL}/ct/${UPDATE_SCRIPT_NAME}.sh")"
"""
self.assertEqual(
managed_installs._extract_helper_slug_from_update_wrapper(wrapper),
"nginxproxymanager",
)
def test_update_script_name_is_a_compatible_fallback(self):
wrapper = "export UPDATE_SCRIPT_NAME='qbittorrent'"
self.assertEqual(
managed_installs._extract_helper_slug_from_update_wrapper(wrapper),
"qbittorrent",
)
def test_dynamic_or_command_chained_assignments_are_rejected(self):
for wrapper in (
'export SCRIPT_SLUG="$(touch /tmp/unsafe)"',
'export SCRIPT_SLUG="jellyfin"; touch /tmp/unsafe',
'export UPDATE_SCRIPT_NAME="${UNTRUSTED}"',
):
with self.subTest(wrapper=wrapper):
self.assertIsNone(
managed_installs._extract_helper_slug_from_update_wrapper(wrapper)
)
def test_wrapper_is_executable_evidence(self):
with patch.object(managed_installs, "_probe_helper_scripts_slug", return_value="jellyfin"):
slug, source = managed_installs._identify_helper_slug("101", "media")
self.assertEqual((slug, source), ("jellyfin", "update_wrapper"))
def test_tag_hostname_is_suggestion_only(self):
with patch.object(managed_installs, "_probe_helper_scripts_slug", return_value=None), \
patch.object(managed_installs, "_probe_lxc_tags", return_value={"community-scripts"}), \
patch.object(managed_installs, "_guess_helper_slug_from_hostname", return_value="jellyfin"):
slug, source = managed_installs._identify_helper_slug("101", "jellyfin")
self.assertEqual((slug, source), ("jellyfin", "tag_hostname"))
def _detect_one(self, source, slug="jellyfin"):
patches = (
patch.object(managed_installs, "_lxc_updates_detection_enabled", return_value=True),
patch.object(managed_installs, "_read_registry", return_value={"items": []}),
patch.object(managed_installs, "_list_pve_lxcs", return_value=[{
"vmid": "101", "status": "running", "name": "jellyfin",
}]),
patch.object(managed_installs, "_get_oci_managed_vmids", return_value={}),
patch.object(managed_installs, "_probe_lxc_is_oci", return_value=False),
patch.object(managed_installs, "_probe_lxc_os", return_value="debian"),
patch.object(managed_installs, "_identify_helper_slug", return_value=(slug, source)),
patch.object(managed_installs, "_fetch_helpers_cache", return_value={
slug: {"name": "Jellyfin", "updateable": True},
}),
)
for ctx in patches:
ctx.start()
try:
return managed_installs._detect_lxc_containers()[0]
finally:
for ctx in reversed(patches):
ctx.stop()
def test_hostname_guess_never_enables_updater(self):
item = self._detect_one("tag_hostname")
self.assertFalse(item["_has_app_updater"])
self.assertEqual(item["_helper_slug_source"], "tag_hostname")
def test_valid_wrapper_enables_updateable_app(self):
item = self._detect_one("update_wrapper")
self.assertTrue(item["_has_app_updater"])
def test_base_os_wrapper_never_enables_app_updater(self):
item = self._detect_one("update_wrapper", slug="debian")
self.assertFalse(item["_has_app_updater"])
class BulkUpdateConfigTests(unittest.TestCase):
def test_os_is_mandatory_and_a_second_target_is_required(self):
ok, error = lxc_apps.validate_bulk_update({"targets": ["app:abc"]})
self.assertFalse(ok)
self.assertIn("OS", error)
ok, error = lxc_apps.validate_bulk_update({"targets": ["os"]})
self.assertFalse(ok)
self.assertIn("application", error)
def test_targets_are_deduplicated_and_normalised(self):
ok, config = lxc_apps.validate_bulk_update({
"targets": ["docker-engine", "os", "app:abc", "docker-engine"],
})
self.assertTrue(ok, config)
self.assertEqual(config["targets"], ["os", "app:abc", "docker-engine"])
def test_only_opaque_docker_units_are_allowed(self):
ok, _ = lxc_apps.validate_bulk_update({
"targets": ["os", "docker-compose:media"],
})
self.assertFalse(ok)
ok, config = lxc_apps.validate_bulk_update({
"targets": ["os", "docker-unit:0123456789abcdefabcd"],
})
self.assertTrue(ok, config)
def test_bulk_config_round_trips_separately_from_schedule(self):
with tempfile.TemporaryDirectory() as temp_dir, \
patch.object(lxc_apps, "_APPS_DIR", temp_dir):
ok, _ = lxc_apps.update_schedule(101, {
"enabled": False,
"cron": "",
"target": "os",
"targets": ["os"],
})
self.assertTrue(ok)
ok, _ = lxc_apps.update_bulk_update(101, {
"targets": ["os", "docker-engine"],
})
self.assertTrue(ok)
self.assertEqual(lxc_apps.get_bulk_update(101)["targets"], ["os", "docker-engine"])
self.assertEqual(lxc_apps.get_schedule(101)["targets"], ["os"])
self.assertTrue(lxc_apps.delete_bulk_update(101))
self.assertIsNone(lxc_apps.get_bulk_update(101))
self.assertIsNotNone(lxc_apps.get_schedule(101))
class ScheduledReleaseTargetTests(unittest.TestCase):
def test_untracked_custom_app_is_not_release_gated(self):
gated, remaining = lxc_apps.partition_scheduled_release_targets(
["app:links"],
[{
"id": "links",
"name": "Links only",
"update_command": "systemctl restart links",
}],
)
self.assertEqual(gated, set())
self.assertEqual(remaining, ["app:links"])
def test_deferred_tracked_app_does_not_block_untracked_or_os(self):
gated, remaining = lxc_apps.partition_scheduled_release_targets(
["os", "app:tracked", "app:links"],
[
{"id": "tracked", "installed_via": "binary"},
{"id": "links", "update_command": "true"},
],
)
self.assertEqual(gated, {"tracked"})
self.assertEqual(remaining, ["os", "app:links"])
def test_legacy_apps_target_keeps_only_untracked_when_gate_defers(self):
gated, remaining = lxc_apps.partition_scheduled_release_targets(
["os", "apps"],
[
{"id": "tracked", "installed_via": "file"},
{"id": "links", "update_command": "true"},
{"id": "docker", "installed_via": "binary", "helper_slug": "docker"},
],
)
self.assertEqual(gated, {"tracked"})
self.assertEqual(remaining, ["os", "app:links"])
class AppCacheWriteThroughContractTests(unittest.TestCase):
def test_successful_mutations_publish_the_complete_sidecar(self):
cache_source = (REPO_ROOT / "AppImage" / "lib" / "lxc-apps-cache.ts").read_text()
panel_source = (REPO_ROOT / "AppImage" / "components" / "lxc-app-panel.tsx").read_text()
server_source = (REPO_ROOT / "AppImage" / "scripts" / "flask_server.py").read_text()
self.assertIn("export function setLxcAppsCached", cache_source)
self.assertGreaterEqual(panel_source.count("setLxcAppsCached(vmid, r, suggestions)"), 6)
self.assertNotIn("_vm_cache_put(_vm_apps_cache", server_source)
self.assertIn("lxc_apps.load_sidecar(vmid)", server_source)
def test_post_apply_revalidates_without_evicting_render_seed(self):
source = (REPO_ROOT / "AppImage" / "components" / "virtual-machines.tsx").read_text()
apply_block = source[source.index("const handleApplyComplete"):source.index("const getAggregateUpdateCheck")]
self.assertIn("void fetchLxcApps(applyVmid)", apply_block)
self.assertNotIn("invalidateLxcApps(applyVmid)", apply_block)
def test_apply_completion_is_owned_by_the_backend_and_idempotent(self):
server_source = (REPO_ROOT / "AppImage" / "scripts" / "flask_server.py").read_text()
terminal_source = (REPO_ROOT / "AppImage" / "scripts" / "flask_terminal_routes.py").read_text()
endpoint_block = server_source[
server_source.index("def api_lxc_updates_applied"):
server_source.index("@app.route('/api/health/thresholds'")
]
finalizer_block = server_source[
server_source.index("def _finalize_lxc_update"):
server_source.index("def _terminal_lxc_update_completed")
]
self.assertIn("set_script_completion_hook(_terminal_lxc_update_completed)", server_source)
self.assertIn("params.get('RUN_ID')", server_source)
self.assertIn("_run_script_completion_hook", terminal_source)
self.assertIn("_lxc_update_finalizations", finalizer_block)
self.assertIn("managed_installs.refresh_lxc(vmid)", finalizer_block)
self.assertNotIn("managed_installs.check_for_updates(force=True)", endpoint_block)
self.assertIn("entity_id=f'{vmid}:{safe_run_id}'", finalizer_block)
def test_scheduled_updates_use_the_shared_finalizer(self):
source = (REPO_ROOT / "AppImage" / "scripts" / "flask_server.py").read_text()
scheduler_block = source[
source.index("def _run_scheduled_update"):
source.index("def _scheduler_loop")
]
self.assertIn("_finalize_lxc_update(", scheduler_block)
self.assertIn("return finish('partial'", scheduler_block)
self.assertIn("before_snapshot=before", scheduler_block)
class DockerStackNotificationTests(unittest.TestCase):
def _docker_app(self):
return {
"id": "docker",
"name": "Docker",
"helper_slug": "docker",
"notifications_enabled": True,
"state": {
"installed_version": "27.4.0",
"latest_version": "29.7.2",
"update_available": True,
},
}
def _inventory(self, reference="portainer/portainer-ce:latest", digest="sha256:new"):
return {
"available": True,
"images": [{
"reference": reference,
"installed_version": "2.19.4",
"available_version": "2.39.6",
"remote_digest": digest,
"update_available": True,
}],
}
def test_engine_and_images_share_one_payload(self):
payload = lxc_apps._docker_stack_notification_payload(
101, self._docker_app(), self._inventory(), "docker",
)
self.assertEqual(payload["count"], 2)
self.assertIn("Docker Engine: 27.4.0 → 29.7.2", payload["details"])
self.assertIn("portainer/portainer-ce:latest: 2.19.4 → 2.39.6", payload["details"])
def test_signature_changes_when_pending_identity_changes_at_same_count(self):
first = lxc_apps._docker_stack_notification_payload(
101, {**self._docker_app(), "state": {}}, self._inventory(), "docker",
)
second = lxc_apps._docker_stack_notification_payload(
101,
{**self._docker_app(), "state": {}},
self._inventory("library/nginx:latest", "sha256:other"),
"docker",
)
self.assertEqual(first["count"], second["count"])
self.assertNotEqual(first["signature"], second["signature"])
def test_generic_app_event_does_not_duplicate_docker_stack_event(self):
emitter = MagicMock()
fake_module = SimpleNamespace(notification_manager=emitter)
with patch.dict(sys.modules, {"notification_manager": fake_module}):
lxc_apps._fire_update_notification(101, self._docker_app())
emitter.emit_event.assert_not_called()
class DockerInventoryCachePolicyTests(unittest.TestCase):
def setUp(self):
self.original_cache = lxc_apps._docker_inventory_cache
lxc_apps._docker_inventory_cache = {}
def tearDown(self):
lxc_apps._docker_inventory_cache = self.original_cache
def _inventory(self, available=True):
return {
"vmid": 101,
"available": available,
"images": [],
"checked_at_unix": 1_000_000,
}
def test_available_inventory_uses_24_hour_cache(self):
self.assertEqual(lxc_apps._DOCKER_INVENTORY_TTL_SEC, 24 * 3600)
lxc_apps._docker_inventory_cache["101"] = self._inventory()
fresh_scan = self._inventory()
fresh_scan["engine_version"] = "new-scan"
with patch.object(lxc_apps.time, "time", return_value=1_000_000 + 3600), \
patch.object(lxc_apps, "_docker_inventory_from_ct", return_value=fresh_scan) as scan:
cached = lxc_apps.get_docker_inventory(101)
self.assertNotIn("engine_version", cached)
scan.assert_not_called()
def test_available_inventory_refreshes_after_24_hours(self):
lxc_apps._docker_inventory_cache["101"] = self._inventory()
fresh_scan = self._inventory()
fresh_scan["engine_version"] = "new-scan"
with patch.object(lxc_apps.time, "time", return_value=1_000_000 + 24 * 3600 + 1), \
patch.object(lxc_apps, "_docker_inventory_from_ct", return_value=fresh_scan) as scan:
refreshed = lxc_apps.get_docker_inventory(101)
self.assertEqual(refreshed["engine_version"], "new-scan")
scan.assert_called_once_with(101)
def test_unavailable_inventory_retries_after_30_seconds(self):
lxc_apps._docker_inventory_cache["101"] = self._inventory(available=False)
fresh_scan = self._inventory(available=True)
with patch.object(lxc_apps.time, "time", return_value=1_000_029), \
patch.object(lxc_apps, "_docker_inventory_from_ct", return_value=fresh_scan) as scan:
lxc_apps.get_docker_inventory(101)
scan.assert_not_called()
with patch.object(lxc_apps.time, "time", return_value=1_000_031), \
patch.object(lxc_apps, "_docker_inventory_from_ct", return_value=fresh_scan) as scan:
lxc_apps.get_docker_inventory(101)
scan.assert_called_once_with(101)
def test_force_failure_preserves_docker_unit_identity_as_refreshing(self):
previous = self._inventory(available=True)
previous.update({
"images": [{"reference": "demo/app:latest", "update_available": True}],
"update_units": [{
"id": "docker-unit:0123456789abcdefabcd",
"display_name": "Demo",
"update_available": True,
}],
})
lxc_apps._docker_inventory_cache["101"] = previous
failed_scan = {
"vmid": 101,
"available": False,
"images": [],
"update_count": 0,
"error": "Docker is not ready",
}
with patch.object(lxc_apps.time, "time", return_value=1_000_100), \
patch.object(lxc_apps, "_docker_inventory_from_ct", return_value=failed_scan):
result = lxc_apps.get_docker_inventory(101, force=True)
self.assertFalse(result["available"])
self.assertTrue(result["refreshing"])
self.assertEqual(result["update_units"][0]["display_name"], "Demo")
self.assertEqual(result["update_units"][0]["id"], "docker-unit:0123456789abcdefabcd")
def test_force_failure_preserves_empty_lifecycle_pending_state(self):
pending = self._inventory(available=False)
pending.update({
"refreshing": True,
"images": [],
"update_units": [],
"error": None,
})
lxc_apps._docker_inventory_cache["101"] = pending
failed_scan = {
"vmid": 101,
"available": False,
"images": [],
"update_units": [],
"update_count": 0,
"error": "Docker is not ready",
}
with patch.object(lxc_apps.time, "time", return_value=1_000_100), \
patch.object(lxc_apps, "_docker_inventory_from_ct", return_value=failed_scan):
result = lxc_apps.get_docker_inventory(101, force=True)
self.assertFalse(result["available"])
self.assertTrue(result["refreshing"])
self.assertEqual(result["images"], [])
self.assertEqual(result["update_units"], [])
def test_lifecycle_transition_keeps_ids_but_clears_old_update_state(self):
previous = self._inventory(available=True)
previous.update({
"images": [{"reference": "demo/app:latest", "update_available": True}],
"update_units": [{
"id": "docker-unit:0123456789abcdefabcd",
"display_name": "Demo",
"update_available": True,
}],
})
lxc_apps._docker_inventory_cache["101"] = previous
pending = lxc_apps.mark_docker_inventory_refreshing(101)
self.assertFalse(pending["available"])
self.assertTrue(pending["refreshing"])
self.assertIsNone(pending["images"][0]["update_available"])
self.assertIsNone(pending["update_units"][0]["update_available"])
self.assertTrue(previous["images"][0]["update_available"])
def test_docker_ps_timeout_is_not_reported_as_an_empty_inventory(self):
with patch.object(lxc_apps, "_pct_exec", side_effect=[
(0, "27.4.0\n", ""),
(124, "", "timed out after 10s"),
]):
result = lxc_apps._docker_inventory_from_ct(101)
self.assertFalse(result["available"])
self.assertEqual(result["images"], [])
self.assertIn("not ready", result["error"])
def test_docker_image_ls_timeout_is_not_reported_as_empty(self):
with patch.object(lxc_apps, "_pct_exec", side_effect=[
(0, "27.4.0\n", ""),
(0, "", ""),
(124, "", "timed out after 15s"),
]):
result = lxc_apps._docker_inventory_from_ct(101)
self.assertFalse(result["available"])
self.assertEqual(result["images"], [])
self.assertIn("timed out", result["error"])
def test_pct_exec_timeout_kills_the_complete_local_process_group(self):
process = MagicMock()
process.pid = 4321
process.communicate.side_effect = [
subprocess.TimeoutExpired(cmd="pct", timeout=1),
("", ""),
]
with patch.object(lxc_apps.subprocess, "Popen", return_value=process), \
patch.object(lxc_apps.os, "killpg") as killpg:
rc, out, err = lxc_apps._pct_exec(101, ["docker", "version"], timeout=1)
self.assertEqual((rc, out), (124, ""))
self.assertIn("timed out", err)
killpg.assert_called_once_with(4321, lxc_apps.signal.SIGKILL)
def test_inventory_does_not_define_disk_persistence(self):
source = (REPO_ROOT / "AppImage" / "scripts" / "lxc_apps.py").read_text()
self.assertNotIn("docker_inventory.json", source)
self.assertNotIn("_save_docker_inventory_disk", source)
def test_daily_collector_and_ui_force_points_are_explicit(self):
notification_source = (REPO_ROOT / "AppImage" / "scripts" / "notification_events.py").read_text()
frontend_source = (REPO_ROOT / "AppImage" / "components" / "virtual-machines.tsx").read_text()
automatic_block = frontend_source[
frontend_source.index("// Docker drift is opt-in"):
frontend_source.index("const refreshDockerInventory")
]
manual_block = frontend_source[
frontend_source.index("const refreshDockerInventory"):
frontend_source.index("const closeCustomCmdEditor")
]
self.assertIn("refresh_docker_inventories(force=True)", notification_source)
self.assertIn("/docker/inventory`", automatic_block)
self.assertNotIn("?force=1", automatic_block)
self.assertIn("/docker/inventory?force=1", manual_block)
def test_startup_and_lifecycle_rebuild_memory_inventory_without_blank_gap(self):
server_source = (REPO_ROOT / "AppImage" / "scripts" / "flask_server.py").read_text()
lifecycle_block = server_source[
server_source.index("def _refresh_started_guest"):
server_source.index("def _schedule_started_guest_refresh")
]
startup_block = server_source[
server_source.index("def _deferred_startup_inits"):
server_source.index("threading.Thread(target=_deferred_startup_inits")
]
self.assertIn("refresh_docker_inventories(force=True)", startup_block)
self.assertIn("mark_docker_inventory_refreshing(vmid)", lifecycle_block)
self.assertIn("get_docker_inventory(vmid, force=True)", lifecycle_block)
self.assertIn("docker_refresh_pending", lifecycle_block)
self.assertIn("time.monotonic() + 7 * 60", lifecycle_block)
self.assertIn("Docker ready in", lifecycle_block)
self.assertLess(
lifecycle_block.index("get_docker_inventory(vmid, force=True)"),
lifecycle_block.index("get_suggestions(vmid, force=True)"),
)
self.assertLess(
lifecycle_block.index("_publish_guest_modal_cache_revision(vmid)"),
lifecycle_block.index("get_suggestions(vmid, force=True)"),
)
def test_manual_docker_refresh_publishes_endpoint_result_without_full_vm_wait(self):
frontend_source = (REPO_ROOT / "AppImage" / "components" / "virtual-machines.tsx").read_text()
manual_block = frontend_source[
frontend_source.index("const refreshDockerInventory"):
frontend_source.index("const closeCustomCmdEditor")
]
self.assertIn("const inventory = await fetchApi<LxcDockerInventory>", manual_block)
self.assertIn("docker_inventory: inventory", manual_block)
self.assertIn("{ revalidate: false }", manual_block)
self.assertNotIn("await mutate()", manual_block)
def test_bulk_ui_does_not_expose_internal_docker_ids_while_refreshing(self):
frontend_source = (REPO_ROOT / "AppImage" / "components" / "virtual-machines.tsx").read_text()
bulk_block = frontend_source[
frontend_source.index("const pendingDockerBulkTargets"):
frontend_source.index("{/* Options card")
]
self.assertIn("vmLxc.bulkUpdate.dockerInventoryPending", bulk_block)
self.assertIn("vmLxc.bulkUpdate.missingDockerTarget", bulk_block)
self.assertIn("pendingDockerBulkTargets.length > 0", bulk_block)
self.assertNotIn("{target}\n", bulk_block)
if __name__ == "__main__":
unittest.main()
tests/lxc_updates/test_update_method_choice.py
+181
View File
@@ -0,0 +1,181 @@
"""Updater consent regressions: temporary sidecars, no guests or network."""
import ast
import copy
from datetime import datetime
import json
import os
from pathlib import Path
import re
import subprocess
import sys
import tempfile
import time
import types
import unittest
import uuid
from unittest.mock import Mock, patch
ROOT = Path(__file__).resolve().parents[2]
sys.path.insert(0, str(ROOT / 'AppImage/scripts'))
import lxc_apps as apps
def routes():
names = {'_scheduled_helper_enabled', '_resolve_bulk_update_plan',
'_compose_scheduled_update_command', '_run_scheduled_update',
'_normalise_schedule_targets'}
nodes = [node for node in ast.parse((ROOT / 'AppImage/scripts/flask_server.py').read_text()).body
if isinstance(node, ast.FunctionDef) and node.name in names]
ns = dict(json=json, os=os, re=re, time=time, uuid=uuid, datetime=datetime,
subprocess=types.SimpleNamespace(run=Mock(return_value=types.SimpleNamespace(returncode=0)),
TimeoutExpired=subprocess.TimeoutExpired),
_DOCKER_ENGINE_INTEGRATED_COMMAND='integrated-docker',
_APPLY_UPDATES_SCRIPT=str(ROOT / 'scripts/lxc/apply_updates.sh'),
_create_lxc_update_log=lambda *_: ('test.log', None),
_fast_guest_status=lambda *_: 'running',
_lxc_update_snapshot=lambda *_: {'ct_name': 'test'},
_lxc_update_target_labels=lambda *_: [],
_append_lxc_update_log=lambda *_: None,
_prune_lxc_update_logs=lambda *_: None,
_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)
return ns
class UpdateChoiceTests(unittest.TestCase):
def setUp(self):
directory = tempfile.TemporaryDirectory()
self.addCleanup(directory.cleanup)
self.addCleanup(patch.stopall)
patch.object(apps, '_APPS_DIR', directory.name).start()
patch.object(apps, 'check_app', return_value=None).start()
self.item = {'type': 'lxc', '_vmid': 101, '_has_app_updater': True,
'_helper_slug_source': 'update_wrapper', '_helper_slug': 'qbittorrent'}
patch.dict(sys.modules, {'lxc_apps': apps, 'managed_installs': types.SimpleNamespace(
get_active_items=lambda: [self.item])}).start()
self.api = routes()
def save(self, records, **extra):
data = {'vmid': 101, 'apps': copy.deepcopy(records), **extra}
self.assertTrue(apps._write_sidecar(101, data))
return apps._read_sidecar(101)
def helper(self, method='helper', **extra):
return {'id': 'qbit', 'name': 'qBittorrent', 'helper_slug': 'qbittorrent',
'update_method': method, **extra}
def test_registration_does_not_enable_helper_even_in_legacy_wildcard_schedule(self):
self.save([], schedule={'enabled': True, 'target': 'both'})
ok, sidecar = apps.add_app(101, {'name': 'qBittorrent', 'helper_slug': 'qbittorrent'})
self.assertTrue(ok, sidecar)
self.assertEqual(sidecar['apps'][0]['update_method'], 'none')
self.assertFalse(self.api['_scheduled_helper_enabled'](101, 'app', ['apps']))
def test_legacy_commands_and_explicit_selections_survive_migration(self):
helper = self.helper()
helper.pop('update_method')
result = self.save([helper, {'id': 'manual', 'update_command': 'my-updater'}],
bulk_update={'targets': ['os', 'app:qbit']})
self.assertEqual([a['update_method'] for a in result['apps']], ['helper', 'custom'])
result = self.save([helper], schedule={'enabled': True, 'target': 'both'})
self.assertEqual(result['apps'][0]['update_method'], 'helper')
result = self.save([helper], schedule={'enabled': False, 'target': 'both'})
self.assertEqual(result['apps'][0]['update_method'], 'none')
self.assertEqual(self.save([helper])['apps'][0]['update_method'], 'none')
def test_choice_survives_editing_ports_and_disable_never_falls_back(self):
self.save([self.helper()], schedule={'enabled': True, 'targets': ['apps']})
record = apps._read_sidecar(101)['apps'][0]
ok, result = apps.update_app(101, 'qbit', {**record, 'ports': [{'port': 8090}]})
self.assertTrue(ok, result)
self.assertEqual(result['apps'][0]['update_method'], 'helper')
ok, result = apps.update_app(101, 'qbit', {**record, 'update_method': 'custom', 'update_command': 'my-updater'})
self.assertTrue(ok, result)
self.assertFalse(self.api['_scheduled_helper_enabled'](101, 'app', ['apps']))
ok, result = apps.update_app(101, 'qbit', {**record, 'update_method': 'none', 'update_command': ''})
self.assertTrue(ok, result)
self.assertFalse(apps.helper_update_selected(101, 'qbittorrent'))
self.assertTrue(result['schedule']['enabled'])
def test_conflicting_or_empty_methods_are_rejected(self):
for payload in ({'update_method': 'custom'}, {'update_method': 'helper'},
{'update_method': 'helper', 'helper_slug': 'qbittorrent', 'update_command': 'true'},
{'update_method': 'none', 'update_command': 'true'}, {'update_method': 'automatic'}):
self.assertFalse(apps.validate_config({'name': 'test', **payload})[0], payload)
def test_multi_app_bulk_keeps_methods_separate(self):
self.save([self.helper(), {'id': 'other', 'name': 'Other', 'update_method': 'custom',
'update_command': '/opt/other/update.sh'},
{'id': 'unconfigured', 'helper_slug': 'jellyfin', 'update_method': 'none'}])
plan = self.api['_resolve_bulk_update_plan'](101, ['os', 'app:qbit', 'app:other'])
self.assertTrue(plan['ok'], plan)
self.assertTrue(plan['run_helper'])
self.assertTrue(plan['allow_helper_with_custom'])
self.assertEqual(plan['update_command'], '/opt/other/update.sh')
plan = self.api['_resolve_bulk_update_plan'](101, ['os', 'app:unconfigured'])
self.assertFalse(plan['ok'])
self.assertFalse(apps.helper_update_selected(101, 'qbittorrent', ['app:other']))
def test_helper_for_one_app_never_authorizes_a_different_helper(self):
self.save([self.helper(), {'id': 'wrong', 'helper_slug': 'jellyfin', 'update_method': 'helper'}])
plan = self.api['_resolve_bulk_update_plan'](101, ['os', 'app:qbit', 'app:wrong'])
self.assertFalse(plan['ok'])
self.assertEqual(plan['unavailable'][0]['target'], 'app:wrong')
def test_legacy_download_guard_covers_bulk_and_schedule_without_rewriting_settings(self):
command = 'PHS_SILENT=1 bash -c "$(wget -qLO - \'https://example.com/odoo.sh2\')"'
self.save([{'id': 'odoo', 'name': 'Odoo', 'update_method': 'custom', 'update_command': command},
{'id': 'other', 'name': 'Other', 'update_method': 'custom', 'update_command': 'echo OTHER'}])
saved_before = Path(apps._sidecar_path(101)).read_bytes()
plan = self.api['_resolve_bulk_update_plan'](101, ['os', 'app:odoo', 'app:other'])
self.assertTrue(plan['ok'])
self.assertIn('updater download failed', plan['update_command'])
self.assertTrue(plan['update_command'].endswith(') && echo OTHER'))
self.api['subprocess'].run.return_value.returncode = 4
result = self.api['_run_scheduled_update'](101, {'targets': ['app:odoo', 'app:other']})
self.assertEqual(result['status'], 'failure')
actual = self.api['subprocess'].run.call_args.kwargs['env']['UPDATE_COMMAND']
self.assertEqual(actual, plan['update_command'])
self.assertEqual(self.api['_finalize_lxc_update'].call_args.kwargs['status'], 'failure')
self.assertEqual(Path(apps._sidecar_path(101)).read_bytes(), saved_before)
self.assertEqual(apps._read_sidecar(101)['apps'][0]['update_command'], command)
def test_detection_and_duplicate_conflicts_do_not_authorize_execution(self):
self.save([self.helper('none')])
self.assertFalse(apps.helper_update_selected(101, 'qbittorrent'))
self.save([self.helper(), {**self.helper('custom'), 'id': 'duplicate', 'update_command': 'my-update'}])
self.assertFalse(apps.helper_update_selected(101, 'qbittorrent'))
self.save([self.helper()])
self.item['_helper_slug_source'] = 'tag_hostname'
self.assertFalse(self.api['_scheduled_helper_enabled'](101, 'app', ['apps']))
def test_disabled_app_schedule_reports_skipped_without_running(self):
self.save([self.helper('none')])
result = self.api['_run_scheduled_update'](101, {'targets': ['app:qbit']})
self.assertEqual(result['status'], 'skipped')
self.assertEqual(result['executed_targets'], [])
self.api['subprocess'].run.assert_not_called()
def test_disabled_app_does_not_block_os_or_custom_app_and_reports_partial(self):
self.save([self.helper('none'), {'id': 'other', 'update_command': 'my-updater'}])
result = self.api['_run_scheduled_update'](101, {'targets': ['os', 'app:qbit', 'app:other']})
self.assertEqual(result['status'], 'partial')
self.assertEqual(result['executed_targets'], ['os', 'app:other'])
env = self.api['subprocess'].run.call_args.kwargs['env']
self.assertEqual(env['RUN_HELPER'], '0')
self.assertEqual(env['UPDATE_COMMAND'], 'my-updater')
def test_wildcard_only_runs_selected_methods(self):
self.save([self.helper(), {'id': 'other', 'update_command': 'my-updater'},
{'id': 'links', 'name': 'Links only', 'update_method': 'none'}])
result = self.api['_run_scheduled_update'](101, {'targets': ['apps']})
self.assertEqual(result['status'], 'success')
self.assertEqual(result['executed_targets'], ['app:qbit', 'app:other'])
env = self.api['subprocess'].run.call_args.kwargs['env']
self.assertEqual(env['RUN_HELPER'], '1')
self.assertEqual(env['UPDATE_COMMAND'], 'my-updater')
if __name__ == '__main__':
unittest.main()
tests/lxc_updates/test_updater_editor.cjs
+280
View File
@@ -0,0 +1,280 @@
// Exercise the actual selector with a tiny JSX/hook harness; no browser or API.
const assert = require('node:assert/strict')
const fs = require('node:fs')
const path = require('node:path')
const vm = require('node:vm')
const {spawnSync} = require('node:child_process')
const os = require('node:os')
const root = path.resolve(__dirname, '../../AppImage')
const ts = require(path.join(root, 'node_modules/typescript'))
const source = fs.readFileSync(path.join(root, 'components/app-updater-editor.tsx'), 'utf8')
const result = ts.transpileModule(source, {compilerOptions: {
module: ts.ModuleKind.CommonJS, target: ts.ScriptTarget.ES2020, jsx: ts.JsxEmit.ReactJSX,
}, reportDiagnostics: true})
assert.equal(result.diagnostics.length, 0)
const flatten = node => !node || typeof node !== 'object' ? [] : [node, ...(
[node.props?.children].flat(Infinity).flatMap(flatten)
)]
// Check the real shared Button and cn/tailwind-merge, not only editor props.
function loadSource(relative) {
const compiled = ts.transpileModule(fs.readFileSync(path.join(root, relative), 'utf8'), {
compilerOptions: {module: ts.ModuleKind.CommonJS, target: ts.ScriptTarget.ES2020, jsx: ts.JsxEmit.ReactJSX},
})
const context = {exports: {}, require: name => name === '@/lib/utils'
? loadSource('lib/utils.ts') : require(path.join(root, 'node_modules', name))}
vm.runInNewContext(compiled.outputText, context)
return context.exports
}
const actualButton = loadSource('components/ui/button.tsx').Button
const navbarSource = fs.readFileSync(path.join(root, 'components/proxmox-dashboard.tsx'), 'utf8')
const appPanelSource = fs.readFileSync(path.join(root, 'components/lxc-app-panel.tsx'), 'utf8')
function assertSaveContrast(save) {
const dom = actualButton.render(save.props, null)
assert.equal(dom.props.disabled, save.props.disabled)
assert.match(dom.props.className, /\btext-white\b/)
assert.match(dom.props.className, /disabled:opacity-50/)
assert(!dom.props.className.includes('disabled:opacity-100'), 'retain the shared disabled appearance')
assert(!dom.props.className.includes('disabled:bg-blue-800'), 'do not override the disabled background')
const background = dom.props.className.split(' ').find(token => /^bg-blue-\d+$/.test(token))
assert.equal(background, 'bg-blue-500')
assert(navbarSource.includes(`data-[state=active]:${background}`), 'match the active navigation tab blue')
}
let englishEditor
for (const locale of ['en', 'es', 'de', 'fr', 'it', 'pt', 'sk', 'sv']) {
const messages = JSON.parse(fs.readFileSync(path.join(root, `messages/${locale}/common.json`)))
let help = null
const t = key => {
const value = key.split('.').reduce((object, part) => object?.[part], messages)
assert.equal(typeof value, 'string', `${locale}: missing ${key}`)
return value
}
const context = {exports: {}, require: name => {
if (name === 'react') return {useId: () => 'test-command', useState: () => [help, value => {help = value}]}
if (name === 'react/jsx-runtime') return {jsx: (type, props) => ({type, props}), jsxs: (type, props) => ({type, props})}
if (name === '@/lib/i18n/provider') return {useT: () => t}
return new Proxy({}, {get: (_, key) => String(key)})
}}
vm.runInNewContext(result.outputText, context)
if (locale === 'en') englishEditor = context.exports.AppUpdaterEditor
let saves = 0, method = 'none', command = ''
const props = {method, command, helperAvailable: true, helperSlug: 'qbittorrent', configured: false,
saving: false, changed: true, onMethodChange: value => {method = value}, onCommandChange: value => {command = value},
onSave: () => {saves++}, onCancel: () => {}, onRemove: () => {}}
const render = overrides => flatten(context.exports.AppUpdaterEditor({...props, method, command, ...overrides}))
let tree = render()
if (locale === 'es') assert.equal(t('vmLxc.updates.updaterChoiceHint'), 'Elige y guarda un método de actualización.')
assert.equal(tree.filter(n => n.props['aria-pressed'] === true).length, 0)
assert.equal(tree.filter(n => n.type === 'Textarea').length, 0)
let save = tree.find(n => n.type === 'Button' && n.props.onClick === props.onSave)
assert.equal(save.props.disabled, true)
assert.match(save.props.className, /bg-blue-500/)
assert.match(save.props.className, /hover:bg-blue-600/)
assert.match(save.props.className, /text-white/)
assertSaveContrast(save)
const helperButton = tree.find(n => n.type === 'Button' && n.props.children === t('vmLxc.updates.helperMethod'))
helperButton.props.onClick()
assert.equal(method, 'helper')
assert.equal(saves, 0, 'selecting must not execute or save')
tree = render()
const helperField = tree.find(n => n.type === 'Textarea')
const canonicalHelper = helperField.props.value
assert.equal(canonicalHelper, 'PHS_SILENT=1 bash -c "$(curl -fsSL https://raw.githubusercontent.com/community-scripts/ProxmoxVE/main/ct/qbittorrent.sh)"')
assert(!canonicalHelper.includes('_proxmenux_updater'), 'internal guards must not appear in the editor')
assert(!helperField.props.readOnly, 'the helper launcher must be editable')
helperField.props.onChange({target: {value: canonicalHelper}})
assert.equal(method, 'helper', 'an unchanged launcher remains the official method')
const editedHelper = canonicalHelper.replace('PHS_SILENT=1', 'PHS_SILENT=0')
helperField.props.onChange({target: {value: editedHelper}})
assert.equal(method, 'custom', 'editing must route execution through the saved custom command')
assert.equal(command, editedHelper)
assert.equal(render().find(n => n.type === 'Textarea').props.value, editedHelper)
helperButton.props.onClick()
assert.equal(render().find(n => n.type === 'Textarea').props.value, canonicalHelper)
assert.equal(command, editedHelper, 'switching back to helper preserves the unsaved custom draft')
command = ''
tree = render()
assert.equal(tree.find(n => n.props.onClick === props.onSave).props.disabled, false)
assertSaveContrast(tree.find(n => n.props.onClick === props.onSave))
for (const overrides of [{saving: true}, {changed: false}]) {
const disabledSave = render(overrides).find(n => n.props.onClick === props.onSave)
assert.equal(disabledSave.props.disabled, true)
assertSaveContrast(disabledSave)
}
const helperInfo = tree.find(n => n.props['aria-label'] === t('vmLxc.updates.helperMethodHelp'))
assert.match(helperInfo.props.className, /text-blue-500/)
helperInfo.props.onClick()
tree = render()
assert.equal(tree.find(n => n.type === 'Dialog').props.open, true)
assert(tree.some(n => n.type === 'a' && n.props.href.endsWith('/ct/qbittorrent.sh')))
assert(tree.some(n => n.type === 'a' && n.props.href === 'https://community-scripts.org/docs/tools/pve/update-apps'))
const helperLinks = tree.filter(n => n.type === 'a')
assert.equal(helperLinks.length, 2)
for (const link of helperLinks) {
assert.match(link.props.className, /text-blue-400 hover:text-blue-300/)
assert(appPanelSource.includes('text-blue-400 hover:text-blue-300'), 'use the same colors as the App web links')
}
assert(tree.some(n => n.type === 'code' && n.props.children === canonicalHelper))
assert(!render({helperSlug: "bad'; touch /tmp/injected"}).some(n => n.type === 'code'))
assert(!render({helperSlug: undefined}).some(n => n.type === 'code'))
assert.equal(tree.filter(n => n.type === 'Textarea').length, 1)
assert.equal(render({helperSlug: undefined}).find(n => n.props.onClick === props.onSave).props.disabled, true)
const customInfo = tree.find(n => n.props['aria-label'] === t('vmLxc.updates.customMethodHelp'))
assert.match(customInfo.props.className, /text-blue-500/)
customInfo.props.onClick()
tree = render()
const headingIndex = tree.findIndex(n => n.type === 'p' && n.props.children === t('vmLxc.updates.customExamplesHeading'))
const descriptionIndex = tree.findIndex(n => n.type === 'DialogDescription')
const firstExampleIndex = tree.findIndex(n => n.type === 'code')
assert(headingIndex > descriptionIndex && headingIndex < firstExampleIndex, 'examples heading follows the introduction')
if (locale === 'es') assert.equal(t('vmLxc.updates.customExamplesHeading'), 'Ejemplos para adaptar:')
const examples = tree.filter(n => n.type === 'code').map(n => n.props.children)
assert.equal(examples.length, 4)
assert(examples.includes('/opt/my-app/update.sh'))
assert(examples[1].includes("curl -fsSL 'https://example.com/my-app/update.sh'"))
assert(examples[1].includes('-o "$script" &&\nbash "$script"'), 'never execute a failed download')
assert(examples[1].includes('trap'), 'clean up the temporary download')
for (const example of examples) {
const syntax = spawnSync('sh', ['-n'], {input: example, encoding: 'utf8'})
assert.equal(syntax.status, 0, syntax.stderr)
}
if (locale === 'en') {
const fixture = fs.mkdtempSync(path.join(os.tmpdir(), 'proxmenux-helper-launcher-'))
try {
// The visible command stays short; the actual backend adds protection.
fs.symlinkSync('/bin/bash', path.join(fixture, 'bash'))
const fetch = '#!/bin/sh\nprintf "%s" "$FETCH_BODY"\nexit "$FETCH_STATUS"\n'
fs.writeFileSync(path.join(fixture, 'curl'), fetch, {mode: 0o755})
for (const tool of ['wget', 'curl']) {
const wget = path.join(fixture, 'wget')
if (tool === 'wget') fs.writeFileSync(wget, fetch, {mode: 0o755})
else fs.unlinkSync(wget)
const command = canonicalHelper.replaceAll('qbittorrent.sh', 'qbittorrent.sh2')
.replace('curl -fsSL', tool === 'wget' ? 'wget -qLO -' : 'curl -fsSL')
const prepared = spawnSync('python3', ['-c',
'import sys; from lxc_apps import protect_download_update_command; sys.stdout.write(protect_download_update_command(sys.stdin.read()))'], {
input: command, encoding: 'utf8', env: {...process.env, PYTHONPATH: path.join(root, 'scripts')},
})
assert.equal(prepared.status, 0, prepared.stderr)
assert(prepared.stdout.includes('updater download failed'), 'both curl and historical wget launchers remain protected')
for (const [status, body, expected] of [[8, '', 1], [8, 'echo SHOULD_NOT_RUN', 1],
[0, '', 1], [0, 'echo SCRIPT_RAN; exit 0', 0], [0, 'echo SCRIPT_RAN; exit 23', 23]]) {
const run = spawnSync('/bin/sh', ['-c', prepared.stdout], {
encoding: 'utf8', env: {...process.env, PATH: fixture, FETCH_STATUS: String(status), FETCH_BODY: body},
})
assert.equal(run.status, expected, run.stderr)
assert(!run.stdout.includes('SHOULD_NOT_RUN'), 'a partial failed download must not execute')
assert.equal(run.stdout.includes('SCRIPT_RAN'), status === 0 && body !== '')
}
}
} finally {
fs.rmSync(fixture, {recursive: true, force: true})
}
}
if (locale === 'en') {
const fixture = fs.mkdtempSync(path.join(os.tmpdir(), 'proxmenux-online-example-'))
try {
const bin = path.join(fixture, 'bin')
fs.mkdirSync(bin)
fs.writeFileSync(path.join(bin, 'curl'), '#!/bin/sh\nexit "$EXAMPLE_FETCH_STATUS"\n', {mode: 0o755})
fs.writeFileSync(path.join(bin, 'bash'), '#!/bin/sh\n: > "$EXAMPLE_EXECUTION_MARKER"\n', {mode: 0o755})
for (const status of [0, 22]) {
const marker = path.join(fixture, `ran-${status}`)
const run = spawnSync('/bin/sh', ['-c', examples[1]], {encoding: 'utf8', env: {
...process.env, PATH: `${bin}:${process.env.PATH}`, TMPDIR: fixture,
EXAMPLE_FETCH_STATUS: String(status), EXAMPLE_EXECUTION_MARKER: marker,
}})
assert.equal(run.status, status, run.stderr)
assert.equal(fs.existsSync(marker), status === 0, 'do not run a failed or incomplete download')
}
assert.deepEqual(fs.readdirSync(fixture).sort(), ['bin', 'ran-0'], 'temporary scripts must be removed')
} finally {
fs.rmSync(fixture, {recursive: true, force: true})
}
}
assert(examples.some(code => code.includes('apt-get install -y --only-upgrade my-package')))
assert(examples.some(code => code.includes('install -b -m 0755 /tmp/my-app.new')))
assert.equal(command, '', 'help examples must not change the saved command')
assert.equal(saves, 0, 'opening help must not execute or save')
assert.match(tree.find(n => n.type === 'DialogContent').props.className, /overflow-y-auto/)
method = 'custom'
tree = render()
assert.equal(tree.find(n => n.props.onClick === props.onSave).props.disabled, true)
tree.find(n => n.type === 'Textarea').props.onChange({target: {value: '/opt/my-app/update.sh'}})
tree = render()
assert.equal(tree.find(n => n.props.onClick === props.onSave).props.disabled, false)
tree = render({helperAvailable: false})
assert(!tree.some(n => n.type === 'Button' && n.props.children === t('vmLxc.updates.helperMethod')))
method = 'helper'
tree = render({helperAvailable: false})
assert.equal(tree.find(n => n.props.onClick === props.onSave).props.disabled, true)
console.log(`PASS ${locale}: explicit selection, navbar blue, dimmed disabled Save, examples heading, local/online scripts, no execution, validation`)
}
// Integrate the editor with the actual parent open/save/cancel functions.
async function testParentRoundTrip() {
const parentSource = fs.readFileSync(path.join(root, 'components/virtual-machines.tsx'), 'utf8')
const ast = ts.createSourceFile('virtual-machines.tsx', parentSource, ts.ScriptTarget.Latest, true, ts.ScriptKind.TSX)
const names = ['openCustomCmdEditor', 'saveCustomCommand', 'closeCustomCmdEditor']
const declarations = {}
function visit(node) {
if (ts.isVariableDeclaration(node) && ts.isIdentifier(node.name) && names.includes(node.name.text)) {
declarations[node.name.text] = `const ${node.getText(ast)};`
}
ts.forEachChild(node, visit)
}
visit(ast)
for (const name of names) assert(declarations[name], `missing actual parent function ${name}`)
const compiled = ts.transpileModule(names.map(n => declarations[n]).join('\n') + '\n' +
names.map(n => `exports.${n} = ${n};`).join('\n'), {compilerOptions: {module: ts.ModuleKind.CommonJS}})
const saved = new Map([
['a', {id: 'a', name: 'Odoo', helper_slug: 'odoo', update_method: 'helper', update_command: ''}],
['b', {id: 'b', name: 'Other app', update_method: 'custom', update_command: '/opt/other/update.sh'}],
])
let writes = 0
const parent = {exports: {}, customCmdDraft: '', updaterMethodDraft: 'none',
setCustomCmdEditingApp: value => {parent.editing = value},
setCustomCmdDraft: value => {parent.customCmdDraft = value},
setUpdaterMethodDraft: value => {parent.updaterMethodDraft = value},
setCustomCmdSaving: value => {parent.saving = value},
patchAppWatch: async (vmid, app, patch) => {
assert.equal(vmid, 101)
saved.set(app.id, {...saved.get(app.id), ...patch})
writes++
}, t: key => key, alert: error => {throw Error(error)},
}
vm.runInNewContext(compiled.outputText, parent)
const render = () => flatten(englishEditor({method: parent.updaterMethodDraft, command: parent.customCmdDraft,
helperAvailable: true, helperSlug: saved.get(parent.editing)?.helper_slug, configured: true,
saving: false, changed: true, onMethodChange: parent.setUpdaterMethodDraft,
onCommandChange: parent.setCustomCmdDraft, onCancel: parent.exports.closeCustomCmdEditor,
onSave: () => {}, onRemove: () => {}}))
parent.exports.openCustomCmdEditor(saved.get('a'))
let field = render().find(n => n.type === 'Textarea')
assert(field.props.value.includes('/ct/odoo.sh'))
const edited = field.props.value.replace('PHS_SILENT=1', 'PHS_SILENT=0')
field.props.onChange({target: {value: edited}})
assert.equal(writes, 0, 'editing must not persist before Save')
parent.exports.closeCustomCmdEditor()
assert.equal(saved.get('a').update_method, 'helper', 'Cancel leaves the original choice unchanged')
parent.exports.openCustomCmdEditor(saved.get('a'))
field = render().find(n => n.type === 'Textarea')
assert(field.props.value.includes('PHS_SILENT=1'))
field.props.onChange({target: {value: edited}})
await parent.exports.saveCustomCommand(101, saved.get('a'))
assert.equal(saved.get('a').update_method, 'custom')
assert.equal(saved.get('a').update_command, edited, 'Save must not discard the edited helper launcher')
parent.exports.openCustomCmdEditor(saved.get('a'))
assert.equal(render().find(n => n.type === 'Textarea').props.value, edited, 'reopen the exact saved custom command')
parent.exports.openCustomCmdEditor(saved.get('b'))
assert.equal(render().find(n => n.type === 'Textarea').props.value, '/opt/other/update.sh', 'keep commands separate per app')
parent.exports.openCustomCmdEditor(saved.get('a'))
parent.setUpdaterMethodDraft('helper')
await parent.exports.saveCustomCommand(101, saved.get('a'))
assert.equal(saved.get('a').update_method, 'helper')
assert.equal(saved.get('a').update_command, '', 'explicitly selecting helper restores the official path')
assert.equal(saved.get('b').update_command, '/opt/other/update.sh')
console.log('PASS actual parent: edit, cancel, save, reopen, per-app isolation, restore helper')
}
testParentRoundTrip().catch(error => {console.error(error); process.exitCode = 1})