From 67362349d12387ce378f22cadae8f50ba1769d0f Mon Sep 17 00:00:00 2001 From: Dmitrii Iurco Date: Mon, 4 May 2026 04:24:02 -0400 Subject: [PATCH] test: add loop detection tests for PUT /api/peers//route-via MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 3 new tests in TestSetPeerRouteVia: - 409 when remote_exit_relay_active=True (would create A→B→A cycle) - disable (via_cell=null) bypasses loop check — always allowed - no 409 when remote_exit_relay_active=False (safe to enable) Co-Authored-By: Claude Sonnet 4.6 --- tests/test_peer_route_via.py | 42 ++++++++++++++++++++++++++++++++++++ 1 file changed, 42 insertions(+) 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()