From 4bf583c0714b889ced519c08c0dc81a6bfccb49e Mon Sep 17 00:00:00 2001 From: Dmitrii Iurco Date: Tue, 21 Apr 2026 01:26:40 -0400 Subject: [PATCH] =?UTF-8?q?fix:=20diagnostics=20tab=20=E2=80=94=20run=20pi?= =?UTF-8?q?ng/traceroute=20in=20cell-wireguard,=20fix=20wrong=20method=20c?= =?UTF-8?q?all?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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/ and GET /live-iptables endpoints. Co-Authored-By: Claude Sonnet 4.6 --- api/app.py | 6 ++- api/routing_manager.py | 89 +++++++++++++------------------------ tests/test_api_endpoints.py | 26 ++++++++--- 3 files changed, 55 insertions(+), 66 deletions(-) diff --git a/api/app.py b/api/app.py index 228d0ee..ecfbc4c 100644 --- a/api/app.py +++ b/api/app.py @@ -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}") diff --git a/api/routing_manager.py b/api/routing_manager.py index 5282557..ebdfd81 100644 --- a/api/routing_manager.py +++ b/api/routing_manager.py @@ -388,67 +388,38 @@ class RoutingManager(BaseServiceManager): } def test_routing_connectivity(self, target_ip: str, via_peer: str = None) -> Dict: - """Test routing connectivity""" - try: - results = {} - - # 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: + """Test routing connectivity by running ping/traceroute in cell-wireguard.""" + WG = 'cell-wireguard' + + 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""" diff --git a/tests/test_api_endpoints.py b/tests/test_api_endpoints.py index 1edd1e6..4d8f0be 100644 --- a/tests/test_api_endpoints.py +++ b/tests/test_api_endpoints.py @@ -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/ (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',