"""Pure-function tests for host_agent staleness filter.""" from dataclasses import dataclass from datetime import datetime, timedelta, timezone from plugins.host_agent.scheduler import _filter_stale @dataclass class FakeReg: host_id: str last_seen_at: datetime | None NOW = datetime(2026, 4, 15, 12, 0, 0, tzinfo=timezone.utc) def test_filter_stale_returns_only_old_rows(): regs = [ FakeReg("fresh", NOW - timedelta(seconds=30)), FakeReg("stale", NOW - timedelta(seconds=600)), FakeReg("never", None), ] stale = _filter_stale(regs, now=NOW, stale_after_seconds=180) ids = {r.host_id for r in stale} assert ids == {"stale"} def test_filter_stale_empty_input(): assert _filter_stale([], now=NOW, stale_after_seconds=180) == [] def test_filter_stale_all_fresh(): regs = [ FakeReg("a", NOW - timedelta(seconds=10)), FakeReg("b", NOW - timedelta(seconds=30)), ] assert _filter_stale(regs, now=NOW, stale_after_seconds=60) == [] def test_filter_stale_boundary_exact_cutoff_is_not_stale(): # A row exactly at the cutoff should not be flagged (strict less-than). regs = [FakeReg("edge", NOW - timedelta(seconds=180))] assert _filter_stale(regs, now=NOW, stale_after_seconds=180) == [] def test_filter_stale_skips_none_last_seen(): regs = [FakeReg("never", None)] assert _filter_stale(regs, now=NOW, stale_after_seconds=1) == []