From 2fa181ffbc293a0097c04ab921853a7b9bdb34b0 Mon Sep 17 00:00:00 2001 From: jokob-sk Date: Wed, 20 Aug 2025 08:40:14 +1000 Subject: [PATCH] api layer v0.2.4 - /nettools endpoint --- server/api_server/api_server_start.py | 21 ++++++- server/api_server/devices_endpoint.py | 14 +++++ server/api_server/nettools_endpoint.py | 21 +++++++ test/test_devices_endpoints.py | 14 +++++ test/test_nettools_endpoints.py | 76 ++++++++++++++++++++++++++ 5 files changed, 145 insertions(+), 1 deletion(-) create mode 100755 server/api_server/nettools_endpoint.py create mode 100755 test/test_nettools_endpoints.py diff --git a/server/api_server/api_server_start.py b/server/api_server/api_server_start.py index d81f20b8..4e5727ee 100755 --- a/server/api_server/api_server_start.py +++ b/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 # -------------------------- diff --git a/server/api_server/devices_endpoint.py b/server/api_server/devices_endpoint.py index ca86a346..eb1960a4 100755 --- a/server/api_server/devices_endpoint.py +++ b/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. diff --git a/server/api_server/nettools_endpoint.py b/server/api_server/nettools_endpoint.py new file mode 100755 index 00000000..c5c49bf4 --- /dev/null +++ b/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 + diff --git a/test/test_devices_endpoints.py b/test/test_devices_endpoints.py index 2409bb8e..1b84ecec 100755 --- a/test/test_devices_endpoints.py +++ b/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) diff --git a/test/test_nettools_endpoints.py b/test/test_nettools_endpoints.py new file mode 100755 index 00000000..3225e50d --- /dev/null +++ b/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", "") + + + \ No newline at end of file