roof/pic
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: diagnostics tab — run ping/traceroute in cell-wireguard, fix wrong method call

Browse Source 
The connectivity endpoint was calling routing_manager.test_connectivity()
(no args, internal health check) instead of test_routing_connectivity(target_ip).
Also ping/traceroute aren't installed in the API container; run them via
docker exec cell-wireguard instead.

Updated test_api_endpoints to mock test_routing_connectivity and cover
the new DELETE /firewall/<id> and GET /live-iptables endpoints.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
Dmitrii Iurco
2026-04-21 01:26:40 -04:00
parent 901094f60a
commit 4bf583c071
3 changed files with 55 additions and 66 deletions
Download Patch File Download Diff File Expand all files Collapse all files
api/app.py
+4 -2
View File
@@ -1638,8 +1638,10 @@ def get_live_iptables():
def test_routing_connectivity():
"""Test routing connectivity."""
try:
data = request.get_json(silent=True)
result = routing_manager.test_connectivity(data)
data = request.get_json(silent=True) or {}
target_ip = data.get('target_ip', '8.8.8.8')
via_peer = data.get('via_peer')
result = routing_manager.test_routing_connectivity(target_ip, via_peer)
return jsonify(result)
except Exception as e:
logger.error(f"Error testing routing connectivity: {e}")
api/routing_manager.py
+29 -58
View File
@@ -388,68 +388,39 @@ class RoutingManager(BaseServiceManager):
}
def test_routing_connectivity(self, target_ip: str, via_peer: str = None) -> Dict:
"""Test routing connectivity"""
try:
results = {}
"""Test routing connectivity by running ping/traceroute in cell-wireguard."""
WG = 'cell-wireguard'
# Test basic connectivity
try:
result = subprocess.run(['ping', '-c', '3', '-W', '5', target_ip],
capture_output=True, text=True, timeout=30)
results['ping'] = {
'success': result.returncode == 0,
'output': result.stdout,
'error': result.stderr
}
except Exception as e:
results['ping'] = {
'success': False,
'output': '',
'error': str(e)
}
# Test traceroute
try:
result = subprocess.run(['traceroute', '-m', '10', target_ip],
capture_output=True, text=True, timeout=30)
results['traceroute'] = {
'success': result.returncode == 0,
'output': result.stdout,
'error': result.stderr
}
except Exception as e:
results['traceroute'] = {
'success': False,
'output': '',
'error': str(e)
}
# Test specific route if via_peer is specified
if via_peer:
try:
# Test route through specific peer
result = subprocess.run(['ping', '-c', '3', '-W', '5', '-I', via_peer, target_ip],
capture_output=True, text=True, timeout=30)
results['peer_route'] = {
'success': result.returncode == 0,
'output': result.stdout,
'error': result.stderr
}
except Exception as e:
results['peer_route'] = {
'success': False,
'output': '',
'error': str(e)
}
return results
except Exception as e:
def _exec(cmd):
result = subprocess.run(
['docker', 'exec', WG] + cmd,
capture_output=True, text=True, timeout=35
)
return {
'ping': {'success': False, 'output': '', 'error': str(e)},
'traceroute': {'success': False, 'output': '', 'error': str(e)}
'success': result.returncode == 0,
'output': result.stdout,
'error': result.stderr,
}
results = {}
try:
results['ping'] = _exec(['ping', '-c', '4', '-W', '3', target_ip])
except Exception as e:
results['ping'] = {'success': False, 'output': '', 'error': str(e)}
try:
results['traceroute'] = _exec(['traceroute', '-m', '10', '-w', '2', target_ip])
except Exception as e:
results['traceroute'] = {'success': False, 'output': '', 'error': str(e)}
if via_peer:
try:
results['peer_route'] = _exec(['ping', '-c', '3', '-W', '3', '-I', via_peer, target_ip])
except Exception as e:
results['peer_route'] = {'success': False, 'output': '', 'error': str(e)}
return results
def get_routing_logs(self, lines: int = 50) -> Dict:
"""Get routing and firewall logs"""
try:
tests/test_api_endpoints.py
+21 -5
View File
@@ -531,7 +531,9 @@ class TestAPIEndpoints(unittest.TestCase):
mock_routing.add_exit_node.return_value = {'result': 'ok'}
mock_routing.add_bridge_route.return_value = {'result': 'ok'}
mock_routing.add_split_route.return_value = {'result': 'ok'}
mock_routing.test_connectivity.return_value = {'success': True}
mock_routing.test_routing_connectivity.return_value = {'ping': {'success': True, 'output': '', 'error': ''}}
mock_routing.remove_firewall_rule.return_value = True
mock_routing.get_live_iptables.return_value = {'filter': '', 'nat': ''}
mock_routing.get_routing_logs.return_value = {'logs': 'logdata'}
# /api/routing/status (GET)
response = self.client.get('/api/routing/status')
@@ -618,12 +620,26 @@ class TestAPIEndpoints(unittest.TestCase):
self.assertEqual(response.status_code, 500)
mock_routing.add_split_route.side_effect = None
# /api/routing/connectivity (POST)
response = self.client.post('/api/routing/connectivity', data=json.dumps({'target': '10.0.0.2'}), content_type='application/json')
response = self.client.post('/api/routing/connectivity', data=json.dumps({'target_ip': '8.8.8.8'}), content_type='application/json')
self.assertEqual(response.status_code, 200)
mock_routing.test_connectivity.side_effect = Exception('fail')
response = self.client.post('/api/routing/connectivity', data=json.dumps({'target': '10.0.0.2'}), content_type='application/json')
mock_routing.test_routing_connectivity.side_effect = Exception('fail')
response = self.client.post('/api/routing/connectivity', data=json.dumps({'target_ip': '8.8.8.8'}), content_type='application/json')
self.assertEqual(response.status_code, 500)
mock_routing.test_connectivity.side_effect = None
mock_routing.test_routing_connectivity.side_effect = None
# /api/routing/firewall/<id> (DELETE)
response = self.client.delete('/api/routing/firewall/fw_1')
self.assertEqual(response.status_code, 200)
mock_routing.remove_firewall_rule.return_value = False
response = self.client.delete('/api/routing/firewall/fw_999')
self.assertEqual(response.status_code, 404)
mock_routing.remove_firewall_rule.return_value = True
# /api/routing/live-iptables (GET)
response = self.client.get('/api/routing/live-iptables')
self.assertEqual(response.status_code, 200)
mock_routing.get_live_iptables.side_effect = Exception('fail')
response = self.client.get('/api/routing/live-iptables')
self.assertEqual(response.status_code, 500)
mock_routing.get_live_iptables.side_effect = None
# /api/routing/logs (GET)
mock_routing.get_logs.return_value = {
'iptables': 'iptables log data',