Spoolman/tests/test_scan_relay.py

"""Tests for the scan relay's debounce cache and reader registry.

Both are pure in-memory state, so they are tested here without a socket or a session. Time is
injected rather than slept through, which is what keeps this file fast enough to be worth
running on every change.
"""

import re
from datetime import datetime, timedelta, timezone

import pytest

from spoolman.scanrelay import MAX_READERS, READER_ID_PATTERN, ScanRelay, derive_reader_id

# Aware, matching the relay's own clock. Injected time and the relay's default must agree:
# register() writes last_seen and readers() prunes against it, so a naive value here would
# only pass because these tests inject every clock -- see the internal-consistency test below.
T0 = datetime(2026, 8, 13, 12, 0, 0, tzinfo=timezone.utc)


@pytest.fixture
def relay() -> ScanRelay:
    return ScanRelay(debounce_window=timedelta(seconds=3), reader_ttl=timedelta(hours=1))


def test_first_scan_broadcasts(relay: ScanRelay):
    assert relay.should_broadcast("04A2", "desk", now=T0) is True


def test_repeat_within_the_window_is_suppressed(relay: ScanRelay):
    """A reader re-reads a tag that is sitting still; subscribers should see one event."""
    relay.should_broadcast("04A2", "desk", now=T0)
    assert relay.should_broadcast("04A2", "desk", now=T0 + timedelta(seconds=1)) is False
    assert relay.should_broadcast("04A2", "desk", now=T0 + timedelta(seconds=2.9)) is False


def test_repeat_after_the_window_broadcasts_again(relay: ScanRelay):
    """Tapping the same tag again later is a new scan, not a duplicate."""
    relay.should_broadcast("04A2", "desk", now=T0)
    assert relay.should_broadcast("04A2", "desk", now=T0 + timedelta(seconds=4)) is True


def test_debounce_is_per_uid_and_per_reader(relay: ScanRelay):
    """Two readers tapping at once, or one reader seeing two tags, are distinct scans."""
    relay.should_broadcast("04A2", "desk", now=T0)
    assert relay.should_broadcast("04A2", "printer", now=T0) is True
    assert relay.should_broadcast("B1C2", "desk", now=T0) is True


def test_debounce_entries_are_pruned(relay: ScanRelay):
    """The cache does not grow without bound as tags come and go."""
    for i in range(50):
        relay.should_broadcast(f"UID{i:04X}", "desk", now=T0)
    assert len(relay._last_broadcast) == 50  # noqa: SLF001
    relay.should_broadcast("LATER", "desk", now=T0 + timedelta(seconds=30))
    assert len(relay._last_broadcast) == 1  # noqa: SLF001


def test_readers_are_listed_most_recently_seen_first(relay: ScanRelay):
    relay.register("desk", "Desk reader", now=T0)
    relay.register("printer", "Voron spool holder", now=T0 + timedelta(minutes=1))

    readers = relay.readers(now=T0 + timedelta(minutes=2))
    assert [r.reader_id for r in readers] == ["printer", "desk"]
    assert readers[0].name == "Voron spool holder"
    assert readers[1].last_seen == T0


def test_registry_is_empty_before_anything_scans(relay: ScanRelay):
    """The registry is not persisted, so this is also the state after a restart."""
    assert relay.readers(now=T0) == []


def test_reader_keeps_its_name_when_a_later_scan_omits_one(relay: ScanRelay):
    """A trimmed-down agent must not blank the label in someone's picker."""
    relay.register("desk", "Desk reader", now=T0)
    relay.register("desk", None, now=T0 + timedelta(minutes=1))

    readers = relay.readers(now=T0 + timedelta(minutes=2))
    assert readers[0].name == "Desk reader"
    assert readers[0].last_seen == T0 + timedelta(minutes=1)


def test_a_new_name_replaces_the_old_one(relay: ScanRelay):
    relay.register("desk", "Desk reader", now=T0)
    relay.register("desk", "Bench reader", now=T0 + timedelta(minutes=1))
    assert relay.readers(now=T0 + timedelta(minutes=2))[0].name == "Bench reader"


def test_readers_past_their_ttl_are_evicted_on_read(relay: ScanRelay):
    """Eviction happens as the response is built: no background task, nothing to shut down."""
    relay.register("desk", None, now=T0)
    assert len(relay.readers(now=T0 + timedelta(minutes=59))) == 1
    assert relay.readers(now=T0 + timedelta(hours=2)) == []


def test_registry_is_capped(relay: ScanRelay):
    """An unauthenticated endpoint cannot be used to grow the registry without bound."""
    for i in range(MAX_READERS + 20):
        relay.register(f"reader-{i}", None, now=T0 + timedelta(seconds=i))

    readers = relay.readers(now=T0 + timedelta(minutes=30))
    assert len(readers) == MAX_READERS
    # The least recently seen are the ones dropped.
    assert readers[0].reader_id == f"reader-{MAX_READERS + 19}"
    assert {r.reader_id for r in readers} == {f"reader-{i}" for i in range(20, MAX_READERS + 20)}


def test_the_relays_own_clock_is_internally_consistent():
    """Exercise the default `now` on every method, which the injected-time tests never do.

    Each method reaching for its own clock is exactly where a naive/aware mismatch hides: the
    stored last_seen and the pruning cutoff are compared, so one clock disagreeing with the
    others makes GET /tag/reader raise rather than return.
    """
    relay = ScanRelay()
    relay.register("desk", "Desk reader")
    assert relay.should_broadcast("04A2", "desk") is True
    assert relay.should_broadcast("04A2", "desk") is False

    listed = relay.readers()
    assert [r.reader_id for r in listed] == ["desk"]


@pytest.mark.parametrize(
    ("host", "expected"),
    [
        ("192.168.1.50", "ip-192-168-1-50"),
        ("10.0.0.1", "ip-10-0-0-1"),
        ("fe80::1", "ip-fe80-1"),
        ("testclient", "ip-testclient"),
        (None, "unknown"),
        ("", "unknown"),
        ("...", "unknown"),
    ],
)
def test_reader_id_is_derived_from_the_client_address(host: str | None, expected: str):
    """An unmodified agent that sends no reader_id still gets a stable id to pair against."""
    assert derive_reader_id(host) == expected


def test_derived_reader_id_fits_the_pattern():
    """Whatever comes out must be usable as a websocket path segment."""
    assert re.match(READER_ID_PATTERN, derive_reader_id("a" * 200))