vh/NetAlertX
1
0
Fork 0
mirror of https://github.com/jokob-sk/NetAlertX.git synced 2026-04-06 02:01:37 -07:00
Code Issues Actions 6 Packages Projects Releases Wiki Activity

TEST: field locking test fixes 6

Browse Source 
Signed-off-by: jokob-sk <jokob.sk@gmail.com>
This commit is contained in:
jokob-sk
2026-01-29 10:02:09 +11:00
parent 46d866b5ee
commit 7faaa630a1
2 changed files with 24 additions and 100 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
server/scan/device_handling.py
View File

@@ -360,11 +360,14 @@ def update_ipv4_ipv6(db):
if records_to_update: if records_to_update:
# We use COALESCE(?, Column) so that if the first arg is NULL, # We use COALESCE(?, Column) so that if the first arg is NULL,
# it keeps the current value of the column. # it keeps the current value of the column.
mylog("none", f"[Update Devices] Updated records_to_update: {records_to_update}")
sql.executemany( sql.executemany(
""" """
UPDATE Devices UPDATE Devices
SET devPrimaryIPv4 = COALESCE(?, devPrimaryIPv4), SET devPrimaryIPv4 = COALESCE(?, NULLIF(devPrimaryIPv4, '')),
devPrimaryIPv6 = COALESCE(?, devPrimaryIPv6) devPrimaryIPv6 = COALESCE(?, NULLIF(devPrimaryIPv6, ''))
WHERE devMac = ? WHERE devMac = ?
""", """,
records_to_update, records_to_update,

117
test/authoritative_fields/test_ip_update_logic.py
View File

@@ -3,9 +3,7 @@ Unit tests for device IP update logic (devPrimaryIPv4/devPrimaryIPv6 handling).
""" """
from unittest.mock import Mock, patch from unittest.mock import Mock, patch
import pytest import pytest
from server.scan import device_handling from server.scan import device_handling
@@ -28,68 +26,29 @@ def mock_device_handling():
yield yield
def test_primary_ipv6_is_set_and_ipv4_preserved(in_memory_db, mock_device_handling): def test_primary_ipv6_is_set_and_ipv4_preserved(scan_db, mock_device_handling):
"""Setting IPv6 in CurrentScan should update devPrimaryIPv6 without changing devPrimaryIPv4.""" """Setting IPv6 in CurrentScan should update devPrimaryIPv6 without changing devPrimaryIPv4."""
cur = in_memory_db.cursor() cur = scan_db.cursor()
# Create device with IPv4 primary # Create device with IPv4 primary
cur.execute( cur.execute(
""" """
INSERT INTO Devices ( INSERT INTO Devices (
devMac, devLastConnection, devPresentLastScan, devLastIP, devMac, devLastConnection, devPresentLastScan, devLastIP,
devPrimaryIPv4, devPrimaryIPv6, devVendor, devParentPort, devPrimaryIPv4, devPrimaryIPv6, devVendor, devType, devName, devIcon
devParentMAC, devSite, devSSID, devType, devName, devIcon ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
""", """,
( ("AA:BB:CC:DD:EE:FF", "2025-01-01 00:00:00", 0, "192.168.1.10", "192.168.1.10", "", "TestVendor", "type", "Device", "icon")
"AA:BB:CC:DD:EE:FF",
"2025-01-01 00:00:00",
0,
"192.168.1.10",
"192.168.1.10",
"",
"TestVendor",
"",
"",
"",
"",
"type",
"Device",
"icon",
),
) )
# CurrentScan with IPv6 # CurrentScan with IPv6
cur.execute( cur.execute(
""" "INSERT INTO CurrentScan (scanMac, scanLastIP, scanSourcePlugin, scanLastConnection) VALUES (?, ?, ?, ?)",
INSERT INTO CurrentScan ( ("AA:BB:CC:DD:EE:FF", "2001:db8::1", "IPv6SCAN", "2025-01-01 01:00:00")
scanMac, scanLastIP, scanVendor, scanSourcePlugin, scanName,
scanLastQuery, scanLastConnection, scanSyncHubNode,
scanSite, scanSSID, scanParentMAC, scanParentPort, scanType
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
""",
(
"AA:BB:CC:DD:EE:FF",
"2001:db8::1",
"",
"",
"",
"",
"2025-01-01 01:00:00",
"",
"",
"",
"",
"",
"",
),
) )
in_memory_db.commit() scan_db.commit()
# Mock DummyDB-like object db = Mock(sql_connection=scan_db, sql=cur)
db = Mock()
db.sql_connection = in_memory_db
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) device_handling.update_ipv4_ipv6(db)
@@ -104,68 +63,29 @@ def test_primary_ipv6_is_set_and_ipv4_preserved(in_memory_db, mock_device_handli
assert row["devPrimaryIPv6"] == "2001:db8::1" assert row["devPrimaryIPv6"] == "2001:db8::1"
def test_primary_ipv4_is_set_and_ipv6_preserved(in_memory_db, mock_device_handling): def test_primary_ipv4_is_set_and_ipv6_preserved(scan_db, mock_device_handling):
"""Setting IPv4 in CurrentScan should update devPrimaryIPv4 without changing devPrimaryIPv6.""" """Setting IPv4 in CurrentScan should update devPrimaryIPv4 without changing devPrimaryIPv6."""
cur = in_memory_db.cursor() cur = scan_db.cursor()
# Create device with IPv6 primary # Create device with IPv6 primary
cur.execute( cur.execute(
""" """
INSERT INTO Devices ( INSERT INTO Devices (
devMac, devLastConnection, devPresentLastScan, devLastIP, devMac, devLastConnection, devPresentLastScan, devLastIP,
devPrimaryIPv4, devPrimaryIPv6, devVendor, devParentPort, devPrimaryIPv4, devPrimaryIPv6, devVendor, devType, devName, devIcon
devParentMAC, devSite, devSSID, devType, devName, devIcon ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
""", """,
( ("11:22:33:44:55:66", "2025-01-01 00:00:00", 0, "2001:db8::2", "", "2001:db8::2", "TestVendor", "type", "Device", "icon")
"11:22:33:44:55:66",
"2025-01-01 00:00:00",
0,
"2001:db8::2",
"",
"2001:db8::2",
"TestVendor",
"",
"",
"",
"",
"type",
"Device",
"icon",
),
) )
# CurrentScan with IPv4 # CurrentScan with IPv4
cur.execute( cur.execute(
""" "INSERT INTO CurrentScan (scanMac, scanLastIP, scanSourcePlugin, scanLastConnection) VALUES (?, ?, ?, ?)",
INSERT INTO CurrentScan ( ("11:22:33:44:55:66", "10.0.0.5", "ARPSCAN", "2025-01-01 02:00:00")
scanMac, scanLastIP, scanVendor, scanSourcePlugin, scanName,
scanLastQuery, scanLastConnection, scanSyncHubNode,
scanSite, scanSSID, scanParentMAC, scanParentPort, scanType
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
""",
(
"11:22:33:44:55:66",
"10.0.0.5",
"",
"",
"",
"",
"2025-01-01 02:00:00",
"",
"",
"",
"",
"",
"",
),
) )
in_memory_db.commit() scan_db.commit()
# Mock DummyDB-like object db = Mock(sql_connection=scan_db, sql=cur)
db = Mock()
db.sql_connection = in_memory_db
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) device_handling.update_ipv4_ipv6(db)
@@ -178,3 +98,4 @@ def test_primary_ipv4_is_set_and_ipv6_preserved(in_memory_db, mock_device_handli
assert row["devLastIP"] == "10.0.0.5" assert row["devLastIP"] == "10.0.0.5"
assert row["devPrimaryIPv4"] == "10.0.0.5" assert row["devPrimaryIPv4"] == "10.0.0.5"
assert row["devPrimaryIPv6"] == "2001:db8::2" assert row["devPrimaryIPv6"] == "2001:db8::2"