menu update
This commit is contained in:
David Schroeder
@@ -0,0 +1,82 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Check reorganized menu routes, including conditional hardware entries."""
|
||||
from pathlib import Path
|
||||
import os
|
||||
import re
|
||||
import subprocess
|
||||
import unittest
|
||||
|
||||
ROOT = Path(__file__).resolve().parents[1]
|
||||
SOURCE = (ROOT / 'proxmenu-scripts.sh').read_text()
|
||||
|
||||
|
||||
def function(name):
|
||||
return name + '() {' + SOURCE.split(name + '() {', 1)[1].split('\n}\n', 1)[0] + '\n}\n'
|
||||
|
||||
|
||||
class MenuLayoutTests(unittest.TestCase):
|
||||
def menu(self, name, selection, dell=False):
|
||||
# Real menu functions, mocked host inspection and terminal selection.
|
||||
script = '''
|
||||
TAPM_POST_PROXMENUX_DETECTED() { return 1; }
|
||||
TAPM_PULSE_RESOURCE_INSTALLED_FROM_CONFIGS() { return 0; }
|
||||
TAPM_PACKAGE_INSTALLED() { return 0; }
|
||||
TAPM_OMSA_SUPPORTED_HARDWARE() { [[ "$TEST_DELL" == yes ]]; }
|
||||
TAPM_HA_NODE_IN_MAINTENANCE() { return 1; }
|
||||
systemctl() { return 0; }
|
||||
dpkg-query() { printf 'install ok installed'; }
|
||||
glances() { :; }
|
||||
SELECT_MENU() {
|
||||
local -n actual_labels="$2" actual_values="$3"
|
||||
[[ ${#actual_labels[@]} == ${#actual_values[@]} ]] || exit 40
|
||||
if [[ ${TEST_VISITED:-0} == 1 ]]; then MENU_SELECTION=back; return; fi
|
||||
TEST_VISITED=1
|
||||
printf 'TITLE:%s\\n' "$1"
|
||||
for i in "${!actual_values[@]}"; do
|
||||
printf 'ROW:%s:%s\\n' "${actual_values[$i]}" "${actual_labels[$i]}"
|
||||
done
|
||||
MENU_SELECTION="$TEST_SELECTION"
|
||||
}
|
||||
'''
|
||||
for action in ['SERVICE_RECOVERY_MENU', 'TAPM_HOST_CONFIGURATION_MENU', 'INSTALL_GLANCES',
|
||||
'INSTALL_OMSA', 'TAPM_NOTIFICATIONS_MENU', 'DETECT_CPU', 'INSTALL_KEEPALIVE',
|
||||
'VIRTIO_MENU', 'TAPM_POST_VZDUMP_MENU', 'TAPM_ORPHAN_DISK_REPORT']:
|
||||
script += action + '() { printf "ACTION:' + action + '\\n"; }\n'
|
||||
script += function(name) + '\n' + name
|
||||
result = subprocess.run(['bash', '-c', script], capture_output=True, text=True,
|
||||
env=dict(os.environ, TEST_SELECTION=selection, TEST_DELL='yes' if dell else 'no'))
|
||||
self.assertEqual(result.returncode, 0, result.stderr)
|
||||
rows = dict(re.findall(r'^ROW:([^:]+):(.*)$', result.stdout, re.M))
|
||||
return rows, result.stdout
|
||||
|
||||
def test_local_service_recovery_is_under_host_configuration(self):
|
||||
rows, output = self.menu('HOST_SETUP_MENU', 'services')
|
||||
self.assertEqual(set(rows), {'post_install', 'services'})
|
||||
self.assertIn('ACTION:SERVICE_RECOVERY_MENU', output)
|
||||
|
||||
def test_cluster_cpu_action_moved_with_its_dispatch(self):
|
||||
rows, output = self.menu('CLUSTER_MENU', 'cpu')
|
||||
self.assertEqual(set(rows), {'maintenance', 'cpu', 'keepalived'})
|
||||
self.assertIn('ACTION:DETECT_CPU', output)
|
||||
|
||||
def test_storage_routes_keep_iso_scan_deferred(self):
|
||||
for choice, action in [('virtio', 'VIRTIO_MENU'), ('vzdump', 'TAPM_POST_VZDUMP_MENU'),
|
||||
('orphan_disks', 'TAPM_ORPHAN_DISK_REPORT')]:
|
||||
rows, output = self.menu('STORAGE_MENU', choice)
|
||||
self.assertEqual(set(rows), {'iso_nfs', 'virtio', 'orphan_disks', 'vzdump'})
|
||||
self.assertIn('ACTION:' + action, output)
|
||||
self.assertEqual(output.count('ACTION:'), 1)
|
||||
|
||||
def test_notifications_and_glances_with_optional_dell_entry(self):
|
||||
for dell in [False, True]:
|
||||
rows, output = self.menu('MONITORING_MENU', 'notifications', dell)
|
||||
self.assertEqual('omsa' in rows, dell)
|
||||
self.assertIn('glances', rows)
|
||||
self.assertIn('ACTION:TAPM_NOTIFICATIONS_MENU', output)
|
||||
rows, output = self.menu('MONITORING_MENU', 'glances', dell)
|
||||
self.assertIn('ACTION:INSTALL_GLANCES', output)
|
||||
self.assertTrue(rows['glances'].startswith('Install '))
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
unittest.main()
|
||||
@@ -0,0 +1,217 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Exercise notification policies without touching a host or sending messages."""
|
||||
import contextlib
|
||||
import importlib.util
|
||||
import io
|
||||
import json
|
||||
import os
|
||||
from pathlib import Path
|
||||
import subprocess
|
||||
import tempfile
|
||||
import unittest
|
||||
from unittest.mock import patch
|
||||
|
||||
ROOT = Path(__file__).resolve().parents[1]
|
||||
spec = importlib.util.spec_from_file_location('notifications', ROOT / 'inc/notifications.py')
|
||||
n = importlib.util.module_from_spec(spec)
|
||||
spec.loader.exec_module(n)
|
||||
|
||||
|
||||
class PolicyTests(unittest.TestCase):
|
||||
def test_external_policy_change_takes_precedence(self):
|
||||
snapshot = dict(policy=dict(mode='version', owner='pve1'), notify={'package-updates': 'always'})
|
||||
self.assertEqual(n.effective_mode(snapshot), 'all')
|
||||
snapshot['notify']['package-updates'] = 'never'
|
||||
self.assertEqual(n.effective_mode(snapshot), 'version')
|
||||
self.assertEqual(n.effective_mode(dict(policy={}, notify={})), 'auto')
|
||||
|
||||
def test_only_newer_ve_versions_qualify(self):
|
||||
rows = [dict(Package='linux-image', OldVersion='1', Version='2'),
|
||||
dict(Package='pve-manager', OldVersion='9.1.9', Version='9.1.10')]
|
||||
with patch.object(n.subprocess, 'run', return_value=subprocess.CompletedProcess([], 0)) as run:
|
||||
self.assertEqual(n.candidate_update(rows), ('9.1.9', '9.1.10'))
|
||||
self.assertEqual(run.call_args.args[0], ['dpkg', '--compare-versions', '9.1.10', 'gt', '9.1.9'])
|
||||
with patch.object(n.subprocess, 'run', return_value=subprocess.CompletedProcess([], 1)):
|
||||
self.assertIsNone(n.candidate_update(rows))
|
||||
with patch.object(n.subprocess, 'run') as run:
|
||||
self.assertIsNone(n.candidate_update(rows[:1]))
|
||||
run.assert_not_called()
|
||||
rows[1]['Version'] = 'invalid\nversion'
|
||||
with self.assertRaises(n.NotificationError):
|
||||
n.candidate_update(rows)
|
||||
|
||||
def test_cancel_never_writes_configuration(self):
|
||||
with patch.object(n, 'current_status', return_value={}), patch.object(n, 'local_node', return_value='pve1'), \
|
||||
patch('builtins.input', return_value='no'), patch.object(n, 'prepare_checker') as prepare, \
|
||||
patch.object(n, 'bridge') as bridge, contextlib.redirect_stdout(io.StringIO()):
|
||||
n.apply_policy('version')
|
||||
prepare.assert_not_called()
|
||||
bridge.assert_not_called()
|
||||
|
||||
def test_failed_timer_setup_does_not_suppress_native_alerts(self):
|
||||
with patch.object(n, 'current_status', return_value={}), patch.object(n, 'local_node', return_value='pve1'), \
|
||||
patch('builtins.input', return_value='yes'), patch.object(n, 'bridge') as bridge, \
|
||||
patch.object(n, 'prepare_checker', side_effect=n.NotificationError('failed')), \
|
||||
contextlib.redirect_stdout(io.StringIO()), self.assertRaises(n.NotificationError):
|
||||
n.apply_policy('version')
|
||||
bridge.assert_not_called()
|
||||
|
||||
def test_disabled_or_nonowner_worker_is_inert(self):
|
||||
for mode, owner in [('disabled', 'pve1'), ('all', 'pve1'), ('version', 'pve2')]:
|
||||
snapshot = dict(policy=dict(mode=mode, owner=owner), notify={'package-updates': 'never'})
|
||||
with patch.object(n, 'bridge', return_value=snapshot), patch.object(n, 'local_node', return_value='pve1'), \
|
||||
patch.object(n, 'api') as api:
|
||||
self.assertEqual(n.check_versions(), 0)
|
||||
api.assert_not_called()
|
||||
|
||||
def test_partial_cluster_check_continues_without_upgrade_alerts(self):
|
||||
snapshot = dict(policy=dict(mode='version', owner='pve1', generation='g'), notify={'package-updates': 'never'})
|
||||
responses = [[{'node': 'pve1', 'status': 'online'}, {'node': 'pve2', 'status': 'offline'},
|
||||
{'node': 'pve3', 'status': 'online'}, {'node': 'pve4', 'status': 'online'}],
|
||||
[], n.NotificationError('API down'),
|
||||
[dict(Package='pve-manager', OldVersion='9.1.9', Version='9.1.10')]]
|
||||
output = io.StringIO()
|
||||
with patch.object(n, 'bridge', side_effect=[snapshot, {'sent': True}]) as bridge, \
|
||||
patch.object(n, 'local_node', return_value='pve1'), patch.object(n, 'api', side_effect=responses), \
|
||||
patch.object(n.subprocess, 'run', return_value=subprocess.CompletedProcess([], 0)), \
|
||||
contextlib.redirect_stdout(output):
|
||||
self.assertEqual(n.check_versions(), 1)
|
||||
self.assertEqual(bridge.call_count, 2)
|
||||
self.assertEqual(bridge.call_args.kwargs['node'], 'pve4')
|
||||
self.assertNotIn('pve-manager', output.getvalue())
|
||||
self.assertIn('offline', output.getvalue())
|
||||
self.assertIn('9.1.9 -> 9.1.10', output.getvalue())
|
||||
|
||||
def test_checker_install_does_not_modify_daily_job(self):
|
||||
with tempfile.TemporaryDirectory() as directory:
|
||||
root = Path(directory)
|
||||
with patch.object(n, 'UNIT_DIR', root / 'units'), patch.object(n, 'TEMPLATE_DIR', root / 'templates'), \
|
||||
patch.object(n, 'command') as command:
|
||||
n.prepare_checker()
|
||||
paths = list(root.rglob('*'))
|
||||
self.assertFalse(any('pve-daily' in p.name for p in paths))
|
||||
for path in paths:
|
||||
if path.is_file():
|
||||
self.assertNotIn('pve-manager', path.read_text())
|
||||
self.assertIn('OnCalendar=hourly', (root / 'units' / (n.SERVICE + '.timer')).read_text())
|
||||
self.assertEqual(command.call_args.args[0], ['systemctl', 'is-active', n.SERVICE + '.timer'])
|
||||
|
||||
def test_delivery_test_requires_explicit_confirmation(self):
|
||||
targets = [{'name': 'smtp-team', 'type': 'smtp'}]
|
||||
with patch.object(n, 'api', return_value=targets) as api, \
|
||||
patch('builtins.input', side_effect=['1', 'no']), contextlib.redirect_stdout(io.StringIO()):
|
||||
n.test_delivery()
|
||||
self.assertEqual(api.call_count, 1)
|
||||
|
||||
|
||||
class BridgeTests(unittest.TestCase):
|
||||
"""Run real bridge code against fake PVE modules and temporary config files."""
|
||||
def setUp(self):
|
||||
self.temp = tempfile.TemporaryDirectory()
|
||||
self.root = Path(self.temp.name)
|
||||
(self.root / 'PVE').mkdir()
|
||||
(self.root / 'priv').mkdir()
|
||||
self.dc = self.root / 'dc.json'
|
||||
self.dc.write_text(json.dumps({'notify': {'package-updates': 'auto', 'replication': 'always'}, 'keyboard': 'en-us'}))
|
||||
modules = {
|
||||
'Cluster': '''package PVE::Cluster;
|
||||
use JSON; use PVE::Tools;
|
||||
sub cfs_update {}
|
||||
sub cfs_read_file { JSON::decode_json(PVE::Tools::file_get_contents("$ENV{FIXTURE}/dc.json")) }
|
||||
sub cfs_write_file { PVE::Tools::file_set_contents("$ENV{FIXTURE}/dc.json", JSON::encode_json($_[1])); }
|
||||
sub cfs_lock_domain { eval { $_[2]->(); }; }
|
||||
sub cfs_lock_file { eval { $_[2]->(); }; }
|
||||
1;
|
||||
''',
|
||||
'DataCenterConfig': 'package PVE::DataCenterConfig; 1;',
|
||||
'INotify': 'package PVE::INotify; sub inotify_init {} sub nodename { "pve1" } 1;',
|
||||
'Tools': '''package PVE::Tools;
|
||||
sub file_get_contents { open(my $f, '<', $_[0]) or die $!; local $/; return <$f>; }
|
||||
sub file_set_contents {
|
||||
die "write failure" if $ENV{FAIL_POLICY} && $_[0] =~ /tapm-update-policy.json$/;
|
||||
open(my $f, '>', $_[0]) or die $!; print $f $_[1]; close $f or die $!;
|
||||
}
|
||||
1;
|
||||
''',
|
||||
'Notify': '''package PVE::Notify;
|
||||
use JSON;
|
||||
sub info {
|
||||
die "delivery failure" if $ENV{FAIL_SEND};
|
||||
open(my $f, '>>', "$ENV{FIXTURE}/sent.jsonl") or die $!;
|
||||
print $f JSON::encode_json([@_]), "\\n"; close $f;
|
||||
}
|
||||
1;
|
||||
''',
|
||||
}
|
||||
for module, source in modules.items():
|
||||
(self.root / 'PVE' / (module + '.pm')).write_text(source)
|
||||
source = (ROOT / 'inc/notifications-bridge.pl').read_text()
|
||||
source = source.replace('/etc/pve', str(self.root)).replace('die "root required" if $> != 0;', '')
|
||||
self.script = self.root / 'bridge.pl'
|
||||
self.script.write_text(source)
|
||||
|
||||
def tearDown(self):
|
||||
self.temp.cleanup()
|
||||
|
||||
def call(self, action, env=None, **kwargs):
|
||||
result = subprocess.run(['perl', '-I', str(self.root), str(self.script)],
|
||||
input=json.dumps(dict(action=action, **kwargs)), text=True, capture_output=True,
|
||||
env=dict(os.environ, FIXTURE=str(self.root), **(env or {})))
|
||||
return result
|
||||
|
||||
def read(self):
|
||||
result = self.call('read')
|
||||
self.assertEqual(result.returncode, 0, result.stderr)
|
||||
return json.loads(result.stdout)
|
||||
|
||||
def apply(self, mode='version', **kwargs):
|
||||
return self.call('apply', expected=self.read(),
|
||||
policy=dict(mode=mode, owner='pve1', generation='g'), **kwargs)
|
||||
|
||||
def send(self, version='9.1.10', **kwargs):
|
||||
return self.call('send', generation='g', node='pve2', installed='9.1.9', available=version, **kwargs)
|
||||
|
||||
def test_apply_preserves_unrelated_settings_and_saves_previous_policy(self):
|
||||
for mode, expected in [('version', 'never'), ('all', 'always'), ('disabled', 'never')]:
|
||||
result = self.apply(mode)
|
||||
self.assertEqual(result.returncode, 0, result.stderr)
|
||||
dc = json.loads(self.dc.read_text())
|
||||
self.assertEqual(dc['notify'], {'package-updates': expected, 'replication': 'always'})
|
||||
self.assertEqual(dc['keyboard'], 'en-us')
|
||||
previous = json.loads((self.root / 'priv/tapm-update-policy.previous.json').read_text())
|
||||
self.assertEqual(previous['policy']['mode'], 'all')
|
||||
|
||||
def test_stale_configuration_is_rejected(self):
|
||||
previous = self.read()
|
||||
self.assertEqual(self.apply('all').returncode, 0)
|
||||
result = self.call('apply', expected=previous, policy=dict(mode='disabled', owner='pve1'))
|
||||
self.assertNotEqual(result.returncode, 0)
|
||||
self.assertEqual(self.read()['notify']['package-updates'], 'always')
|
||||
|
||||
def test_failed_policy_save_restores_native_notifications(self):
|
||||
result = self.apply(env={'FAIL_POLICY': '1'})
|
||||
self.assertNotEqual(result.returncode, 0)
|
||||
self.assertEqual(self.read()['notify']['package-updates'], 'auto')
|
||||
|
||||
def test_deduplication_and_generic_message(self):
|
||||
self.assertEqual(self.apply().returncode, 0)
|
||||
for version, expected in [('9.1.10', True), ('9.1.10', False), ('9.1.11', True), ('9.1.10', False)]:
|
||||
result = self.send(version)
|
||||
self.assertEqual(result.returncode, 0, result.stderr)
|
||||
self.assertEqual(json.loads(result.stdout)['sent'], expected)
|
||||
messages = (self.root / 'sent.jsonl').read_text()
|
||||
self.assertEqual(len(messages.splitlines()), 2)
|
||||
self.assertNotIn('pve-manager', messages)
|
||||
self.assertEqual(json.loads(messages.splitlines()[0])[2], {'type': 'package-updates', 'hostname': 'pve2'})
|
||||
|
||||
def test_failed_delivery_is_retried_and_policy_change_stops_old_worker(self):
|
||||
self.assertEqual(self.apply().returncode, 0)
|
||||
self.assertNotEqual(self.send(env={'FAIL_SEND': '1'}).returncode, 0)
|
||||
self.assertFalse((self.root / 'priv/tapm-update-receipts.json').exists())
|
||||
self.assertTrue(json.loads(self.send().stdout)['sent'])
|
||||
self.assertEqual(self.apply('disabled').returncode, 0)
|
||||
self.assertFalse(json.loads(self.send('9.1.11').stdout)['sent'])
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
unittest.main()
|
||||
@@ -0,0 +1,12 @@
|
||||
#!/usr/bin/env bash
|
||||
# Include the Python behavior suites in the standard validation entry point.
|
||||
set -u -o pipefail
|
||||
TEST_ROOT="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)"
|
||||
failures=0
|
||||
# Hyphenated filenames are not importable by unittest discovery.
|
||||
for test_file in "${TEST_ROOT}"/tests/test-*.py; do
|
||||
if ! PYTHONDONTWRITEBYTECODE=1 python3 "$test_file"; then
|
||||
(( failures += 1 ))
|
||||
fi
|
||||
done
|
||||
(( failures == 0 ))
|
||||
+4
-4
@@ -29,7 +29,7 @@ class UITests(unittest.TestCase):
|
||||
def test_plain_wrapped_menu_and_task(self):
|
||||
output = self.shell('''
|
||||
TAPM_UI_RULE
|
||||
TAPM_UI_MENU_META 'Cluster & Maintenance' orphan_disks
|
||||
TAPM_UI_MENU_META 'Storage, ISOs & Backups' orphan_disks
|
||||
TAPM_UI_MENU_ROW 5 $'Find \\e[36morphaned\\e[39m and unused disks [Read-only]' 1
|
||||
TAPM_UI_TASK 'Find orphaned and unused disks'
|
||||
''', width=40)
|
||||
@@ -39,12 +39,12 @@ TAPM_UI_TASK 'Find orphaned and unused disks'
|
||||
|
||||
def test_submenus_and_choices_do_not_start_tasks(self):
|
||||
cases = [('Main Menu', 'cluster', 'submenu'),
|
||||
('Cluster & Maintenance', 'services', 'submenu'),
|
||||
('Host Configuration & Recovery', 'services', 'submenu'),
|
||||
('Select ISO Storage', 'local', 'choice'),
|
||||
('Customize TAPM Host Profile', 'utilities', 'choice'),
|
||||
('Customize TAPM Host Profile', 'apply', 'task'),
|
||||
('Cluster & Maintenance', 'orphan_disks', 'task'),
|
||||
('Cluster & Maintenance', 'back', 'navigation')]
|
||||
('Storage, ISOs & Backups', 'orphan_disks', 'task'),
|
||||
('Cluster & High Availability', 'back', 'navigation')]
|
||||
for title, value, expected in cases:
|
||||
with self.subTest(title=title, value=value):
|
||||
output = self.shell(f"TAPM_UI_MENU_META '{title}' '{value}'\nprintf '%s' \"$TAPM_UI_KIND\"")
|
||||
|
||||
Reference in New Issue
Block a user