roof
d54844cd44
A3 — Peer add atomicity: track firewall_applied flag and call
clear_peer_rules() during rollback so partial peer-add failures
don't leave stale iptables rules behind. Added test.
A2 — Pending config flag: instead of clearing before spawning the
helper container (fire-and-forget), set applying=True and let the
helper clear it on success by writing to cell_config.json via a
mounted /app/data volume. On API restart after a failed apply,
_recover_pending_apply() resets the applying flag so the UI shows
pending changes and the user can retry. GET /api/config/pending now
includes the applying field.
A5 (foundation) — Extract all manager instantiation into managers.py.
app.py re-exports every name so existing test patches (patch('app.X'))
continue to work unchanged. 1021 unit tests pass.
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
195 lines
7.6 KiB
Python
195 lines
7.6 KiB
Python
#!/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()
|