From af2a89f4ffb777422293f4ec3437ee289fb2e7dd Mon Sep 17 00:00:00 2001 From: jokob-sk Date: Thu, 29 Jan 2026 09:43:40 +1100 Subject: [PATCH] TEST: field locking test fixes 5 Signed-off-by: jokob-sk --- test/authoritative_fields/conftest.py | 102 +++++++++++++++ .../test_field_lock_scan_integration.py | 118 ------------------ .../test_ip_update_logic.py | 76 ----------- 3 files changed, 102 insertions(+), 194 deletions(-) create mode 100644 test/authoritative_fields/conftest.py diff --git a/test/authoritative_fields/conftest.py b/test/authoritative_fields/conftest.py new file mode 100644 index 00000000..fb99747c --- /dev/null +++ b/test/authoritative_fields/conftest.py @@ -0,0 +1,102 @@ +import pytest +import sqlite3 + + +@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() + + # 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, + 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, + + -- 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 + ) + """) + + # 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() + yield conn + conn.close() diff --git a/test/authoritative_fields/test_field_lock_scan_integration.py b/test/authoritative_fields/test_field_lock_scan_integration.py index c6749d03..f800eb80 100644 --- a/test/authoritative_fields/test_field_lock_scan_integration.py +++ b/test/authoritative_fields/test_field_lock_scan_integration.py @@ -19,124 +19,6 @@ import pytest from server.scan import device_handling -@pytest.fixture -def scan_db(): - """Create an in-memory SQLite database with full device schema.""" - conn = sqlite3.connect(":memory:") - conn.row_factory = sqlite3.Row - cur = conn.cursor() - - # Create Devices table with source tracking - cur.execute( - """ - CREATE TABLE Devices ( - devMac TEXT PRIMARY KEY, - devLastConnection TEXT, - devPresentLastScan INTEGER DEFAULT 0, - devForceStatus TEXT, - devLastIP TEXT, - devName TEXT, - devNameSource TEXT DEFAULT 'NEWDEV', - devVendor TEXT, - devVendorSource TEXT DEFAULT 'NEWDEV', - devLastIPSource TEXT DEFAULT 'NEWDEV', - devType TEXT, - devIcon TEXT, - devParentPort TEXT, - devParentPortSource TEXT DEFAULT 'NEWDEV', - devParentMAC TEXT, - devParentMACSource TEXT DEFAULT 'NEWDEV', - devSite TEXT, - devSiteSource TEXT DEFAULT 'NEWDEV', - devSSID TEXT, - devSSIDSource TEXT DEFAULT 'NEWDEV', - devFQDN TEXT, - devFQDNSource TEXT DEFAULT 'NEWDEV', - devParentRelType TEXT, - devParentRelTypeSource TEXT DEFAULT 'NEWDEV', - devVlan TEXT, - devVlanSource TEXT DEFAULT 'NEWDEV', - devPrimaryIPv4 TEXT, - devPrimaryIPv6 TEXT - ) - """ - ) - - # Create 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 - ) - """ - ) - - cur.execute( - """ - CREATE TABLE Events ( - eve_MAC TEXT, - eve_IP TEXT, - eve_DateTime TEXT, - eve_EventType TEXT, - eve_AdditionalInfo TEXT, - eve_PendingAlertEmail INTEGER - ) - """ - ) - - cur.execute( - """ - CREATE TABLE Sessions ( - ses_MAC TEXT, - ses_IP TEXT, - ses_EventTypeConnection TEXT, - ses_DateTimeConnection TEXT, - ses_EventTypeDisconnection TEXT, - ses_DateTimeDisconnection TEXT, - ses_StillConnected INTEGER, - ses_AdditionalInfo TEXT - ) - """ - ) - - # Add the View logic provided - 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 - LEFT JOIN RankedScans r - ON d.devMac = r.scanMac - WHERE r.rn = 1; - """) - - conn.commit() - yield conn - conn.close() - - @pytest.fixture def mock_device_handlers(): """Mock device_handling helper functions.""" diff --git a/test/authoritative_fields/test_ip_update_logic.py b/test/authoritative_fields/test_ip_update_logic.py index c735c17b..4e118079 100644 --- a/test/authoritative_fields/test_ip_update_logic.py +++ b/test/authoritative_fields/test_ip_update_logic.py @@ -2,7 +2,6 @@ Unit tests for device IP update logic (devPrimaryIPv4/devPrimaryIPv6 handling). """ -import sqlite3 from unittest.mock import Mock, patch import pytest @@ -10,81 +9,6 @@ import pytest from server.scan import device_handling -@pytest.fixture -def in_memory_db(): - """Create an in-memory SQLite database for testing.""" - conn = sqlite3.connect(":memory:") - conn.row_factory = sqlite3.Row - cur = conn.cursor() - - cur.execute( - """ - CREATE TABLE Devices ( - devMac TEXT PRIMARY KEY, - devLastConnection TEXT, - devPresentLastScan INTEGER, - devForceStatus TEXT, - devLastIP TEXT, - devPrimaryIPv4 TEXT, - devPrimaryIPv6 TEXT, - devVendor TEXT, - devParentPort TEXT, - devParentMAC TEXT, - devSite TEXT, - devSSID TEXT, - devType TEXT, - devName TEXT, - devIcon TEXT - ) - """ - ) - - 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 - ) - """ - ) - - # Add the View logic provided - 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 - LEFT JOIN RankedScans r - ON d.devMac = r.scanMac - WHERE r.rn = 1; - """) - - conn.commit() - yield conn - conn.close() - - @pytest.fixture def mock_device_handling(): """Mock device_handling dependencies."""