95 lines
3 KiB
Python
95 lines
3 KiB
Python
from datetime import datetime, timedelta, timezone
|
|
from types import SimpleNamespace
|
|
|
|
from app.models.dashboard import DashboardDeviceCheckHistory
|
|
from app.services.dashboard_monitor import _intervals_from_rows, _is_object_allowed, _series_from_rows
|
|
|
|
|
|
def _obj(obj_id: int, tag_ids: list[int]):
|
|
return SimpleNamespace(id=obj_id, tags=[SimpleNamespace(id=t) for t in tag_ids])
|
|
|
|
|
|
def test_object_allowed_respects_object_mode() -> None:
|
|
obj = _obj(1, [10, 20])
|
|
|
|
allowed, mode, reason = _is_object_allowed(obj, {1: "include"}, {})
|
|
assert allowed is True
|
|
assert mode == "include"
|
|
assert reason == "object_include_policy"
|
|
|
|
allowed, mode, reason = _is_object_allowed(obj, {1: "exclude"}, {})
|
|
assert allowed is False
|
|
assert mode == "exclude"
|
|
assert reason == "object_exclude_policy"
|
|
|
|
|
|
def test_object_allowed_respects_tag_policy() -> None:
|
|
obj = _obj(1, [10, 20])
|
|
|
|
allowed, mode, reason = _is_object_allowed(obj, {}, {10: False, 20: False})
|
|
assert allowed is False
|
|
assert mode == "inherit"
|
|
assert reason == "tag_policy_disabled"
|
|
|
|
allowed, mode, reason = _is_object_allowed(obj, {}, {10: False, 20: True})
|
|
assert allowed is True
|
|
assert mode == "inherit"
|
|
assert reason == "tag_policy_enabled"
|
|
|
|
|
|
def test_object_allowed_default_when_no_tag_policy() -> None:
|
|
obj = _obj(2, [100])
|
|
allowed, mode, reason = _is_object_allowed(obj, {}, {})
|
|
assert allowed is True
|
|
assert mode == "inherit"
|
|
assert reason == "inherit_default"
|
|
|
|
|
|
def test_series_and_intervals_for_ping_history() -> None:
|
|
start = datetime(2026, 5, 5, 0, 0, tzinfo=timezone.utc)
|
|
end = start + timedelta(hours=3)
|
|
|
|
rows = [
|
|
DashboardDeviceCheckHistory(
|
|
device_id=1,
|
|
check_key="ping",
|
|
status="online",
|
|
ping_ms=12,
|
|
checked_at=start + timedelta(minutes=10),
|
|
duration_ms=20,
|
|
error=None,
|
|
source="dashboard_monitor",
|
|
),
|
|
DashboardDeviceCheckHistory(
|
|
device_id=1,
|
|
check_key="ping",
|
|
status="offline",
|
|
ping_ms=None,
|
|
checked_at=start + timedelta(minutes=70),
|
|
duration_ms=30,
|
|
error="timeout",
|
|
source="dashboard_monitor",
|
|
),
|
|
DashboardDeviceCheckHistory(
|
|
device_id=1,
|
|
check_key="ping",
|
|
status="online",
|
|
ping_ms=25,
|
|
checked_at=start + timedelta(minutes=130),
|
|
duration_ms=25,
|
|
error=None,
|
|
source="dashboard_monitor",
|
|
),
|
|
]
|
|
|
|
series = _series_from_rows(rows, start_at=start, end_at=end, bucket_sec=3600)
|
|
assert len(series) == 3
|
|
assert series[0]["availability_pct"] == 100.0
|
|
assert series[1]["availability_pct"] == 0.0
|
|
assert series[2]["availability_pct"] == 100.0
|
|
|
|
intervals = _intervals_from_rows(rows, end_at=end)
|
|
assert len(intervals) == 3
|
|
assert intervals[0]["status"] == "online"
|
|
assert intervals[1]["status"] == "offline"
|
|
assert intervals[2]["status"] == "online"
|