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

TEST: field locking test fixes 5

Browse Source 
Signed-off-by: jokob-sk <jokob.sk@gmail.com>
This commit is contained in:
jokob-sk
2026-01-29 09:43:40 +11:00
parent e649bcfe25
commit af2a89f4ff
3 changed files with 102 additions and 194 deletions
Download Patch File Download Diff File Expand all files Collapse all files

102
test/authoritative_fields/conftest.py Normal file
View File

@@ -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()

118
test/authoritative_fields/test_field_lock_scan_integration.py
View File

@@ -19,124 +19,6 @@ import pytest
from server.scan import device_handling 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 @pytest.fixture
def mock_device_handlers(): def mock_device_handlers():
"""Mock device_handling helper functions.""" """Mock device_handling helper functions."""

76
test/authoritative_fields/test_ip_update_logic.py
View File

@@ -2,7 +2,6 @@
Unit tests for device IP update logic (devPrimaryIPv4/devPrimaryIPv6 handling). Unit tests for device IP update logic (devPrimaryIPv4/devPrimaryIPv6 handling).
""" """
import sqlite3
from unittest.mock import Mock, patch from unittest.mock import Mock, patch
import pytest import pytest
@@ -10,81 +9,6 @@ import pytest
from server.scan import device_handling 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 @pytest.fixture
def mock_device_handling(): def mock_device_handling():
"""Mock device_handling dependencies.""" """Mock device_handling dependencies."""