David Schroeder
218 lines
11 KiB
Python
218 lines
11 KiB
Python
#!/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()
|