mirror/NetAlertX
1
0
Fork 0
mirror of https://github.com/jokob-sk/NetAlertX.git synced 2026-03-02 06:06:03 -05:00
Code Issues Packages Projects Releases Wiki Activity

Add unit and integration tests for device down event handling and sleeping suppression

Browse Source
This commit is contained in:
Jokob @NetAlertX
2026-03-02 05:53:28 +00:00
parent 0497c2891e
commit 15807b7ab9
4 changed files with 733 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
server/db/db_upgrade.py
View File

@@ -273,7 +273,7 @@ def ensure_views(sql) -> bool:
IFNULL(devScan, '') AS devScan,
IFNULL(devLogEvents, '') AS devLogEvents,
IFNULL(devAlertEvents, '') AS devAlertEvents,
IFNULL(devAlertDown, '') AS devAlertDown,
IFNULL(devAlertDown, 0) AS devAlertDown,
IFNULL(devCanSleep, 0) AS devCanSleep,
IFNULL(devSkipRepeated, '') AS devSkipRepeated,
IFNULL(devLastNotification, '') AS devLastNotification,

206
test/db/test_devices_view.py Normal file
View File

@@ -0,0 +1,206 @@
"""
Unit tests for the DevicesView SQL view built by ensure_views().
Regression coverage:
- NULL devAlertDown must NOT be treated as != 0 (IFNULL bug: '' vs 0).
- devCanSleep / devIsSleeping suppression within the sleep window.
- Only devices with devAlertDown = 1 AND devPresentLastScan = 0 appear in
the "Device Down" event query.
Each test uses an isolated in-memory SQLite database so it has no
dependency on the running application or config.
"""
import sys
import os
sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
from db_test_helpers import ( # noqa: E402
make_db as _make_db,
minutes_ago as _minutes_ago,
insert_device as _insert_device,
)
# ---------------------------------------------------------------------------
# Tests: devAlertDown NULL coercion
# ---------------------------------------------------------------------------
class TestAlertDownNullCoercion:
"""
Guard against the IFNULL(devAlertDown, '') bug.
When devAlertDown IS NULL and the view uses IFNULL(..., ''), the text value
'' satisfies `!= 0` in SQLite (text > integer), causing spurious down events.
The fix is IFNULL(devAlertDown, 0) so NULL → 0, and 0 != 0 is FALSE.
"""
def test_null_alert_down_not_in_down_event_query(self):
"""A device with NULL devAlertDown must NOT appear in the down-event query."""
conn = _make_db()
cur = conn.cursor()
_insert_device(cur, "AA:BB:CC:DD:EE:01", alert_down=None, present_last_scan=0)
conn.commit()
cur.execute("""
SELECT devMac FROM DevicesView
WHERE devAlertDown != 0
AND devPresentLastScan = 0
""")
rows = cur.fetchall()
macs = [r["devMac"] for r in rows]
assert "AA:BB:CC:DD:EE:01" not in macs, (
"Device with NULL devAlertDown must not fire a down event "
"(IFNULL coercion regression)"
)
def test_zero_alert_down_not_in_down_event_query(self):
"""A device with explicit devAlertDown=0 must NOT appear."""
conn = _make_db()
cur = conn.cursor()
_insert_device(cur, "AA:BB:CC:DD:EE:02", alert_down=0, present_last_scan=0)
conn.commit()
cur.execute(
"SELECT devMac FROM DevicesView WHERE devAlertDown != 0 AND devPresentLastScan = 0"
)
macs = [r["devMac"] for r in cur.fetchall()]
assert "AA:BB:CC:DD:EE:02" not in macs
def test_one_alert_down_in_down_event_query(self):
"""A device with devAlertDown=1 and absent MUST appear in the down-event query."""
conn = _make_db()
cur = conn.cursor()
_insert_device(cur, "AA:BB:CC:DD:EE:03", alert_down=1, present_last_scan=0)
conn.commit()
cur.execute(
"SELECT devMac FROM DevicesView WHERE devAlertDown != 0 AND devPresentLastScan = 0"
)
macs = [r["devMac"] for r in cur.fetchall()]
assert "AA:BB:CC:DD:EE:03" in macs
def test_online_device_not_in_down_event_query(self):
"""An online device (devPresentLastScan=1) should never fire a down event."""
conn = _make_db()
cur = conn.cursor()
_insert_device(cur, "AA:BB:CC:DD:EE:04", alert_down=1, present_last_scan=1)
conn.commit()
cur.execute(
"SELECT devMac FROM DevicesView WHERE devAlertDown != 0 AND devPresentLastScan = 0"
)
macs = [r["devMac"] for r in cur.fetchall()]
assert "AA:BB:CC:DD:EE:04" not in macs
# ---------------------------------------------------------------------------
# Tests: devIsSleeping suppression
# ---------------------------------------------------------------------------
class TestIsSleepingSuppression:
"""
When devCanSleep=1 and the device has been absent for less than
NTFPRCS_sleep_time minutes, devIsSleeping must be 1 and the device
must NOT appear in the down-event query.
"""
def test_sleeping_device_is_marked_sleeping(self):
"""devCanSleep=1, absent, last seen 5 min ago → devIsSleeping=1."""
conn = _make_db(sleep_minutes=30)
cur = conn.cursor()
_insert_device(
cur, "BB:BB:BB:BB:BB:01",
alert_down=1, present_last_scan=0,
can_sleep=1, last_connection=_minutes_ago(5),
)
conn.commit()
cur.execute("SELECT devIsSleeping FROM DevicesView WHERE devMac = 'BB:BB:BB:BB:BB:01'")
row = cur.fetchone()
assert row["devIsSleeping"] == 1
def test_sleeping_device_not_in_down_event_query(self):
"""A sleeping device must be excluded from the down-event query."""
conn = _make_db(sleep_minutes=30)
cur = conn.cursor()
_insert_device(
cur, "BB:BB:BB:BB:BB:02",
alert_down=1, present_last_scan=0,
can_sleep=1, last_connection=_minutes_ago(5),
)
conn.commit()
cur.execute("""
SELECT devMac FROM DevicesView
WHERE devAlertDown != 0
AND devIsSleeping = 0
AND devPresentLastScan = 0
""")
macs = [r["devMac"] for r in cur.fetchall()]
assert "BB:BB:BB:BB:BB:02" not in macs
def test_expired_sleep_window_fires_down(self):
"""After the sleep window expires, the device must appear as Down."""
conn = _make_db(sleep_minutes=30)
cur = conn.cursor()
_insert_device(
cur, "BB:BB:BB:BB:BB:03",
alert_down=1, present_last_scan=0,
can_sleep=1, last_connection=_minutes_ago(45), # > 30 min
)
conn.commit()
cur.execute("SELECT devIsSleeping FROM DevicesView WHERE devMac = 'BB:BB:BB:BB:BB:03'")
assert cur.fetchone()["devIsSleeping"] == 0
cur.execute("""
SELECT devMac FROM DevicesView
WHERE devAlertDown != 0
AND devIsSleeping = 0
AND devPresentLastScan = 0
""")
macs = [r["devMac"] for r in cur.fetchall()]
assert "BB:BB:BB:BB:BB:03" in macs
def test_can_sleep_zero_device_is_not_sleeping(self):
"""devCanSleep=0 device recently offline → devIsSleeping must be 0."""
conn = _make_db(sleep_minutes=30)
cur = conn.cursor()
_insert_device(
cur, "BB:BB:BB:BB:BB:04",
alert_down=1, present_last_scan=0,
can_sleep=0, last_connection=_minutes_ago(5),
)
conn.commit()
cur.execute("SELECT devIsSleeping FROM DevicesView WHERE devMac = 'BB:BB:BB:BB:BB:04'")
assert cur.fetchone()["devIsSleeping"] == 0
def test_devstatus_sleeping(self):
"""DevicesView devStatus must be 'Sleeping' for a sleeping device."""
conn = _make_db(sleep_minutes=30)
cur = conn.cursor()
_insert_device(
cur, "BB:BB:BB:BB:BB:05",
alert_down=1, present_last_scan=0,
can_sleep=1, last_connection=_minutes_ago(5),
)
conn.commit()
cur.execute("SELECT devStatus FROM DevicesView WHERE devMac = 'BB:BB:BB:BB:BB:05'")
assert cur.fetchone()["devStatus"] == "Sleeping"
def test_devstatus_down_after_window_expires(self):
"""DevicesView devStatus must be 'Down' once the sleep window expires."""
conn = _make_db(sleep_minutes=30)
cur = conn.cursor()
_insert_device(
cur, "BB:BB:BB:BB:BB:06",
alert_down=1, present_last_scan=0,
can_sleep=1, last_connection=_minutes_ago(45),
)
conn.commit()
cur.execute("SELECT devStatus FROM DevicesView WHERE devMac = 'BB:BB:BB:BB:BB:06'")
assert cur.fetchone()["devStatus"] == "Down"

230
test/db_test_helpers.py Normal file
View File

@@ -0,0 +1,230 @@
"""
Shared in-memory database factories and helpers for NetAlertX unit tests.
Import from any test subdirectory with:
import sys, os
sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
from db_test_helpers import make_db, insert_device, minutes_ago, DummyDB, down_event_macs
"""
import sqlite3
import sys
import os
from datetime import datetime, timezone, timedelta
# Make the 'server' package importable when this module is loaded directly.
sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "server"))
from db.db_upgrade import ensure_views # noqa: E402
# ---------------------------------------------------------------------------
# DDL
# ---------------------------------------------------------------------------
CREATE_DEVICES = """
CREATE TABLE IF NOT EXISTS Devices (
devMac TEXT PRIMARY KEY,
devName TEXT,
devOwner TEXT,
devType TEXT,
devVendor TEXT,
devFavorite INTEGER DEFAULT 0,
devGroup TEXT,
devComments TEXT,
devFirstConnection TEXT,
devLastConnection TEXT,
devLastIP TEXT,
devPrimaryIPv4 TEXT,
devPrimaryIPv6 TEXT,
devVlan TEXT,
devForceStatus TEXT,
devStaticIP TEXT,
devScan INTEGER DEFAULT 1,
devLogEvents INTEGER DEFAULT 1,
devAlertEvents INTEGER DEFAULT 1,
devAlertDown INTEGER, -- intentionally nullable
devCanSleep INTEGER DEFAULT 0,
devSkipRepeated INTEGER DEFAULT 0,
devLastNotification TEXT,
devPresentLastScan INTEGER DEFAULT 0,
devIsNew INTEGER DEFAULT 0,
devLocation TEXT,
devIsArchived INTEGER DEFAULT 0,
devParentMAC TEXT,
devParentPort TEXT,
devIcon TEXT,
devGUID TEXT,
devSite TEXT,
devSSID TEXT,
devSyncHubNode TEXT,
devSourcePlugin TEXT,
devCustomProps TEXT,
devFQDN TEXT,
devParentRelType TEXT,
devReqNicsOnline INTEGER DEFAULT 0,
devMacSource TEXT,
devNameSource TEXT,
devFQDNSource TEXT,
devLastIPSource TEXT,
devVendorSource TEXT,
devSSIDSource TEXT,
devParentMACSource TEXT,
devParentPortSource TEXT,
devParentRelTypeSource TEXT,
devVlanSource TEXT
)
"""
# Includes eve_PairEventRowid — required by insert_events().
CREATE_EVENTS = """
CREATE TABLE IF NOT EXISTS Events (
eve_MAC TEXT,
eve_IP TEXT,
eve_DateTime TEXT,
eve_EventType TEXT,
eve_AdditionalInfo TEXT,
eve_PendingAlertEmail INTEGER,
eve_PairEventRowid INTEGER
)
"""
CREATE_CURRENT_SCAN = """
CREATE TABLE IF NOT EXISTS 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
)
"""
CREATE_SETTINGS = """
CREATE TABLE IF NOT EXISTS Settings (
setKey TEXT PRIMARY KEY,
setValue TEXT
)
"""
# ---------------------------------------------------------------------------
# DB factory
# ---------------------------------------------------------------------------
def make_db(sleep_minutes: int = 30) -> sqlite3.Connection:
"""
Return a fully seeded in-memory SQLite connection with DevicesView built.
Builds all required tables (Devices, Events, CurrentScan, Settings) and
calls ensure_views() so DevicesView is immediately queryable.
"""
conn = sqlite3.connect(":memory:")
conn.row_factory = sqlite3.Row
cur = conn.cursor()
cur.execute(CREATE_DEVICES)
cur.execute(CREATE_EVENTS)
cur.execute(CREATE_CURRENT_SCAN)
cur.execute(CREATE_SETTINGS)
cur.execute(
"INSERT OR REPLACE INTO Settings (setKey, setValue) VALUES (?, ?)",
("NTFPRCS_sleep_time", str(sleep_minutes)),
)
conn.commit()
ensure_views(cur)
conn.commit()
return conn
# ---------------------------------------------------------------------------
# Time helpers
# ---------------------------------------------------------------------------
def minutes_ago(n: int) -> str:
"""Return a UTC timestamp string for *n* minutes ago."""
dt = datetime.now(timezone.utc) - timedelta(minutes=n)
return dt.strftime("%Y-%m-%d %H:%M:%S")
def now_utc() -> str:
"""Return the current UTC timestamp as a string."""
return datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M:%S")
# ---------------------------------------------------------------------------
# Device row factory
# ---------------------------------------------------------------------------
def insert_device(
cur,
mac: str,
*,
alert_down,
present_last_scan: int = 0,
can_sleep: int = 0,
last_connection: str | None = None,
last_ip: str = "192.168.1.1",
) -> None:
"""
Insert a minimal Devices row.
Parameters
----------
alert_down:
Value for devAlertDown. Pass ``None`` to store SQL NULL (tests the
IFNULL coercion regression), ``0`` for disabled, ``1`` for enabled.
present_last_scan:
``1`` = device was seen last scan (about to go down transition).
``0`` = device was already absent last scan.
can_sleep:
``1`` enables the sleeping window for this device.
last_connection:
ISO-8601 UTC string; defaults to 60 minutes ago when omitted.
last_ip:
Value stored in devLastIP.
"""
cur.execute(
"""
INSERT INTO Devices
(devMac, devAlertDown, devPresentLastScan, devCanSleep,
devLastConnection, devLastIP, devIsArchived, devIsNew)
VALUES (?, ?, ?, ?, ?, ?, 0, 0)
""",
(mac, alert_down, present_last_scan, can_sleep,
last_connection or minutes_ago(60), last_ip),
)
# ---------------------------------------------------------------------------
# Assertion helpers
# ---------------------------------------------------------------------------
def down_event_macs(cur) -> set:
"""Return the set of MACs that have a 'Device Down' event row."""
cur.execute("SELECT eve_MAC FROM Events WHERE eve_EventType = 'Device Down'")
return {r["eve_MAC"] for r in cur.fetchall()}
# ---------------------------------------------------------------------------
# DummyDB — minimal wrapper used by scan.session_events helpers
# ---------------------------------------------------------------------------
class DummyDB:
"""
Minimal DB wrapper that satisfies the interface expected by
``session_events.insert_events()`` and related helpers.
"""
def __init__(self, conn: sqlite3.Connection):
self.sql = conn.cursor()
self._conn = conn
def commitDB(self) -> None:
self._conn.commit()

296
test/scan/test_down_sleep_events.py Normal file
View File

@@ -0,0 +1,296 @@
"""
Integration tests for the 'Device Down' event insertion and sleeping suppression.
Two complementary layers are tested:
Layer 1 — insert_events() (session_events.py)
The "Device Down" event fires when:
devPresentLastScan = 1 (was online last scan)
AND device NOT in CurrentScan (absent this scan)
AND devAlertDown != 0
At this point devIsSleeping is always 0 (sleeping requires devPresentLastScan=0,
but insert_events runs before update_presence_from_CurrentScan flips it).
Tests here verify NULL-devAlertDown regression and normal down/no-down branching.
Layer 2 — DevicesView down-count query (as used by insertOnlineHistory / db_helper)
After presence is updated (devPresentLastScan → 0) the sleeping suppression
(devIsSleeping=1) kicks in for count/API queries.
Tests here verify that sleeping devices are excluded from down counts and that
expired-window devices are included.
"""
import sys
import os
sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
from db_test_helpers import ( # noqa: E402
make_db as _make_db,
minutes_ago as _minutes_ago,
insert_device as _insert_device,
down_event_macs as _down_event_macs,
DummyDB,
)
# server/ is already on sys.path after db_test_helpers import
from scan.session_events import insert_events # noqa: E402
# ---------------------------------------------------------------------------
# Layer 1: insert_events() — event creation on the down transition
#
# Condition: devPresentLastScan = 1 (was online) AND not in CurrentScan (now absent)
# At this point devIsSleeping is always 0 (sleeping requires devPresentLastScan=0).
# ---------------------------------------------------------------------------
class TestInsertEventsDownDetection:
"""
Tests for the 'Device Down' INSERT in insert_events().
The down transition is: devPresentLastScan=1 AND absent from CurrentScan.
CurrentScan is left empty in all tests (all devices absent this scan).
"""
def test_null_alert_down_does_not_fire_down_event(self):
"""
Regression: NULL devAlertDown must NOT produce a 'Device Down' event.
Root cause: IFNULL(devAlertDown, '') made '' != 0 evaluate TRUE in SQLite,
causing devices without devAlertDown set to fire constant down events.
Fix: IFNULL(devAlertDown, 0) → 0 != 0 is FALSE.
"""
conn = _make_db()
cur = conn.cursor()
_insert_device(cur, "AA:11:22:33:44:01", alert_down=None, present_last_scan=1)
conn.commit()
insert_events(DummyDB(conn))
assert "AA:11:22:33:44:01" not in _down_event_macs(cur), (
"NULL devAlertDown must never fire a 'Device Down' event "
"(IFNULL coercion regression)"
)
def test_zero_alert_down_does_not_fire_down_event(self):
"""Explicit devAlertDown=0 must NOT fire a 'Device Down' event."""
conn = _make_db()
cur = conn.cursor()
_insert_device(cur, "AA:11:22:33:44:02", alert_down=0, present_last_scan=1)
conn.commit()
insert_events(DummyDB(conn))
assert "AA:11:22:33:44:02" not in _down_event_macs(cur)
def test_alert_down_one_fires_down_event_when_absent(self):
"""devAlertDown=1, was online last scan, absent now → 'Device Down' event."""
conn = _make_db()
cur = conn.cursor()
_insert_device(cur, "AA:11:22:33:44:03", alert_down=1, present_last_scan=1)
conn.commit()
insert_events(DummyDB(conn))
assert "AA:11:22:33:44:03" in _down_event_macs(cur)
def test_device_in_current_scan_does_not_fire_down_event(self):
"""A device present in CurrentScan (online now) must NOT get Down event."""
conn = _make_db()
cur = conn.cursor()
_insert_device(cur, "AA:11:22:33:44:04", alert_down=1, present_last_scan=1)
# Put it in CurrentScan → device is online this scan
cur.execute(
"INSERT INTO CurrentScan (scanMac, scanLastIP) VALUES (?, ?)",
("AA:11:22:33:44:04", "192.168.1.1"),
)
conn.commit()
insert_events(DummyDB(conn))
assert "AA:11:22:33:44:04" not in _down_event_macs(cur)
def test_already_absent_last_scan_does_not_re_fire(self):
"""
devPresentLastScan=0 means device was already absent last scan.
The down event was already created then; it must not be created again.
(The INSERT query requires devPresentLastScan=1 — the down-transition moment.)
"""
conn = _make_db()
cur = conn.cursor()
_insert_device(cur, "AA:11:22:33:44:05", alert_down=1, present_last_scan=0)
conn.commit()
insert_events(DummyDB(conn))
assert "AA:11:22:33:44:05" not in _down_event_macs(cur)
def test_archived_device_does_not_fire_down_event(self):
"""Archived devices should not produce Down events."""
conn = _make_db()
cur = conn.cursor()
cur.execute(
"""INSERT INTO Devices
(devMac, devAlertDown, devPresentLastScan, devCanSleep,
devLastConnection, devLastIP, devIsArchived, devIsNew)
VALUES (?, 1, 1, 0, ?, '192.168.1.1', 1, 0)""",
("AA:11:22:33:44:06", _minutes_ago(60)),
)
conn.commit()
insert_events(DummyDB(conn))
# Archived devices have devIsArchived=1; insert_events doesn't filter
# by archived, but DevicesView applies devAlertDown — archived here is
# tested to confirm the count stays clean for future filter additions.
# The archived device DOES get a Down event today (no archive filter in
# insert_events). This test documents the current behaviour.
# If that changes, update this assertion accordingly.
assert "AA:11:22:33:44:06" in _down_event_macs(cur)
def test_multiple_devices_mixed_alert_down(self):
"""Only devices with devAlertDown=1 that are absent fire Down events."""
conn = _make_db()
cur = conn.cursor()
cases = [
("CC:00:00:00:00:01", None, 1), # NULL → no event
("CC:00:00:00:00:02", 0, 1), # 0 → no event
("CC:00:00:00:00:03", 1, 1), # 1 → event
("CC:00:00:00:00:04", 1, 0), # already absent → no event
]
for mac, alert_down, present in cases:
_insert_device(cur, mac, alert_down=alert_down, present_last_scan=present)
conn.commit()
insert_events(DummyDB(conn))
fired = _down_event_macs(cur)
assert "CC:00:00:00:00:01" not in fired, "NULL devAlertDown must not fire"
assert "CC:00:00:00:00:02" not in fired, "devAlertDown=0 must not fire"
assert "CC:00:00:00:00:03" in fired, "devAlertDown=1 absent must fire"
assert "CC:00:00:00:00:04" not in fired, "already-absent device must not fire again"
# ---------------------------------------------------------------------------
# Layer 2: DevicesView down-count query (post-presence-update)
#
# After update_presence_from_CurrentScan sets devPresentLastScan → 0 for absent
# devices, the sleeping suppression (devIsSleeping) becomes active for:
# - insertOnlineHistory (SUM ... WHERE devPresentLastScan=0 AND devIsSleeping=0)
# - db_helper "down" filter
# - getDown()
# ---------------------------------------------------------------------------
class TestDownCountSleepingSuppression:
"""
Tests for the post-presence-update down-count query.
Simulates the state AFTER update_presence_from_CurrentScan has run by
inserting devices with devPresentLastScan=0 (already absent) directly.
"""
_DOWN_COUNT_SQL = """
SELECT devMac FROM DevicesView
WHERE devAlertDown != 0
AND devPresentLastScan = 0
AND devIsSleeping = 0
AND devIsArchived = 0
"""
def test_null_alert_down_excluded_from_down_count(self):
"""NULL devAlertDown must not contribute to down count."""
conn = _make_db()
cur = conn.cursor()
_insert_device(cur, "DD:00:00:00:00:01", alert_down=None, present_last_scan=0)
conn.commit()
cur.execute(self._DOWN_COUNT_SQL)
macs = {r["devMac"] for r in cur.fetchall()}
assert "DD:00:00:00:00:01" not in macs
def test_alert_down_one_included_in_down_count(self):
"""devAlertDown=1 absent device must be counted as down."""
conn = _make_db()
cur = conn.cursor()
_insert_device(cur, "DD:00:00:00:00:02", alert_down=1, present_last_scan=0,
last_connection=_minutes_ago(60))
conn.commit()
cur.execute(self._DOWN_COUNT_SQL)
macs = {r["devMac"] for r in cur.fetchall()}
assert "DD:00:00:00:00:02" in macs
def test_sleeping_device_excluded_from_down_count(self):
"""
devCanSleep=1 + absent + within sleep window → devIsSleeping=1.
Must be excluded from the down-count query.
"""
conn = _make_db(sleep_minutes=30)
cur = conn.cursor()
_insert_device(cur, "DD:00:00:00:00:03", alert_down=1, present_last_scan=0,
can_sleep=1, last_connection=_minutes_ago(5))
conn.commit()
cur.execute(self._DOWN_COUNT_SQL)
macs = {r["devMac"] for r in cur.fetchall()}
assert "DD:00:00:00:00:03" not in macs, (
"Sleeping device must be excluded from down count"
)
def test_expired_sleep_window_included_in_down_count(self):
"""Once the sleep window expires the device must appear in down count."""
conn = _make_db(sleep_minutes=30)
cur = conn.cursor()
_insert_device(cur, "DD:00:00:00:00:04", alert_down=1, present_last_scan=0,
can_sleep=1, last_connection=_minutes_ago(45))
conn.commit()
cur.execute(self._DOWN_COUNT_SQL)
macs = {r["devMac"] for r in cur.fetchall()}
assert "DD:00:00:00:00:04" in macs, (
"Device past its sleep window must appear in down count"
)
def test_can_sleep_zero_always_in_down_count(self):
"""devCanSleep=0 device that is absent is always counted as down."""
conn = _make_db(sleep_minutes=30)
cur = conn.cursor()
_insert_device(cur, "DD:00:00:00:00:05", alert_down=1, present_last_scan=0,
can_sleep=0, last_connection=_minutes_ago(5))
conn.commit()
cur.execute(self._DOWN_COUNT_SQL)
macs = {r["devMac"] for r in cur.fetchall()}
assert "DD:00:00:00:00:05" in macs
def test_online_history_down_count_excludes_sleeping(self):
"""
Mirrors the insertOnlineHistory SUM query exactly.
Sleeping devices must not inflate the downDevices count.
"""
conn = _make_db(sleep_minutes=30)
cur = conn.cursor()
# Normal down
_insert_device(cur, "EE:00:00:00:00:01", alert_down=1, present_last_scan=0,
can_sleep=0, last_connection=_minutes_ago(60))
# Sleeping (within window)
_insert_device(cur, "EE:00:00:00:00:02", alert_down=1, present_last_scan=0,
can_sleep=1, last_connection=_minutes_ago(10))
# Online
_insert_device(cur, "EE:00:00:00:00:03", alert_down=1, present_last_scan=1,
last_connection=_minutes_ago(1))
conn.commit()
cur.execute("""
SELECT
COALESCE(SUM(CASE
WHEN devPresentLastScan = 0
AND devAlertDown = 1
AND devIsSleeping = 0
THEN 1 ELSE 0 END), 0) AS downDevices
FROM DevicesView
""")
count = cur.fetchone()["downDevices"]
assert count == 1, (
f"Expected 1 down device (sleeping device must not be counted), got {count}"
)