test: add loop detection tests for PUT /api/peers/<peer>/route-via
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 <noreply@anthropic.com>
This commit is contained in:
@@ -156,5 +156,47 @@ class TestSetPeerRouteVia(unittest.TestCase):
|
|||||||
self.assertEqual(relay_calls[1].args, ('new-exit', True))
|
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__':
|
if __name__ == '__main__':
|
||||||
unittest.main()
|
unittest.main()
|
||||||
|
|||||||
Reference in New Issue
Block a user