#!/usr/bin/env python3 """ Tests for POST /api/config/apply. The route reads _pending_restart from config_manager, spawns a background thread/process, clears the pending flag, and returns 200. We mock subprocess.Popen / subprocess.run and docker.from_env so the tests run without Docker, and we capture what command-line arguments would be used. """ import sys import json import threading import unittest from pathlib import Path from unittest.mock import patch, MagicMock, call api_dir = Path(__file__).parent.parent / 'api' sys.path.insert(0, str(api_dir)) from app import app, _set_pending_restart, _clear_pending_restart, config_manager class TestConfigApplyRoute(unittest.TestCase): def setUp(self): app.config['TESTING'] = True self.client = app.test_client() _clear_pending_restart() def tearDown(self): _clear_pending_restart() # ── No pending changes ───────────────────────────────────────────────── def test_apply_with_no_pending_returns_200(self): r = self.client.post('/api/config/apply') self.assertEqual(r.status_code, 200) def test_apply_with_no_pending_returns_no_changes_message(self): r = self.client.post('/api/config/apply') data = json.loads(r.data) self.assertIn('message', data) self.assertIn('No pending', data['message']) # ── Pending changes present ──────────────────────────────────────────── @patch('subprocess.Popen') @patch('docker.from_env') def test_apply_with_pending_returns_200(self, mock_docker, mock_popen): mock_docker.side_effect = Exception('no docker in test') mock_popen.return_value = MagicMock() _set_pending_restart(['dns_port: 53 → 5353'], ['*']) r = self.client.post('/api/config/apply') self.assertEqual(r.status_code, 200) @patch('subprocess.Popen') @patch('docker.from_env') def test_apply_with_pending_returns_restart_in_progress(self, mock_docker, mock_popen): mock_docker.side_effect = Exception('no docker in test') mock_popen.return_value = MagicMock() _set_pending_restart(['something changed'], ['*']) r = self.client.post('/api/config/apply') data = json.loads(r.data) self.assertTrue(data.get('restart_in_progress')) # ── Pending state marked "applying" after apply (not immediately cleared) ─ @patch('threading.Thread') @patch('docker.from_env') def test_apply_sets_applying_flag(self, mock_docker, mock_thread): mock_docker.side_effect = Exception('no docker in test') # Don't actually start the thread so we don't need subprocess mock_thread.return_value = MagicMock() _set_pending_restart(['config changed'], ['*']) self.client.post('/api/config/apply') pending = config_manager.configs.get('_pending_restart', {}) # The route now marks needs_restart=True + applying=True instead of clearing # immediately. The helper container clears the flag on success; if the helper # fails, needs_restart stays set so the UI continues showing pending changes. self.assertTrue(pending.get('needs_restart', False)) self.assertTrue(pending.get('applying', False)) # ── needs_network_recreate=True → helper script includes 'down' ──────── @patch('subprocess.Popen') @patch('docker.from_env') def test_apply_network_recreate_spawns_popen_with_down_command( self, mock_docker, mock_popen): mock_docker.side_effect = Exception('no docker in test') mock_popen.return_value = MagicMock() # Set up a wildcard pending change that also requires network recreation _set_pending_restart(['ip_range changed'], ['*']) config_manager.configs['_pending_restart']['network_recreate'] = True r = self.client.post('/api/config/apply') self.assertEqual(r.status_code, 200) # Wait for background thread to call Popen import time for _ in range(20): if mock_popen.called: break time.sleep(0.1) self.assertTrue(mock_popen.called, 'Expected subprocess.Popen to be called for wildcard restart') args, kwargs = mock_popen.call_args cmd = args[0] # cmd is the full docker run ... sh -c 'script' script_arg = cmd[-1] # the -c argument self.assertIn('down', script_arg, f'Expected "down" in helper script when network_recreate=True, got: {script_arg}') # ── needs_network_recreate=False → helper script uses only 'up -d' ───── @patch('subprocess.Popen') @patch('docker.from_env') def test_apply_no_network_recreate_spawns_popen_without_down( self, mock_docker, mock_popen): mock_docker.side_effect = Exception('no docker in test') mock_popen.return_value = MagicMock() _set_pending_restart(['port changed'], ['*']) # network_recreate defaults to False self.client.post('/api/config/apply') import time for _ in range(20): if mock_popen.called: break time.sleep(0.1) self.assertTrue(mock_popen.called) args, _ = mock_popen.call_args script_arg = args[0][-1] self.assertNotIn(' down', script_arg, f'Did not expect "down" in helper script when network_recreate=False') self.assertIn('up -d', script_arg) # ── Specific containers (not wildcard) ───────────────────────────────── @patch('subprocess.run') @patch('docker.from_env') def test_apply_specific_containers_uses_subprocess_run( self, mock_docker, mock_run): mock_docker.side_effect = Exception('no docker in test') mock_run.return_value = MagicMock(returncode=0, stderr='') _set_pending_restart(['dns port changed'], ['dns']) r = self.client.post('/api/config/apply') self.assertEqual(r.status_code, 200) # Give the daemon thread a moment to call subprocess.run import time for _ in range(30): # Look for the compose call specifically (may not be the last call) compose_calls = [ c for c in mock_run.call_args_list if 'compose' in (c.args[0] if c.args else []) ] if compose_calls: break time.sleep(0.1) compose_calls = [ c for c in mock_run.call_args_list if c.args and 'compose' in c.args[0] ] self.assertTrue( len(compose_calls) > 0, f'Expected a subprocess.run call containing "compose"; got calls: {mock_run.call_args_list}' ) cmd = compose_calls[-1].args[0] self.assertIn('up', cmd) self.assertIn('-d', cmd) self.assertIn('dns', cmd) # ── Exception in route body returns 500 ─────────────────────────────── @patch('app.config_manager') def test_apply_returns_500_on_unexpected_exception(self, mock_cm): mock_cm.configs = MagicMock() mock_cm.configs.get.side_effect = Exception('unexpected failure') r = self.client.post('/api/config/apply') self.assertEqual(r.status_code, 500) self.assertIn('error', json.loads(r.data)) if __name__ == '__main__': unittest.main()