api layer v0.2.4 - /nettools/speedtest endpoint

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
 # --------------------------

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

test/test_nettools_endpoints.py

@@ -72,6 +72,26 @@ def test_wakeonlan_device(client, api_token, test_mac):
         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
     create_dummy(client, api_token, test_mac)