mirror/NetAlertX
1
0
Fork 0
mirror of https://github.com/jokob-sk/NetAlertX.git synced 2026-01-30 09:31:26 -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): def update_ipv4_ipv6(db):
""" """
Fill devPrimaryIPv4 and devPrimaryIPv6 based on devLastIP. Fill devPrimaryIPv4 and devPrimaryIPv6 based on devLastIP.
Skips empty devLastIP. Skips empty devLastIP and preserves existing values for the other version.
""" """
sql = db.sql sql = db.sql
mylog("debug", "[Update Devices] Updating devPrimaryIPv4 / devPrimaryIPv6 from devLastIP") mylog("debug", "[Update Devices] Updating devPrimaryIPv4 / devPrimaryIPv6 from devLastIP")
devices = sql.execute("SELECT devMac, devLastIP FROM Devices").fetchall() devices = sql.execute("SELECT devMac, devLastIP FROM Devices").fetchall()
@@ -342,8 +341,9 @@ def update_ipv4_ipv6(db):
for device in devices: for device in devices:
last_ip = device["devLastIP"] last_ip = device["devLastIP"]
# Keeping your specific skip logic
if not last_ip or last_ip.lower() in ("", "null", "(unknown)", "(Unknown)"): if not last_ip or last_ip.lower() in ("", "null", "(unknown)", "(Unknown)"):
continue # skip empty continue
ipv4, ipv6 = None, None ipv4, ipv6 = None, None
try: try:
@@ -353,13 +353,20 @@ def update_ipv4_ipv6(db):
else: else:
ipv6 = last_ip ipv6 = last_ip
except ValueError: 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: 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( sql.executemany(
"UPDATE Devices SET devPrimaryIPv4 = ?, devPrimaryIPv6 = ? WHERE devMac = ?", """
UPDATE Devices
SET devPrimaryIPv4 = COALESCE(?, devPrimaryIPv4),
devPrimaryIPv6 = COALESCE(?, devPrimaryIPv6)
WHERE devMac = ?
""",
records_to_update, 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 db.sql = cur
device_handling.update_devices_data_from_scan(db) device_handling.update_devices_data_from_scan(db)
device_handling.update_ipv4_ipv6(db)
row = cur.execute( row = cur.execute(
"SELECT devLastIP, devPrimaryIPv4, devPrimaryIPv6 FROM Devices WHERE devMac = ?", "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 db.sql = cur
device_handling.update_devices_data_from_scan(db) device_handling.update_devices_data_from_scan(db)
device_handling.update_ipv4_ipv6(db)
row = cur.execute( row = cur.execute(
"SELECT devLastIP, devPrimaryIPv4, devPrimaryIPv6 FROM Devices WHERE devMac = ?", "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 db.sql = cur
device_handling.update_devices_data_from_scan(db) device_handling.update_devices_data_from_scan(db)
device_handling.update_ipv4_ipv6(db)
row = cur.execute( row = cur.execute(
"SELECT devPrimaryIPv4 FROM Devices WHERE devMac = ?", "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 db.sql = cur
device_handling.update_devices_data_from_scan(db) device_handling.update_devices_data_from_scan(db)
device_handling.update_ipv4_ipv6(db)
row1 = cur.execute( row1 = cur.execute(
"SELECT devPrimaryIPv4, devPrimaryIPv6 FROM Devices WHERE devMac = ?", "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 db.sql = cur
device_handling.update_devices_data_from_scan(db) device_handling.update_devices_data_from_scan(db)
device_handling.update_ipv4_ipv6(db)
row2 = cur.execute( row2 = cur.execute(
"SELECT devPrimaryIPv4, devPrimaryIPv6 FROM Devices WHERE devMac = ?", "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 db.sql = cur
device_handling.update_devices_data_from_scan(db) device_handling.update_devices_data_from_scan(db)
device_handling.update_ipv4_ipv6(db)
for idx, expected_ipv4 in enumerate(ipv4_addresses): for idx, expected_ipv4 in enumerate(ipv4_addresses):
mac = f"AA:BB:CC:DD:EE:{idx:02X}" 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 db.sql = cur
device_handling.update_devices_data_from_scan(db) device_handling.update_devices_data_from_scan(db)
device_handling.update_ipv4_ipv6(db)
for idx, expected_ipv6 in enumerate(ipv6_addresses): for idx, expected_ipv6 in enumerate(ipv6_addresses):
mac = f"BB:BB:CC:DD:EE:{idx:02X}" mac = f"BB:BB:CC:DD:EE:{idx:02X}"