mirror/NetAlertX
1
0
Fork 0
mirror of https://github.com/jokob-sk/NetAlertX.git synced 2026-01-29 00:51:29 -05:00
Code Issues Packages Projects Releases Wiki Activity

BE+TEST: field locking test fixes 3, ipv4/6 handling fix

Browse Source 
Signed-off-by: jokob-sk <jokob.sk@gmail.com>
This commit is contained in:
jokob-sk
2026-01-29 08:27:27 +11:00
parent 0fd3bd6974
commit dc2a56aac3
2 changed files with 20 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files

19
server/scan/device_handling.py
View File

@@ -331,10 +331,9 @@ def update_devices_data_from_scan(db):
def update_ipv4_ipv6(db):
"""
Fill devPrimaryIPv4 and devPrimaryIPv6 based on devLastIP.
Skips empty devLastIP.
Skips empty devLastIP and preserves existing values for the other version.
"""
sql = db.sql
mylog("debug", "[Update Devices] Updating devPrimaryIPv4 / devPrimaryIPv6 from devLastIP")
devices = sql.execute("SELECT devMac, devLastIP FROM Devices").fetchall()
@@ -342,8 +341,9 @@ def update_ipv4_ipv6(db):
for device in devices:
last_ip = device["devLastIP"]
# Keeping your specific skip logic
if not last_ip or last_ip.lower() in ("", "null", "(unknown)", "(Unknown)"):
continue # skip empty
continue
ipv4, ipv6 = None, None
try:
@@ -353,13 +353,20 @@ def update_ipv4_ipv6(db):
else:
ipv6 = last_ip
except ValueError:
continue # invalid IP, skip
continue
records_to_update.append([ipv4, ipv6, device["devMac"]])
records_to_update.append((ipv4, ipv6, device["devMac"]))
if records_to_update:
# We use COALESCE(?, Column) so that if the first arg is NULL,
# it keeps the current value of the column.
sql.executemany(
"UPDATE Devices SET devPrimaryIPv4 = ?, devPrimaryIPv6 = ? WHERE devMac = ?",
"""
UPDATE Devices
SET devPrimaryIPv4 = COALESCE(?, devPrimaryIPv4),
devPrimaryIPv6 = COALESCE(?, devPrimaryIPv6)
WHERE devMac = ?
""",
records_to_update,
)

7
test/authoritative_fields/test_ip_format_and_locking.py
View File

@@ -151,6 +151,7 @@ def test_valid_ipv4_format_accepted(ip_test_db, mock_ip_handlers):
db.sql = cur
device_handling.update_devices_data_from_scan(db)
device_handling.update_ipv4_ipv6(db)
row = cur.execute(
"SELECT devLastIP, devPrimaryIPv4, devPrimaryIPv6 FROM Devices WHERE devMac = ?",
@@ -225,6 +226,7 @@ def test_valid_ipv6_format_accepted(ip_test_db, mock_ip_handlers):
db.sql = cur
device_handling.update_devices_data_from_scan(db)
device_handling.update_ipv4_ipv6(db)
row = cur.execute(
"SELECT devLastIP, devPrimaryIPv4, devPrimaryIPv6 FROM Devices WHERE devMac = ?",
@@ -302,6 +304,7 @@ def test_invalid_ip_values_rejected(ip_test_db, mock_ip_handlers):
db.sql = cur
device_handling.update_devices_data_from_scan(db)
device_handling.update_ipv4_ipv6(db)
row = cur.execute(
"SELECT devPrimaryIPv4 FROM Devices WHERE devMac = ?",
@@ -376,6 +379,7 @@ def test_ipv4_ipv6_mixed_in_multiple_scans(ip_test_db, mock_ip_handlers):
db.sql = cur
device_handling.update_devices_data_from_scan(db)
device_handling.update_ipv4_ipv6(db)
row1 = cur.execute(
"SELECT devPrimaryIPv4, devPrimaryIPv6 FROM Devices WHERE devMac = ?",
@@ -415,6 +419,7 @@ def test_ipv4_ipv6_mixed_in_multiple_scans(ip_test_db, mock_ip_handlers):
db.sql = cur
device_handling.update_devices_data_from_scan(db)
device_handling.update_ipv4_ipv6(db)
row2 = cur.execute(
"SELECT devPrimaryIPv4, devPrimaryIPv6 FROM Devices WHERE devMac = ?",
@@ -470,6 +475,7 @@ def test_ipv4_address_format_variations(ip_test_db, mock_ip_handlers):
db.sql = cur
device_handling.update_devices_data_from_scan(db)
device_handling.update_ipv4_ipv6(db)
for idx, expected_ipv4 in enumerate(ipv4_addresses):
mac = f"AA:BB:CC:DD:EE:{idx:02X}"
@@ -524,6 +530,7 @@ def test_ipv6_address_format_variations(ip_test_db, mock_ip_handlers):
db.sql = cur
device_handling.update_devices_data_from_scan(db)
device_handling.update_ipv4_ipv6(db)
for idx, expected_ipv6 in enumerate(ipv6_addresses):
mac = f"BB:BB:CC:DD:EE:{idx:02X}"