pic/tests/test_pending_restart.py

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