vh/NetAlertX
1
0
Fork 0
mirror of https://github.com/jokob-sk/NetAlertX.git synced 2026-04-03 08:41:35 -07:00
Code Issues Actions 4 Packages Projects Releases Wiki Activity

FIX: lowercase MAC normalization across project v0.3

Browse Source 
Signed-off-by: jokob-sk <jokob.sk@gmail.com>
This commit is contained in:
jokob-sk
2026-02-07 14:08:14 +11:00
parent 1bacb59044
commit 8538c87fef
12 changed files with 72 additions and 66 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
test/api_endpoints/test_dbquery_endpoints.py
View File

@@ -26,7 +26,7 @@ def client():
@pytest.fixture(scope="session") @pytest.fixture(scope="session")
def test_mac(): def test_mac():
# Generate a unique MAC for each test run # Generate a unique MAC for each test run
return "AA:BB:CC:" + ":".join(f"{random.randint(0, 255):02X}" for _ in range(3)) return "aa:bb:cc:" + ":".join(f"{random.randint(0, 255):02X}" for _ in range(3))
def auth_headers(token): def auth_headers(token):

2
test/api_endpoints/test_devices_endpoints.py
View File

@@ -24,7 +24,7 @@ def client():
@pytest.fixture @pytest.fixture
def test_mac(): def test_mac():
# Generate a unique MAC for each test run # Generate a unique MAC for each test run
return "AA:BB:CC:" + ":".join(f"{random.randint(0, 255):02X}" for _ in range(3)) return "aa:bb:cc:" + ":".join(f"{random.randint(0, 255):02X}" for _ in range(3))
def auth_headers(token): def auth_headers(token):

4
test/api_endpoints/test_events_endpoints.py
View File

@@ -26,7 +26,7 @@ def client():
@pytest.fixture @pytest.fixture
def test_mac(): def test_mac():
# Generate a unique MAC for each test run # Generate a unique MAC for each test run
return "AA:BB:CC:" + ":".join(f"{random.randint(0, 255):02X}" for _ in range(3)) return "aa:bb:cc:" + ":".join(f"{random.randint(0, 255):02X}" for _ in range(3))
def auth_headers(token): def auth_headers(token):
@@ -115,7 +115,7 @@ def test_get_events_totals(client, api_token):
def test_delete_all_events(client, api_token, test_mac): def test_delete_all_events(client, api_token, test_mac):
# create two events # create two events
create_event(client, api_token, test_mac) create_event(client, api_token, test_mac)
create_event(client, api_token, "FF:FF:FF:FF:FF:FF") create_event(client, api_token, "ff:ff:ff:ff:ff:ff")
resp = list_events(client, api_token) resp = list_events(client, api_token)
# At least the two we created should be present # At least the two we created should be present

2
test/api_endpoints/test_graphq_endpoints.py
View File

@@ -23,7 +23,7 @@ def client():
@pytest.fixture @pytest.fixture
def test_mac(): def test_mac():
# Generate a unique MAC for each test run # Generate a unique MAC for each test run
return "AA:BB:CC:" + ":".join(f"{random.randint(0, 255):02X}" for _ in range(3)) return "aa:bb:cc:" + ":".join(f"{random.randint(0, 255):02X}" for _ in range(3))
def auth_headers(token): def auth_headers(token):

2
test/api_endpoints/test_history_endpoints.py
View File

@@ -24,7 +24,7 @@ def client():
@pytest.fixture @pytest.fixture
def test_mac(): def test_mac():
# Generate a unique MAC for each test run # Generate a unique MAC for each test run
return "AA:BB:CC:" + ":".join(f"{random.randint(0, 255):02X}" for _ in range(3)) return "aa:bb:cc:" + ":".join(f"{random.randint(0, 255):02X}" for _ in range(3))
def auth_headers(token): def auth_headers(token):

2
test/api_endpoints/test_mcp_tools_endpoints.py
View File

@@ -200,7 +200,7 @@ def test_set_device_alias_not_found(mock_update_col, client, api_token):
mock_update_col.return_value = {"success": False, "error": "Device not found"} mock_update_col.return_value = {"success": False, "error": "Device not found"}
payload = {"alias": "New Device Name"} payload = {"alias": "New Device Name"}
response = client.post("/device/FF:FF:FF:FF:FF:FF/set-alias", json=payload, headers=auth_headers(api_token)) response = client.post("/device/ff:ff:ff:ff:ff:ff/set-alias", json=payload, headers=auth_headers(api_token))
assert response.status_code == 200 assert response.status_code == 200
data = response.get_json() data = response.get_json()

2
test/api_endpoints/test_nettools_endpoints.py
View File

@@ -19,7 +19,7 @@ def client():
@pytest.fixture @pytest.fixture
def test_mac(): def test_mac():
# Generate a unique MAC for each test run # Generate a unique MAC for each test run
return "AA:BB:CC:" + ":".join(f"{random.randint(0, 255):02X}" for _ in range(3)) return "aa:bb:cc:" + ":".join(f"{random.randint(0, 255):02X}" for _ in range(3))
def auth_headers(token): def auth_headers(token):

2
test/api_endpoints/test_sessions_endpoints.py
View File

@@ -26,7 +26,7 @@ def client():
@pytest.fixture @pytest.fixture
def test_mac(): def test_mac():
# Generate a unique MAC for each test run # Generate a unique MAC for each test run
return "AA:BB:CC:" + ":".join(f"{random.randint(0, 255):02X}" for _ in range(3)) return "aa:bb:cc:" + ":".join(f"{random.randint(0, 255):02X}" for _ in range(3))
def auth_headers(token): def auth_headers(token):

2
test/api_endpoints/test_settings_endpoints.py
View File

@@ -24,7 +24,7 @@ def client():
@pytest.fixture @pytest.fixture
def test_mac(): def test_mac():
# Generate a unique MAC for each test run # Generate a unique MAC for each test run
return "AA:BB:CC:" + ":".join(f"{random.randint(0, 255):02X}" for _ in range(3)) return "aa:bb:cc:" + ":".join(f"{random.randint(0, 255):02X}" for _ in range(3))
def auth_headers(token): def auth_headers(token):

70
test/scan/test_field_lock_scan_integration.py
View File

@@ -159,7 +159,7 @@ def test_create_new_devices_sets_sources(scan_db_for_new_devices):
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
""", """,
( (
"AA:BB:CC:DD:EE:10", "aa:bb:cc:dd:ee:10",
"DeviceOne", "DeviceOne",
"AcmeVendor", "AcmeVendor",
"ARPSCAN", "ARPSCAN",
@@ -176,7 +176,7 @@ def test_create_new_devices_sets_sources(scan_db_for_new_devices):
settings = { settings = {
"NEWDEV_devType": "default-type", "NEWDEV_devType": "default-type",
"NEWDEV_devParentMAC": "FF:FF:FF:FF:FF:FF", "NEWDEV_devParentMAC": "ff:ff:ff:ff:ff:ff",
"NEWDEV_devOwner": "owner", "NEWDEV_devOwner": "owner",
"NEWDEV_devGroup": "group", "NEWDEV_devGroup": "group",
"NEWDEV_devComments": "", "NEWDEV_devComments": "",
@@ -216,7 +216,7 @@ def test_create_new_devices_sets_sources(scan_db_for_new_devices):
devVlanSource devVlanSource
FROM Devices WHERE devMac = ? FROM Devices WHERE devMac = ?
""", """,
("AA:BB:CC:DD:EE:10",), ("aa:bb:cc:dd:ee:10",),
).fetchone() ).fetchone()
assert row["devMacSource"] == "ARPSCAN" assert row["devMacSource"] == "ARPSCAN"
@@ -245,7 +245,7 @@ def test_scan_updates_newdev_device_name(scan_db, mock_device_handlers):
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
""", """,
( (
"AA:BB:CC:DD:EE:01", "aa:bb:cc:dd:ee:01",
"2025-01-01 00:00:00", "2025-01-01 00:00:00",
0, 0,
"192.168.1.1", "192.168.1.1",
@@ -273,7 +273,7 @@ def test_scan_updates_newdev_device_name(scan_db, mock_device_handlers):
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
""", """,
( (
"AA:BB:CC:DD:EE:01", "aa:bb:cc:dd:ee:01",
"192.168.1.1", "192.168.1.1",
"TestVendor", "TestVendor",
"NBTSCAN", "NBTSCAN",
@@ -299,7 +299,7 @@ def test_scan_updates_newdev_device_name(scan_db, mock_device_handlers):
row = cur.execute( row = cur.execute(
"SELECT devName FROM Devices WHERE devMac = ?", "SELECT devName FROM Devices WHERE devMac = ?",
("AA:BB:CC:DD:EE:01",), ("aa:bb:cc:dd:ee:01",),
).fetchone() ).fetchone()
# Name SHOULD be updated from NEWDEV # Name SHOULD be updated from NEWDEV
@@ -320,7 +320,7 @@ def test_scan_does_not_update_user_field_name(scan_db, mock_device_handlers):
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
""", """,
( (
"AA:BB:CC:DD:EE:02", "aa:bb:cc:dd:ee:02",
"2025-01-01 00:00:00", "2025-01-01 00:00:00",
0, 0,
"192.168.1.2", "192.168.1.2",
@@ -348,7 +348,7 @@ def test_scan_does_not_update_user_field_name(scan_db, mock_device_handlers):
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
""", """,
( (
"AA:BB:CC:DD:EE:02", "aa:bb:cc:dd:ee:02",
"192.168.1.2", "192.168.1.2",
"TestVendor", "TestVendor",
"NBTSCAN", "NBTSCAN",
@@ -374,7 +374,7 @@ def test_scan_does_not_update_user_field_name(scan_db, mock_device_handlers):
row = cur.execute( row = cur.execute(
"SELECT devName FROM Devices WHERE devMac = ?", "SELECT devName FROM Devices WHERE devMac = ?",
("AA:BB:CC:DD:EE:02",), ("aa:bb:cc:dd:ee:02",),
).fetchone() ).fetchone()
# Name should NOT be updated because it's USER-owned # Name should NOT be updated because it's USER-owned
@@ -395,7 +395,7 @@ def test_scan_does_not_update_locked_field(scan_db, mock_device_handlers):
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
""", """,
( (
"AA:BB:CC:DD:EE:03", "aa:bb:cc:dd:ee:03",
"2025-01-01 00:00:00", "2025-01-01 00:00:00",
0, 0,
"192.168.1.3", "192.168.1.3",
@@ -423,7 +423,7 @@ def test_scan_does_not_update_locked_field(scan_db, mock_device_handlers):
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
""", """,
( (
"AA:BB:CC:DD:EE:03", "aa:bb:cc:dd:ee:03",
"192.168.1.3", "192.168.1.3",
"TestVendor", "TestVendor",
"NBTSCAN", "NBTSCAN",
@@ -449,7 +449,7 @@ def test_scan_does_not_update_locked_field(scan_db, mock_device_handlers):
row = cur.execute( row = cur.execute(
"SELECT devName FROM Devices WHERE devMac = ?", "SELECT devName FROM Devices WHERE devMac = ?",
("AA:BB:CC:DD:EE:03",), ("aa:bb:cc:dd:ee:03",),
).fetchone() ).fetchone()
# Name should NOT be updated because it's LOCKED # Name should NOT be updated because it's LOCKED
@@ -470,7 +470,7 @@ def test_scan_updates_empty_vendor_field(scan_db, mock_device_handlers):
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
""", """,
( (
"AA:BB:CC:DD:EE:04", "aa:bb:cc:dd:ee:04",
"2025-01-01 00:00:00", "2025-01-01 00:00:00",
0, 0,
"192.168.1.4", "192.168.1.4",
@@ -498,7 +498,7 @@ def test_scan_updates_empty_vendor_field(scan_db, mock_device_handlers):
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
""", """,
( (
"AA:BB:CC:DD:EE:04", "aa:bb:cc:dd:ee:04",
"192.168.1.4", "192.168.1.4",
"Apple Inc.", "Apple Inc.",
"ARPSCAN", "ARPSCAN",
@@ -524,7 +524,7 @@ def test_scan_updates_empty_vendor_field(scan_db, mock_device_handlers):
row = cur.execute( row = cur.execute(
"SELECT devVendor FROM Devices WHERE devMac = ?", "SELECT devVendor FROM Devices WHERE devMac = ?",
("AA:BB:CC:DD:EE:04",), ("aa:bb:cc:dd:ee:04",),
).fetchone() ).fetchone()
# Vendor SHOULD be updated # Vendor SHOULD be updated
@@ -546,7 +546,7 @@ def test_scan_updates_ip_addresses(scan_db, mock_device_handlers):
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
""", """,
( (
"AA:BB:CC:DD:EE:05", "aa:bb:cc:dd:ee:05",
"2025-01-01 00:00:00", "2025-01-01 00:00:00",
0, 0,
"", "",
@@ -576,7 +576,7 @@ def test_scan_updates_ip_addresses(scan_db, mock_device_handlers):
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
""", """,
( (
"AA:BB:CC:DD:EE:05", "aa:bb:cc:dd:ee:05",
"192.168.1.100", "192.168.1.100",
"Vendor", "Vendor",
"ARPSCAN", "ARPSCAN",
@@ -603,7 +603,7 @@ def test_scan_updates_ip_addresses(scan_db, mock_device_handlers):
row = cur.execute( row = cur.execute(
"SELECT devLastIP, devPrimaryIPv4, devPrimaryIPv6 FROM Devices WHERE devMac = ?", "SELECT devLastIP, devPrimaryIPv4, devPrimaryIPv6 FROM Devices WHERE devMac = ?",
("AA:BB:CC:DD:EE:05",), ("aa:bb:cc:dd:ee:05",),
).fetchone() ).fetchone()
# IPv4 should be set # IPv4 should be set
@@ -627,7 +627,7 @@ def test_scan_updates_ipv6_without_changing_ipv4(scan_db, mock_device_handlers):
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
""", """,
( (
"AA:BB:CC:DD:EE:06", "aa:bb:cc:dd:ee:06",
"2025-01-01 00:00:00", "2025-01-01 00:00:00",
0, 0,
"192.168.1.101", "192.168.1.101",
@@ -657,7 +657,7 @@ def test_scan_updates_ipv6_without_changing_ipv4(scan_db, mock_device_handlers):
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
""", """,
( (
"AA:BB:CC:DD:EE:06", "aa:bb:cc:dd:ee:06",
"fe80::1", "fe80::1",
"Vendor", "Vendor",
"ARPSCAN", "ARPSCAN",
@@ -684,7 +684,7 @@ def test_scan_updates_ipv6_without_changing_ipv4(scan_db, mock_device_handlers):
row = cur.execute( row = cur.execute(
"SELECT devPrimaryIPv4, devPrimaryIPv6 FROM Devices WHERE devMac = ?", "SELECT devPrimaryIPv4, devPrimaryIPv6 FROM Devices WHERE devMac = ?",
("AA:BB:CC:DD:EE:06",), ("aa:bb:cc:dd:ee:06",),
).fetchone() ).fetchone()
# IPv4 should remain, IPv6 should be set # IPv4 should remain, IPv6 should be set
@@ -706,7 +706,7 @@ def test_scan_updates_presence_status(scan_db, mock_device_handlers):
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
""", """,
( (
"AA:BB:CC:DD:EE:07", "aa:bb:cc:dd:ee:07",
"2025-01-01 00:00:00", "2025-01-01 00:00:00",
1, # Was online 1, # Was online
"192.168.1.102", "192.168.1.102",
@@ -737,7 +737,7 @@ def test_scan_updates_presence_status(scan_db, mock_device_handlers):
row = cur.execute( row = cur.execute(
"SELECT devPresentLastScan FROM Devices WHERE devMac = ?", "SELECT devPresentLastScan FROM Devices WHERE devMac = ?",
("AA:BB:CC:DD:EE:07",), ("aa:bb:cc:dd:ee:07",),
).fetchone() ).fetchone()
# Device should be marked as offline # Device should be marked as offline
@@ -750,10 +750,10 @@ def test_scan_multiple_devices_mixed_sources(scan_db, mock_device_handlers):
devices_data = [ devices_data = [
# (MAC, Name, NameSource, Vendor, VendorSource) # (MAC, Name, NameSource, Vendor, VendorSource)
("AA:BB:CC:DD:EE:11", "Device1", "NEWDEV", "", "NEWDEV"), # Both updatable ("aa:bb:cc:dd:ee:11", "Device1", "NEWDEV", "", "NEWDEV"), # Both updatable
("AA:BB:CC:DD:EE:12", "My Device", "USER", "OldVendor", "NEWDEV"), # Name protected ("aa:bb:cc:dd:ee:12", "My Device", "USER", "OldVendor", "NEWDEV"), # Name protected
("AA:BB:CC:DD:EE:13", "Locked Device", "LOCKED", "", "NEWDEV"), # Name locked ("aa:bb:cc:dd:ee:13", "Locked Device", "LOCKED", "", "NEWDEV"), # Name locked
("AA:BB:CC:DD:EE:14", "Device4", "ARPSCAN", "", "NEWDEV"), # Name from plugin ("aa:bb:cc:dd:ee:14", "Device4", "ARPSCAN", "", "NEWDEV"), # Name from plugin
] ]
for mac, name, name_src, vendor, vendor_src in devices_data: for mac, name, name_src, vendor, vendor_src in devices_data:
@@ -786,10 +786,10 @@ def test_scan_multiple_devices_mixed_sources(scan_db, mock_device_handlers):
# Scan discovers all devices with new data # Scan discovers all devices with new data
scan_entries = [ scan_entries = [
("AA:BB:CC:DD:EE:11", "192.168.1.1", "Apple Inc.", "ScanPlugin", "ScannedDevice1"), ("aa:bb:cc:dd:ee:11", "192.168.1.1", "Apple Inc.", "ScanPlugin", "ScannedDevice1"),
("AA:BB:CC:DD:EE:12", "192.168.1.2", "Samsung", "ScanPlugin", "ScannedDevice2"), ("aa:bb:cc:dd:ee:12", "192.168.1.2", "Samsung", "ScanPlugin", "ScannedDevice2"),
("AA:BB:CC:DD:EE:13", "192.168.1.3", "Sony", "ScanPlugin", "ScannedDevice3"), ("aa:bb:cc:dd:ee:13", "192.168.1.3", "Sony", "ScanPlugin", "ScannedDevice3"),
("AA:BB:CC:DD:EE:14", "192.168.1.4", "LG", "ScanPlugin", "ScannedDevice4"), ("aa:bb:cc:dd:ee:14", "192.168.1.4", "LG", "ScanPlugin", "ScannedDevice4"),
] ]
for mac, ip, vendor, scan_method, name in scan_entries: for mac, ip, vendor, scan_method, name in scan_entries:
@@ -815,10 +815,10 @@ def test_scan_multiple_devices_mixed_sources(scan_db, mock_device_handlers):
# Check results # Check results
results = { results = {
"AA:BB:CC:DD:EE:11": {"name": "Device1", "vendor": "Apple Inc."}, # Name already set, won't update "aa:bb:cc:dd:ee:11": {"name": "Device1", "vendor": "Apple Inc."}, # Name already set, won't update
"AA:BB:CC:DD:EE:12": {"name": "My Device", "vendor": "Samsung"}, # Name protected (USER) "aa:bb:cc:dd:ee:12": {"name": "My Device", "vendor": "Samsung"}, # Name protected (USER)
"AA:BB:CC:DD:EE:13": {"name": "Locked Device", "vendor": "Sony"}, # Name locked "aa:bb:cc:dd:ee:13": {"name": "Locked Device", "vendor": "Sony"}, # Name locked
"AA:BB:CC:DD:EE:14": {"name": "Device4", "vendor": "LG"}, # Name already from plugin, won't update "aa:bb:cc:dd:ee:14": {"name": "Device4", "vendor": "LG"}, # Name already from plugin, won't update
} }
for mac, expected in results.items(): for mac, expected in results.items():

28
test/scan/test_ip_format_and_locking.py
View File

@@ -24,10 +24,10 @@ def mock_ip_handlers():
def test_valid_ipv4_format_accepted(scan_db, mock_ip_handlers): def test_valid_ipv4_format_accepted(scan_db, mock_ip_handlers):
"""Valid IPv4 address should be accepted and set as primary IPv4.""" """Valid IPv4 address should be accepted and set as primary IPv4."""
cur = scan_db.cursor() cur = scan_db.cursor()
cur.execute("INSERT INTO Devices (devMac, devName) VALUES (?, ?)", ("AA:BB:CC:DD:EE:01", "Device1")) cur.execute("INSERT INTO Devices (devMac, devName) VALUES (?, ?)", ("ff:ff:cc:dd:ee:01", "Device1"))
cur.execute( cur.execute(
"INSERT INTO CurrentScan (scanMac, scanLastIP, scanSourcePlugin, scanLastConnection) VALUES (?, ?, ?, ?)", "INSERT INTO CurrentScan (scanMac, scanLastIP, scanSourcePlugin, scanLastConnection) VALUES (?, ?, ?, ?)",
("AA:BB:CC:DD:EE:01", "192.168.1.100", "ARPSCAN", "2025-01-01 01:00:00") ("ff:ff:cc:dd:ee:01", "192.168.1.100", "ARPSCAN", "2025-01-01 01:00:00")
) )
scan_db.commit() scan_db.commit()
@@ -35,7 +35,7 @@ def test_valid_ipv4_format_accepted(scan_db, mock_ip_handlers):
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)
row = cur.execute("SELECT devLastIP, devPrimaryIPv4 FROM Devices WHERE devMac = ?", ("AA:BB:CC:DD:EE:01",)).fetchone() row = cur.execute("SELECT devLastIP, devPrimaryIPv4 FROM Devices WHERE devMac = ?", ("ff:ff:cc:dd:ee:01",)).fetchone()
assert row["devLastIP"] == "192.168.1.100" assert row["devLastIP"] == "192.168.1.100"
assert row["devPrimaryIPv4"] == "192.168.1.100" assert row["devPrimaryIPv4"] == "192.168.1.100"
@@ -43,10 +43,10 @@ def test_valid_ipv4_format_accepted(scan_db, mock_ip_handlers):
def test_valid_ipv6_format_accepted(scan_db, mock_ip_handlers): def test_valid_ipv6_format_accepted(scan_db, mock_ip_handlers):
"""Valid IPv6 address should be accepted and set as primary IPv6.""" """Valid IPv6 address should be accepted and set as primary IPv6."""
cur = scan_db.cursor() cur = scan_db.cursor()
cur.execute("INSERT INTO Devices (devMac) VALUES (?)", ("AA:BB:CC:DD:EE:02",)) cur.execute("INSERT INTO Devices (devMac) VALUES (?)", ("ff:ff:cc:dd:ee:02",))
cur.execute( cur.execute(
"INSERT INTO CurrentScan (scanMac, scanLastIP, scanSourcePlugin, scanLastConnection) VALUES (?, ?, ?, ?)", "INSERT INTO CurrentScan (scanMac, scanLastIP, scanSourcePlugin, scanLastConnection) VALUES (?, ?, ?, ?)",
("AA:BB:CC:DD:EE:02", "fe80::1", "ARPSCAN", "2025-01-01 01:00:00") ("ff:ff:cc:dd:ee:02", "fe80::1", "ARPSCAN", "2025-01-01 01:00:00")
) )
scan_db.commit() scan_db.commit()
@@ -54,21 +54,21 @@ def test_valid_ipv6_format_accepted(scan_db, mock_ip_handlers):
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)
row = cur.execute("SELECT devPrimaryIPv6 FROM Devices WHERE devMac = ?", ("AA:BB:CC:DD:EE:02",)).fetchone() row = cur.execute("SELECT devPrimaryIPv6 FROM Devices WHERE devMac = ?", ("ff:ff:cc:dd:ee:02",)).fetchone()
assert row["devPrimaryIPv6"] == "fe80::1" assert row["devPrimaryIPv6"] == "fe80::1"
def test_invalid_ip_values_rejected(scan_db, mock_ip_handlers): def test_invalid_ip_values_rejected(scan_db, mock_ip_handlers):
"""Invalid IP values like (unknown), null, empty should be rejected.""" """Invalid IP values like (unknown), null, empty should be rejected."""
cur = scan_db.cursor() cur = scan_db.cursor()
cur.execute("INSERT INTO Devices (devMac, devPrimaryIPv4) VALUES (?, ?)", ("AA:BB:CC:DD:EE:03", "192.168.1.50")) cur.execute("INSERT INTO Devices (devMac, devPrimaryIPv4) VALUES (?, ?)", ("ff:ff:cc:dd:ee:03", "192.168.1.50"))
invalid_ips = ["", "null", "(unknown)", "(Unknown)"] invalid_ips = ["", "null", "(unknown)", "(Unknown)"]
for invalid_ip in invalid_ips: for invalid_ip in invalid_ips:
cur.execute("DELETE FROM CurrentScan") cur.execute("DELETE FROM CurrentScan")
cur.execute( cur.execute(
"INSERT INTO CurrentScan (scanMac, scanLastIP, scanSourcePlugin, scanLastConnection) VALUES (?, ?, ?, ?)", "INSERT INTO CurrentScan (scanMac, scanLastIP, scanSourcePlugin, scanLastConnection) VALUES (?, ?, ?, ?)",
("AA:BB:CC:DD:EE:03", invalid_ip, "ARPSCAN", "2025-01-01 01:00:00") ("ff:ff:cc:dd:ee:03", invalid_ip, "ARPSCAN", "2025-01-01 01:00:00")
) )
scan_db.commit() scan_db.commit()
@@ -76,7 +76,7 @@ def test_invalid_ip_values_rejected(scan_db, mock_ip_handlers):
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)
row = cur.execute("SELECT devPrimaryIPv4 FROM Devices WHERE devMac = ?", ("AA:BB:CC:DD:EE:03",)).fetchone() row = cur.execute("SELECT devPrimaryIPv4 FROM Devices WHERE devMac = ?", ("ff:ff:cc:dd:ee:03",)).fetchone()
assert row["devPrimaryIPv4"] == "192.168.1.50", f"Failed on {invalid_ip}" assert row["devPrimaryIPv4"] == "192.168.1.50", f"Failed on {invalid_ip}"
@@ -89,7 +89,7 @@ def test_ipv4_then_ipv6_scan_updates_primary_ips(scan_db, mock_ip_handlers):
cur = scan_db.cursor() cur = scan_db.cursor()
# 1️⃣ Create device # 1️⃣ Create device
cur.execute("INSERT INTO Devices (devMac) VALUES (?)", ("AA:BB:CC:DD:EE:04",)) cur.execute("INSERT INTO Devices (devMac) VALUES (?)", ("ff:ff:cc:dd:ee:04",))
scan_db.commit() scan_db.commit()
db = Mock(sql_connection=scan_db, sql=cur) db = Mock(sql_connection=scan_db, sql=cur)
@@ -97,7 +97,7 @@ def test_ipv4_then_ipv6_scan_updates_primary_ips(scan_db, mock_ip_handlers):
# 2️⃣ First scan: IPv4 # 2️⃣ First scan: IPv4
cur.execute( cur.execute(
"INSERT INTO CurrentScan (scanMac, scanLastIP, scanSourcePlugin, scanLastConnection) VALUES (?, ?, ?, ?)", "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") ("ff:ff:cc:dd:ee:04", "192.168.1.100", "ARPSCAN", "2025-01-01 01:00:00")
) )
scan_db.commit() scan_db.commit()
@@ -109,7 +109,7 @@ def test_ipv4_then_ipv6_scan_updates_primary_ips(scan_db, mock_ip_handlers):
cur.execute("DELETE FROM CurrentScan") cur.execute("DELETE FROM CurrentScan")
cur.execute( cur.execute(
"INSERT INTO CurrentScan (scanMac, scanLastIP, scanSourcePlugin, scanLastConnection) VALUES (?, ?, ?, ?)", "INSERT INTO CurrentScan (scanMac, scanLastIP, scanSourcePlugin, scanLastConnection) VALUES (?, ?, ?, ?)",
("AA:BB:CC:DD:EE:04", "fe80::1", "IPv6SCAN", "2025-01-01 02:00:00") ("ff:ff:cc:dd:ee:04", "fe80::1", "IPv6SCAN", "2025-01-01 02:00:00")
) )
scan_db.commit() scan_db.commit()
@@ -120,7 +120,7 @@ def test_ipv4_then_ipv6_scan_updates_primary_ips(scan_db, mock_ip_handlers):
# 4️⃣ Verify results # 4️⃣ Verify results
row = cur.execute( row = cur.execute(
"SELECT devLastIP, devPrimaryIPv4, devPrimaryIPv6 FROM Devices WHERE devMac = ?", "SELECT devLastIP, devPrimaryIPv4, devPrimaryIPv6 FROM Devices WHERE devMac = ?",
("AA:BB:CC:DD:EE:04",) ("ff:ff:cc:dd:ee:04",)
).fetchone() ).fetchone()
assert row["devLastIP"] == "fe80::1" # Latest scan IP (IPv6) assert row["devLastIP"] == "fe80::1" # Latest scan IP (IPv6)
@@ -134,7 +134,7 @@ def test_ipv4_address_format_variations(scan_db, mock_ip_handlers):
ipv4_addresses = ["1.1.1.1", "127.0.0.1", "192.168.1.1", "255.255.255.255"] ipv4_addresses = ["1.1.1.1", "127.0.0.1", "192.168.1.1", "255.255.255.255"]
for idx, ipv4 in enumerate(ipv4_addresses): for idx, ipv4 in enumerate(ipv4_addresses):
mac = f"AA:BB:CC:DD:11:{idx:02X}" mac = f"AA:BB:CC:DD:11:{idx:02X}".lower()
cur.execute("INSERT INTO Devices (devMac) VALUES (?)", (mac,)) cur.execute("INSERT INTO Devices (devMac) VALUES (?)", (mac,))
cur.execute("INSERT INTO CurrentScan (scanMac, scanLastIP, scanSourcePlugin, scanLastConnection) VALUES (?, ?, ?, ?)", cur.execute("INSERT INTO CurrentScan (scanMac, scanLastIP, scanSourcePlugin, scanLastConnection) VALUES (?, ?, ?, ?)",
(mac, ipv4, "SCAN", "2025-01-01 01:00:00")) (mac, ipv4, "SCAN", "2025-01-01 01:00:00"))

20
test/test_plugin_helper.py
View File

@@ -2,23 +2,29 @@ from front.plugins.plugin_helper import is_mac, normalize_mac
def test_is_mac_accepts_wildcard(): def test_is_mac_accepts_wildcard():
assert is_mac("AA:BB:CC:*") is True # is_mac checks structure, so it should still return True
assert is_mac("aa-bb-cc:*") is True # mixed separator assert is_mac("aa:bb:cc:*") is True
assert is_mac("AA-BB-CC:*") is True # mixed case/separator should still be recognized
assert is_mac("00:11:22:33:44:55") is True assert is_mac("00:11:22:33:44:55") is True
assert is_mac("00-11-22-33-44-55") is True assert is_mac("00-11-22-33-44-55") is True
assert is_mac("not-a-mac") is False assert is_mac("not-a-mac") is False
def test_normalize_mac_preserves_wildcard(): def test_normalize_mac_preserves_wildcard():
assert normalize_mac("aa:bb:cc:*") == "AA:BB:CC:*" # UPDATED: Expected results are now lowercase to match the DB standard
assert normalize_mac("aa-bb-cc-*") == "AA:BB:CC:*" assert normalize_mac("aa:bb:cc:*") == "aa:bb:cc:*"
assert normalize_mac("AA-BB-CC-*") == "aa:bb:cc:*"
# Call once and assert deterministic result # Call once and assert deterministic result
result = normalize_mac("aabbcc*") result = normalize_mac("aabbcc*")
assert result == "AA:BB:CC:*", f"Expected 'AA:BB:CC:*' but got '{result}'" assert result == "aa:bb:cc:*", f"Expected 'aa:bb:cc:*' but got '{result}'"
assert normalize_mac("aa:bb:cc:dd:ee:ff") == "aa:bb:cc:dd:ee:ff"
# Ensure full MACs are lowercase too
assert normalize_mac("AA:BB:CC:DD:EE:FF") == "aa:bb:cc:dd:ee:ff"
def test_normalize_mac_preserves_internet_root(): def test_normalize_mac_preserves_internet_root():
# Stays lowercase
assert normalize_mac("internet") == "internet" assert normalize_mac("internet") == "internet"
assert normalize_mac("Internet") == "internet" assert normalize_mac("Internet") == "internet"
assert normalize_mac("INTERNET") == "internet" assert normalize_mac("INTERNET") == "internet"