#!/usr/bin/env python3 """ Tests for PUT /api/peers//route-via. Covers: - missing via_cell field → 400 - peer not found → 404 - connected cell not found → 404 - enable route-via: calls wg methods and set_exit_relay_active(True) - disable route-via: clears wg methods and set_exit_relay_active(False) - switching from one exit cell to another removes old routing first """ 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 _PEER = { 'peer': 'alice', 'ip': '10.0.0.5', 'public_key': 'alicepubkey=', 'route_via': None, 'internet_access': True, 'service_access': [], } _CELL_LINK = { 'cell_name': 'exit-cell', 'public_key': 'exitcellpubkey=', 'vpn_subnet': '10.1.0.0/24', 'dns_ip': '10.1.0.1', 'remote_exit_relay_active': False, } class TestSetPeerRouteVia(unittest.TestCase): def setUp(self): app.config['TESTING'] = True self.client = app.test_client() def _put(self, peer_name, body): return self.client.put( f'/api/peers/{peer_name}/route-via', data=json.dumps(body), content_type='application/json', ) @patch('app.cell_link_manager') @patch('app.wireguard_manager') @patch('app.peer_registry') def test_missing_via_cell_returns_400(self, mock_reg, mock_wg, mock_clm): r = self._put('alice', {}) self.assertEqual(r.status_code, 400) data = json.loads(r.data) self.assertIn('via_cell', data['error']) @patch('app.cell_link_manager') @patch('app.wireguard_manager') @patch('app.peer_registry') def test_peer_not_found_returns_404(self, mock_reg, mock_wg, mock_clm): mock_reg.get_peer.return_value = None r = self._put('nobody', {'via_cell': 'exit-cell'}) self.assertEqual(r.status_code, 404) @patch('app.cell_link_manager') @patch('app.wireguard_manager') @patch('app.peer_registry') def test_unknown_cell_returns_404(self, mock_reg, mock_wg, mock_clm): mock_reg.get_peer.return_value = _PEER mock_clm.list_connections.return_value = [] r = self._put('alice', {'via_cell': 'no-such-cell'}) self.assertEqual(r.status_code, 404) @patch('app.cell_link_manager') @patch('app.wireguard_manager') @patch('app.peer_registry') def test_enable_route_via_calls_wg_methods(self, mock_reg, mock_wg, mock_clm): mock_reg.get_peer.return_value = _PEER mock_reg.set_route_via.return_value = dict(_PEER, route_via='exit-cell') mock_clm.list_connections.return_value = [_CELL_LINK] r = self._put('alice', {'via_cell': 'exit-cell'}) self.assertEqual(r.status_code, 200) mock_wg.update_cell_peer_allowed_ips.assert_called_once_with( 'exitcellpubkey=', '10.1.0.0/24', add_default_route=True) mock_wg.apply_peer_route_via.assert_called_once_with( '10.0.0.5', via_wg_ip='10.1.0.1') @patch('app.cell_link_manager') @patch('app.wireguard_manager') @patch('app.peer_registry') def test_enable_route_via_signals_exit_relay_active(self, mock_reg, mock_wg, mock_clm): mock_reg.get_peer.return_value = _PEER mock_reg.set_route_via.return_value = dict(_PEER, route_via='exit-cell') mock_clm.list_connections.return_value = [_CELL_LINK] self._put('alice', {'via_cell': 'exit-cell'}) mock_clm.set_exit_relay_active.assert_called_once_with('exit-cell', True) @patch('app.cell_link_manager') @patch('app.wireguard_manager') @patch('app.peer_registry') def test_disable_route_via_clears_old_routing(self, mock_reg, mock_wg, mock_clm): peer_with_via = dict(_PEER, route_via='exit-cell') mock_reg.get_peer.return_value = peer_with_via mock_reg.set_route_via.return_value = dict(_PEER, route_via=None) mock_clm.list_connections.return_value = [_CELL_LINK] r = self._put('alice', {'via_cell': None}) self.assertEqual(r.status_code, 200) mock_wg.update_cell_peer_allowed_ips.assert_called_once_with( 'exitcellpubkey=', '10.1.0.0/24', add_default_route=False) mock_wg.remove_peer_route_via.assert_called_once_with('10.0.0.5') mock_clm.set_exit_relay_active.assert_called_once_with('exit-cell', False) @patch('app.cell_link_manager') @patch('app.wireguard_manager') @patch('app.peer_registry') def test_enable_route_via_persists_field(self, mock_reg, mock_wg, mock_clm): mock_reg.get_peer.return_value = _PEER updated = dict(_PEER, route_via='exit-cell') mock_reg.set_route_via.return_value = updated mock_clm.list_connections.return_value = [_CELL_LINK] r = self._put('alice', {'via_cell': 'exit-cell'}) self.assertEqual(r.status_code, 200) mock_reg.set_route_via.assert_called_once_with('alice', 'exit-cell') data = json.loads(r.data) self.assertEqual(data['peer']['route_via'], 'exit-cell') @patch('app.cell_link_manager') @patch('app.wireguard_manager') @patch('app.peer_registry') def test_switch_exit_cell_removes_old_before_adding_new(self, mock_reg, mock_wg, mock_clm): old_link = dict(_CELL_LINK, cell_name='old-exit', public_key='oldpubkey=', vpn_subnet='10.2.0.0/24', dns_ip='10.2.0.1') new_link = dict(_CELL_LINK, cell_name='new-exit', public_key='newpubkey=') peer_with_old = dict(_PEER, route_via='old-exit') mock_reg.get_peer.return_value = peer_with_old mock_reg.set_route_via.return_value = dict(_PEER, route_via='new-exit') mock_clm.list_connections.return_value = [old_link, new_link] self._put('alice', {'via_cell': 'new-exit'}) calls = mock_wg.update_cell_peer_allowed_ips.call_args_list # First call removes old (add_default_route=False), second adds new (True) self.assertEqual(len(calls), 2) self.assertFalse(calls[0].kwargs['add_default_route']) self.assertTrue(calls[1].kwargs['add_default_route']) # set_exit_relay_active called twice: False for old, True for new relay_calls = mock_clm.set_exit_relay_active.call_args_list self.assertEqual(len(relay_calls), 2) self.assertEqual(relay_calls[0].args, ('old-exit', False)) self.assertEqual(relay_calls[1].args, ('new-exit', True)) @patch('app.cell_link_manager') @patch('app.wireguard_manager') @patch('app.peer_registry') def test_loop_detection_returns_409(self, mock_reg, mock_wg, mock_clm): """If the target cell already routes peers through us (remote_exit_relay_active=True), enabling route-via would create an A→B→A loop — expect 409.""" loop_link = dict(_CELL_LINK, remote_exit_relay_active=True) mock_reg.get_peer.return_value = _PEER mock_clm.list_connections.return_value = [loop_link] r = self._put('alice', {'via_cell': 'exit-cell'}) self.assertEqual(r.status_code, 409) data = json.loads(r.data) self.assertIn('loop', data['error'].lower()) mock_wg.update_cell_peer_allowed_ips.assert_not_called() @patch('app.cell_link_manager') @patch('app.wireguard_manager') @patch('app.peer_registry') def test_loop_detection_skipped_when_disabling(self, mock_reg, mock_wg, mock_clm): """Disabling route-via (via_cell=null) must NOT be blocked by the loop check.""" loop_link = dict(_CELL_LINK, remote_exit_relay_active=True) peer_with_via = dict(_PEER, route_via='exit-cell') mock_reg.get_peer.return_value = peer_with_via mock_reg.set_route_via.return_value = dict(_PEER, route_via=None) mock_clm.list_connections.return_value = [loop_link] r = self._put('alice', {'via_cell': None}) self.assertEqual(r.status_code, 200) @patch('app.cell_link_manager') @patch('app.wireguard_manager') @patch('app.peer_registry') def test_no_loop_when_remote_exit_relay_false(self, mock_reg, mock_wg, mock_clm): """remote_exit_relay_active=False means no loop — route-via should succeed.""" safe_link = dict(_CELL_LINK, remote_exit_relay_active=False) mock_reg.get_peer.return_value = _PEER mock_reg.set_route_via.return_value = dict(_PEER, route_via='exit-cell') mock_clm.list_connections.return_value = [safe_link] mock_clm.set_exit_relay_active.return_value = safe_link r = self._put('alice', {'via_cell': 'exit-cell'}) self.assertEqual(r.status_code, 200) if __name__ == '__main__': unittest.main()