feat: Implement forced device status updates and enhance related tests

This commit is contained in:
Jokob @NetAlertX
2026-01-21 09:21:55 +00:00
parent 9f1d04bcd4
commit fcbe4ae88a
7 changed files with 156 additions and 13 deletions

View File

@@ -156,7 +156,7 @@ function getDeviceData() {
},
// Group for other fields like static IP, archived status, etc.
DevDetail_DisplayFields_Title: {
data: ["devStaticIP", "devIsNew", "devFavorite", "devIsArchived"],
data: ["devStaticIP", "devIsNew", "devFavorite", "devIsArchived", "devForceStatus"],
docs: "https://docs.netalertx.com/DEVICE_DISPLAY_SETTINGS",
iconClass: "fa fa-list-check",
inputGroupClasses: "field-group display-group col-lg-4 col-sm-6 col-xs-12",
@@ -295,8 +295,8 @@ function getDeviceData() {
const currentSource = deviceData[sourceField] || "NEWDEV";
const sourceTitle = getString("FieldLock_Source_Label") + currentSource;
const sourceColor = currentSource === "USER" ? "text-warning" : (currentSource === "LOCKED" ? "text-danger" : "text-muted");
inlineControl += `<span class="input-group-addon ${sourceColor}" title="${sourceTitle}">
<i class="fa-solid fa-tag"></i> ${currentSource}
inlineControl += `<span class="input-group-addon pointer ${sourceColor}" title="${sourceTitle}">
${currentSource}
</span>`;
}
@@ -594,14 +594,17 @@ function toggleFieldLock(mac, fieldName) {
lockBtn.find("i").attr("class", `fa-solid ${lockIcon}`);
lockBtn.attr("title", lockTitle);
// Update source indicator if locked
if (shouldLock) {
const sourceIndicator = lockBtn.next();
if (sourceIndicator.hasClass("input-group-addon")) {
sourceIndicator.text("LOCKED");
sourceIndicator.attr("class", "input-group-addon text-danger");
sourceIndicator.attr("title", getString("FieldLock_Source_Label") + "LOCKED");
}
// Update local source state
deviceData[sourceField] = shouldLock ? "LOCKED" : "";
// Update source indicator
const sourceIndicator = lockBtn.next();
if (sourceIndicator.hasClass("input-group-addon")) {
const sourceValue = shouldLock ? "LOCKED" : "NEWDEV";
const sourceClass = shouldLock ? "input-group-addon text-danger" : "input-group-addon text-muted";
sourceIndicator.text(sourceValue);
sourceIndicator.attr("class", sourceClass);
sourceIndicator.attr("title", getString("FieldLock_Source_Label") + sourceValue);
}
showMessage(shouldLock ? getString("FieldLock_Locked") : getString("FieldLock_Unlocked"), 3000, "modal_green");

View File

@@ -1947,9 +1947,9 @@
},
"default_value": "dont_force",
"options": [
"dont_force" ,
"online",
"offline",
"dont_force"
"offline"
],
"localized": [
"name",

View File

@@ -243,6 +243,9 @@ def update_devices_data_from_scan(db):
# Update devPresentLastScan based on NICs presence
update_devPresentLastScan_based_on_nics(db)
# Force device status if configured
update_devPresentLastScan_based_on_force_status(db)
# Guess ICONS
recordsToUpdate = []
@@ -865,6 +868,72 @@ def update_devPresentLastScan_based_on_nics(db):
return len(updates)
# -------------------------------------------------------------------------------
# Force devPresentLastScan based on devForceStatus
def update_devPresentLastScan_based_on_force_status(db):
"""
Forces devPresentLastScan in the Devices table based on devForceStatus.
devForceStatus values:
- "online" -> devPresentLastScan = 1
- "offline" -> devPresentLastScan = 0
- "dont_force" or empty -> no change
Args:
db: A database object with `.execute()` and `.fetchone()` methods.
Returns:
int: Number of devices updated.
"""
sql = db.sql
online_count_row = sql.execute(
"""
SELECT COUNT(*) AS cnt
FROM Devices
WHERE LOWER(COALESCE(devForceStatus, '')) = 'online'
AND devPresentLastScan != 1
"""
).fetchone()
online_updates = online_count_row["cnt"] if online_count_row else 0
offline_count_row = sql.execute(
"""
SELECT COUNT(*) AS cnt
FROM Devices
WHERE LOWER(COALESCE(devForceStatus, '')) = 'offline'
AND devPresentLastScan != 0
"""
).fetchone()
offline_updates = offline_count_row["cnt"] if offline_count_row else 0
if online_updates > 0:
sql.execute(
"""
UPDATE Devices
SET devPresentLastScan = 1
WHERE LOWER(COALESCE(devForceStatus, '')) = 'online'
"""
)
if offline_updates > 0:
sql.execute(
"""
UPDATE Devices
SET devPresentLastScan = 0
WHERE LOWER(COALESCE(devForceStatus, '')) = 'offline'
"""
)
total_updates = online_updates + offline_updates
if total_updates > 0:
mylog("debug", f"[Update Devices] Forced devPresentLastScan for {total_updates} devices")
db.commitDB()
return total_updates
# -------------------------------------------------------------------------------
# Check if the variable contains a valid MAC address or "Internet"
def check_mac_or_internet(input_str):

View File

@@ -33,6 +33,7 @@ def scan_db():
devMac TEXT PRIMARY KEY,
devLastConnection TEXT,
devPresentLastScan INTEGER DEFAULT 0,
devForceStatus TEXT,
devLastIP TEXT,
devName TEXT,
devNameSource TEXT DEFAULT 'NEWDEV',
@@ -93,6 +94,7 @@ def mock_device_handlers():
with patch.multiple(
device_handling,
update_devPresentLastScan_based_on_nics=Mock(return_value=0),
update_devPresentLastScan_based_on_force_status=Mock(return_value=0),
query_MAC_vendor=Mock(return_value=-1),
guess_icon=Mock(return_value="icon"),
guess_type=Mock(return_value="type"),

View File

@@ -0,0 +1,65 @@
"""Tests for forced device status updates."""
import sqlite3
from server.scan import device_handling
class DummyDB:
"""Minimal DB wrapper compatible with device_handling helpers."""
def __init__(self, conn):
self.sql = conn.cursor()
self._conn = conn
def commitDB(self):
self._conn.commit()
def test_force_status_updates_present_flag():
"""Forced status should override devPresentLastScan for online/offline values."""
conn = sqlite3.connect(":memory:")
conn.row_factory = sqlite3.Row
cur = conn.cursor()
cur.execute(
"""
CREATE TABLE Devices (
devMac TEXT PRIMARY KEY,
devPresentLastScan INTEGER,
devForceStatus TEXT
)
"""
)
cur.executemany(
"""
INSERT INTO Devices (devMac, devPresentLastScan, devForceStatus)
VALUES (?, ?, ?)
""",
[
("AA:AA:AA:AA:AA:01", 0, "online"),
("AA:AA:AA:AA:AA:02", 1, "offline"),
("AA:AA:AA:AA:AA:03", 1, "dont_force"),
("AA:AA:AA:AA:AA:04", 0, None),
("AA:AA:AA:AA:AA:05", 0, "ONLINE"),
],
)
conn.commit()
db = DummyDB(conn)
updated = device_handling.update_devPresentLastScan_based_on_force_status(db)
rows = {
row["devMac"]: row["devPresentLastScan"]
for row in cur.execute("SELECT devMac, devPresentLastScan FROM Devices")
}
assert updated == 3
assert rows["AA:AA:AA:AA:AA:01"] == 1
assert rows["AA:AA:AA:AA:AA:02"] == 0
assert rows["AA:AA:AA:AA:AA:03"] == 1
assert rows["AA:AA:AA:AA:AA:04"] == 0
assert rows["AA:AA:AA:AA:AA:05"] == 1
conn.close()

View File

@@ -29,6 +29,7 @@ def ip_test_db():
devMac TEXT PRIMARY KEY,
devLastConnection TEXT,
devPresentLastScan INTEGER,
devForceStatus TEXT,
devLastIP TEXT,
devLastIpSource TEXT DEFAULT 'NEWDEV',
devPrimaryIPv4 TEXT,
@@ -78,6 +79,7 @@ def mock_ip_handlers():
with patch.multiple(
device_handling,
update_devPresentLastScan_based_on_nics=Mock(return_value=0),
update_devPresentLastScan_based_on_force_status=Mock(return_value=0),
query_MAC_vendor=Mock(return_value=-1),
guess_icon=Mock(return_value="icon"),
guess_type=Mock(return_value="type"),

View File

@@ -23,6 +23,7 @@ def in_memory_db():
devMac TEXT PRIMARY KEY,
devLastConnection TEXT,
devPresentLastScan INTEGER,
devForceStatus TEXT,
devLastIP TEXT,
devPrimaryIPv4 TEXT,
devPrimaryIPv6 TEXT,
@@ -69,6 +70,7 @@ def mock_device_handling():
with patch.multiple(
device_handling,
update_devPresentLastScan_based_on_nics=Mock(return_value=0),
update_devPresentLastScan_based_on_force_status=Mock(return_value=0),
query_MAC_vendor=Mock(return_value=-1),
guess_icon=Mock(return_value="icon"),
guess_type=Mock(return_value="type"),