mirror of
https://github.com/MacRimi/ProxMenux.git
synced 2026-11-01 10:22:08 +00:00
215 lines
10 KiB
Python
215 lines
10 KiB
Python
"""Actual wrapper + worker, with every machine effect replaced by local fixtures.
|
|
|
|
Only fixed filesystem locations are redirected. Package commands and reboot
|
|
are inert executables which record calls; no root, network or host update.
|
|
"""
|
|
import os
|
|
from pathlib import Path
|
|
import re
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import time
|
|
import unittest
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
|
|
|
|
class HostUpdateShell(unittest.TestCase):
|
|
def test_packaging_copies_required_modules_with_no_host_probe_on_import(self):
|
|
source = (ROOT / 'AppImage/scripts/build_appimage.sh').read_text().splitlines()
|
|
copies = [f'cp "$SCRIPT_DIR/{name}.py" "$APP_DIR/usr/bin/"'
|
|
for name in ('host_update_jobs', 'kernel_reboot')]
|
|
for line in copies:
|
|
self.assertIn(line, source)
|
|
app = self.folder / 'AppDir'
|
|
(app / 'usr/bin').mkdir(parents=True)
|
|
result = subprocess.run(['/bin/bash', '-c', '\n'.join(copies)],
|
|
env=dict(os.environ, SCRIPT_DIR=str(ROOT / 'AppImage/scripts'), APP_DIR=str(app)),
|
|
capture_output=True, text=True, timeout=3)
|
|
self.assertEqual(result.returncode, 0, result.stderr)
|
|
code = ('import sys; from unittest.mock import patch; sys.path.insert(0,sys.argv[1]); '
|
|
'guard=patch("subprocess.Popen",side_effect=AssertionError("forbidden probe")); '
|
|
'guard.start(); import host_update_jobs,kernel_reboot; '
|
|
'assert host_update_jobs.list_jobs()==[]')
|
|
result = subprocess.run([sys.executable, '-I', '-B', '-c', code, str(app / 'usr/bin')],
|
|
capture_output=True, text=True, timeout=3)
|
|
self.assertEqual(result.returncode, 0, result.stderr)
|
|
|
|
def setUp(self):
|
|
self.temp = tempfile.TemporaryDirectory(prefix='host-update-shell-')
|
|
self.addCleanup(self.temp.cleanup)
|
|
self.folder = Path(self.temp.name)
|
|
self.bin = self.folder / 'bin'
|
|
self.bin.mkdir()
|
|
scripts = self.folder / 'scripts'
|
|
(scripts / 'utilities').mkdir(parents=True)
|
|
(scripts / 'global').mkdir()
|
|
self.calls = self.folder / 'calls'
|
|
self.calls.touch()
|
|
self.lock = self.folder / 'update.lock'
|
|
for name in ('utilities/proxmox_update.sh', 'global/update-pve-safe.sh'):
|
|
text = (ROOT / 'scripts' / name).read_text()
|
|
text = text.replace('/usr/local/share/proxmenux', str(self.folder))
|
|
text = text.replace('/run/proxmenux-host-update.lock', str(self.lock))
|
|
text = text.replace('/var/run/proxmenux-update-', str(self.folder / 'marker-'))
|
|
text = text.replace('/var/run/reboot-required', str(self.folder / 'reboot-required'))
|
|
text = text.replace('/var/log/proxmox-update-', str(self.folder / 'update-'))
|
|
text = text.replace('/tmp/proxmenux_screen_capture_', str(self.folder / 'capture-'))
|
|
(scripts / name).write_text(text)
|
|
(self.folder / 'utils.sh').write_text('''
|
|
load_language() { :; }
|
|
initialize_cache() { :; }
|
|
translate() { printf '%s' "$1"; }
|
|
msg_error() { echo "$*"; }
|
|
msg_warn() { echo "$*"; }
|
|
msg_info() { echo "$*"; }
|
|
msg_info2() { echo "$*"; }
|
|
msg_ok() { echo "$*"; }
|
|
msg_title() { echo "$*"; }
|
|
msg_success() { echo "$*"; }
|
|
show_proxmenux_logo() { :; }
|
|
LOCAL_SCRIPTS="$FIXTURE/scripts"
|
|
''')
|
|
# Keep the real confirmation contract added upstream; stub only the
|
|
# package/host effects, not the updater's shared decision helper.
|
|
confirmation = re.search(r'^pmx_confirm_update\(\) \{\n.*?^\}\n',
|
|
(ROOT / 'scripts/global/common-functions.sh').read_text(),
|
|
re.MULTILINE | re.DOTALL)
|
|
self.assertIsNotNone(confirmation)
|
|
(scripts / 'global/common-functions.sh').write_text('''
|
|
pmx_format_upgradable() { cat; }
|
|
pmx_dpkg_changes_since() { echo 'package changes fixture'; }
|
|
lvm_repair_check() { echo lvm-poststep >> "$CALLS"; }
|
|
pmx_rebuild_dkms_after_kernel() { echo dkms-poststep >> "$CALLS"; }
|
|
''' + confirmation.group(0))
|
|
(scripts / 'global/utils-install-functions.sh').write_text('ensure_repositories() { return 0; }\n')
|
|
commands = {
|
|
'pveversion': "echo 'pve-manager/9.2.24/test (running kernel: 7.0.14-25-pve)'",
|
|
'df': "printf 'Filesystem 1K-blocks Used Available Use%% Mounted\\nfixture 9999999 1 9999999 1%% /\\n'",
|
|
'uname': "echo 7.0.14-25-pve",
|
|
'dpkg-query': "printf 'install ok installed\\tproxmox-kernel-7.0.14-26-pve-signed\\n'",
|
|
'apt-cache': "echo 'Candidate: 9.2.24'",
|
|
'apt-get': 'echo "apt-get $*" >> "$CALLS"; exit 0',
|
|
'apt': '''echo "apt $*" >> "$CALLS"
|
|
if [[ "$1" == list ]]; then
|
|
printf 'Listing...\\npve-manager/stable 9.2.24 amd64 [upgradable from: 9.2.21]\\n'
|
|
else
|
|
touch "$FIXTURE/upgrade-started"
|
|
while [[ ! -e "$FIXTURE/release" ]]; do sleep .02; done
|
|
exit "${UPGRADE_EXIT:-0}"
|
|
fi''',
|
|
'dialog': 'echo dialog >> "$CALLS"; exit "${CONFIRM_EXIT:-0}"',
|
|
'whiptail': '''echo "whiptail $*" >> "$CALLS"
|
|
if [[ "$1" == --title && "$2" == "Proxmox Update" ]]; then exit "${CONFIRM_EXIT:-0}"; fi
|
|
exit 1''',
|
|
'reboot': 'echo FORBIDDEN-REBOOT >> "$CALLS"; exit 99',
|
|
'clear': ':',
|
|
}
|
|
for name, body in commands.items():
|
|
path = self.bin / name
|
|
path.write_text('#!/bin/bash\n' + body + '\n')
|
|
path.chmod(0o700)
|
|
self.env = dict(os.environ, PATH=str(self.bin)+':'+os.environ['PATH'],
|
|
FIXTURE=str(self.folder), CALLS=str(self.calls),
|
|
EXECUTION_MODE='web', PROXMENUX_HOST_UPDATE_JOB='1')
|
|
self.wrapper = scripts / 'utilities/proxmox_update.sh'
|
|
self.procs = []
|
|
self.addCleanup(self.finish)
|
|
|
|
def finish(self):
|
|
(self.folder / 'release').touch()
|
|
for proc in self.procs:
|
|
if proc.poll() is None:
|
|
proc.communicate('\n', timeout=8)
|
|
|
|
def start(self, **env):
|
|
proc = subprocess.Popen(['/bin/bash', str(self.wrapper)], env=dict(self.env, **env),
|
|
stdin=subprocess.PIPE, stdout=subprocess.PIPE,
|
|
stderr=subprocess.STDOUT, text=True)
|
|
self.procs.append(proc)
|
|
return proc
|
|
|
|
def wait_upgrade(self):
|
|
until = time.monotonic() + 5
|
|
while not (self.folder / 'upgrade-started').exists():
|
|
self.assertLess(time.monotonic(), until, 'inert upgrade did not start')
|
|
time.sleep(.01)
|
|
|
|
def test_web_full_flow_finishes_poststeps_and_footer_without_reboot_or_keypress(self):
|
|
proc = self.start()
|
|
self.wait_upgrade()
|
|
(self.folder / 'release').touch()
|
|
output, _ = proc.communicate(timeout=8)
|
|
self.assertEqual(proc.returncode, 0, output)
|
|
calls = self.calls.read_text()
|
|
self.assertIn('lvm-poststep', calls)
|
|
self.assertIn('dkms-poststep', calls)
|
|
self.assertIn('apt-get -y autoremove', calls)
|
|
self.assertIn('apt-get -y autoclean', calls)
|
|
self.assertNotIn('FORBIDDEN-REBOOT', calls)
|
|
self.assertIn('whiptail --title Proxmox Update --yesno', calls)
|
|
self.assertNotIn('whiptail --title Reboot Required', calls)
|
|
self.assertIn('Reboot Required', output)
|
|
self.assertIn('You can reboot later manually.', output)
|
|
self.assertFalse((self.folder / 'marker-in-progress').exists())
|
|
self.assertTrue((self.folder / 'marker-just-finished').exists())
|
|
log = next(self.folder.glob('update-*.log')).read_text()
|
|
self.assertIn('--- Packages changed (exit 0) ---', log)
|
|
self.assertIn('Finished:', log)
|
|
|
|
def test_cancel_before_upgrade_has_no_upgrade_cleanup_or_reboot(self):
|
|
proc = self.start(CONFIRM_EXIT='1')
|
|
output, _ = proc.communicate(timeout=5)
|
|
self.assertEqual(proc.returncode, 130, output)
|
|
calls = self.calls.read_text()
|
|
for forbidden in ('full-upgrade', 'autoremove', 'autoclean', 'REBOOT', 'poststep'):
|
|
self.assertNotIn(forbidden, calls)
|
|
|
|
def test_failure_keeps_failure_and_never_offers_reboot(self):
|
|
(self.folder / 'release').touch()
|
|
proc = self.start(UPGRADE_EXIT='1')
|
|
output, _ = proc.communicate(timeout=5)
|
|
self.assertEqual(proc.returncode, 1, output)
|
|
for forbidden in ('autoremove', 'autoclean', 'REBOOT', 'poststep', 'whiptail --title Reboot Required'):
|
|
self.assertNotIn(forbidden, self.calls.read_text())
|
|
|
|
def test_old_menu_cli_contract_retains_explicit_reboot_prompt(self):
|
|
(self.folder / 'release').touch()
|
|
proc = self.start(PROXMENUX_HOST_UPDATE_JOB='', EXECUTION_MODE='cli')
|
|
output, _ = proc.communicate('\n', timeout=5)
|
|
self.assertEqual(proc.returncode, 0, output)
|
|
self.assertIn('whiptail --title Reboot Required --yesno', self.calls.read_text())
|
|
self.assertNotIn('FORBIDDEN-REBOOT', self.calls.read_text())
|
|
|
|
def test_shared_lock_rejects_second_legacy_or_web_caller_before_any_package_call(self):
|
|
first = self.start()
|
|
self.wait_upgrade()
|
|
before = self.calls.read_text()
|
|
for mode in ('cli', 'web'):
|
|
second = self.start(EXECUTION_MODE=mode)
|
|
output, _ = second.communicate(timeout=3)
|
|
self.assertEqual(second.returncode, 75, output)
|
|
self.assertEqual(self.calls.read_text(), before)
|
|
self.assertIn('already running', output)
|
|
self.assertIsNone(first.poll())
|
|
|
|
def test_lock_does_not_follow_symlink_or_truncate_existing_file(self):
|
|
target = self.folder / 'protected'
|
|
target.write_text('preserve')
|
|
self.lock.symlink_to(target)
|
|
proc = self.start()
|
|
output, _ = proc.communicate(timeout=3)
|
|
self.assertEqual(proc.returncode, 75, output)
|
|
self.assertEqual(target.read_text(), 'preserve')
|
|
self.assertEqual(self.calls.read_text(), '')
|
|
self.lock.unlink()
|
|
self.lock.write_text('existing metadata')
|
|
proc = self.start(CONFIRM_EXIT='1')
|
|
proc.communicate(timeout=5)
|
|
self.assertEqual(self.lock.read_text(), 'existing metadata')
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main(verbosity=2)
|