mirror of
https://github.com/jokob-sk/NetAlertX.git
synced 2026-09-12 22:25:49 -04:00
fix: address review feedback on NIC presence logic and tests
- Replace max() with explicit if/else for readability - Use db_test_helpers (make_db, make_device_dict, insert_device_from_dict, DummyDB) instead of local mock DB objects in test_nic_presence.py - Lowercase all MAC addresses in tests Co-authored-by: jokob-sk <96159884+jokob-sk@users.noreply.github.com>
This commit is contained in:
copilot-swe-agent[bot]andjokob-sk authored and GitHub committed
1 parent
283bea21c9
commit
f52cc50705
2 files changed
+82
-98
No files matched your search
@@ -1270,12 +1270,17 @@ def update_devPresentLastScan_based_on_nics(db):
|
||||
if nics:
|
||||
nic_statuses = [nic.get("devPresentLastScan") == 1 for nic in nics]
|
||||
if req_all:
|
||||
nic_derived = int(all(nic_statuses))
|
||||
nic_online = all(nic_statuses)
|
||||
else:
|
||||
nic_derived = int(any(nic_statuses))
|
||||
# NIC children can only raise a parent's presence, never lower it
|
||||
# when the parent itself was directly detected as present this scan.
|
||||
new_present = max(original, nic_derived)
|
||||
nic_online = any(nic_statuses)
|
||||
|
||||
if original == 1:
|
||||
# Parent was directly detected this scan — NIC children cannot
|
||||
# force it offline. Leave new_present = original (no change).
|
||||
pass
|
||||
else:
|
||||
# Parent was not directly detected — NICs determine presence.
|
||||
new_present = 1 if nic_online else 0
|
||||
|
||||
# Only add update if changed
|
||||
if original != new_present:
|
||||
|
||||
@@ -5,9 +5,12 @@ relationship had its own directly-detected presence overwritten by the NIC
|
||||
child's absence, producing an endless one-directional Connected event stream.
|
||||
"""
|
||||
|
||||
import sqlite3
|
||||
import sys
|
||||
import os
|
||||
|
||||
import pytest
|
||||
sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
|
||||
|
||||
from db_test_helpers import make_db, make_device_dict, insert_device_from_dict, DummyDB
|
||||
|
||||
from server.scan import device_handling
|
||||
|
||||
@@ -16,51 +19,19 @@ from server.scan import device_handling
|
||||
# Helpers
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def _make_db(rows):
|
||||
"""Return a DummyDB-compatible object populated with the given Devices rows."""
|
||||
conn = sqlite3.connect(":memory:")
|
||||
conn.row_factory = sqlite3.Row
|
||||
cur = conn.cursor()
|
||||
cur.execute(
|
||||
"""
|
||||
CREATE TABLE Devices (
|
||||
devMac TEXT PRIMARY KEY,
|
||||
devPresentLastScan INTEGER DEFAULT 0,
|
||||
devParentMAC TEXT,
|
||||
devParentRelType TEXT DEFAULT '',
|
||||
devReqNicsOnline INTEGER DEFAULT 0
|
||||
)
|
||||
"""
|
||||
)
|
||||
cur.executemany(
|
||||
"""
|
||||
INSERT INTO Devices (devMac, devPresentLastScan, devParentMAC,
|
||||
devParentRelType, devReqNicsOnline)
|
||||
VALUES (:mac, :present, :parent_mac, :rel_type, :req_all)
|
||||
""",
|
||||
rows,
|
||||
)
|
||||
conn.commit()
|
||||
|
||||
class DummyDB:
|
||||
def __init__(self, connection):
|
||||
self.sql = connection.cursor()
|
||||
self._conn = connection
|
||||
|
||||
def commitDB(self):
|
||||
self._conn.commit()
|
||||
|
||||
db = DummyDB(conn)
|
||||
# Re-use conn cursor for later reads
|
||||
db._raw_conn = conn
|
||||
return db
|
||||
def _setup(devices: list[dict]):
|
||||
"""Return a DummyDB seeded with the given device dicts."""
|
||||
conn = make_db()
|
||||
for dev in devices:
|
||||
insert_device_from_dict(conn, dev)
|
||||
return DummyDB(conn)
|
||||
|
||||
|
||||
def _present(db, mac):
|
||||
row = db._raw_conn.execute(
|
||||
def _present(db: DummyDB, mac: str) -> int:
|
||||
row = db._conn.execute(
|
||||
"SELECT devPresentLastScan FROM Devices WHERE devMac = ?", (mac,)
|
||||
).fetchone()
|
||||
return row[0]
|
||||
return row["devPresentLastScan"]
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
@@ -68,32 +39,34 @@ def _present(db, mac):
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
class TestNicChildDoesNotForcePresentParentDown:
|
||||
"""Parent was directly detected this scan; absent NIC must NOT override that."""
|
||||
"""Parent was directly detected this scan; an absent NIC must not override that."""
|
||||
|
||||
def test_any_mode_absent_nic_does_not_clear_present_parent(self):
|
||||
"""Bug: req_all=0, parent present=1, nic present=0 → parent must stay 1."""
|
||||
db = _make_db([
|
||||
{"mac": "AA:AA:AA:AA:AA:01", "present": 1,
|
||||
"parent_mac": "", "rel_type": "", "req_all": 0},
|
||||
{"mac": "BB:BB:BB:BB:BB:01", "present": 0,
|
||||
"parent_mac": "AA:AA:AA:AA:AA:01", "rel_type": "nic", "req_all": 0},
|
||||
db = _setup([
|
||||
make_device_dict("aa:aa:aa:aa:aa:01", devPresentLastScan=1,
|
||||
devParentMAC="", devParentRelType="", devReqNicsOnline=0),
|
||||
make_device_dict("bb:bb:bb:bb:bb:01", devPresentLastScan=0,
|
||||
devParentMAC="aa:aa:aa:aa:aa:01",
|
||||
devParentRelType="nic", devReqNicsOnline=0),
|
||||
])
|
||||
device_handling.update_devPresentLastScan_based_on_nics(db)
|
||||
assert _present(db, "AA:AA:AA:AA:AA:01") == 1, (
|
||||
assert _present(db, "aa:aa:aa:aa:aa:01") == 1, (
|
||||
"Parent directly detected as present must not be forced offline "
|
||||
"by an absent NIC child."
|
||||
)
|
||||
|
||||
def test_req_all_mode_absent_nic_does_not_clear_present_parent(self):
|
||||
"""Bug: req_all=1, parent present=1, nic present=0 → parent must stay 1."""
|
||||
db = _make_db([
|
||||
{"mac": "AA:AA:AA:AA:AA:02", "present": 1,
|
||||
"parent_mac": "", "rel_type": "", "req_all": 1},
|
||||
{"mac": "BB:BB:BB:BB:BB:02", "present": 0,
|
||||
"parent_mac": "AA:AA:AA:AA:AA:02", "rel_type": "nic", "req_all": 0},
|
||||
db = _setup([
|
||||
make_device_dict("aa:aa:aa:aa:aa:02", devPresentLastScan=1,
|
||||
devParentMAC="", devParentRelType="", devReqNicsOnline=1),
|
||||
make_device_dict("bb:bb:bb:bb:bb:02", devPresentLastScan=0,
|
||||
devParentMAC="aa:aa:aa:aa:aa:02",
|
||||
devParentRelType="nic", devReqNicsOnline=0),
|
||||
])
|
||||
device_handling.update_devPresentLastScan_based_on_nics(db)
|
||||
assert _present(db, "AA:AA:AA:AA:AA:02") == 1
|
||||
assert _present(db, "aa:aa:aa:aa:aa:02") == 1
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
@@ -101,52 +74,58 @@ class TestNicChildDoesNotForcePresentParentDown:
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
class TestNicRaisesAbsentParent:
|
||||
"""NIC children should be able to mark a parent present when it wasn't seen directly."""
|
||||
"""NIC children should be able to mark a parent present when it was not seen directly."""
|
||||
|
||||
def test_any_mode_online_nic_raises_absent_parent(self):
|
||||
db = _make_db([
|
||||
{"mac": "AA:AA:AA:AA:AA:03", "present": 0,
|
||||
"parent_mac": "", "rel_type": "", "req_all": 0},
|
||||
{"mac": "BB:BB:BB:BB:BB:03", "present": 1,
|
||||
"parent_mac": "AA:AA:AA:AA:AA:03", "rel_type": "nic", "req_all": 0},
|
||||
db = _setup([
|
||||
make_device_dict("aa:aa:aa:aa:aa:03", devPresentLastScan=0,
|
||||
devParentMAC="", devParentRelType="", devReqNicsOnline=0),
|
||||
make_device_dict("bb:bb:bb:bb:bb:03", devPresentLastScan=1,
|
||||
devParentMAC="aa:aa:aa:aa:aa:03",
|
||||
devParentRelType="nic", devReqNicsOnline=0),
|
||||
])
|
||||
device_handling.update_devPresentLastScan_based_on_nics(db)
|
||||
assert _present(db, "AA:AA:AA:AA:AA:03") == 1
|
||||
assert _present(db, "aa:aa:aa:aa:aa:03") == 1
|
||||
|
||||
def test_req_all_mode_all_nics_online_raises_absent_parent(self):
|
||||
db = _make_db([
|
||||
{"mac": "AA:AA:AA:AA:AA:04", "present": 0,
|
||||
"parent_mac": "", "rel_type": "", "req_all": 1},
|
||||
{"mac": "BB:BB:BB:BB:BB:04a", "present": 1,
|
||||
"parent_mac": "AA:AA:AA:AA:AA:04", "rel_type": "nic", "req_all": 0},
|
||||
{"mac": "BB:BB:BB:BB:BB:04b", "present": 1,
|
||||
"parent_mac": "AA:AA:AA:AA:AA:04", "rel_type": "nic", "req_all": 0},
|
||||
db = _setup([
|
||||
make_device_dict("aa:aa:aa:aa:aa:04", devPresentLastScan=0,
|
||||
devParentMAC="", devParentRelType="", devReqNicsOnline=1),
|
||||
make_device_dict("bb:bb:bb:bb:bb:04", devPresentLastScan=1,
|
||||
devParentMAC="aa:aa:aa:aa:aa:04",
|
||||
devParentRelType="nic", devReqNicsOnline=0),
|
||||
make_device_dict("cc:cc:cc:cc:cc:04", devPresentLastScan=1,
|
||||
devParentMAC="aa:aa:aa:aa:aa:04",
|
||||
devParentRelType="nic", devReqNicsOnline=0),
|
||||
])
|
||||
device_handling.update_devPresentLastScan_based_on_nics(db)
|
||||
assert _present(db, "AA:AA:AA:AA:AA:04") == 1
|
||||
assert _present(db, "aa:aa:aa:aa:aa:04") == 1
|
||||
|
||||
def test_req_all_mode_partial_nics_does_not_raise_absent_parent(self):
|
||||
"""In req_all mode, if not all NICs are online, an absent parent stays absent."""
|
||||
db = _make_db([
|
||||
{"mac": "AA:AA:AA:AA:AA:05", "present": 0,
|
||||
"parent_mac": "", "rel_type": "", "req_all": 1},
|
||||
{"mac": "BB:BB:BB:BB:BB:05a", "present": 1,
|
||||
"parent_mac": "AA:AA:AA:AA:AA:05", "rel_type": "nic", "req_all": 0},
|
||||
{"mac": "BB:BB:BB:BB:BB:05b", "present": 0,
|
||||
"parent_mac": "AA:AA:AA:AA:AA:05", "rel_type": "nic", "req_all": 0},
|
||||
"""req_all=1: if not all NICs are online, an absent parent stays absent."""
|
||||
db = _setup([
|
||||
make_device_dict("aa:aa:aa:aa:aa:05", devPresentLastScan=0,
|
||||
devParentMAC="", devParentRelType="", devReqNicsOnline=1),
|
||||
make_device_dict("bb:bb:bb:bb:bb:05", devPresentLastScan=1,
|
||||
devParentMAC="aa:aa:aa:aa:aa:05",
|
||||
devParentRelType="nic", devReqNicsOnline=0),
|
||||
make_device_dict("cc:cc:cc:cc:cc:05", devPresentLastScan=0,
|
||||
devParentMAC="aa:aa:aa:aa:aa:05",
|
||||
devParentRelType="nic", devReqNicsOnline=0),
|
||||
])
|
||||
device_handling.update_devPresentLastScan_based_on_nics(db)
|
||||
assert _present(db, "AA:AA:AA:AA:AA:05") == 0
|
||||
assert _present(db, "aa:aa:aa:aa:aa:05") == 0
|
||||
|
||||
def test_any_mode_all_nics_absent_leaves_parent_absent(self):
|
||||
db = _make_db([
|
||||
{"mac": "AA:AA:AA:AA:AA:06", "present": 0,
|
||||
"parent_mac": "", "rel_type": "", "req_all": 0},
|
||||
{"mac": "BB:BB:BB:BB:BB:06", "present": 0,
|
||||
"parent_mac": "AA:AA:AA:AA:AA:06", "rel_type": "nic", "req_all": 0},
|
||||
db = _setup([
|
||||
make_device_dict("aa:aa:aa:aa:aa:06", devPresentLastScan=0,
|
||||
devParentMAC="", devParentRelType="", devReqNicsOnline=0),
|
||||
make_device_dict("bb:bb:bb:bb:bb:06", devPresentLastScan=0,
|
||||
devParentMAC="aa:aa:aa:aa:aa:06",
|
||||
devParentRelType="nic", devReqNicsOnline=0),
|
||||
])
|
||||
device_handling.update_devPresentLastScan_based_on_nics(db)
|
||||
assert _present(db, "AA:AA:AA:AA:AA:06") == 0
|
||||
assert _present(db, "aa:aa:aa:aa:aa:06") == 0
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
@@ -155,13 +134,13 @@ class TestNicRaisesAbsentParent:
|
||||
|
||||
class TestNoNicChildren:
|
||||
def test_parent_with_no_nics_unchanged(self):
|
||||
db = _make_db([
|
||||
{"mac": "AA:AA:AA:AA:AA:07", "present": 1,
|
||||
"parent_mac": "", "rel_type": "", "req_all": 0},
|
||||
{"mac": "AA:AA:AA:AA:AA:08", "present": 0,
|
||||
"parent_mac": "", "rel_type": "", "req_all": 0},
|
||||
db = _setup([
|
||||
make_device_dict("aa:aa:aa:aa:aa:07", devPresentLastScan=1,
|
||||
devParentMAC="", devParentRelType="", devReqNicsOnline=0),
|
||||
make_device_dict("aa:aa:aa:aa:aa:08", devPresentLastScan=0,
|
||||
devParentMAC="", devParentRelType="", devReqNicsOnline=0),
|
||||
])
|
||||
updated = device_handling.update_devPresentLastScan_based_on_nics(db)
|
||||
assert updated == 0
|
||||
assert _present(db, "AA:AA:AA:AA:AA:07") == 1
|
||||
assert _present(db, "AA:AA:AA:AA:AA:08") == 0
|
||||
assert _present(db, "aa:aa:aa:aa:aa:07") == 1
|
||||
assert _present(db, "aa:aa:aa:aa:aa:08") == 0
|
||||
Reference in new issue
Block a user