ledgrab/server/tests/test_discovery_watcher.py

"""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"