api layer v0.2.4 - /nettools endpoint

server/api_server/api_server_start.py

@@ -3,10 +3,11 @@ from flask import Flask, request, jsonify, Response
 from flask_cors import CORS
 from .graphql_endpoint import devicesSchema
 from .device_endpoint import get_device_data, set_device_data, delete_device, delete_device_events, reset_device_props, copy_device, update_device_column
-from .devices_endpoint import delete_unknown_devices, delete_all_with_empty_macs, delete_devices, export_devices, import_csv, devices_totals, devices_by_status
+from .devices_endpoint import get_all_devices, delete_unknown_devices, delete_all_with_empty_macs, delete_devices, export_devices, import_csv, devices_totals, devices_by_status
 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
 from .sync_endpoint import handle_sync_post, handle_sync_get
 import sys
 
@@ -28,6 +29,7 @@ CORS(
         r"/device/*": {"origins": "*"},
         r"/devices/*": {"origins": "*"},
         r"/history/*": {"origins": "*"},
+        r"/nettools/*": {"origins": "*"},
         r"/events/*": {"origins": "*"}
     },
     supports_credentials=True,
@@ -129,6 +131,12 @@ def api_update_device_column(mac):
 # Devices Collections
 # --------------------------
 
+@app.route("/devices", methods=["GET"])
+def api_get_devices():
+    if not is_authorized():
+        return jsonify({"error": "Forbidden"}), 403
+    return get_all_devices()
+
 @app.route("/devices", methods=["DELETE"])
 def api_delete_devices():
     if not is_authorized():
@@ -181,6 +189,17 @@ def api_devices_by_status():
 
     return devices_by_status(status)
 
+# --------------------------
+# Net tools
+# --------------------------
+@app.route("/nettools/wakeonlan", methods=["POST"])
+def api_wakeonlan():
+    if not is_authorized():
+        return jsonify({"error": "Forbidden"}), 403
+
+    mac = request.json.get("devMac")
+    return wakeonlan(mac)
+
 # --------------------------
 # Online history
 # --------------------------

server/api_server/devices_endpoint.py

@@ -27,6 +27,20 @@ from db.db_helper import get_table_json, get_device_condition_by_status
 # Device Endpoints Functions
 # --------------------------
 
+def get_all_devices():
+    """Retrieve all devices from the database."""
+    conn = get_temp_db_connection()
+    cur = conn.cursor()
+    cur.execute("SELECT * FROM Devices")
+    rows = cur.fetchall()
+
+    # Convert rows to list of dicts using column names
+    columns = [col[0] for col in cur.description]
+    devices = [dict(zip(columns, row)) for row in rows]
+
+    conn.close()
+    return jsonify({"success": True, "devices": devices})
+
 def delete_devices(macs):
     """
     Delete devices from the Devices table.

server/api_server/nettools_endpoint.py

@@ -0,0 +1,21 @@
+import subprocess
+import re
+from flask import jsonify
+
+def wakeonlan(mac):  
+
+    # Validate MAC
+    if not re.match(r'^([0-9A-Fa-f]{2}[:-]){5}([0-9A-Fa-f]{2})$', mac):
+        return jsonify({"success": False, "error": f"Invalid MAC: {mac}"}), 400
+
+    try:
+        result = subprocess.run(
+            ["wakeonlan", mac],
+            capture_output=True,
+            text=True,
+            check=True
+        )
+        return jsonify({"success": True, "message": "WOL packet sent", "output": result.stdout.strip()})
+    except subprocess.CalledProcessError as e:
+        return jsonify({"success": False, "error": "Failed to send WOL packet", "details": e.stderr.strip()}), 500
+

test/test_devices_endpoints.py

@@ -41,6 +41,20 @@ def create_dummy(client, api_token, test_mac):
     }
     resp = client.post(f"/device/{test_mac}", json=payload, headers=auth_headers(api_token))
 
+def test_get_all_devices(client, api_token, test_mac):
+    # Ensure there is at least one device
+    create_dummy(client, api_token, test_mac)
+    
+    # Fetch all devices
+    resp = client.get("/devices", headers=auth_headers(api_token))
+    assert resp.status_code == 200
+    assert resp.json.get("success") is True
+    devices = resp.json.get("devices")
+    assert isinstance(devices, list)
+    # Ensure our test device is in the list
+    assert any(d["devMac"] == test_mac for d in devices)
+
+
 def test_delete_devices_with_macs(client, api_token, test_mac):
     # First create device so it exists
     create_dummy(client, api_token, test_mac)

test/test_nettools_endpoints.py

@@ -0,0 +1,76 @@
+import sys
+import pathlib
+import sqlite3
+import base64
+import random
+import string
+import uuid
+import pytest
+
+INSTALL_PATH = "/app"
+sys.path.extend([f"{INSTALL_PATH}/front/plugins", f"{INSTALL_PATH}/server"])
+
+from helper import timeNowTZ, get_setting_value
+from api_server.api_server_start import app
+
+@pytest.fixture(scope="session")
+def api_token():
+    return get_setting_value("API_TOKEN")
+
+@pytest.fixture
+def client():
+    with app.test_client() as client:
+        yield client
+
+@pytest.fixture
+def test_mac():
+    # Generate a unique MAC for each test run
+    return "AA:BB:CC:" + ":".join(f"{random.randint(0,255):02X}" for _ in range(3))
+
+def auth_headers(token):
+    return {"Authorization": f"Bearer {token}"}
+
+
+def create_dummy(client, api_token, test_mac):
+    payload = {
+        "createNew": True,
+        "devName": "Test Device",
+        "devOwner": "Unit Test",
+        "devType": "Router",
+        "devVendor": "TestVendor",
+    }
+    resp = client.post(f"/device/{test_mac}", json=payload, headers=auth_headers(api_token))
+
+def test_wakeonlan_device(client, api_token, test_mac):
+    # 1. Ensure at least one device exists
+    create_dummy(client, api_token, test_mac)
+
+    # 2. Fetch all devices
+    resp = client.get("/devices", headers=auth_headers(api_token))
+    assert resp.status_code == 200
+    devices = resp.json.get("devices", [])
+    assert len(devices) > 0
+
+    # 3. Pick the first device (or the test device)
+    device_mac = devices[0]["devMac"]
+
+    # 4. Call the wakeonlan endpoint
+    resp = client.post(
+        "/nettools/wakeonlan",
+        json={"devMac": device_mac},
+        headers=auth_headers(api_token)
+    )
+
+    # 5. Conditional assertions based on MAC
+    if device_mac.lower() == 'internet' or device_mac == test_mac:
+        # For athe dummy "internet" or test MAC, expect a 400 response
+        assert resp.status_code == 400
+    else:
+        # For any other MAC, expect a 200 response
+        assert resp.status_code == 200
+        data = resp.json
+        assert data.get("success") is True
+        assert "WOL packet sent" in data.get("message", "")
+        
+   
+