71 lines
2.2 KiB
Python
71 lines
2.2 KiB
Python
"""Tests for device partitioning and parsing."""
|
|
|
|
from datetime import datetime, timedelta, timezone
|
|
|
|
from cnmaestro.devices import (
|
|
_parse_last_sync,
|
|
device_from_api,
|
|
partition_for_cleanup,
|
|
)
|
|
|
|
|
|
NOW = datetime(2026, 5, 9, 12, 0, tzinfo=timezone.utc)
|
|
|
|
|
|
def make(mac, status, last_sync, **extra):
|
|
return device_from_api(
|
|
{"mac": mac, "name": f"d-{mac}", "product": extra.pop("product", "ePMP"),
|
|
"status": status, "last_sync": last_sync, **extra}
|
|
)
|
|
|
|
|
|
def test_parse_last_sync_epoch_ms():
|
|
dt = _parse_last_sync(1715000000000)
|
|
assert dt is not None
|
|
assert dt.tzinfo is not None
|
|
|
|
|
|
def test_parse_last_sync_iso():
|
|
dt = _parse_last_sync("2026-05-08T12:00:00Z")
|
|
assert dt == datetime(2026, 5, 8, 12, 0, tzinfo=timezone.utc)
|
|
|
|
|
|
def test_parse_last_sync_none():
|
|
assert _parse_last_sync(None) is None
|
|
assert _parse_last_sync("") is None
|
|
assert _parse_last_sync(0) is None
|
|
|
|
|
|
def test_partition_buckets():
|
|
online_d = make("AA:01", "online", int((NOW - timedelta(minutes=5)).timestamp() * 1000))
|
|
recent_d = make("AA:02", "offline", int((NOW - timedelta(hours=1)).timestamp() * 1000))
|
|
stale_d = make("AA:03", "offline", int((NOW - timedelta(days=3)).timestamp() * 1000))
|
|
never_d = make("AA:04", "offline", None)
|
|
|
|
p = partition_for_cleanup(
|
|
[online_d, recent_d, stale_d, never_d],
|
|
threshold_hours=24,
|
|
now=NOW,
|
|
)
|
|
|
|
assert [d.mac for d in p.online] == ["AA:01"]
|
|
assert [d.mac for d in p.offline_recent] == ["AA:02"]
|
|
assert [d.mac for d in p.offline_stale] == ["AA:03"]
|
|
assert [d.mac for d in p.never_synced] == ["AA:04"]
|
|
assert {d.mac for d in p.candidates} == {"AA:03", "AA:04"}
|
|
assert p.total == 4
|
|
|
|
|
|
def test_partition_threshold_boundary():
|
|
"""A device offline exactly at the threshold is not yet stale."""
|
|
just_under = make(
|
|
"AA:05", "offline",
|
|
int((NOW - timedelta(hours=23, minutes=59)).timestamp() * 1000),
|
|
)
|
|
just_over = make(
|
|
"AA:06", "offline",
|
|
int((NOW - timedelta(hours=24, minutes=1)).timestamp() * 1000),
|
|
)
|
|
p = partition_for_cleanup([just_under, just_over], threshold_hours=24, now=NOW)
|
|
assert [d.mac for d in p.offline_recent] == ["AA:05"]
|
|
assert [d.mac for d in p.offline_stale] == ["AA:06"]
|