#!/usr/bin/env python3 """Tests for the pending-restart system: helpers, endpoints, and cancel.""" import sys import os import json import unittest import tempfile import shutil from pathlib import Path from unittest.mock import patch, MagicMock api_dir = Path(__file__).parent.parent / 'api' sys.path.insert(0, str(api_dir)) from app import app, _set_pending_restart, _clear_pending_restart, _collect_service_ports, config_manager class TestSetPendingRestart(unittest.TestCase): def setUp(self): _clear_pending_restart() def tearDown(self): _clear_pending_restart() def test_sets_needs_restart(self): _set_pending_restart(['something changed']) p = config_manager.configs.get('_pending_restart', {}) self.assertTrue(p['needs_restart']) def test_stores_changes(self): _set_pending_restart(['port x changed']) p = config_manager.configs['_pending_restart'] self.assertIn('port x changed', p['changes']) def test_default_containers_is_wildcard(self): _set_pending_restart(['some change']) p = config_manager.configs['_pending_restart'] self.assertEqual(p['containers'], ['*']) def test_specific_containers_stored(self): _set_pending_restart(['dns port changed'], ['dns']) p = config_manager.configs['_pending_restart'] self.assertEqual(p['containers'], ['dns']) def test_accumulates_changes_on_second_call(self): _set_pending_restart(['change A'], ['dns']) _set_pending_restart(['change B'], ['mail']) p = config_manager.configs['_pending_restart'] self.assertIn('change A', p['changes']) self.assertIn('change B', p['changes']) def test_accumulates_containers_on_second_call(self): _set_pending_restart(['A'], ['dns']) _set_pending_restart(['B'], ['mail']) p = config_manager.configs['_pending_restart'] self.assertIn('dns', p['containers']) self.assertIn('mail', p['containers']) def test_wildcard_absorbs_specific(self): _set_pending_restart(['ip range changed'], ['*']) _set_pending_restart(['port changed'], ['dns']) p = config_manager.configs['_pending_restart'] self.assertEqual(p['containers'], ['*']) def test_specific_escalates_to_wildcard(self): _set_pending_restart(['A'], ['dns']) _set_pending_restart(['B'], ['*']) p = config_manager.configs['_pending_restart'] self.assertEqual(p['containers'], ['*']) def test_no_duplicate_containers(self): _set_pending_restart(['A'], ['mail']) _set_pending_restart(['B'], ['mail']) p = config_manager.configs['_pending_restart'] self.assertEqual(p['containers'].count('mail'), 1) class TestClearPendingRestart(unittest.TestCase): def test_clears_flag(self): _set_pending_restart(['something']) _clear_pending_restart() p = config_manager.configs.get('_pending_restart', {}) self.assertFalse(p.get('needs_restart', False)) def test_clears_changes(self): _set_pending_restart(['something']) _clear_pending_restart() p = config_manager.configs.get('_pending_restart', {}) self.assertEqual(p.get('changes', []), []) def test_clears_containers(self): _set_pending_restart(['something'], ['dns']) _clear_pending_restart() p = config_manager.configs.get('_pending_restart', {}) self.assertEqual(p.get('containers', []), []) class TestCollectServicePorts(unittest.TestCase): def test_extracts_dns_port(self): cfg = {'network': {'dns_port': 5353}} ports = _collect_service_ports(cfg) self.assertEqual(ports['dns_port'], 5353) def test_extracts_wg_port_from_wireguard(self): cfg = {'wireguard': {'port': 12345}} ports = _collect_service_ports(cfg) self.assertEqual(ports['wg_port'], 12345) def test_extracts_wg_port_from_identity_fallback(self): cfg = {'_identity': {'wireguard_port': 9999}} ports = _collect_service_ports(cfg) self.assertEqual(ports['wg_port'], 9999) def test_wireguard_port_takes_priority_over_identity(self): cfg = {'wireguard': {'port': 12345}, '_identity': {'wireguard_port': 9999}} ports = _collect_service_ports(cfg) self.assertEqual(ports['wg_port'], 12345) def test_extracts_email_ports(self): cfg = {'email': {'smtp_port': 2525, 'submission_port': 465, 'imap_port': 1993, 'webmail_port': 9000}} ports = _collect_service_ports(cfg) self.assertEqual(ports['mail_smtp_port'], 2525) self.assertEqual(ports['mail_submission_port'], 465) self.assertEqual(ports['mail_imap_port'], 1993) self.assertEqual(ports['rainloop_port'], 9000) def test_extracts_calendar_port(self): cfg = {'calendar': {'port': 5233}} ports = _collect_service_ports(cfg) self.assertEqual(ports['radicale_port'], 5233) def test_extracts_files_ports(self): cfg = {'files': {'port': 8181, 'manager_port': 9090}} ports = _collect_service_ports(cfg) self.assertEqual(ports['webdav_port'], 8181) self.assertEqual(ports['filegator_port'], 9090) def test_missing_keys_not_in_result(self): ports = _collect_service_ports({}) self.assertNotIn('dns_port', ports) self.assertNotIn('wg_port', ports) def test_empty_sections_not_in_result(self): ports = _collect_service_ports({'email': {}}) self.assertNotIn('mail_smtp_port', ports) class TestGetPendingEndpoint(unittest.TestCase): def setUp(self): app.config['TESTING'] = True self.client = app.test_client() _clear_pending_restart() def tearDown(self): _clear_pending_restart() def test_returns_not_pending_by_default(self): r = self.client.get('/api/config/pending') self.assertEqual(r.status_code, 200) data = json.loads(r.data) self.assertFalse(data['needs_restart']) def test_returns_pending_state(self): _set_pending_restart(['dns port: 53 → 5353'], ['dns']) r = self.client.get('/api/config/pending') data = json.loads(r.data) self.assertTrue(data['needs_restart']) self.assertIn('dns port: 53 → 5353', data['changes']) self.assertIn('dns', data['containers']) def test_returns_containers_field(self): _set_pending_restart(['x'], ['wireguard']) data = json.loads(self.client.get('/api/config/pending').data) self.assertIn('containers', data) self.assertEqual(data['containers'], ['wireguard']) class TestCancelPendingEndpoint(unittest.TestCase): def setUp(self): app.config['TESTING'] = True self.client = app.test_client() _clear_pending_restart() def tearDown(self): _clear_pending_restart() def test_cancel_clears_pending(self): _set_pending_restart(['something'], ['dns']) r = self.client.delete('/api/config/pending') self.assertEqual(r.status_code, 200) p = config_manager.configs.get('_pending_restart', {}) self.assertFalse(p.get('needs_restart', False)) def test_cancel_returns_message(self): _set_pending_restart(['something']) data = json.loads(self.client.delete('/api/config/pending').data) self.assertIn('message', data) def test_cancel_idempotent_when_nothing_pending(self): r = self.client.delete('/api/config/pending') self.assertEqual(r.status_code, 200) def test_get_after_cancel_shows_not_pending(self): _set_pending_restart(['x']) self.client.delete('/api/config/pending') data = json.loads(self.client.get('/api/config/pending').data) self.assertFalse(data['needs_restart']) self.assertEqual(data['changes'], []) class TestPortChangeDetection(unittest.TestCase): """Test that port changes always trigger pending restart, even on first save.""" def setUp(self): app.config['TESTING'] = True self.client = app.test_client() _clear_pending_restart() # Remove any stored service configs so we start clean for key in ('calendar', 'files', 'wireguard', 'network', 'email'): config_manager.configs.pop(key, None) def tearDown(self): _clear_pending_restart() for key in ('calendar', 'files', 'wireguard', 'network', 'email'): config_manager.configs.pop(key, None) def _put_config(self, payload): return self.client.put('/api/config', data=json.dumps(payload), content_type='application/json') def test_calendar_port_first_save_marks_pending(self): """First-time calendar port save should still queue pending restart.""" r = self._put_config({'calendar': {'port': 5233}}) self.assertEqual(r.status_code, 200) p = config_manager.configs.get('_pending_restart', {}) self.assertTrue(p.get('needs_restart'), 'pending restart not set on first calendar port save') self.assertIn('radicale', p.get('containers', [])) def test_files_port_first_save_marks_pending(self): """First-time files (webdav) port save should queue pending restart.""" r = self._put_config({'files': {'port': 8181}}) self.assertEqual(r.status_code, 200) p = config_manager.configs.get('_pending_restart', {}) self.assertTrue(p.get('needs_restart')) self.assertIn('webdav', p.get('containers', [])) def test_files_manager_port_first_save_marks_pending(self): r = self._put_config({'files': {'manager_port': 9090}}) self.assertEqual(r.status_code, 200) p = config_manager.configs.get('_pending_restart', {}) self.assertTrue(p.get('needs_restart')) self.assertIn('filegator', p.get('containers', [])) def test_multiple_service_port_changes_accumulate_containers(self): """Saving two services should accumulate both containers in pending.""" self._put_config({'calendar': {'port': 5233}}) self._put_config({'files': {'port': 8181}}) p = config_manager.configs.get('_pending_restart', {}) self.assertTrue(p.get('needs_restart')) containers = p.get('containers', []) self.assertIn('radicale', containers) self.assertIn('webdav', containers) def test_same_port_as_default_no_pending(self): """Saving the default port value should NOT trigger pending restart.""" r = self._put_config({'calendar': {'port': 5232}}) # 5232 is default self.assertEqual(r.status_code, 200) p = config_manager.configs.get('_pending_restart', {}) self.assertFalse(p.get('needs_restart', False)) if __name__ == '__main__': unittest.main()