mirror of
https://github.com/jokob-sk/NetAlertX.git
synced 2026-01-30 01:21:17 -05:00
TEST: field locking test fixes 8
Signed-off-by: jokob-sk <jokob.sk@gmail.com>
This commit is contained in:
@@ -80,30 +80,52 @@ def test_invalid_ip_values_rejected(scan_db, mock_ip_handlers):
|
||||
assert row["devPrimaryIPv4"] == "192.168.1.50", f"Failed on {invalid_ip}"
|
||||
|
||||
|
||||
def test_ipv4_ipv6_mixed_in_multiple_scans(scan_db, mock_ip_handlers):
|
||||
"""Multiple scans with different IP types should set both primary fields correctly."""
|
||||
def test_ipv4_then_ipv6_scan_updates_primary_ips(scan_db, mock_ip_handlers):
|
||||
"""
|
||||
Test that multiple scans with different IP types correctly update:
|
||||
- devLastIP to the latest scan
|
||||
- devPrimaryIPv4 and devPrimaryIPv6 appropriately
|
||||
"""
|
||||
cur = scan_db.cursor()
|
||||
|
||||
# 1️⃣ Create device
|
||||
cur.execute("INSERT INTO Devices (devMac) VALUES (?)", ("AA:BB:CC:DD:EE:04",))
|
||||
|
||||
# Scan 1: IPv4
|
||||
cur.execute("INSERT INTO CurrentScan (scanMac, scanLastIP, scanSourcePlugin, scanLastConnection) VALUES (?, ?, ?, ?)",
|
||||
("AA:BB:CC:DD:EE:04", "192.168.1.100", "ARPSCAN", "2025-01-01 01:00:00"))
|
||||
scan_db.commit()
|
||||
|
||||
db = Mock(sql_connection=scan_db, sql=cur)
|
||||
device_handling.update_devices_data_from_scan(db)
|
||||
device_handling.update_ipv4_ipv6(db)
|
||||
|
||||
# Scan 2: IPv6
|
||||
cur.execute("DELETE FROM CurrentScan")
|
||||
cur.execute("INSERT INTO CurrentScan (scanMac, scanLastIP, scanSourcePlugin, scanLastConnection) VALUES (?, ?, ?, ?)",
|
||||
("AA:BB:CC:DD:EE:04", "fe80::1", "IPv6SCAN", "2025-01-01 02:00:00"))
|
||||
# 2️⃣ First scan: IPv4
|
||||
cur.execute(
|
||||
"INSERT INTO CurrentScan (scanMac, scanLastIP, scanSourcePlugin, scanLastConnection) VALUES (?, ?, ?, ?)",
|
||||
("AA:BB:CC:DD:EE:04", "192.168.1.100", "ARPSCAN", "2025-01-01 01:00:00")
|
||||
)
|
||||
scan_db.commit()
|
||||
device_handling.update_devices_data_from_scan(db)
|
||||
device_handling.update_ipv4_ipv6(db)
|
||||
|
||||
row = cur.execute("SELECT devPrimaryIPv4, devPrimaryIPv6 FROM Devices WHERE devMac = ?", ("AA:BB:CC:DD:EE:04",)).fetchone()
|
||||
assert row["devPrimaryIPv4"] == "192.168.1.100"
|
||||
assert row["devPrimaryIPv6"] == "fe80::1"
|
||||
with patch("server.scan.device_handling.get_plugin_authoritative_settings", return_value={}):
|
||||
device_handling.update_devices_data_from_scan(db)
|
||||
device_handling.update_ipv4_ipv6(db)
|
||||
|
||||
# 3️⃣ Second scan: IPv6
|
||||
cur.execute("DELETE FROM CurrentScan")
|
||||
cur.execute(
|
||||
"INSERT INTO CurrentScan (scanMac, scanLastIP, scanSourcePlugin, scanLastConnection) VALUES (?, ?, ?, ?)",
|
||||
("AA:BB:CC:DD:EE:04", "fe80::1", "IPv6SCAN", "2025-01-01 02:00:00")
|
||||
)
|
||||
scan_db.commit()
|
||||
|
||||
with patch("server.scan.device_handling.get_plugin_authoritative_settings", return_value={}):
|
||||
device_handling.update_devices_data_from_scan(db)
|
||||
device_handling.update_ipv4_ipv6(db)
|
||||
|
||||
# 4️⃣ Verify results
|
||||
row = cur.execute(
|
||||
"SELECT devLastIP, devPrimaryIPv4, devPrimaryIPv6 FROM Devices WHERE devMac = ?",
|
||||
("AA:BB:CC:DD:EE:04",)
|
||||
).fetchone()
|
||||
|
||||
assert row["devLastIP"] == "fe80::1" # Latest scan IP (IPv6)
|
||||
assert row["devPrimaryIPv4"] == "192.168.1.100" # IPv4 preserved
|
||||
assert row["devPrimaryIPv6"] == "fe80::1" # IPv6 set
|
||||
|
||||
|
||||
def test_ipv4_address_format_variations(scan_db, mock_ip_handlers):
|
||||
|
||||
@@ -26,17 +26,22 @@ def mock_device_handling():
|
||||
yield
|
||||
|
||||
|
||||
def test_primary_ipv6_is_set_and_ipv4_preserved(scan_db, mock_device_handling):
|
||||
def test_ipv6_update_preserves_ipv4(scan_db, mock_device_handling):
|
||||
"""
|
||||
Test that when a device already has a primary IPv4 and receives an IPv6 scan:
|
||||
- devLastIP is updated to the latest IP (IPv6)
|
||||
- devPrimaryIPv6 is set to the new IPv6
|
||||
- devPrimaryIPv4 is preserved
|
||||
"""
|
||||
cur = scan_db.cursor()
|
||||
|
||||
# 1. Create device
|
||||
# 1️⃣ Create device with IPv4
|
||||
cur.execute(
|
||||
"INSERT INTO Devices (devMac, devLastIP, devPrimaryIPv4, devName) VALUES (?, ?, ?, ?)",
|
||||
("AA:BB:CC:DD:EE:FF", "192.168.1.10", "192.168.1.10", "Device")
|
||||
)
|
||||
|
||||
# 2. Insert Scan with a SPECIFIC plugin name
|
||||
# This is critical so the logic inside update_devices_data_from_scan finds it
|
||||
# 2️⃣ Insert a scan reporting IPv6
|
||||
cur.execute(
|
||||
"""
|
||||
INSERT INTO CurrentScan (scanMac, scanLastIP, scanSourcePlugin, scanLastConnection)
|
||||
@@ -48,20 +53,20 @@ def test_primary_ipv6_is_set_and_ipv4_preserved(scan_db, mock_device_handling):
|
||||
|
||||
db = Mock(sql_connection=scan_db, sql=cur)
|
||||
|
||||
# We must mock get_plugin_authoritative_settings to allow updates from "TEST_PLUGIN"
|
||||
# 3️⃣ Mock plugin authoritative settings
|
||||
with patch("server.scan.device_handling.get_plugin_authoritative_settings", return_value={}):
|
||||
device_handling.update_devices_data_from_scan(db)
|
||||
device_handling.update_ipv4_ipv6(db)
|
||||
|
||||
# 4️⃣ Verify the device fields
|
||||
row = cur.execute(
|
||||
"SELECT devLastIP, devPrimaryIPv4, devPrimaryIPv6 FROM Devices WHERE devMac = ?",
|
||||
("AA:BB:CC:DD:EE:FF",),
|
||||
).fetchone()
|
||||
|
||||
# These were failing because devLastIP remained 192.168.1.10
|
||||
assert row["devLastIP"] == "2001:db8::1"
|
||||
assert row["devPrimaryIPv6"] == "2001:db8::1"
|
||||
assert row["devPrimaryIPv4"] == "192.168.1.10"
|
||||
assert row["devLastIP"] == "2001:db8::1" # Latest IP is now IPv6
|
||||
assert row["devPrimaryIPv6"] == "2001:db8::1" # IPv6 field updated
|
||||
assert row["devPrimaryIPv4"] == "192.168.1.10" # IPv4 preserved
|
||||
|
||||
|
||||
def test_primary_ipv4_is_set_and_ipv6_preserved(scan_db, mock_device_handling):
|
||||
|
||||
Reference in New Issue
Block a user