BE: Fix: NIC-Covered Parent Devices Miss Connected/Down Reconnected Events #1821

This commit is contained in:
jokob-sk committed 2026-09-30 22:01:04 +10:00
1 parent 5ac9c53142
commit fbbdefb898
6 files changed
+89 -9

No files matched your search

+22
View File
@@ -948,6 +948,28 @@ class TestInsertEventsNicDerivedReconnect:
assert _event_types_for(conn, "aa:22:00:00:00:09") == []
def test_sibling_parent_with_absent_nics_not_falsely_reconnected(self):
"""Regression: an alias collision inside nic_derived_presence_condition()
used to collapse its per-row check into a table-wide tautology - any
NIC-covered parent anywhere in Devices made every other absent parent
look NIC-derived-present too. A single-parent-per-DB test can't catch
this (there's nothing else in the table to falsely match against), so
this puts two sibling parents in one DB on purpose."""
conn = _make_db()
_setup_parent_with_nics(
conn, "aa:22:00:00:00:10", [("bb:22:00:00:00:10", True)],
parent_present_last_scan=0,
)
_setup_parent_with_nics(
conn, "aa:22:00:00:00:11", [("bb:22:00:00:00:11", False)],
parent_present_last_scan=0,
)
insert_events(DummyDB(conn))
assert _event_types_for(conn, "aa:22:00:00:00:10") == ["Connected"]
assert _event_types_for(conn, "aa:22:00:00:00:11") == []
def test_orphan_pairing_closes_end_to_end(self):
"""The reporter's exact scenario (issue #1821):
absent -> Disconnected -> NIC-present -> Connected -> absent again ->
+39 -2
View File
@@ -75,8 +75,8 @@ class TestNicDerivedHelperCorrectness:
def test_returns_expected_sql_fragment(self):
result = nic_derived_presence_condition("devMac")
assert "EXISTS (" in result
assert "SELECT 1 FROM Devices AS nic_parent" in result
assert "nic_parent.devMac = devMac" in result
assert "SELECT 1 FROM Devices AS nic_presence_parent" in result
assert "nic_presence_parent.devMac = devMac" in result
assert "devReqNicsOnline" in result
@pytest.mark.parametrize("bad_value", [
@@ -96,6 +96,14 @@ class TestNicDerivedHelperCorrectness:
with pytest.raises(ValueError):
nic_derived_presence_condition(bad_value)
@pytest.mark.parametrize("bad_value", ["nic_presence_parent", "nic_presence_parent.devMac"])
def test_rejects_own_internal_alias(self, bad_value):
"""A caller referencing this function's own internal alias would hit
the exact shadowing bug that motivated the nic_parent ->
nic_presence_parent rename - see this module's docstring."""
with pytest.raises(ValueError):
nic_derived_presence_condition(bad_value)
class TestQualifiedColumnExecutesCorrectly:
"""Executes the fragment, not just checks the generated SQL text - proves
@@ -119,6 +127,35 @@ class TestQualifiedColumnExecutesCorrectly:
"into a table-wide 'does anything assert presence' check"
)
def test_nic_derived_condition_correlates_when_caller_aliases_nic_parent(self):
"""Regression: nic_derived_presence_condition() used to alias its own
inner scan as nic_parent too, so a caller that (like insert_events()'s
NIC-derived reconnect query) aliases its own row as nic_parent got
mac_column="nic_parent.devMac" shadowed by the helper's own inner
alias - collapsing the EXISTS into a table-wide tautology instead of
a per-row check."""
conn = sqlite3.connect(":memory:")
conn.execute("""CREATE TABLE Devices (devMac TEXT, devReqNicsOnline INTEGER,
devParentMAC TEXT, devParentRelType TEXT)""")
conn.execute("CREATE TABLE CurrentScan (scanMac TEXT, scanPresence INTEGER)")
conn.execute("INSERT INTO Devices VALUES ('aa', 0, NULL, NULL)")
conn.execute("INSERT INTO Devices VALUES ('aa-nic', 0, 'aa', 'nic')")
conn.execute("INSERT INTO CurrentScan VALUES ('aa-nic', 1)") # aa's NIC is present
conn.execute("INSERT INTO Devices VALUES ('bb', 0, NULL, NULL)")
conn.execute("INSERT INTO Devices VALUES ('bb-nic', 0, 'bb', 'nic')") # bb's NIC absent
conn.commit()
condition = nic_derived_presence_condition("nic_parent.devMac")
rows = conn.execute(
f"""SELECT devMac, {condition} AS is_nic_present FROM Devices AS nic_parent
WHERE devParentMAC IS NULL"""
).fetchall()
assert dict(rows) == {"aa": 1, "bb": 0}, (
"each outer row must be checked against its own NIC children, not "
"collapse into a table-wide 'does any device have NIC presence' check"
)
def _call_count(func, target_name="current_scan_presence_condition"):
"""Count calls to target_name within func's own source (AST-based, not