#!/usr/bin/env python3 """ Unit tests for WireGuard-specific Flask endpoints in api/app.py. Covers routes that were not already tested in test_api_endpoints.py: POST /api/wireguard/check-port GET /api/wireguard/server-config POST /api/wireguard/refresh-ip GET /api/wireguard/peers/statuses POST /api/wireguard/apply-enforcement POST /api/wireguard/network/setup GET /api/wireguard/network/status """ import sys import json import unittest 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 class TestWireGuardEndpoints(unittest.TestCase): def setUp(self): app.config['TESTING'] = True self.client = app.test_client() # ── POST /api/wireguard/check-port ───────────────────────────────────── @patch('app.wireguard_manager') def test_check_port_returns_port_open_true(self, mock_wg): mock_wg.check_port_open.return_value = True mock_wg._get_configured_port.return_value = 51820 r = self.client.post('/api/wireguard/check-port') self.assertEqual(r.status_code, 200) data = json.loads(r.data) self.assertIn('port_open', data) self.assertIn('port', data) self.assertTrue(data['port_open']) self.assertEqual(data['port'], 51820) @patch('app.wireguard_manager') def test_check_port_returns_port_open_false(self, mock_wg): mock_wg.check_port_open.return_value = False mock_wg._get_configured_port.return_value = 51820 r = self.client.post('/api/wireguard/check-port') self.assertEqual(r.status_code, 200) data = json.loads(r.data) self.assertFalse(data['port_open']) @patch('app.wireguard_manager') def test_check_port_returns_500_on_exception(self, mock_wg): mock_wg.check_port_open.side_effect = Exception('socket error') r = self.client.post('/api/wireguard/check-port') self.assertEqual(r.status_code, 500) self.assertIn('error', json.loads(r.data)) # ── GET /api/wireguard/server-config ─────────────────────────────────── @patch('app.wireguard_manager') def test_server_config_returns_config_dict(self, mock_wg): mock_wg.get_server_config.return_value = { 'public_key': 'PUBKEY==', 'endpoint': '1.2.3.4:51820', 'port': 51820, } r = self.client.get('/api/wireguard/server-config') self.assertEqual(r.status_code, 200) data = json.loads(r.data) self.assertIn('public_key', data) self.assertIn('endpoint', data) @patch('app.wireguard_manager') def test_server_config_returns_500_on_exception(self, mock_wg): mock_wg.get_server_config.side_effect = RuntimeError('wg not running') r = self.client.get('/api/wireguard/server-config') self.assertEqual(r.status_code, 500) self.assertIn('error', json.loads(r.data)) # ── POST /api/wireguard/refresh-ip ───────────────────────────────────── @patch('app.wireguard_manager') def test_refresh_ip_returns_external_ip_and_endpoint(self, mock_wg): mock_wg.get_external_ip.return_value = '203.0.113.10' mock_wg._get_configured_port.return_value = 51820 r = self.client.post('/api/wireguard/refresh-ip') self.assertEqual(r.status_code, 200) data = json.loads(r.data) self.assertEqual(data['external_ip'], '203.0.113.10') self.assertEqual(data['port'], 51820) self.assertEqual(data['endpoint'], '203.0.113.10:51820') @patch('app.wireguard_manager') def test_refresh_ip_endpoint_is_none_when_ip_unavailable(self, mock_wg): mock_wg.get_external_ip.return_value = None mock_wg._get_configured_port.return_value = 51820 r = self.client.post('/api/wireguard/refresh-ip') self.assertEqual(r.status_code, 200) data = json.loads(r.data) self.assertIsNone(data['endpoint']) @patch('app.wireguard_manager') def test_refresh_ip_passes_force_refresh_true(self, mock_wg): mock_wg.get_external_ip.return_value = '1.2.3.4' mock_wg._get_configured_port.return_value = 51820 self.client.post('/api/wireguard/refresh-ip') mock_wg.get_external_ip.assert_called_once_with(force_refresh=True) @patch('app.wireguard_manager') def test_refresh_ip_returns_500_on_exception(self, mock_wg): mock_wg.get_external_ip.side_effect = Exception('network error') r = self.client.post('/api/wireguard/refresh-ip') self.assertEqual(r.status_code, 500) self.assertIn('error', json.loads(r.data)) # ── GET /api/wireguard/peers/statuses ────────────────────────────────── @patch('app.wireguard_manager') def test_peer_statuses_returns_dict_keyed_by_public_key(self, mock_wg): mock_wg.get_all_peer_statuses.return_value = { 'KEY1==': {'latest_handshake': 1700000000, 'transfer_rx': 1024}, 'KEY2==': {'latest_handshake': 1700000100, 'transfer_rx': 2048}, } r = self.client.get('/api/wireguard/peers/statuses') self.assertEqual(r.status_code, 200) data = json.loads(r.data) self.assertIsInstance(data, dict) self.assertIn('KEY1==', data) self.assertIn('KEY2==', data) @patch('app.wireguard_manager') def test_peer_statuses_returns_empty_dict_when_no_peers(self, mock_wg): mock_wg.get_all_peer_statuses.return_value = {} r = self.client.get('/api/wireguard/peers/statuses') self.assertEqual(r.status_code, 200) self.assertEqual(json.loads(r.data), {}) @patch('app.wireguard_manager') def test_peer_statuses_returns_500_on_exception(self, mock_wg): mock_wg.get_all_peer_statuses.side_effect = Exception('wg show failed') r = self.client.get('/api/wireguard/peers/statuses') self.assertEqual(r.status_code, 500) self.assertIn('error', json.loads(r.data)) # ── POST /api/wireguard/apply-enforcement ────────────────────────────── @patch('app.firewall_manager') @patch('app.peer_registry') def test_apply_enforcement_returns_ok_and_peer_count(self, mock_reg, mock_fw): mock_reg.list_peers.return_value = [ {'name': 'peer1', 'public_key': 'K1=='}, {'name': 'peer2', 'public_key': 'K2=='}, ] mock_fw.apply_all_peer_rules.return_value = None mock_fw.apply_all_dns_rules.return_value = None r = self.client.post('/api/wireguard/apply-enforcement') self.assertEqual(r.status_code, 200) data = json.loads(r.data) self.assertTrue(data['ok']) self.assertEqual(data['peers'], 2) @patch('app.firewall_manager') @patch('app.peer_registry') def test_apply_enforcement_calls_both_rule_functions(self, mock_reg, mock_fw): mock_reg.list_peers.return_value = [] mock_fw.apply_all_peer_rules.return_value = None mock_fw.apply_all_dns_rules.return_value = None self.client.post('/api/wireguard/apply-enforcement') mock_fw.apply_all_peer_rules.assert_called_once() mock_fw.apply_all_dns_rules.assert_called_once() @patch('app.peer_registry') def test_apply_enforcement_returns_500_on_exception(self, mock_reg): mock_reg.list_peers.side_effect = Exception('registry error') r = self.client.post('/api/wireguard/apply-enforcement') self.assertEqual(r.status_code, 500) self.assertIn('error', json.loads(r.data)) # ── POST /api/wireguard/network/setup ────────────────────────────────── @patch('app.wireguard_manager') def test_network_setup_returns_200_on_success(self, mock_wg): mock_wg.setup_network_configuration.return_value = True r = self.client.post('/api/wireguard/network/setup') self.assertEqual(r.status_code, 200) data = json.loads(r.data) self.assertIn('message', data) @patch('app.wireguard_manager') def test_network_setup_returns_500_when_manager_returns_false(self, mock_wg): mock_wg.setup_network_configuration.return_value = False r = self.client.post('/api/wireguard/network/setup') self.assertEqual(r.status_code, 500) self.assertIn('error', json.loads(r.data)) @patch('app.wireguard_manager') def test_network_setup_returns_500_on_exception(self, mock_wg): mock_wg.setup_network_configuration.side_effect = Exception('iptables fail') r = self.client.post('/api/wireguard/network/setup') self.assertEqual(r.status_code, 500) self.assertIn('error', json.loads(r.data)) # ── GET /api/wireguard/network/status ────────────────────────────────── @patch('app.wireguard_manager') def test_network_status_returns_200_with_status_dict(self, mock_wg): mock_wg.get_network_status.return_value = { 'ip_forwarding': True, 'nat_active': True, 'interface': 'wg0', } r = self.client.get('/api/wireguard/network/status') self.assertEqual(r.status_code, 200) data = json.loads(r.data) self.assertIn('ip_forwarding', data) @patch('app.wireguard_manager') def test_network_status_returns_500_on_exception(self, mock_wg): mock_wg.get_network_status.side_effect = Exception('iproute error') r = self.client.get('/api/wireguard/network/status') self.assertEqual(r.status_code, 500) self.assertIn('error', json.loads(r.data)) class TestWireGuardPortPropagation(unittest.TestCase): """ Test that changing wireguard_port via the identity config path calls wireguard_manager.apply_config (writes wg0.conf), not just update_config (which only saves a JSON file and never touches wg0.conf). """ def setUp(self): app.config['TESTING'] = True self.client = app.test_client() @patch('routes.config._set_pending_restart') @patch('app.wireguard_manager') @patch('app.config_manager') def test_wireguard_port_identity_change_calls_apply_config( self, mock_cm, mock_wg, mock_pending ): """wireguard_port in identity update must call apply_config, not just update_config.""" mock_cm.configs = { '_identity': {'wireguard_port': 51820, 'ip_range': '10.0.0.0/24'}, 'wireguard': {'port': 51820}, } mock_cm.service_schemas = {} mock_cm.update_service_config.return_value = None mock_cm._save_all_configs.return_value = None mock_wg.apply_config.return_value = {'restarted': [], 'warnings': []} mock_pending.return_value = None r = self.client.put( '/api/config', data=json.dumps({'wireguard_port': 51821}), content_type='application/json', ) self.assertEqual(r.status_code, 200) mock_wg.apply_config.assert_called_once_with({'port': 51821}) @patch('routes.config._set_pending_restart') @patch('app.wireguard_manager') @patch('app.config_manager') def test_wireguard_port_same_value_does_not_call_apply_config( self, mock_cm, mock_wg, mock_pending ): """apply_config must NOT be called when the new port equals the current port.""" mock_cm.configs = { '_identity': {'wireguard_port': 51820, 'ip_range': '10.0.0.0/24'}, 'wireguard': {'port': 51820}, } mock_cm.service_schemas = {} mock_cm.update_service_config.return_value = None mock_cm._save_all_configs.return_value = None mock_pending.return_value = None r = self.client.put( '/api/config', data=json.dumps({'wireguard_port': 51820}), content_type='application/json', ) self.assertEqual(r.status_code, 200) mock_wg.apply_config.assert_not_called() class TestApplyPendingConfigForceRecreate(unittest.TestCase): """ POST /api/config/apply for specific containers (not '*') must pass --force-recreate to docker compose so that port-binding changes actually take effect even if Docker's config-hash comparison misses them. The config-hash issue arises from Docker file bind-mounts: the env file inside the container is mounted to a specific inode; if the host file was ever replaced (new inode), the container's bind-mount stays on the old inode and docker compose sees stale values. --force-recreate bypasses the hash comparison entirely. """ def setUp(self): app.config['TESTING'] = True self.client = app.test_client() @patch('routes.config._clear_pending_restart') @patch('app.config_manager') def test_apply_pending_uses_force_recreate(self, mock_cm, mock_clear): """apply_pending_config for specific containers must include --force-recreate.""" mock_cm.configs = { '_pending_restart': { 'needs_restart': True, 'containers': ['wireguard'], 'network_recreate': False, } } captured_target = {} def patched_thread(target=None, daemon=False, **kw): captured_target['fn'] = target t = MagicMock() t.start = lambda: None return t with patch('routes.config.threading.Thread', side_effect=patched_thread): r = self.client.post('/api/config/apply') self.assertEqual(r.status_code, 200) self.assertIn('fn', captured_target) # Execute the captured _do_apply and verify subprocess call includes --force-recreate with patch('subprocess.run') as mock_run, \ patch('time.sleep'): mock_run.return_value = MagicMock(returncode=0, stderr='') captured_target['fn']() call_args = mock_run.call_args self.assertIsNotNone(call_args, 'subprocess.run was not called in _do_apply') cmd = call_args[0][0] self.assertIn('--force-recreate', cmd, f'--force-recreate missing from docker compose command: {cmd}') self.assertIn('wireguard', cmd) @patch('routes.config._clear_pending_restart') @patch('app.config_manager') def test_apply_pending_all_services_no_force_recreate(self, mock_cm, mock_clear): """All-services restart ('*') uses a helper container (Popen), not subprocess.run.""" mock_cm.configs = { '_pending_restart': { 'needs_restart': True, 'containers': ['*'], 'network_recreate': False, } } captured_target = {} def patched_thread(target=None, daemon=False, **kw): captured_target['fn'] = target t = MagicMock() t.start = lambda: None return t with patch('routes.config.threading.Thread', side_effect=patched_thread): r = self.client.post('/api/config/apply') self.assertEqual(r.status_code, 200) self.assertIn('fn', captured_target) # For '*', _do_apply spawns a helper container via Popen, not subprocess.run with patch('subprocess.Popen') as mock_popen, \ patch('subprocess.run') as mock_run: mock_popen.return_value = MagicMock() captured_target['fn']() mock_run.assert_not_called() mock_popen.assert_called_once() if __name__ == '__main__': unittest.main()