api layer v0.2 - /devices

server/api_server/api_server_start.py

@@ -2,7 +2,8 @@ import threading
 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
+from .device_endpoint import get_device_data, set_device_data, delete_device, delete_device_events, reset_device_props
+from .devices_endpoint import delete_unknown_devices, delete_all_with_empty_macs, delete_devices
 from .prometheus_endpoint import getMetricStats
 from .sync_endpoint import handle_sync_post, handle_sync_get
 import sys
@@ -22,7 +23,8 @@ CORS(
     app,
     resources={
         r"/metrics": {"origins": "*"},
-        r"/device/*": {"origins": "*"}
+        r"/device/*": {"origins": "*"},
+        r"/devices/*": {"origins": "*"}
     },
     supports_credentials=True,
     allow_headers=["Authorization", "Content-Type"]
@@ -92,14 +94,17 @@ def api_reset_device_props(mac):
     return reset_device_props(mac, request.json)
 
 # --------------------------
-# Device Collections
+# Devices Collections
 # --------------------------
 
 @app.route("/devices", methods=["DELETE"])
-def api_delete_all_devices():
+def api_delete_devices():
     if not is_authorized():
         return jsonify({"error": "Forbidden"}), 403
-    return delete_all_devices()
+    
+    macs = request.json.get("macs") if request.is_json else None
+
+    return delete_devices(macs)
 
 @app.route("/devices/empty-macs", methods=["DELETE"])
 def api_delete_all_empty_macs():
@@ -119,6 +124,7 @@ def api_get_devices_totals():
         return jsonify({"error": "Forbidden"}), 403
     return get_devices_totals()
 
+
 # --------------------------
 # Device Events / History
 # --------------------------

server/api_server/device_endpoint.py

@@ -261,7 +261,6 @@ def delete_device_events(mac):
 
 def reset_device_props(mac, data=None):
     """Reset device custom properties to default."""
-    from .helpers import get_setting_value
     default_props = get_setting_value("NEWDEV_devCustomProps")
     conn = get_temp_db_connection()
     cur = conn.cursor()

server/api_server/devices_endpoint.py

@@ -0,0 +1,75 @@
+#!/usr/bin/env python
+
+import json
+import subprocess
+import argparse
+import os
+import pathlib
+import sys
+from datetime import datetime
+from flask import jsonify, request
+
+# Register NetAlertX directories
+INSTALL_PATH="/app"
+sys.path.extend([f"{INSTALL_PATH}/front/plugins", f"{INSTALL_PATH}/server"])
+
+from database import get_temp_db_connection
+from helper import row_to_json, get_date_from_period, is_random_mac, format_date, get_setting_value
+
+
+# --------------------------
+# Device Endpoints Functions
+# --------------------------
+
+def delete_devices(macs):
+    """
+    Delete devices from the Devices table.
+    - If `macs` is None → delete ALL devices.
+    - If `macs` is a list → delete only matching MACs (supports wildcard '*').
+    """
+
+    conn = get_temp_db_connection()
+    cur = conn.cursor()
+
+    if not macs:
+        # No MACs provided → delete all
+        cur.execute("DELETE FROM Devices")
+        conn.commit()
+        conn.close()
+        return jsonify({"success": True, "deleted": "all"})
+
+    deleted_count = 0
+
+    for mac in macs:
+        if "*" in mac:
+            # Wildcard matching
+            sql_pattern = mac.replace("*", "%")
+            cur.execute("DELETE FROM Devices WHERE devMAC LIKE ?", (sql_pattern,))
+        else:
+            # Exact match
+            cur.execute("DELETE FROM Devices WHERE devMAC = ?", (mac,))
+        deleted_count += cur.rowcount
+
+    conn.commit()
+    conn.close()
+
+    return jsonify({"success": True, "deleted_count": deleted_count})
+
+def delete_all_with_empty_macs():
+    """Delete devices with empty MAC addresses."""
+    conn = get_temp_db_connection()
+    cur = conn.cursor()
+    cur.execute("DELETE FROM Devices WHERE devMAC IS NULL OR devMAC = ''")
+    deleted = cur.rowcount
+    conn.commit()
+    conn.close()
+    return jsonify({"success": True, "deleted": deleted})
+
+def delete_unknown_devices():
+    """Delete devices marked as unknown."""
+    conn = get_temp_db_connection()
+    cur = conn.cursor()
+    cur.execute("""DELETE FROM Devices WHERE devName='(unknown)' OR devName='(name not found)'""")
+    conn.commit()
+    conn.close()
+    return jsonify({"success": True, "deleted": cur.rowcount})

server/database.py

@@ -209,6 +209,8 @@ def get_temp_db_connection():
     Returns a new SQLite connection with Row factory.
     Should be used per-thread/request to avoid cross-thread issues.
     """
-    conn = sqlite3.connect(fullDbPath)
+    conn = sqlite3.connect(fullDbPath, timeout=5, isolation_level=None)
+    conn.execute("PRAGMA journal_mode=WAL;")
+    conn.execute("PRAGMA busy_timeout=5000;")  # 5s wait before giving up
     conn.row_factory = sqlite3.Row
     return conn

test/test_devices_endpoints.py

@@ -0,0 +1,70 @@
+import sys
+import pathlib
+import sqlite3
+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 test_delete_devices_with_macs(client, api_token, test_mac):
+
+    # First create device so it exists
+    payload = {
+        "createNew": True,
+        "name": "Test Device",
+        "owner": "Unit Test",
+        "type": "Router",
+        "vendor": "TestVendor",
+    }
+    resp = client.post(f"/device/{test_mac}", json=payload, headers=auth_headers(api_token))
+    
+    client.post(f"/device/{test_mac}", json={"createNew": True}, headers=auth_headers(api_token))
+
+    # Delete by MAC
+    resp = client.delete("/devices", json={"macs": [test_mac]}, headers=auth_headers(api_token))
+    assert resp.status_code == 200
+    assert resp.json.get("success") is True
+
+def test_delete_test_devices(client, api_token, test_mac):
+
+    # Delete by MAC
+    resp = client.delete("/devices", json={"macs": ["AA:BB:CC:*"]}, headers=auth_headers(api_token))
+    assert resp.status_code == 200
+    assert resp.json.get("success") is True
+
+
+def test_delete_all_empty_macs(client, api_token):
+    resp = client.delete("/devices/empty-macs", headers=auth_headers(api_token))
+    assert resp.status_code == 200
+    # Expect success flag in response
+    assert resp.json.get("success") is True
+
+
+def test_delete_unknown_devices(client, api_token):
+    resp = client.delete("/devices/unknown", headers=auth_headers(api_token))
+    assert resp.status_code == 200
+    assert resp.json.get("success") is True
+