feat(notifications): device event notifications (snack + Web Notifications)

Surface device connection state changes (configured target online/offline) and discovery events (new WLED on LAN, new serial port, devices that disappear) through a configurable per-event channel matrix: none / snack / OS / both. - Backend: long-running mDNS browser + 10 s serial poller in core/devices/discovery_watcher.py, gated by user pref. Reuses the existing device_health_changed event for online/offline transitions. New GET/PUT /api/v1/preferences/notifications endpoint with Pydantic v2 schema (channel matrix + background-discovery flag + grace/debounce). 13 new tests, full suite still 899 passing. - Frontend: features/notifications-watcher.ts with startup-grace + flap-debounce + bulk-coalesce pipeline. Web Notifications API for the OS channel (no platform-specific code, works in PWA shell). New "Notifications" tab in Settings with 4 IconSelect rows + bg toggle + permission row + test button. en/ru/zh translations. Defaults: device_offline=both (urgent), online/discovered=snack, lost=none, background discovery on. Already-configured devices are filtered from discovery events to avoid double-notifications.
2026-04-25 17:49:20 +03:00
parent 8e109f32b9
commit 8aa3a323d6
14 changed files with 1451 additions and 2 deletions
@@ -0,0 +1,198 @@
+"""Tests for the background discovery watcher.
+
+We mock zeroconf and ``list_serial_ports`` so the tests run in milliseconds
+without touching the network or hardware. The mDNS path is exercised via
+the public state-change handler; the serial path is exercised via the
+internal one-shot poll method.
+"""
+
+from __future__ import annotations
+
+from unittest.mock import patch
+
+import pytest
+from zeroconf import ServiceStateChange
+
+from ledgrab.core.devices.discovery_watcher import DiscoveryWatcher
+from ledgrab.core.devices.serial_transport import SerialPortInfo
+
+
+class _FakeDevice:
+    def __init__(self, url: str) -> None:
+        self.url = url
+
+
+class _FakeStore:
+    def __init__(self, devices: list[_FakeDevice] | None = None) -> None:
+        self._devices = devices or []
+
+    def get_all_devices(self):
+        return list(self._devices)
+
+
+@pytest.fixture
+def captured_events():
+    events: list[dict] = []
+    return events
+
+
+@pytest.fixture
+def make_watcher(captured_events):
+    def _factory(store: _FakeStore | None = None) -> DiscoveryWatcher:
+        return DiscoveryWatcher(
+            device_store=store or _FakeStore(),
+            fire_event=captured_events.append,
+        )
+
+    return _factory
+
+
+# -----------------------------------------------------------------------
+# Serial polling
+# -----------------------------------------------------------------------
+
+
+def test_seed_ports_do_not_emit_events(make_watcher, captured_events):
+    """Ports already plugged in at startup must not generate notifications.
+
+    Otherwise every server start would dump a "device discovered" toast
+    for every COM port on the machine.
+    """
+    watcher = make_watcher()
+    seed = [SerialPortInfo(device="COM3", description="USB-Serial CH340")]
+
+    # First-poll behaviour models the seed: the watcher's startup path
+    # populates _serial_seen without firing events. We exercise that
+    # behaviour by populating the dict directly and then running the
+    # poll, asserting no events fire because nothing changed.
+    from ledgrab.core.devices.discovery_watcher import _DiscoveredEntry
+
+    for p in seed:
+        watcher._serial_seen[p.device] = _DiscoveredEntry(
+            key=p.device, url=p.device, name=p.description, device_type="serial"
+        )
+
+    with patch(
+        "ledgrab.core.devices.discovery_watcher.list_serial_ports",
+        return_value=seed,
+    ):
+        watcher._poll_serial_once()
+
+    assert captured_events == []
+
+
+def test_new_serial_port_emits_discovered(make_watcher, captured_events):
+    """A port appearing between polls fires device_discovered."""
+    watcher = make_watcher()
+
+    # First poll: seed COM3.
+    with patch(
+        "ledgrab.core.devices.discovery_watcher.list_serial_ports",
+        return_value=[SerialPortInfo(device="COM3", description="CH340")],
+    ):
+        watcher._poll_serial_once()
+    captured_events.clear()  # discovered for COM3 above is fine; we test the next change
+
+    # Second poll: COM4 has appeared.
+    with patch(
+        "ledgrab.core.devices.discovery_watcher.list_serial_ports",
+        return_value=[
+            SerialPortInfo(device="COM3", description="CH340"),
+            SerialPortInfo(device="COM4", description="CP2102"),
+        ],
+    ):
+        watcher._poll_serial_once()
+
+    assert len(captured_events) == 1
+    evt = captured_events[0]
+    assert evt["type"] == "device_discovered"
+    assert evt["device_type"] == "serial"
+    assert evt["url"] == "COM4"
+    assert evt["name"] == "CP2102"
+
+
+def test_disappearing_serial_port_emits_lost(make_watcher, captured_events):
+    """A port that vanishes between polls fires device_lost."""
+    watcher = make_watcher()
+
+    with patch(
+        "ledgrab.core.devices.discovery_watcher.list_serial_ports",
+        return_value=[SerialPortInfo(device="COM4", description="CP2102")],
+    ):
+        watcher._poll_serial_once()
+    captured_events.clear()
+
+    with patch(
+        "ledgrab.core.devices.discovery_watcher.list_serial_ports",
+        return_value=[],
+    ):
+        watcher._poll_serial_once()
+
+    assert len(captured_events) == 1
+    assert captured_events[0]["type"] == "device_lost"
+    assert captured_events[0]["url"] == "COM4"
+
+
+def test_configured_devices_are_filtered_from_discovery(make_watcher, captured_events):
+    """If a port matches a configured device URL, suppress discovery events.
+
+    The user already gets device_health_changed for known devices; firing
+    a redundant 'discovered' event the first time the port re-appears
+    would be confusing noise.
+    """
+    store = _FakeStore([_FakeDevice("COM5")])
+    watcher = make_watcher(store)
+
+    # COM5 appears for the first time — but it's already configured,
+    # so no event should fire.
+    with patch(
+        "ledgrab.core.devices.discovery_watcher.list_serial_ports",
+        return_value=[SerialPortInfo(device="COM5", description="ESP32")],
+    ):
+        watcher._poll_serial_once()
+
+    assert captured_events == []
+
+
+def test_serial_enumeration_failure_is_swallowed(make_watcher, captured_events):
+    """A blown enumeration call shouldn't crash the loop or fire bogus events."""
+    watcher = make_watcher()
+
+    with patch(
+        "ledgrab.core.devices.discovery_watcher.list_serial_ports",
+        side_effect=OSError("device busy"),
+    ):
+        watcher._poll_serial_once()  # must not raise
+
+    assert captured_events == []
+
+
+# -----------------------------------------------------------------------
+# mDNS state-change dispatcher
+# -----------------------------------------------------------------------
+
+
+def test_mdns_removed_emits_lost_only_for_unconfigured(make_watcher, captured_events):
+    """ServiceStateChange.Removed → device_lost, but only when the URL
+    isn't already in the user's device store."""
+    watcher = make_watcher()
+
+    # Seed a previously-resolved entry.
+    from ledgrab.core.devices.discovery_watcher import _DiscoveredEntry
+
+    watcher._wled_seen["wled-foo._wled._tcp.local."] = _DiscoveredEntry(
+        key="wled-foo._wled._tcp.local.",
+        url="http://192.168.1.55",
+        name="wled-foo",
+        device_type="wled",
+    )
+
+    watcher._on_wled_state_change(
+        service_type="_wled._tcp.local.",
+        name="wled-foo._wled._tcp.local.",
+        state_change=ServiceStateChange.Removed,
+    )
+
+    assert len(captured_events) == 1
+    assert captured_events[0]["type"] == "device_lost"
+    assert captured_events[0]["device_type"] == "wled"