diff --git a/server/api_server/api_server_start.py b/server/api_server/api_server_start.py index 104aebe7..44d970ac 100755 --- a/server/api_server/api_server_start.py +++ b/server/api_server/api_server_start.py @@ -7,7 +7,7 @@ from .devices_endpoint import get_all_devices, delete_unknown_devices, delete_al from .events_endpoint import delete_events, delete_events_30, get_events from .history_endpoint import delete_online_history from .prometheus_endpoint import getMetricStats -from .nettools_endpoint import wakeonlan, traceroute +from .nettools_endpoint import wakeonlan, traceroute, speedtest from .sync_endpoint import handle_sync_post, handle_sync_get import sys @@ -207,6 +207,12 @@ def api_traceroute(): ip = request.json.get("devLastIP") return traceroute(ip) +@app.route("/nettools/speedtest", methods=["GET"]) +def api_speedtest(): + if not is_authorized(): + return jsonify({"error": "Forbidden"}), 403 + return speedtest() + # -------------------------- # Online history # -------------------------- diff --git a/server/api_server/nettools_endpoint.py b/server/api_server/nettools_endpoint.py index a044cfa2..00f3b6cd 100755 --- a/server/api_server/nettools_endpoint.py +++ b/server/api_server/nettools_endpoint.py @@ -1,8 +1,13 @@ import subprocess import re +import sys import ipaddress from flask import jsonify +# Register NetAlertX directories +INSTALL_PATH = "/app" +sys.path.extend([f"{INSTALL_PATH}/front/plugins", f"{INSTALL_PATH}/server"]) + def wakeonlan(mac): # Validate MAC @@ -65,3 +70,29 @@ def traceroute(ip): "error": "Traceroute failed", "details": e.stderr.strip() }), 500 + + +def speedtest(): + """ + API endpoint to run a speedtest using speedtest-cli. + Returns JSON with the test output or error. + """ + try: + # Run speedtest-cli command + result = subprocess.run( + [f"{INSTALL_PATH}/back/speedtest-cli", "--secure", "--simple"], + capture_output=True, + text=True, + check=True + ) + + # Return each line as a list + output_lines = result.stdout.strip().split("\n") + return jsonify({"success": True, "output": output_lines}) + + except subprocess.CalledProcessError as e: + return jsonify({ + "success": False, + "error": "Speedtest failed", + "details": e.stderr.strip() + }), 500 \ No newline at end of file diff --git a/test/test_nettools_endpoints.py b/test/test_nettools_endpoints.py index 8fbefa25..1ed50928 100755 --- a/test/test_nettools_endpoints.py +++ b/test/test_nettools_endpoints.py @@ -71,6 +71,26 @@ def test_wakeonlan_device(client, api_token, test_mac): data = resp.json assert data.get("success") is True assert "WOL packet sent" in data.get("message", "") + +def test_speedtest_endpoint(client, api_token): + # 1. Call the speedtest endpoint + resp = client.get("/nettools/speedtest", headers=auth_headers(api_token)) + + # 2. Assertions + if resp.status_code == 403: + # Unauthorized access + data = resp.json + assert "error" in data + assert data["error"] == "Forbidden" + else: + # Expect success + assert resp.status_code == 200 + data = resp.json + assert data.get("success") is True + assert "output" in data + assert isinstance(data["output"], list) + # Optionally check that output lines are strings + assert all(isinstance(line, str) for line in data["output"]) def test_traceroute_device(client, api_token, test_mac): # 1. Ensure at least one device exists @@ -106,4 +126,4 @@ def test_traceroute_device(client, api_token, test_mac): assert "output" in data assert isinstance(data["output"], str) - \ No newline at end of file +