49 lines
1.4 KiB
Python
49 lines
1.4 KiB
Python
"""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) == []
|