#!/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()