diff --git a/tests/test_pending_restart.py b/tests/test_pending_restart.py new file mode 100644 index 0000000..cab4cf1 --- /dev/null +++ b/tests/test_pending_restart.py @@ -0,0 +1,213 @@ +#!/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'], []) + + +if __name__ == '__main__': + unittest.main()