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"