diff --git a/tests/test_peer_route_via.py b/tests/test_peer_route_via.py index b8884ce..e7e7804 100644 --- a/tests/test_peer_route_via.py +++ b/tests/test_peer_route_via.py @@ -156,5 +156,47 @@ class TestSetPeerRouteVia(unittest.TestCase): 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()