mirror/NetAlertX
1
0 Fork 0
mirror of https://github.com/jokob-sk/NetAlertX.git synced 2026-10-28 03:51:08 -04:00
Code Issues Packages Projects Releases Wiki Activity

BE: conditional device import support, notification supression support during scan #1721

Browse Source
This commit is contained in:
jokob-sk committed 2026-09-10 18:41:07 +10:00
1 parent 34993ed77b
commit de730e85e9
13 files changed
+359 -778
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

docs/PIHOLE_GUIDE.md
+2 -2
View File
@@ -1,6 +1,6 @@
# Integration with Pi-hole
NetAlertX includes four plugins for integrating with an existing Pi-hole installation. The first plugin imports devices through the Pi-hole v6 API, the second parses the `dhcp.leases` file generated by Pi-hole, the third reads the Pi-hole SQLite database directly, and the fourth flags devices with a blocked-query spike (a common malware/compromised-device signature) and, alongside that, imports devices from one or two Pi-hole v6 instances at once. You can use any of these approaches individually or combine them with each other and other [plugins](/docs/PLUGINS_OVERVIEW.md).
NetAlertX includes four plugins for integrating with an existing Pi-hole installation. The first plugin imports devices through the Pi-hole v6 API, the second parses the `dhcp.leases` file generated by Pi-hole, the third reads the Pi-hole SQLite database directly, and the fourth flags devices with a blocked-query spike (a common malware/compromised-device signature) and, alongside that, imports devices from one or two Pi-hole v6 instances at once. You can use any of these approaches individually or combine them with each other and other [plugins](/PLUGINS_OVERVIEW).
## Approach 1: `PIHOLEAPI` Plugin - Import devices directly from the Pi-hole v6 API
@@ -93,4 +93,4 @@ No changes are required - this plugin only talks to Pi-hole's API, the same as `
---
Explore other [plugins](/docs/PLUGINS_OVERVIEW.md) to discover additional information about your network, or learn how to scan [remote networks](./REMOTE_NETWORKS.md).
Explore other [plugins](./PLUGINS_OVERVIEW.md) to discover additional information about your network, or learn how to scan [remote networks](./REMOTE_NETWORKS.md).
docs/README.md
+1 -1
View File
@@ -36,7 +36,7 @@ There is also an in-app Help / FAQ section that should be answering frequently a
- [Better name resolution with Reverse DNS](./REVERSE_DNS.md)
- [Network treemap configuration](./NETWORK_TREE.md)
- [Backups](./BACKUPS.md)
- [Plugins overview](/docs/PLUGINS_OVERVIEW.md)
- [Plugins overview](./PLUGINS_OVERVIEW.md)
#### 🐛 Debugging help & tips
docs/SYNOLOGY_GUIDE.md
+1 -1
View File
@@ -80,7 +80,7 @@ volumes:
![Build](./img/SYNOLOGY/09_Run_and_build.png)
10. Navigate to `<Synology URL>:20211` (or your custom port).
11. Read the [Subnets](./SUBNETS.md) and [Plugins](/docs/PLUGINS_OVERVIEW.md) docs to complete your setup.
11. Read the [Subnets](./SUBNETS.md) and [Plugins](./PLUGINS_OVERVIEW.md) docs to complete your setup.
## Solving permission issues
server/plugins/README.md
+1 -1
View File
@@ -1 +1 @@
Plugins docs have been relocated. Check the new [Plugins Overview location](/docs/PLUGINS_OVERVIEW.md).
Plugins docs have been relocated. Check the new [Plugins Overview location](./PLUGINS_OVERVIEW.md).
server/plugins/_publisher_email/README.md
+1 -1
View File
@@ -1,6 +1,6 @@
## Overview
A simple EMail (SMTP) notification gateway publisher. Check the [SMTP docs](/docs/SMTP.md) for additional help.
A simple EMail (SMTP) notification gateway publisher. Check the [SMTP docs](./SMTP.md) for additional help.
### Usage
server/plugins/notification_processing/README.md
+1 -1
View File
@@ -56,5 +56,5 @@ Certain notifications can be disabled per device:
- Review the **Settings** page to configure which notification types should be enabled.
- Ensure that device-specific alert settings align with your requirements.
For additional details, check the [Notifications Guide](/docs/NOTIFICATIONS.md).
For additional details, check the [Notifications Guide](/NOTIFICATIONS).
server/plugins/sync/README.md
+1 -1
View File
@@ -7,7 +7,7 @@ The synchronization plugin is designed to synchronize data across multiple insta
> [!TIP]
> This plugin is usually used if you need to monitor inaccessible networks (WAN, VLAN etc.). Read the [Remote networks documentation](/docs/REMOTE_NETWORKS.md) for more details about these scenarios.
> This plugin is usually used if you need to monitor inaccessible networks (WAN, VLAN etc.). Read the [Remote networks documentation](./REMOTE_NETWORKS.md) for more details about these scenarios.
### Synchronization Modes
server/scan/session_events.py
+12 -2
View File
@@ -227,6 +227,11 @@ def insert_events(db):
# - quiet_agg: unrestricted by scanPresence on purpose - a plugin's quiet
# preference counts even from a row that isn't the one asserting
# presence (most-restrictive-wins is a separate axis from presence).
# This runs before create_new_devices(), so a legitimately new device
# (scanCreatesDevice=1 this cycle) has no Devices row yet either - the
# final WHERE clause treats "will be created this cycle" the same as
# "already exists" rather than requiring a Devices row, or a
# scanCreatesDevice=0-only MAC would suppress the mainline case too.
sql.execute(f""" INSERT OR IGNORE INTO Events (eveMac, eveIp, eveDateTime,
eveEventType, eveAdditionalInfo,
evePendingAlertEmail)
@@ -245,12 +250,17 @@ def insert_events(db):
) present_agg
JOIN (
SELECT scanMac,
MAX(CASE WHEN scanNotificationMode = 'quiet' THEN 1 ELSE 0 END) AS scanQuiet
MAX(CASE WHEN scanNotificationMode = 'quiet' THEN 1 ELSE 0 END) AS scanQuiet,
MAX(scanCreatesDevice) AS scanCreates
FROM CurrentScan
GROUP BY scanMac
) quiet_agg ON quiet_agg.scanMac = present_agg.scanMac
LEFT JOIN LatestEventsPerMAC AS last_event ON present_agg.scanMac = last_event.eveMac
WHERE last_event.devPresentLastScan = 0 OR last_event.eveMac IS NULL
WHERE (last_event.devPresentLastScan = 0 OR last_event.eveMac IS NULL)
AND (
quiet_agg.scanCreates = 1
OR EXISTS (SELECT 1 FROM Devices WHERE devMac = present_agg.scanMac)
)
""")
# Check disconnections
test/scan/conftest.py
+18 -135
View File
@@ -1,143 +1,26 @@
import os
import sys
import pytest
import sqlite3
sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
from db_test_helpers import make_db # noqa: E402
@pytest.fixture
def scan_db():
"""Centralized in-memory SQLite database for all integration tests."""
conn = sqlite3.connect(":memory:")
conn.row_factory = sqlite3.Row
cur = conn.cursor()
"""Centralized in-memory SQLite database for scan integration tests.
# 1. Comprehensive Devices Table
cur.execute("""
CREATE TABLE Devices (
devMac TEXT PRIMARY KEY,
devLastConnection TEXT,
devFirstConnection TEXT,
devPresentLastScan INTEGER DEFAULT 0,
devForceStatus TEXT,
devLastIP TEXT,
devPrimaryIPv4 TEXT,
devPrimaryIPv6 TEXT,
devVendor TEXT,
devParentPort TEXT,
devParentMAC TEXT,
devParentRelType TEXT,
devSite TEXT,
devSSID TEXT,
devType TEXT,
devName TEXT,
devIcon TEXT,
devGUID TEXT,
devFQDN TEXT,
devSyncHubNode TEXT,
devOwner TEXT,
devGroup TEXT,
devLocation TEXT,
devComments TEXT,
devCustomProps TEXT,
devIsArchived INTEGER DEFAULT 0,
devIsNew INTEGER DEFAULT 1,
devFavorite INTEGER DEFAULT 0,
devScan INTEGER DEFAULT 1,
devAlertDown INTEGER DEFAULT 0,
devAlertEvents INTEGER DEFAULT 1,
-- Authoritative Metadata Columns
devMacSource TEXT,
devNameSource TEXT,
devVendorSource TEXT,
devLastIPSource TEXT,
devTypeSource TEXT,
devSSIDSource TEXT,
devParentMACSource TEXT,
devParentPortSource TEXT,
devParentRelTypeSource TEXT,
devFQDNSource TEXT,
devVlanSource TEXT,
-- Field Locking Columns
devNameLocked INTEGER DEFAULT 0,
devTypeLocked INTEGER DEFAULT 0,
devIconLocked INTEGER DEFAULT 0
)
""")
# 2. CurrentScan Table
cur.execute("""
CREATE TABLE CurrentScan (
scanMac TEXT,
scanLastIP TEXT,
scanVendor TEXT,
scanSourcePlugin TEXT,
scanName TEXT,
scanLastQuery TEXT,
scanLastConnection TEXT,
scanSyncHubNode TEXT,
scanSite TEXT,
scanSSID TEXT,
scanParentMAC TEXT,
scanParentPort TEXT,
scanType TEXT,
scanCreatesDevice INTEGER NOT NULL DEFAULT 1,
scanNotificationMode TEXT NOT NULL DEFAULT 'normal',
scanPresence INTEGER NOT NULL DEFAULT 1
)
""")
# 3. Events Table
cur.execute("""
CREATE TABLE Events (
eve_MAC TEXT,
eve_IP TEXT,
eve_DateTime TEXT,
eve_EventType TEXT,
eve_AdditionalInfo TEXT,
eve_PendingAlertEmail INTEGER
)
""")
# 4. LatestEventsPerMAC View
cur.execute("""DROP VIEW IF EXISTS LatestEventsPerMAC;""")
cur.execute("""
CREATE VIEW LatestEventsPerMAC AS
WITH RankedEvents AS (
SELECT
e.*,
ROW_NUMBER() OVER (PARTITION BY e.eve_MAC ORDER BY e.eve_DateTime DESC) AS row_num
FROM Events AS e
)
SELECT
e.eve_MAC,
e.eve_EventType,
e.eve_DateTime,
e.eve_PendingAlertEmail,
d.devPresentLastScan,
c.scanLastIP
FROM RankedEvents AS e
LEFT JOIN Devices AS d ON e.eve_MAC = d.devMac
LEFT JOIN CurrentScan AS c ON e.eve_MAC = c.scanMac
WHERE e.row_num = 1;
""")
# 3. LatestDeviceScan View (Inner Join for Online Devices)
cur.execute("""
CREATE VIEW LatestDeviceScan AS
WITH RankedScans AS (
SELECT
c.*,
ROW_NUMBER() OVER (
PARTITION BY c.scanMac, c.scanSourcePlugin
ORDER BY c.scanLastConnection DESC
) AS rn
FROM CurrentScan c
)
SELECT d.*, r.* FROM Devices d
INNER JOIN RankedScans r ON d.devMac = r.scanMac
WHERE r.rn = 1;
""")
conn.commit()
Delegates to db_test_helpers.make_db() - the canonical schema/views
shared by the rest of the suite - rather than a local copy. The local
copy this replaced had drifted from production: Events used pre-rename
column names (eve_MAC instead of eveMac, see migrate_to_camelcase()),
and Devices carried devNameLocked/devTypeLocked/devIconLocked/
devTypeSource columns that don't exist in server/db/schema/app.sql and
that no production code reads or writes - field locking is actually
expressed via the LOCKED/USER/NEWDEV/<plugin> sentinel values already
stored in the real *Source columns, not separate boolean columns.
"""
conn = make_db()
yield conn
conn.close()
test/scan/test_field_lock_scan_integration.py
+208 -542
View File
@@ -11,13 +11,23 @@ Tests all combinations of field sources (LOCKED, USER, NEWDEV, plugin name)
with realistic scan data.
"""
import sqlite3
import os
import sys
from unittest.mock import Mock, patch
import pytest
from server.scan import device_handling
sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
from db_test_helpers import ( # noqa: E402
DummyDB,
insert_current_scan_row_from_dict,
insert_device_from_dict,
make_current_scan_dict,
make_device_dict,
)
@pytest.fixture
def mock_device_handlers():
@@ -40,142 +50,21 @@ def mock_device_handlers():
yield
@pytest.fixture
def scan_db_for_new_devices():
"""Create an in-memory SQLite database for create_new_devices tests."""
conn = sqlite3.connect(":memory:")
conn.row_factory = sqlite3.Row
cur = conn.cursor()
cur.execute(
"""
CREATE TABLE Devices (
devMac TEXT PRIMARY KEY,
devName TEXT,
devVendor TEXT,
devLastIP TEXT,
devPrimaryIPv4 TEXT,
devPrimaryIPv6 TEXT,
devFirstConnection TEXT,
devLastConnection TEXT,
devSyncHubNode TEXT,
devGUID TEXT,
devParentMAC TEXT,
devParentPort TEXT,
devSite TEXT,
devSSID TEXT,
devType TEXT,
devSourcePlugin TEXT,
devMacSource TEXT,
devNameSource TEXT,
devFQDNSource TEXT,
devLastIPSource TEXT,
devVendorSource TEXT,
devSSIDSource TEXT,
devParentMACSource TEXT,
devParentPortSource TEXT,
devParentRelTypeSource TEXT,
devVlanSource TEXT,
devAlertEvents INTEGER,
devAlertDown INTEGER,
devPresentLastScan INTEGER,
devIsArchived INTEGER,
devIsNew INTEGER,
devSkipRepeated INTEGER,
devScan INTEGER,
devOwner TEXT,
devFavorite INTEGER,
devGroup TEXT,
devComments TEXT,
devLogEvents INTEGER,
devLocation TEXT,
devCustomProps TEXT,
devParentRelType TEXT,
devReqNicsOnline INTEGER
)
"""
)
cur.execute(
"""
CREATE TABLE CurrentScan (
scanMac TEXT,
scanName TEXT,
scanVendor TEXT,
scanSourcePlugin TEXT,
scanLastIP TEXT,
scanSyncHubNode TEXT,
scanParentMAC TEXT,
scanParentPort TEXT,
scanSite TEXT,
scanSSID TEXT,
scanType TEXT,
scanCreatesDevice INTEGER NOT NULL DEFAULT 1,
scanNotificationMode TEXT NOT NULL DEFAULT 'normal',
scanPresence INTEGER NOT NULL DEFAULT 1
)
"""
)
cur.execute(
"""
CREATE TABLE Events (
eveMac TEXT,
eveIp TEXT,
eveDateTime TEXT,
eveEventType TEXT,
eveAdditionalInfo TEXT,
evePendingAlertEmail INTEGER
)
"""
)
cur.execute(
"""
CREATE TABLE Sessions (
sesMac TEXT,
sesIp TEXT,
sesEventTypeConnection TEXT,
sesDateTimeConnection TEXT,
sesEventTypeDisconnection TEXT,
sesDateTimeDisconnection TEXT,
sesStillConnected INTEGER,
sesAdditionalInfo TEXT
)
"""
)
conn.commit()
yield conn
conn.close()
def test_create_new_devices_sets_sources(scan_db_for_new_devices):
def test_create_new_devices_sets_sources(scan_db):
"""New device insert initializes source fields from scan method."""
cur = scan_db_for_new_devices.cursor()
cur.execute(
"""
INSERT INTO CurrentScan (
scanMac, scanName, scanVendor, scanSourcePlugin, scanLastIP,
scanSyncHubNode, scanParentMAC, scanParentPort,
scanSite, scanSSID, scanType
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
""",
(
insert_current_scan_row_from_dict(
scan_db,
make_current_scan_dict(
"aa:bb:cc:dd:ee:10",
"DeviceOne",
"AcmeVendor",
"ARPSCAN",
"192.168.1.10",
"",
"11:22:33:44:55:66",
"1",
"",
"MyWifi",
"",
scanName="DeviceOne",
scanVendor="AcmeVendor",
scanSourcePlugin="ARPSCAN",
scanLastIP="192.168.1.10",
scanParentMAC="11:22:33:44:55:66",
scanParentPort="1",
scanSSID="MyWifi",
),
)
scan_db_for_new_devices.commit()
settings = {
"NEWDEV_devType": "default-type",
@@ -189,22 +78,16 @@ def test_create_new_devices_sets_sources(scan_db_for_new_devices):
"SYNC_node_name": "SYNCNODE",
}
def get_setting_value_side_effect(key):
return settings.get(key, "")
db = Mock()
db.sql_connection = scan_db_for_new_devices
db.sql = cur
db.commitDB = scan_db_for_new_devices.commit
db = DummyDB(scan_db)
with patch.multiple(
device_handling,
get_setting_value=Mock(side_effect=get_setting_value_side_effect),
get_setting_value=Mock(side_effect=lambda key: settings.get(key, "")),
safe_int=Mock(return_value=0),
):
device_handling.create_new_devices(db)
row = cur.execute(
row = scan_db.execute(
"""
SELECT
devMacSource,
@@ -234,33 +117,19 @@ def test_create_new_devices_sets_sources(scan_db_for_new_devices):
assert row["devVlanSource"] == "NEWDEV"
def test_create_new_devices_ignores_dangling_newdev_parentmac(scan_db_for_new_devices):
def test_create_new_devices_ignores_dangling_newdev_parentmac(scan_db):
"""A stale NEWDEV_devParentMAC pointing to a since-deleted device is treated as unset,
instead of seeding the new device with another dangling Parent Node reference."""
cur = scan_db_for_new_devices.cursor()
cur.execute(
"""
INSERT INTO CurrentScan (
scanMac, scanName, scanVendor, scanSourcePlugin, scanLastIP,
scanSyncHubNode, scanParentMAC, scanParentPort,
scanSite, scanSSID, scanType
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
""",
(
insert_current_scan_row_from_dict(
scan_db,
make_current_scan_dict(
"aa:bb:cc:dd:ee:11",
"DeviceTwo",
"AcmeVendor",
"ARPSCAN",
"192.168.1.11",
"",
"", # no parent reported by the scan itself
"",
"",
"",
"",
scanName="DeviceTwo",
scanVendor="AcmeVendor",
scanSourcePlugin="ARPSCAN",
scanLastIP="192.168.1.11",
),
)
scan_db_for_new_devices.commit()
settings = {
"NEWDEV_devType": "default-type",
@@ -275,10 +144,7 @@ def test_create_new_devices_ignores_dangling_newdev_parentmac(scan_db_for_new_de
"SYNC_node_name": "SYNCNODE",
}
db = Mock()
db.sql_connection = scan_db_for_new_devices
db.sql = cur
db.commitDB = scan_db_for_new_devices.commit
db = DummyDB(scan_db)
with patch.multiple(
device_handling,
@@ -287,7 +153,7 @@ def test_create_new_devices_ignores_dangling_newdev_parentmac(scan_db_for_new_de
):
device_handling.create_new_devices(db)
row = cur.execute(
row = scan_db.execute(
"SELECT devParentMAC FROM Devices WHERE devMac = ?", ("aa:bb:cc:dd:ee:11",)
).fetchone()
@@ -296,71 +162,43 @@ def test_create_new_devices_ignores_dangling_newdev_parentmac(scan_db_for_new_de
def test_scan_updates_newdev_device_name(scan_db, mock_device_handlers):
"""Scanner discovers name for device with NEWDEV source."""
cur = scan_db.cursor()
# Device with empty name (NEWDEV)
cur.execute(
"""
INSERT INTO Devices (
devMac, devLastConnection, devPresentLastScan, devLastIP,
devName, devNameSource, devVendor, devVendorSource, devLastIPSource,
devType, devIcon, devParentPort, devParentMAC, devSite, devSSID
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
""",
(
insert_device_from_dict(
scan_db,
make_device_dict(
"aa:bb:cc:dd:ee:01",
"2025-01-01 00:00:00",
0,
"192.168.1.1",
"", # No name yet
"NEWDEV", # Default/unset
"TestVendor",
"NEWDEV",
"ARPSCAN",
"type",
"icon",
"",
"",
"",
"",
devLastConnection="2025-01-01 00:00:00",
devPresentLastScan=0,
devLastIP="192.168.1.1",
devName="", # No name yet
devNameSource="NEWDEV",
devVendor="TestVendor",
devVendorSource="NEWDEV",
devLastIPSource="ARPSCAN",
),
)
# Scanner discovers name
cur.execute(
"""
INSERT INTO CurrentScan (
scanMac, scanLastIP, scanVendor, scanSourcePlugin, scanName,
scanLastQuery, scanLastConnection, scanSyncHubNode,
scanSite, scanSSID, scanParentMAC, scanParentPort, scanType
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
""",
(
insert_current_scan_row_from_dict(
scan_db,
make_current_scan_dict(
"aa:bb:cc:dd:ee:01",
"192.168.1.1",
"TestVendor",
"NBTSCAN",
"DiscoveredDevice",
"",
"2025-01-01 01:00:00",
"",
"",
"",
"",
"",
"",
scanLastIP="192.168.1.1",
scanVendor="TestVendor",
scanSourcePlugin="NBTSCAN",
scanName="DiscoveredDevice",
scanLastQuery="",
scanLastConnection="2025-01-01 01:00:00",
),
)
scan_db.commit()
db = Mock()
db.sql_connection = scan_db
db.sql = cur
db.sql = scan_db.cursor()
# Run scan update
device_handling.update_devices_data_from_scan(db)
row = cur.execute(
row = scan_db.execute(
"SELECT devName FROM Devices WHERE devMac = ?",
("aa:bb:cc:dd:ee:01",),
).fetchone()
@@ -371,71 +209,43 @@ def test_scan_updates_newdev_device_name(scan_db, mock_device_handlers):
def test_scan_does_not_update_user_field_name(scan_db, mock_device_handlers):
"""Scanner cannot override devName when source is USER."""
cur = scan_db.cursor()
# Device with USER-edited name
cur.execute(
"""
INSERT INTO Devices (
devMac, devLastConnection, devPresentLastScan, devLastIP,
devName, devNameSource, devVendor, devVendorSource, devLastIPSource,
devType, devIcon, devParentPort, devParentMAC, devSite, devSSID
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
""",
(
insert_device_from_dict(
scan_db,
make_device_dict(
"aa:bb:cc:dd:ee:02",
"2025-01-01 00:00:00",
0,
"192.168.1.2",
"My Custom Device",
"USER", # User-owned
"TestVendor",
"NEWDEV",
"ARPSCAN",
"type",
"icon",
"",
"",
"",
"",
devLastConnection="2025-01-01 00:00:00",
devPresentLastScan=0,
devLastIP="192.168.1.2",
devName="My Custom Device",
devNameSource="USER", # User-owned
devVendor="TestVendor",
devVendorSource="NEWDEV",
devLastIPSource="ARPSCAN",
),
)
# Scanner tries to update name
cur.execute(
"""
INSERT INTO CurrentScan (
scanMac, scanLastIP, scanVendor, scanSourcePlugin, scanName,
scanLastQuery, scanLastConnection, scanSyncHubNode,
scanSite, scanSSID, scanParentMAC, scanParentPort, scanType
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
""",
(
insert_current_scan_row_from_dict(
scan_db,
make_current_scan_dict(
"aa:bb:cc:dd:ee:02",
"192.168.1.2",
"TestVendor",
"NBTSCAN",
"ScannedDevice",
"",
"2025-01-01 01:00:00",
"",
"",
"",
"",
"",
"",
scanLastIP="192.168.1.2",
scanVendor="TestVendor",
scanSourcePlugin="NBTSCAN",
scanName="ScannedDevice",
scanLastQuery="",
scanLastConnection="2025-01-01 01:00:00",
),
)
scan_db.commit()
db = Mock()
db.sql_connection = scan_db
db.sql = cur
db.sql = scan_db.cursor()
# Run scan update
device_handling.update_devices_data_from_scan(db)
row = cur.execute(
row = scan_db.execute(
"SELECT devName FROM Devices WHERE devMac = ?",
("aa:bb:cc:dd:ee:02",),
).fetchone()
@@ -446,71 +256,43 @@ def test_scan_does_not_update_user_field_name(scan_db, mock_device_handlers):
def test_scan_does_not_update_locked_field(scan_db, mock_device_handlers):
"""Scanner cannot override LOCKED devName."""
cur = scan_db.cursor()
# Device with LOCKED name
cur.execute(
"""
INSERT INTO Devices (
devMac, devLastConnection, devPresentLastScan, devLastIP,
devName, devNameSource, devVendor, devVendorSource, devLastIPSource,
devType, devIcon, devParentPort, devParentMAC, devSite, devSSID
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
""",
(
insert_device_from_dict(
scan_db,
make_device_dict(
"aa:bb:cc:dd:ee:03",
"2025-01-01 00:00:00",
0,
"192.168.1.3",
"Important Device",
"LOCKED", # Locked
"TestVendor",
"NEWDEV",
"ARPSCAN",
"type",
"icon",
"",
"",
"",
"",
devLastConnection="2025-01-01 00:00:00",
devPresentLastScan=0,
devLastIP="192.168.1.3",
devName="Important Device",
devNameSource="LOCKED", # Locked
devVendor="TestVendor",
devVendorSource="NEWDEV",
devLastIPSource="ARPSCAN",
),
)
# Scanner tries to update name
cur.execute(
"""
INSERT INTO CurrentScan (
scanMac, scanLastIP, scanVendor, scanSourcePlugin, scanName,
scanLastQuery, scanLastConnection, scanSyncHubNode,
scanSite, scanSSID, scanParentMAC, scanParentPort, scanType
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
""",
(
insert_current_scan_row_from_dict(
scan_db,
make_current_scan_dict(
"aa:bb:cc:dd:ee:03",
"192.168.1.3",
"TestVendor",
"NBTSCAN",
"Unknown",
"",
"2025-01-01 01:00:00",
"",
"",
"",
"",
"",
"",
scanLastIP="192.168.1.3",
scanVendor="TestVendor",
scanSourcePlugin="NBTSCAN",
scanName="Unknown",
scanLastQuery="",
scanLastConnection="2025-01-01 01:00:00",
),
)
scan_db.commit()
db = Mock()
db.sql_connection = scan_db
db.sql = cur
db.sql = scan_db.cursor()
# Run scan update
device_handling.update_devices_data_from_scan(db)
row = cur.execute(
row = scan_db.execute(
"SELECT devName FROM Devices WHERE devMac = ?",
("aa:bb:cc:dd:ee:03",),
).fetchone()
@@ -521,71 +303,43 @@ def test_scan_does_not_update_locked_field(scan_db, mock_device_handlers):
def test_scan_updates_empty_vendor_field(scan_db, mock_device_handlers):
"""Scan updates vendor when it's empty/NULL."""
cur = scan_db.cursor()
# Device with empty vendor
cur.execute(
"""
INSERT INTO Devices (
devMac, devLastConnection, devPresentLastScan, devLastIP,
devName, devNameSource, devVendor, devVendorSource, devLastIPSource,
devType, devIcon, devParentPort, devParentMAC, devSite, devSSID
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
""",
(
insert_device_from_dict(
scan_db,
make_device_dict(
"aa:bb:cc:dd:ee:04",
"2025-01-01 00:00:00",
0,
"192.168.1.4",
"Device",
"NEWDEV",
"", # Empty vendor
"NEWDEV",
"ARPSCAN",
"type",
"icon",
"",
"",
"",
"",
devLastConnection="2025-01-01 00:00:00",
devPresentLastScan=0,
devLastIP="192.168.1.4",
devName="Device",
devNameSource="NEWDEV",
devVendor="", # Empty vendor
devVendorSource="NEWDEV",
devLastIPSource="ARPSCAN",
),
)
# Scan discovers vendor
cur.execute(
"""
INSERT INTO CurrentScan (
scanMac, scanLastIP, scanVendor, scanSourcePlugin, scanName,
scanLastQuery, scanLastConnection, scanSyncHubNode,
scanSite, scanSSID, scanParentMAC, scanParentPort, scanType
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
""",
(
insert_current_scan_row_from_dict(
scan_db,
make_current_scan_dict(
"aa:bb:cc:dd:ee:04",
"192.168.1.4",
"Apple Inc.",
"ARPSCAN",
"",
"",
"2025-01-01 01:00:00",
"",
"",
"",
"",
"",
"",
scanLastIP="192.168.1.4",
scanVendor="Apple Inc.",
scanSourcePlugin="ARPSCAN",
scanName="",
scanLastQuery="",
scanLastConnection="2025-01-01 01:00:00",
),
)
scan_db.commit()
db = Mock()
db.sql_connection = scan_db
db.sql = cur
db.sql = scan_db.cursor()
# Run scan update
device_handling.update_devices_data_from_scan(db)
row = cur.execute(
row = scan_db.execute(
"SELECT devVendor FROM Devices WHERE devMac = ?",
("aa:bb:cc:dd:ee:04",),
).fetchone()
@@ -596,75 +350,46 @@ def test_scan_updates_empty_vendor_field(scan_db, mock_device_handlers):
def test_scan_updates_ip_addresses(scan_db, mock_device_handlers):
"""Scan updates IPv4 and IPv6 addresses correctly."""
cur = scan_db.cursor()
# Device with empty IPs
cur.execute(
"""
INSERT INTO Devices (
devMac, devLastConnection, devPresentLastScan, devLastIP,
devName, devNameSource, devVendor, devVendorSource, devLastIPSource,
devType, devIcon, devParentPort, devParentMAC, devSite, devSSID,
devPrimaryIPv4, devPrimaryIPv6
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
""",
(
insert_device_from_dict(
scan_db,
make_device_dict(
"aa:bb:cc:dd:ee:05",
"2025-01-01 00:00:00",
0,
"",
"Device",
"NEWDEV",
"Vendor",
"NEWDEV",
"NEWDEV",
"type",
"icon",
"",
"",
"",
"",
"", # No IPv4
"", # No IPv6
devLastConnection="2025-01-01 00:00:00",
devPresentLastScan=0,
devLastIP="",
devName="Device",
devNameSource="NEWDEV",
devVendor="Vendor",
devVendorSource="NEWDEV",
devLastIPSource="NEWDEV",
devPrimaryIPv4="", # No IPv4
devPrimaryIPv6="", # No IPv6
),
)
# Scan discovers IPv4
cur.execute(
"""
INSERT INTO CurrentScan (
scanMac, scanLastIP, scanVendor, scanSourcePlugin, scanName,
scanLastQuery, scanLastConnection, scanSyncHubNode,
scanSite, scanSSID, scanParentMAC, scanParentPort, scanType
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
""",
(
insert_current_scan_row_from_dict(
scan_db,
make_current_scan_dict(
"aa:bb:cc:dd:ee:05",
"192.168.1.100",
"Vendor",
"ARPSCAN",
"",
"",
"2025-01-01 01:00:00",
"",
"",
"",
"",
"",
"",
scanLastIP="192.168.1.100",
scanVendor="Vendor",
scanSourcePlugin="ARPSCAN",
scanName="",
scanLastQuery="",
scanLastConnection="2025-01-01 01:00:00",
),
)
scan_db.commit()
db = Mock()
db.sql_connection = scan_db
db.sql = cur
db.sql = scan_db.cursor()
# Run scan update
device_handling.update_devices_data_from_scan(db)
device_handling.update_ipv4_ipv6(db)
row = cur.execute(
row = scan_db.execute(
"SELECT devLastIP, devPrimaryIPv4, devPrimaryIPv6 FROM Devices WHERE devMac = ?",
("aa:bb:cc:dd:ee:05",),
).fetchone()
@@ -677,75 +402,46 @@ def test_scan_updates_ip_addresses(scan_db, mock_device_handlers):
def test_scan_updates_ipv6_without_changing_ipv4(scan_db, mock_device_handlers):
"""Scan updates IPv6 without overwriting IPv4."""
cur = scan_db.cursor()
# Device with IPv4 already set
cur.execute(
"""
INSERT INTO Devices (
devMac, devLastConnection, devPresentLastScan, devLastIP,
devName, devNameSource, devVendor, devVendorSource, devLastIPSource,
devType, devIcon, devParentPort, devParentMAC, devSite, devSSID,
devPrimaryIPv4, devPrimaryIPv6
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
""",
(
insert_device_from_dict(
scan_db,
make_device_dict(
"aa:bb:cc:dd:ee:06",
"2025-01-01 00:00:00",
0,
"192.168.1.101",
"Device",
"NEWDEV",
"Vendor",
"NEWDEV",
"NEWDEV",
"type",
"icon",
"",
"",
"",
"",
"192.168.1.101", # IPv4 already set
"", # No IPv6
devLastConnection="2025-01-01 00:00:00",
devPresentLastScan=0,
devLastIP="192.168.1.101",
devName="Device",
devNameSource="NEWDEV",
devVendor="Vendor",
devVendorSource="NEWDEV",
devLastIPSource="NEWDEV",
devPrimaryIPv4="192.168.1.101", # IPv4 already set
devPrimaryIPv6="", # No IPv6
),
)
# Scan discovers IPv6
cur.execute(
"""
INSERT INTO CurrentScan (
scanMac, scanLastIP, scanVendor, scanSourcePlugin, scanName,
scanLastQuery, scanLastConnection, scanSyncHubNode,
scanSite, scanSSID, scanParentMAC, scanParentPort, scanType
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
""",
(
insert_current_scan_row_from_dict(
scan_db,
make_current_scan_dict(
"aa:bb:cc:dd:ee:06",
"fe80::1",
"Vendor",
"ARPSCAN",
"",
"",
"2025-01-01 01:00:00",
"",
"",
"",
"",
"",
"",
scanLastIP="fe80::1",
scanVendor="Vendor",
scanSourcePlugin="ARPSCAN",
scanName="",
scanLastQuery="",
scanLastConnection="2025-01-01 01:00:00",
),
)
scan_db.commit()
db = Mock()
db.sql_connection = scan_db
db.sql = cur
db.sql = scan_db.cursor()
# Run scan update
device_handling.update_devices_data_from_scan(db)
device_handling.update_ipv4_ipv6(db)
row = cur.execute(
row = scan_db.execute(
"SELECT devPrimaryIPv4, devPrimaryIPv6 FROM Devices WHERE devMac = ?",
("aa:bb:cc:dd:ee:06",),
).fetchone()
@@ -757,48 +453,32 @@ def test_scan_updates_ipv6_without_changing_ipv4(scan_db, mock_device_handlers):
def test_scan_updates_presence_status(scan_db, mock_device_handlers):
"""Scan correctly updates devPresentLastScan status."""
cur = scan_db.cursor()
# Device not in current scan (offline)
cur.execute(
"""
INSERT INTO Devices (
devMac, devLastConnection, devPresentLastScan, devLastIP,
devName, devNameSource, devVendor, devVendorSource, devLastIPSource,
devType, devIcon, devParentPort, devParentMAC, devSite, devSSID
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
""",
(
insert_device_from_dict(
scan_db,
make_device_dict(
"aa:bb:cc:dd:ee:07",
"2025-01-01 00:00:00",
1, # Was online
"192.168.1.102",
"Device",
"NEWDEV",
"Vendor",
"NEWDEV",
"ARPSCAN",
"type",
"icon",
"",
"",
"",
"",
devLastConnection="2025-01-01 00:00:00",
devPresentLastScan=1, # Was online
devLastIP="192.168.1.102",
devName="Device",
devNameSource="NEWDEV",
devVendor="Vendor",
devVendorSource="NEWDEV",
devLastIPSource="ARPSCAN",
),
)
# Note: No CurrentScan entry for this MAC - device is offline
scan_db.commit()
db = Mock()
db.sql_connection = scan_db
db.sql = cur
db.sql = scan_db.cursor()
# Run scan update
device_handling.update_devices_data_from_scan(db)
device_handling.update_presence_from_CurrentScan(db)
row = cur.execute(
row = scan_db.execute(
"SELECT devPresentLastScan FROM Devices WHERE devMac = ?",
("aa:bb:cc:dd:ee:07",),
).fetchone()
@@ -809,8 +489,6 @@ def test_scan_updates_presence_status(scan_db, mock_device_handlers):
def test_scan_multiple_devices_mixed_sources(scan_db, mock_device_handlers):
"""Scan with multiple devices having different source combinations."""
cur = scan_db.cursor()
devices_data = [
# (MAC, Name, NameSource, Vendor, VendorSource)
("aa:bb:cc:dd:ee:11", "Device1", "NEWDEV", "", "NEWDEV"), # Both updatable
@@ -820,30 +498,18 @@ def test_scan_multiple_devices_mixed_sources(scan_db, mock_device_handlers):
]
for mac, name, name_src, vendor, vendor_src in devices_data:
cur.execute(
"""
INSERT INTO Devices (
devMac, devLastConnection, devPresentLastScan, devLastIP,
devName, devNameSource, devVendor, devVendorSource, devLastIPSource,
devType, devIcon, devParentPort, devParentMAC, devSite, devSSID
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
""",
(
insert_device_from_dict(
scan_db,
make_device_dict(
mac,
"2025-01-01 00:00:00",
0,
"192.168.1.1",
name,
name_src,
vendor,
vendor_src,
"ARPSCAN",
"type",
"icon",
"",
"",
"",
"",
devLastConnection="2025-01-01 00:00:00",
devPresentLastScan=0,
devLastIP="192.168.1.1",
devName=name,
devNameSource=name_src,
devVendor=vendor,
devVendorSource=vendor_src,
devLastIPSource="ARPSCAN",
),
)
@@ -856,22 +522,22 @@ def test_scan_multiple_devices_mixed_sources(scan_db, mock_device_handlers):
]
for mac, ip, vendor, scan_method, name in scan_entries:
cur.execute(
"""
INSERT INTO CurrentScan (
scanMac, scanLastIP, scanVendor, scanSourcePlugin, scanName,
scanLastQuery, scanLastConnection, scanSyncHubNode,
scanSite, scanSSID, scanParentMAC, scanParentPort, scanType
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
""",
(mac, ip, vendor, scan_method, name, "", "2025-01-01 01:00:00", "", "", "", "", "", ""),
insert_current_scan_row_from_dict(
scan_db,
make_current_scan_dict(
mac,
scanLastIP=ip,
scanVendor=vendor,
scanSourcePlugin=scan_method,
scanName=name,
scanLastQuery="",
scanLastConnection="2025-01-01 01:00:00",
),
)
scan_db.commit()
db = Mock()
db.sql_connection = scan_db
db.sql = cur
db.sql = scan_db.cursor()
# Run scan update
device_handling.update_devices_data_from_scan(db)
@@ -885,7 +551,7 @@ def test_scan_multiple_devices_mixed_sources(scan_db, mock_device_handlers):
}
for mac, expected in results.items():
row = cur.execute(
row = scan_db.execute(
"SELECT devName, devVendor FROM Devices WHERE devMac = ?",
(mac,),
).fetchone()
test/scan/test_force_status.py
+23 -38
View File
@@ -1,58 +1,43 @@
"""Tests for forced device status updates."""
import sqlite3
import os
import sys
from server.scan import device_handling
class DummyDB:
"""Minimal DB wrapper compatible with device_handling helpers."""
def __init__(self, conn):
self.sql = conn.cursor()
self._conn = conn
def commitDB(self):
self._conn.commit()
sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
from db_test_helpers import ( # noqa: E402
DummyDB,
insert_device_from_dict,
make_db,
make_device_dict,
)
def test_force_status_updates_present_flag():
"""Forced status should override devPresentLastScan for online/offline values."""
conn = sqlite3.connect(":memory:")
conn.row_factory = sqlite3.Row
cur = conn.cursor()
conn = make_db()
cur.execute(
"""
CREATE TABLE Devices (
devMac TEXT PRIMARY KEY,
devPresentLastScan INTEGER,
devForceStatus TEXT
for mac, present_last_scan, force_status in [
("AA:AA:AA:AA:AA:01", 0, "online"),
("AA:AA:AA:AA:AA:02", 1, "offline"),
("AA:AA:AA:AA:AA:03", 1, "dont_force"),
("AA:AA:AA:AA:AA:04", 0, None),
("AA:AA:AA:AA:AA:05", 0, "ONLINE"),
]:
insert_device_from_dict(
conn,
make_device_dict(
mac, devPresentLastScan=present_last_scan, devForceStatus=force_status
),
)
"""
)
cur.executemany(
"""
INSERT INTO Devices (devMac, devPresentLastScan, devForceStatus)
VALUES (?, ?, ?)
""",
[
("AA:AA:AA:AA:AA:01", 0, "online"),
("AA:AA:AA:AA:AA:02", 1, "offline"),
("AA:AA:AA:AA:AA:03", 1, "dont_force"),
("AA:AA:AA:AA:AA:04", 0, None),
("AA:AA:AA:AA:AA:05", 0, "ONLINE"),
],
)
conn.commit()
db = DummyDB(conn)
updated = device_handling.update_devPresentLastScan_based_on_force_status(db)
rows = {
row["devMac"]: row["devPresentLastScan"]
for row in cur.execute("SELECT devMac, devPresentLastScan FROM Devices")
for row in conn.execute("SELECT devMac, devPresentLastScan FROM Devices")
}
assert updated == 3
test/scan/test_scan_presence.py
+73
View File
@@ -149,6 +149,79 @@ class TestNewConnectionsRespectsPresence:
assert rows == [], "scanPresence=0 must not generate a Connected event"
class TestNewConnectionsNoOrphanForCreatesDeviceZero:
"""A never-before-seen MAC reported only by scanCreatesDevice=0 rows must
not get a Connected event either - create_new_devices() will never turn
it into a Devices row, so the event would be a permanent orphan (shows up
in Events_Devices via its LEFT JOIN with every devName/devVendor field
NULL). Distinct from TestNewConnectionsRespectsPresence: this MAC does
assert presence, it's the creation gate that must suppress it."""
def test_unknown_mac_creates_device_zero_no_connected_event(self):
conn = make_db()
insert_current_scan_row_from_dict(
conn, make_current_scan_dict(MAC, scanPresence=1, scanCreatesDevice=0)
)
db = DummyDB(conn)
insert_events(db)
rows = conn.execute(
"SELECT * FROM Events WHERE eveMac = ? AND eveEventType IN ('Connected','Down Reconnected')",
(MAC,),
).fetchall()
assert rows == [], (
"a MAC with no Devices row that will never get one "
"(scanCreatesDevice=0 on every contributing row) must not get a Connected event"
)
def test_unknown_mac_creates_device_one_still_connects(self):
"""Regression guard for the fix above: insert_events() runs before
create_new_devices(), so a legitimately new device (scanCreatesDevice=1)
also has no Devices row yet at this point - it must still get a
Connected event same-cycle as its New Device event, unaffected by the
orphan-suppression fix."""
conn = make_db()
insert_current_scan_row_from_dict(
conn, make_current_scan_dict(MAC, scanPresence=1, scanCreatesDevice=1)
)
db = DummyDB(conn)
insert_events(db)
rows = conn.execute(
"SELECT * FROM Events WHERE eveMac = ? AND eveEventType = 'Connected'",
(MAC,),
).fetchall()
assert len(rows) == 1, (
"a brand-new device that create_new_devices() will create later "
"this same cycle must still get its Connected event"
)
def test_existing_device_reconnect_via_creates_device_zero_row_still_connects(self):
"""An already-existing device (created in a prior cycle) reconnecting
this cycle only via an enrich-only-for-creation plugin must still get
a Connected event - scanCreatesDevice only gates origination of new
devices, not updates/reconnects of ones that already exist."""
conn = make_db()
insert_device(conn, MAC, alert_down=1, present_last_scan=0)
insert_current_scan_row_from_dict(
conn, make_current_scan_dict(MAC, scanPresence=1, scanCreatesDevice=0)
)
db = DummyDB(conn)
insert_events(db)
rows = conn.execute(
"SELECT * FROM Events WHERE eveMac = ? AND eveEventType = 'Connected'",
(MAC,),
).fetchall()
assert len(rows) == 1, (
"scanCreatesDevice=0 must not block a Connected event for a device "
"that already exists"
)
class TestOnlineToPresenceZeroTransitionClosesSession:
"""The regression this PRD review round specifically caught: a device
going from online to a scanPresence=0-only report must still get a
test/test_device_atomicity.py
+17 -53
View File
@@ -18,9 +18,11 @@ from unittest.mock import patch
# Add server and plugins to path
sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..', 'server'))
sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..', 'server', 'plugins'))
sys.path.insert(0, os.path.dirname(__file__))
from models.device_instance import DeviceInstance # noqa: E402 [flake8 lint suppression]
from plugin_helper import normalize_mac # noqa: E402 [flake8 lint suppression]
from db_test_helpers import CREATE_DEVICES # noqa: E402 [flake8 lint suppression]
class TestDeviceAtomicity(unittest.TestCase):
@@ -37,51 +39,15 @@ class TestDeviceAtomicity(unittest.TestCase):
conn.row_factory = sqlite3.Row
cur = conn.cursor()
# Create Devices table with source-tracking columns
cur.execute("""
CREATE TABLE Devices (
devMac TEXT PRIMARY KEY,
devName TEXT,
devOwner TEXT,
devType TEXT,
devVendor TEXT,
devIcon TEXT,
devFavorite INTEGER DEFAULT 0,
devGroup TEXT,
devLocation TEXT,
devComments TEXT,
devParentMAC TEXT,
devParentPort TEXT,
devSSID TEXT,
devSite TEXT,
devStaticIP INTEGER DEFAULT 0,
devScan INTEGER DEFAULT 0,
devAlertEvents INTEGER DEFAULT 0,
devAlertDown INTEGER DEFAULT 0,
devCanSleep INTEGER DEFAULT 0,
devParentRelType TEXT DEFAULT 'default',
devReqNicsOnline INTEGER DEFAULT 0,
devSkipRepeated INTEGER DEFAULT 0,
devIsNew INTEGER DEFAULT 0,
devIsArchived INTEGER DEFAULT 0,
devLastConnection TEXT,
devFirstConnection TEXT,
devLastIP TEXT,
devGUID TEXT,
devCustomProps TEXT,
devSourcePlugin TEXT,
devNameSource TEXT,
devTypeSource TEXT,
devVendorSource TEXT,
devIconSource TEXT,
devGroupSource TEXT,
devLocationSource TEXT,
devCommentsSource TEXT,
devMacSource TEXT,
devVlan TEXT,
devForceStatus STRING DEFAULT NULL
)
""")
# Create Devices table with source-tracking columns - reuses the
# canonical schema shared by the rest of the suite rather than a
# local copy (the local copy this replaced had drifted: it declared
# devTypeSource/devIconSource/devGroupSource/devLocationSource/
# devCommentsSource, none of which exist in
# server/db/schema/app.sql or are read/written by any production
# code - source tracking only exists for the fields CREATE_DEVICES
# actually declares, e.g. devNameSource/devVendorSource/devMacSource).
cur.execute(CREATE_DEVICES)
conn.commit()
conn.close()
@@ -147,10 +113,9 @@ class TestDeviceAtomicity(unittest.TestCase):
cur = conn.cursor()
cur.execute("""
INSERT INTO Devices (
devMac, devName, devOwner, devType,
devNameSource, devTypeSource
) VALUES (?, ?, ?, ?, ?, ?)
""", (test_mac, "Old Name", "Old Owner", "Desktop", "PLUGIN", "PLUGIN"))
devMac, devName, devOwner, devType, devNameSource
) VALUES (?, ?, ?, ?, ?)
""", (test_mac, "Old Name", "Old Owner", "Desktop", "PLUGIN"))
conn.commit()
conn.close()
@@ -193,10 +158,9 @@ class TestDeviceAtomicity(unittest.TestCase):
cur = conn.cursor()
cur.execute("""
INSERT INTO Devices (
devMac, devName, devOwner, devType,
devNameSource, devTypeSource
) VALUES (?, ?, ?, ?, ?, ?)
""", (test_mac, "Original Name", "Original Owner", "Desktop", "PLUGIN", "PLUGIN"))
devMac, devName, devOwner, devType, devNameSource
) VALUES (?, ?, ?, ?, ?)
""", (test_mac, "Original Name", "Original Owner", "Desktop", "PLUGIN"))
conn.commit()
conn.close()